Dspy Langwatch

LangWatch — Auto-Tracing + Real-Time Optimizer Progress for DSPy

Guide the user through setting up LangWatch for automatic DSPy tracing and live optimizer progress tracking.

What is LangWatch

LangWatch is an open-source LLMOps platform with two distinct DSPy integrations:

1. Auto-tracing (inference): automatically captures module inputs/outputs, LM calls, and retrieval queries 2. Optimizer progress tracking (unique feature): streams live step-by-step scores, predictor states, and cost as optimizers run

No other observability tool (Langtrace, Phoenix, Weave, MLflow) patches DSPy optimizers to stream live progress.

• Cloud: Managed at app.langwatch.ai (free tier available)
• Self-hosted: Docker Compose, Helm chart, enterprise on-prem
• Open source: github.com/langwatch/langwatch

When to use LangWatch

Use LangWatch when:

• You run long optimization passes and want to see progress in real-time
• You want auto-tracing of DSPy inference with no manual decorators
• You want a dashboard showing optimizer scores, cost, and predictor state as they happen
• You need both inference tracing AND optimizer monitoring in one tool

Do NOT use LangWatch when:

• You only need tracing and want the simplest one-line setup — see /dspy-langtrace
• You want a local trace viewer with built-in evals — see /dspy-phoenix
• Your team already uses W&B for experiment tracking — see /dspy-weave
• You need a model registry and full ML lifecycle — see /dspy-mlflow

Setup

Install

pip install langwatch
# Or pin DSPy version compatibility:
pip install langwatch[dspy]

Cloud setup (quickest)

1. Sign up at app.langwatch.ai 2. Create a project and copy your API key 3. Set the environment variable:

export LANGWATCH_API_KEY="your-key"

Self-hosted setup

Docker Compose

git clone https://github.com/langwatch/langwatch.git
cd langwatch
docker compose up -d

Then point your SDK at your local instance:

export LANGWATCH_ENDPOINT="http://localhost:5560"

Helm chart (Kubernetes)

LangWatch provides a Helm chart for production Kubernetes deployments. See the LangWatch docs for Helm values and configuration.

Integration 1: Auto-Tracing (Inference)

Use @langwatch.trace() and autotrack_dspy() to automatically capture all DSPy calls during inference.

What gets traced

Basic auto-tracing

import langwatch
import dspy

dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))

@langwatch.trace()
def answer_question(question):
    langwatch.get_current_trace().autotrack_dspy()

    program = dspy.ChainOfThought("question -> answer")
    return program(question=question)

result = answer_question("What is DSPy?")
# View traces at app.langwatch.ai (or your self-hosted URL)

Tracing a full pipeline

import langwatch
import dspy

dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))

class RAGPipeline(dspy.Module):
    def __init__(self):
        self.retrieve = dspy.Retrieve(k=3)
        self.answer = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        context = self.retrieve(question).passages
        return self.answer(context=context, question=question)

pipeline = RAGPipeline()

@langwatch.trace()
def handle_query(question):
    langwatch.get_current_trace().autotrack_dspy()
    return pipeline(question=question)

result = handle_query("How do refunds work?")
# LangWatch captures:
#   - The RAGPipeline call
#   - The Retrieve call (query, passages)
#   - The ChainOfThought LM call (prompt, response, tokens)

Adding metadata to traces

@langwatch.trace()
def handle_query(user_id, question):
    trace = langwatch.get_current_trace()
    trace.autotrack_dspy()
    trace.update(metadata={"user_id": user_id, "environment": "production"})
    return pipeline(question=question)

Integration 2: Optimizer Progress Tracking (Unique Feature)

LangWatch patches DSPy optimizer classes to stream live step-by-step progress. This is LangWatch's killer feature — no other tool does this.

What the optimizer dashboard shows

• Live scores: see each trial's score as it completes
• Predictor states: which instructions and demos the optimizer is testing
• LM calls: every call the optimizer makes during search
• Cost tracking: running cost total as the optimizer runs
• Progress bar: how far through the optimization you are

Supported optimizers

Setup optimizer tracking

import langwatch.dspy
import dspy

dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))

trainset = [...]  # your training examples

def metric(example, prediction, trace=None):
    return prediction.answer.strip().lower() == example.answer.strip().lower()

program = dspy.ChainOfThought("question -> answer")
optimizer = dspy.MIPROv2(metric=metric, auto="medium")

# Initialize LangWatch optimizer tracking
langwatch.dspy.init(
    experiment="mipro-medium-run1",
    optimizer=optimizer,
)

# Run optimization — progress streams to the LangWatch dashboard
optimized = optimizer.compile(program, trainset=trainset)
# Watch live progress at app.langwatch.ai

Tracking BootstrapFewShot

import langwatch.dspy
import dspy

dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))

program = dspy.ChainOfThought("question -> answer")
optimizer = dspy.BootstrapFewShot(metric=metric, max_bootstrapped_demos=4)

langwatch.dspy.init(
    experiment="bootstrap-4demos",
    optimizer=optimizer,
)

optimized = optimizer.compile(program, trainset=trainset)

Comparing multiple optimizer runs

Run multiple experiments with different names — they appear side-by-side in the LangWatch dashboard:

import langwatch.dspy
import dspy

dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))

experiments = [
    ("bootstrap-4", dspy.BootstrapFewShot, {"metric": metric, "max_bootstrapped_demos": 4}),
    ("bootstrap-8", dspy.BootstrapFewShot, {"metric": metric, "max_bootstrapped_demos": 8}),
    ("mipro-light", dspy.MIPROv2, {"metric": metric, "auto": "light"}),
    ("mipro-medium", dspy.MIPROv2, {"metric": metric, "auto": "medium"}),
]

