Tutorial Weave: Observabilitas dan Evaluasi Aplikasi LLM dengan Weights & Biases

Membangun aplikasi berbasis Large Language Model (LLM) semakin mudah, tetapi memantau, men-debug, dan mengevaluasi kualitas output-nya tetap menjadi tantangan besar. Bagaimana Anda tahu apakah prompt yang baru memberikan hasil yang lebih baik? Bagaimana melacak setiap panggilan LLM di production? Di sinilah Weave dari Weights & Biases hadir sebagai solusi.

Weave adalah framework open-source untuk tracing, evaluasi, dan monitoring aplikasi LLM. Berbeda dengan W&B tradisional yang fokus pada experiment tracking untuk training model, Weave dirancang khusus untuk siklus hidup aplikasi LLM mulai dari development hingga production.

Dalam tutorial ini, kita akan mempelajari cara menggunakan Weave untuk melacak panggilan LLM, membuat evaluasi otomatis, mengelola dataset, dan memantau aplikasi LLM di production.

Instalasi dan Setup

Instalasi Weave

pip install weave openai

Weave mendukung Python 3.9+ dan terintegrasi dengan berbagai provider LLM seperti OpenAI, Anthropic, Google, Mistral, dan lainnya.

Konfigurasi API Key

Anda memerlukan akun Weights & Biases. Daftar gratis di wandb.ai, lalu dapatkan API key Anda.

export WANDBAPIKEY="your-api-key-here"
export OPENAIAPIKEY="your-openai-key-here"

Inisialisasi Project

import weave

weave.init("my-llm-project")

Setelah inisialisasi, semua operasi yang di-trace akan muncul di dashboard Weave pada platform W&B.

Basic Usage: Tracing LLM Calls

Auto-Tracing dengan Integrasi Built-in

Weave secara otomatis melacak panggilan ke provider LLM populer tanpa perlu kode tambahan.

import weave
from openai import OpenAI

weave.init("tracing-demo")
client = OpenAI()

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[
        {"role": "system", "content": "Kamu adalah asisten yang membantu."},
        {"role": "user", "content": "Jelaskan apa itu machine learning dalam 2 kalimat."}
    ]
)

print(response.choices[0].message.content)

Hanya dengan menambahkan weave.init(), setiap panggilan OpenAI secara otomatis di-trace. Anda bisa melihat input, output, latency, token usage, dan cost di dashboard Weave.

Custom Tracing dengan Decorator @weave.op

Untuk fungsi kustom Anda sendiri, gunakan decorator @weave.op:

import weave
import json
from openai import OpenAI

weave.init("custom-tracing")
client = OpenAI()

