Weave Tutorial: LLM Application Observability and Evaluation with Weights & Biases

Building applications powered by Large Language Models (LLMs) is becoming increasingly accessible, but monitoring, debugging, and evaluating the quality of their outputs remains a significant challenge. How do you know if a new prompt performs better? How do you trace every LLM call in production? This is where Weave by Weights & Biases comes in.

Weave is an open-source framework for tracing, evaluating, and monitoring LLM applications. Unlike traditional W&B which focuses on experiment tracking for model training, Weave is specifically designed for the LLM application lifecycle from development to production.

In this tutorial, we will learn how to use Weave to trace LLM calls, create automated evaluations, manage datasets, and monitor LLM applications in production.

Installation and Setup

Installing Weave

pip install weave openai

Weave supports Python 3.9+ and integrates with various LLM providers including OpenAI, Anthropic, Google, Mistral, and more.

Configuring API Keys

You will need a Weights & Biases account. Sign up for free at wandb.ai and get your API key.

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

Initializing a Project

import weave

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

After initialization, all traced operations will appear in the Weave dashboard on the W&B platform.

Basic Usage: Tracing LLM Calls

Auto-Tracing with Built-in Integrations

Weave automatically traces calls to popular LLM providers without any additional code.

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": "You are a helpful assistant."},
        {"role": "user", "content": "Explain what machine learning is in 2 sentences."}
    ]
)

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

By simply adding weave.init(), every OpenAI call is automatically traced. You can view inputs, outputs, latency, token usage, and cost in the Weave dashboard.

Custom Tracing with the @weave.op Decorator

For your custom functions, use the @weave.op decorator:

import weave
import json
from openai import OpenAI

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

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

    """Extract entities from text using an LLM."""
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Extract entities (people, places, organizations) from the text. Return in JSON format."
            },
            {"role": "user", "content": text}
        ],
        responseformat={"type": "jsonobject"}

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

@weave.op
def processdocument(document: str) -> dict:
    """Process a document: extract entities and summarize."""
    entities = extractentities(document)

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

result = processdocument(

    "Sundar Pichai met with Tim Cook in San Francisco to discuss "
    "the partnership between Google and Apple on AI initiatives."
)
print(json.dumps(result, indent=2))

The @weave.op decorator creates hierarchical traces, allowing you to see how processdocument calls extractentities, complete with inputs and outputs at every level.

import weave

weave.init("metadata-demo")

@weave.op
def classifysentiment(text: str, language: str = "en") -> dict:
    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": f"Classify the sentiment of the following text. Answer with: positive, negative, or neutral."
            },
            {"role": "user", "content": text}
        ]
    )
    return {
        "text": text,
        "language": language,
        "sentiment": response.choices[0].message.content
    }

result = classifysentiment("This product is amazing and high quality!", language="en")

print(result)

import weave

weave.init("evaluation-demo")

dataset = weave.Dataset(
    name="sentiment-test-set",
    rows=[
        {"text": "The food at this restaurant is absolutely delicious!", "expected": "positive"},
        {"text": "The service was terrible and extremely slow.", "expected": "negative"},
        {"text": "The product price is standard for the market.", "expected": "neutral"},
        {"text": "I am disappointed with the quality of the item.", "expected": "negative"},
        {"text": "The support team is very responsive and helpful.", "expected": "positive"},
        {"text": "Delivery arrived within the estimated timeframe.", "expected": "neutral"},
        {"text": "The app crashes frequently and is unstable.", "expected": "negative"},
        {"text": "The product design is modern and elegant.", "expected": "positive"},
    ]
)

weave.publish(dataset)

import weave
from openai import OpenAI

class SentimentClassifier(weave.Model):
    modelname: str = "gpt-4o-mini"
    systemprompt: str = "Classify the sentiment of the following text. Answer ONLY with one word: positive, negative, or neutral."


    @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}

@weave.op
def accuracyscorer(expected: str, modeloutput: dict) -> dict:

    """Check if the sentiment prediction matches the expected value."""
    predicted = modeloutput.get("sentiment", "").strip().lower()
    expectedclean = expected.strip().lower()

    match = predicted == expectedclean
    return {"correct": match}

@weave.op
def formatscorer(modeloutput: dict) -> dict:
    """Check if the output follows the expected format."""
    sentiment = modeloutput.get("sentiment", "")

    validsentiments = ["positive", "negative", "neutral"]
    isvalid = sentiment.strip().lower() in validsentiments
    return {"validformat": isvalid}

Running an Evaluation

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())

async def comparemodels():
    models = [
        SentimentClassifier(
            modelname="gpt-4o-mini",

            systemprompt="Classify the sentiment: positive, negative, or neutral."
        ),
        SentimentClassifier(
            modelname="gpt-4o-mini",

            systemprompt="You are an expert sentiment analyst. Analyze the following text and determine its sentiment. Answer ONLY: positive, negative, or neutral."
        ),
        SentimentClassifier(
            modelname="gpt-4o",

            systemprompt="Classify the sentiment: positive, negative, or neutral."
        ),
    ]

    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())