for name, opt_class, kwargs in experiments:
    program = dspy.ChainOfThought("question -> answer")
    optimizer = opt_class(**kwargs)
    langwatch.dspy.init(experiment=name, optimizer=optimizer)
    optimized = optimizer.compile(program, trainset=trainset)

LangWatch vs Langtrace vs Phoenix vs Weave vs MLflow

Decision guide

What do you need?
|
+- Watch optimizer progress live? -> LangWatch (this skill)
+- Easiest auto-tracing setup? -> Langtrace (/dspy-langtrace)
+- Tracing + evals (local)? -> Phoenix (/dspy-phoenix)
+- Tracing + experiment tracking (cloud)? -> Weave (/dspy-weave)
+- Full ML lifecycle + model registry? -> MLflow (/dspy-mlflow)

Cross-references

• Langtrace (auto-instrumentation, easiest one-line setup) — /dspy-langtrace
• Arize Phoenix (open-source with evals) — /dspy-phoenix
• W&B Weave (team dashboards, experiment tracking) — /dspy-weave
• MLflow (full ML lifecycle, model registry) — /dspy-mlflow
• Lightweight experiment tracking (JSONL-based, no extra tools) — /ai-tracking-experiments
• Production monitoring — /ai-monitoring
• For worked examples, see examples.md

LangWatch Examples

Trace a pipeline at inference

import langwatch
import dspy

dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))

class SupportBot(dspy.Module):
    def __init__(self):
        self.retrieve = dspy.Retrieve(k=5)
        self.answer = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        context = self.retrieve(question).passages
        return self.answer(context=context, question=question)

bot = SupportBot()

@langwatch.trace()
def handle_support(question):
    langwatch.get_current_trace().autotrack_dspy()
    return bot(question=question)

# Run queries — all automatically traced
questions = [
    "How do I reset my password?",
    "What's your refund policy?",
    "Can I upgrade my plan mid-cycle?",
]

for q in questions:
    result = handle_support(question=q)
    print(f"Q: {q}\nA: {result.answer}\n")

Find slow requests in the LangWatch UI

1. Go to app.langwatch.ai (or your self-hosted URL) 2. Open your project 3. Sort traces by latency (descending) 4. Click a slow trace to see the span tree 5. Check which step is the bottleneck:

• Retrieve slow? Vector DB may need optimization or query is too broad
• LM call slow? Model may be overloaded or prompt is too long

Watch MIPROv2 optimization live

import langwatch.dspy
import dspy
from dspy.evaluate import Evaluate

dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))

# Prepare data
trainset = [
    dspy.Example(question="What is Python?", answer="A programming language").with_inputs("question"),
    dspy.Example(question="What is DSPy?", answer="A framework for programming LMs").with_inputs("question"),
    # ... more examples
]
devset = trainset[:10]

def metric(example, prediction, trace=None):
    return prediction.answer.strip().lower() == example.answer.strip().lower()

# Set up optimizer with LangWatch tracking
program = dspy.ChainOfThought("question -> answer")
optimizer = dspy.MIPROv2(metric=metric, auto="medium")

langwatch.dspy.init(
    experiment="mipro-medium-support-bot",
    optimizer=optimizer,
)

# Start optimization — open app.langwatch.ai to watch live
optimized = optimizer.compile(program, trainset=trainset)

# The dashboard shows:
# - Each trial's score as it completes
# - Which instructions/demos the optimizer tested
# - Running cost total
# - Progress through the optimization

# Evaluate the result
evaluator = Evaluate(devset=devset, metric=metric, num_threads=4)
score = evaluator(optimized)
print(f"Final score: {score:.1f}%")

Compare optimizer strategies side-by-side

import langwatch.dspy
import dspy

dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))

def metric(example, prediction, trace=None):
    return prediction.answer.strip().lower() == example.answer.strip().lower()

# Run three experiments — each appears in the LangWatch dashboard
strategies = [
    {
        "name": "bootstrap-quick",
        "optimizer_class": dspy.BootstrapFewShot,
        "kwargs": {"metric": metric, "max_bootstrapped_demos": 4},
    },
    {
        "name": "mipro-light",
        "optimizer_class": dspy.MIPROv2,
        "kwargs": {"metric": metric, "auto": "light"},
    },
    {
        "name": "mipro-medium",
        "optimizer_class": dspy.MIPROv2,
        "kwargs": {"metric": metric, "auto": "medium"},
    },
]

results = {}
for strategy in strategies:
    program = dspy.ChainOfThought("question -> answer")
    optimizer = strategy["optimizer_class"](**strategy["kwargs"])

    langwatch.dspy.init(
        experiment=strategy["name"],
        optimizer=optimizer,
    )

    optimized = optimizer.compile(program, trainset=trainset)
    results[strategy["name"]] = optimized

# Open app.langwatch.ai — all three experiments are visible
# Compare scores, cost, and convergence speed across strategies

Trace with metadata in production

import langwatch
import dspy

dspy.configure(lm=dspy.LM("openai/gpt-4o-mini"))

bot = SupportBot()

@langwatch.trace()
def handle_support(user_id, plan, question):
    trace = langwatch.get_current_trace()
    trace.autotrack_dspy()
    trace.update(metadata={
        "user_id": user_id,
        "plan": plan,
        "source": "api",
    })
    return bot(question=question)

# In production — filter traces by plan or user_id in the dashboard
result = handle_support("user-456", "enterprise", "How do I set up SSO?")