@weave.op
def extractentities(text: str) -> dict:

    """Ekstrak entitas dari teks menggunakan LLM."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Ekstrak entitas (orang, tempat, organisasi) dari teks. Kembalikan dalam format JSON."
            },
            {"role": "user", "content": text}
        ],
        responseformat={"type": "jsonobject"}

    )
    return json.loads(response.choices[0].message.content)

@weave.op
def processdocument(document: str) -> dict:
    """Proses dokumen: ekstrak entitas dan buat ringkasan."""
    entities = extractentities(document)

    return {
        "entities": entities,
        "entitycount": sum(len(v) for v in entities.values() if isinstance(v, list))
    }

result = processdocument(

    "Joko Widodo bertemu dengan Joe Biden di Jakarta untuk membahas "
    "kerjasama ASEAN dan investasi di Indonesia."
)
print(json.dumps(result, indent=2, ensureascii=False))

Decorator @weave.op membuat hierarchical trace, sehingga Anda bisa melihat bagaimana processdocument memanggil extractentities, lengkap dengan input/output di setiap level.

Tracing dengan Metadata Tambahan

Anda bisa menambahkan atribut kustom ke trace untuk mempermudah filtering dan analisis:

import weave

weave.init("metadata-demo")

@weave.op
def classifysentiment(text: str, language: str = "id") -> dict:

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": f"Klasifikasikan sentimen teks berikut dalam bahasa {'Indonesia' if language == 'id' else 'Inggris'}. Jawab dengan: positif, negatif, atau netral."
            },
            {"role": "user", "content": text}
        ]
    )
    return {
        "text": text,
        "language": language,
        "sentiment": response.choices[0].message.content
    }

result = classifysentiment("Produk ini sangat bagus dan berkualitas!", language="id")
print(result)

Evaluasi Otomatis dengan Weave Evaluation

Salah satu fitur terkuat Weave adalah framework evaluasi yang memungkinkan Anda mengukur kualitas output LLM secara sistematis.

Membuat Dataset Evaluasi

import weave

weave.init("evaluation-demo")

dataset = weave.Dataset(
    name="sentiment-test-set",
    rows=[
        {"text": "Makanan di restoran ini enak sekali!", "expected": "positif"},
        {"text": "Pelayanannya sangat buruk dan lambat.", "expected": "negatif"},
        {"text": "Harga produk ini standar di pasaran.", "expected": "netral"},
        {"text": "Saya kecewa dengan kualitas barangnya.", "expected": "negatif"},
        {"text": "Tim support sangat responsif dan membantu.", "expected": "positif"},
        {"text": "Pengiriman sesuai estimasi waktu.", "expected": "netral"},
        {"text": "Aplikasinya sering crash dan tidak stabil.", "expected": "negatif"},
        {"text": "Desain produknya modern dan elegan.", "expected": "positif"},
    ]
)

weave.publish(dataset)

Mendefinisikan Model untuk Evaluasi

Weave menggunakan konsep Model sebagai wrapper untuk fungsi yang ingin Anda evaluasi:

import weave
from openai import OpenAI

class SentimentClassifier(weave.Model):
    modelname: str = "gpt-4o-mini"

    systemprompt: str = "Klasifikasikan sentimen teks berikut. Jawab HANYA dengan satu kata: positif, negatif, atau netral."

    @weave.op
    def predict(self, text: str) -> dict:
        client = OpenAI()
        response = client.chat.completions.create(
            model=self.modelname,

            messages=[
                {"role": "system", "content": self.systemprompt},
                {"role": "user", "content": text}
            ]
        )
        result = response.choices[0].message.content.strip().lower()
        return {"sentiment": result}

Membuat Scorer

Scorer adalah fungsi yang menilai kualitas output model:

@weave.op
def accuracyscorer(expected: str, modeloutput: dict) -> dict:
    """Mengecek apakah prediksi sentimen sesuai dengan expected."""
    predicted = modeloutput.get("sentiment", "").strip().lower()

    expectedclean = expected.strip().lower()
    match = predicted == expectedclean

    return {"correct": match}

@weave.op
def formatscorer(modeloutput: dict) -> dict:

    """Mengecek apakah output sesuai format yang diharapkan."""
    sentiment = modeloutput.get("sentiment", "")
    validsentiments = ["positif", "negatif", "netral"]

    isvalid = sentiment.strip().lower() in validsentiments

    return {"validformat": isvalid}

import asyncio

async def runevaluation():
    model = SentimentClassifier()

    evaluation = weave.Evaluation(
        dataset=dataset,
        scorers=[accuracyscorer, formatscorer]
    )

    results = await evaluation.evaluate(model)
    print(f"Accuracy: {results['accuracyscorer']['correct']['truecount']}/{len(dataset.rows)}")
    print(f"Format validity: {results['formatscorer']['validformat']['truecount']}/{len(dataset.rows)}")

    return results

results = asyncio.run(runevaluation())

Membandingkan Model atau Prompt

Kekuatan evaluasi Weave terletak pada kemampuan membandingkan berbagai konfigurasi:

async def comparemodels():
    models = [
        SentimentClassifier(
            modelname="gpt-4o-mini",
            systemprompt="Klasifikasikan sentimen: positif, negatif, atau netral."

        ),
        SentimentClassifier(
            modelname="gpt-4o-mini",
            systemprompt="Kamu adalah ahli analisis sentimen bahasa Indonesia. Analisis teks berikut dan tentukan sentimennya. Jawab HANYA: positif, negatif, atau netral."

        ),
        SentimentClassifier(
            modelname="gpt-4o",
            systemprompt="Klasifikasikan sentimen: positif, negatif, atau netral."

        ),
    ]

    evaluation = weave.Evaluation(
        dataset=dataset,
        scorers=[accuracyscorer, formatscorer]

    )

    for model in models:
        print(f"\nEvaluating: {model.modelname} with prompt variant...")
        results = await evaluation.evaluate(model)
        print(f"Results: {results}")

asyncio.run(comparemodels())

@weave.op
def llmjudgescorer(text: str, modeloutput: dict) -> dict:
    """Gunakan LLM untuk menilai kualitas klasifikasi sentimen."""
    client = OpenAI()
    predicted = modeloutput.get("sentiment", "")


    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "Kamu adalah evaluator yang menilai apakah klasifikasi sentimen sudah tepat. "
                    "Nilai dari 1-5 (1=sangat salah, 5=sangat tepat). "
                    "Jawab HANYA dengan angka."
                )
            },
            {
                "role": "user",
                "content": f"Teks: {text}\nKlasifikasi: {predicted}"
            }
        ]
    )
    score = int(response.choices[0].message.content.strip())
    return {"judgescore": score, "isgood": score >= 4}

import weave
import numpy as np
from openai import OpenAI

weave.init("rag-tracing")
client = OpenAI()

@weave.op
def getembedding(text: str) -> list:
    """Dapatkan embedding dari teks."""
    response = client.embeddings.create(
        model="text-embedding-3-small",
        input=text
    )
    return response.data[0].embedding

@weave.op
def searchdocuments(query: str, documents: list[dict], topk: int = 3) -> list[dict]:
    """Cari dokumen yang relevan berdasarkan similarity."""
    queryembedding = getembedding(query)

    scoreddocs = []

    for doc in documents:
        docembedding = getembedding(doc["content"])

        similarity = np.dot(queryembedding, docembedding) / (

            np.linalg.norm(queryembedding) * np.linalg.norm(docembedding)

        )
        scoreddocs.append({doc, "score": float(similarity)})


    scoreddocs.sort(key=lambda x: x["score"], reverse=True)

    return scoreddocs[:topk]


@weave.op
def generateanswer(query: str, contextdocs: list[dict]) -> str:

    """Generate jawaban berdasarkan konteks dokumen."""
    context = "\n\n".join([
        f"Dokumen {i+1}: {doc['content']}"
        for i, doc in enumerate(contextdocs)
    ])

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Jawab pertanyaan berdasarkan konteks yang diberikan. Jika informasi tidak ada di konteks, katakan tidak tersedia."
            },
            {
                "role": "user",
                "content": f"Konteks:\n{context}\n\nPertanyaan: {query}"
            }
        ]
    )
    return response.choices[0].message.content

@weave.op
def ragpipeline(query: str, knowledgebase: list[dict]) -> dict:
    """Pipeline RAG lengkap: retrieve lalu generate."""
    relevantdocs = searchdocuments(query, knowledgebase)

    answer = generateanswer(query, relevantdocs)

    return {
        "query": query,
        "retrieveddocs": relevantdocs,

        "answer": answer
    }

knowledgebase = [
    {"content": "Python adalah bahasa pemrograman yang dibuat oleh Guido van Rossum pada tahun 1991."},
    {"content": "Machine learning adalah cabang AI yang memungkinkan komputer belajar dari data."},
    {"content": "Django adalah framework web Python yang mengikuti pola MVT (Model-View-Template)."},
    {"content": "FastAPI adalah framework Python modern untuk membangun API dengan performa tinggi."},
    {"content": "PyTorch adalah library deep learning yang dikembangkan oleh Meta AI."},
]

result = ragpipeline("Apa itu FastAPI?", knowledgebase)
print(result["answer"])

ragdataset = weave.Dataset(
    name="rag-test-set",
    rows=[
        {
            "query": "Siapa pembuat Python?",
            "expectedanswer": "Guido van Rossum",
            "knowledgebase": knowledgebase
        },
        {
            "query": "Apa itu FastAPI?",
            "expectedanswer": "framework Python modern untuk membangun API",

            "knowledgebase": knowledgebase

        },
        {
            "query": "Framework deep learning dari Meta?",
            "expectedanswer": "PyTorch",
            "knowledgebase": knowledgebase
        },
    ]
)

@weave.op
def answerrelevancescorer(expectedanswer: str, modeloutput: dict) -> dict:
    """Cek apakah jawaban mengandung informasi yang diharapkan."""
    answer = modeloutput.get("answer", "").lower()

    expected = expectedanswer.lower()
    containsexpected = expected in answer

    return {"relevant": containsexpected}

@weave.op
def retrievalscorer(modeloutput: dict) -> dict:
    """Evaluasi kualitas retrieval."""
    docs = modeloutput.get("retrieveddocs", [])
    avgscore = sum(d["score"] for d in docs) / len(docs) if docs else 0

    return {"avgretrievalscore": round(avgscore, 4)}

class RAGModel(weave.Model):
    @weave.op
    def predict(self, query: str, knowledgebase: list) -> dict:

        return ragpipeline(query, knowledgebase)


async def evaluaterag():
    model = RAGModel()
    evaluation = weave.Evaluation(
        dataset=ragdataset,

        scorers=[answerrelevancescorer, retrievalscorer]
    )
    results = await evaluation.evaluate(model)
    print(f"RAG Evaluation Results: {results}")

asyncio.run(evaluaterag())

import weave
from fastapi import FastAPI
from pydantic import BaseModel

weave.init("production-monitoring")
app = FastAPI()

class QueryRequest(BaseModel):
    question: str

class QueryResponse(BaseModel):
    answer: str
    sources: list[str]

@weave.op
def handlequery(question: str) -> dict:
    result = ragpipeline(question, knowledgebase)
    return {
        "answer": result["answer"],
        "sources": [d["content"][:50] for d in result["retrieveddocs"]]

    }

@app.post("/query", responsemodel=QueryResponse)
async def queryendpoint(request: QueryRequest):

    result = handlequery(request.question)
    return QueryResponse(result)

Dengan setup ini, setiap request ke API Anda secara otomatis di-trace di Weave. Anda bisa memantau latency, error rate, token usage, dan kualitas respons secara real-time melalui dashboard.

Best Practices

1. Organisasi Project

Gunakan nama project yang deskriptif dan konsisten untuk memudahkan navigasi:

weave.init("company/product-chatbot-v2")

2. Trace Setiap Langkah Penting

Jangan hanya trace panggilan LLM. Trace juga langkah preprocessing, retrieval, dan postprocessing:

@weave.op
def preprocessquery(query: str) -> str:

    query = query.strip().lower()
    return query

@weave.op
def postprocessresponse(response: str) -> str:
    response = response.strip()
    if not response.endswith("."):
        response += "."
    return response

3. Versioning Model dengan weave.Model

Gunakan weave.Model untuk melacak konfigurasi model Anda:

class MyAssistant(weave.Model):
    modelname: str = "gpt-4o-mini"

    temperature: float = 0.7
    maxtokens: int = 500
    systemprompt: str = "Kamu adalah asisten yang membantu."


    @weave.op
    def predict(self, usermessage: str) -> str:
        client = OpenAI()
        response = client.chat.completions.create(
            model=self.modelname,

            temperature=self.temperature,
            maxtokens=self.maxtokens,

            messages=[
                {"role": "system", "content": self.systemprompt},
                {"role": "user", "content": usermessage}

            ]
        )
        return response.choices[0].message.content

async def cievaluation():
    """Jalankan evaluasi sebagai bagian dari CI/CD pipeline."""
    model = SentimentClassifier()
    evaluation = weave.Evaluation(
        dataset=dataset,
        scorers=[accuracyscorer]

    )
    results = await evaluation.evaluate(model)

    accuracy = results["accuracyscorer"]["correct"]["truefraction"]

    assert accuracy >= 0.8, f"Accuracy {accuracy} di bawah threshold 0.8"
    print(f"CI evaluation passed with accuracy: {accuracy}")

evaluation = weave.Evaluation(
    dataset=dataset,
    scorers=[
        accuracyscorer,
        formatscorer,

        llmjudgescorer,

        latencyscorer,
    ]
)

Integrasi dengan Provider LLM Lain

Weave mendukung berbagai provider LLM secara native:

# Anthropic
from anthropic import Anthropic
client = Anthropic()

Google Gemini
import google.generativeai as genai
model = genai.GenerativeModel("gemini-pro")

Mistral
from mistralai.client import MistralClient
client = MistralClient()

LiteLLM (multi-provider)
import litellm
response = litellm.completion(model="gpt-4o-mini", messages=[...])

Semua panggilan di atas secara otomatis di-trace oleh Weave tanpa kode tambahan.

Kesimpulan

Weave dari Weights & Biases adalah alat yang sangat berharga untuk pengembangan aplikasi LLM yang serius. Dengan fitur tracing otomatis, framework evaluasi yang fleksibel, dan kemampuan monitoring production, Weave membantu Anda membangun aplikasi LLM yang lebih andal dan berkualitas tinggi.

Poin-poin utama yang telah kita pelajari:

• Tracing: Gunakan weave.init() untuk auto-trace dan @weave.op untuk custom tracing
• Evaluasi: Buat dataset, definisikan scorer, dan jalankan evaluasi sistematik menggunakan weave.Evaluation
• Model Versioning: Gunakan weave.Model untuk melacak konfigurasi dan membandingkan performa
• Production Monitoring: Integrasikan Weave ke aplikasi production untuk monitoring real-time
• Perbandingan: Bandingkan model, prompt, dan konfigurasi secara side-by-side di dashboard

Mulailah dengan menginstal Weave (pip install weave) dan tambahkan tracing ke proyek LLM Anda. Anda akan terkejut betapa banyak insight yang bisa Anda dapatkan tentang perilaku aplikasi LLM Anda.