Each evaluation is saved in the Weave dashboard, allowing you to compare results visually in one place.

Advanced Usage

LLM-as-Judge Scorer

For more complex evaluations, you can use an LLM as a judge:

@weave.op
def llmjudgescorer(text: str, modeloutput: dict) -> dict:

    """Use an LLM to judge the quality of sentiment classification."""
    client = OpenAI()
    predicted = modeloutput.get("sentiment", "")

    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {
                "role": "system",
                "content": (
                    "You are an evaluator assessing whether a sentiment classification is correct. "
                    "Rate from 1-5 (1=completely wrong, 5=perfectly correct). "
                    "Answer ONLY with a number."
                )
            },
            {
                "role": "user",
                "content": f"Text: {text}\nClassification: {predicted}"
            }
        ]
    )
    score = int(response.choices[0].message.content.strip())
    return {"judgescore": score, "isgood": score >= 4}

Tracing a RAG Pipeline

Weave is especially useful for tracing complex RAG pipelines:

import weave
import numpy as np
from openai import OpenAI

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

@weave.op
def getembedding(text: str) -> list:

    """Get embedding from text."""
    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]:

    """Search for relevant documents based on 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 an answer based on context documents."""
    context = "\n\n".join([
        f"Document {i+1}: {doc['content']}"
        for i, doc in enumerate(contextdocs)

    ])

    response = client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {
                "role": "system",
                "content": "Answer the question based on the given context. If the information is not in the context, say it is not available."
            },
            {
                "role": "user",
                "content": f"Context:\n{context}\n\nQuestion: {query}"
            }
        ]
    )
    return response.choices[0].message.content

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

    """Complete RAG pipeline: retrieve then generate."""
    relevantdocs = searchdocuments(query, knowledgebase)
    answer = generateanswer(query, relevantdocs)
    return {
        "query": query,
        "retrieveddocs": relevantdocs,
        "answer": answer
    }

knowledgebase = [

    {"content": "Python is a programming language created by Guido van Rossum in 1991."},
    {"content": "Machine learning is a branch of AI that enables computers to learn from data."},
    {"content": "Django is a Python web framework that follows the MVT (Model-View-Template) pattern."},
    {"content": "FastAPI is a modern Python framework for building high-performance APIs."},
    {"content": "PyTorch is a deep learning library developed by Meta AI."},
]

result = ragpipeline("What is FastAPI?", knowledgebase)

print(result["answer"])

ragdataset = weave.Dataset(
    name="rag-test-set",
    rows=[
        {
            "query": "Who created Python?",
            "expectedanswer": "Guido van Rossum",

            "knowledgebase": knowledgebase

        },
        {
            "query": "What is FastAPI?",
            "expectedanswer": "modern Python framework for building APIs",
            "knowledgebase": knowledgebase
        },
        {
            "query": "Which deep learning framework is from Meta?",
            "expectedanswer": "PyTorch",

            "knowledgebase": knowledgebase

        },
    ]
)

@weave.op
def answerrelevancescorer(expectedanswer: str, modeloutput: dict) -> dict:

    """Check if the answer contains the expected information."""
    answer = modeloutput.get("answer", "").lower()
    expected = expectedanswer.lower()

    containsexpected = expected in answer
    return {"relevant": containsexpected}


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

    """Evaluate retrieval quality."""
    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)

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

@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

class MyAssistant(weave.Model):
    modelname: str = "gpt-4o-mini"
    temperature: float = 0.7
    maxtokens: int = 500

    systemprompt: str = "You are a helpful assistant."

    @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

Every time a parameter changes, Weave automatically creates a new version, allowing you to track which changes affected output quality.

4. Evaluate Regularly

Create CI/CD pipelines that run evaluations whenever prompts or configurations change:

async def cievaluation():
    """Run evaluation as part of the 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} below threshold 0.8"
    print(f"CI evaluation passed with accuracy: {accuracy}")

5. Use Multiple Scorers

Combine various scorers for comprehensive evaluation:

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

        formatscorer,
        llmjudgescorer,
        latency_scorer,
    ]
)

Integration with Other LLM Providers

Weave natively supports various LLM providers:

# 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=[...])

All of the above calls are automatically traced by Weave without any additional code.

Conclusion

Weave by Weights & Biases is an invaluable tool for serious LLM application development. With automatic tracing, a flexible evaluation framework, and production monitoring capabilities, Weave helps you build more reliable and higher-quality LLM applications.

Key takeaways from this tutorial:

• Tracing: Use weave.init() for auto-tracing and @weave.op for custom tracing
• Evaluation: Create datasets, define scorers, and run systematic evaluations using weave.Evaluation
• Model Versioning: Use weave.Model to track configurations and compare performance
• Production Monitoring: Integrate Weave into production applications for real-time monitoring
• Comparison: Compare models, prompts, and configurations side-by-side in the dashboard

Start by installing Weave (pip install weave) and add tracing to your LLM project. You will be surprised how many insights you can gain about your LLM application's behavior.