Ai Tracing Requests

See What Your AI Did on a Specific Request

Guide the user through tracing and debugging individual AI requests. The goal: for any request, see every LM call, retrieval step, intermediate result, token count, and latency.

How tracing differs from monitoring

Step 1: Understand the situation

Ask the user: 1. What happened? A specific wrong answer, slow response, or unexpected behavior? 2. What does your pipeline look like? Single module or multi-step pipeline? Which DSPy modules? 3. Where is this running? Local development, staging, or production?

Then decide the approach:

Step 2: Quick debugging (no extra tools needed)

Inspect the last LM calls

The fastest way to see what happened:

import dspy

# Run your program
result = my_program(question="What is our refund policy?")

# See the last 5 LM calls — shows full prompts and responses
dspy.inspect_history(n=5)

# Save history to a file for later analysis (DSPy 3.2+)
dspy.inspect_history(n=10, file_path="debug_trace.txt")

This shows:

• The full prompt sent to the LM (including system message, few-shot examples, input)
• The LM's raw response
• How DSPy parsed the response into fields

Time individual steps

import time

result = my_program(question="test")

# Quick manual timing
start = time.time()
step1_result = my_program.step1(question="test")
step1_time = time.time() - start
print(f"Step 1: {step1_time:.2f}s")

start = time.time()
step2_result = my_program.step2(context=step1_result.context, question="test")
step2_time = time.time() - start
print(f"Step 2: {step2_time:.2f}s")

JSONL trace logging

For persistent traces without any extra dependencies:

import json
import time
from datetime import datetime

class TracedProgram(dspy.Module):
    """Wraps any DSPy program to log per-step traces to JSONL."""
    def __init__(self, program, log_path="traces.jsonl"):
        self.program = program
        self.log_path = log_path

    def forward(self, **kwargs):
        trace_id = datetime.now().strftime("%Y%m%d_%H%M%S_%f")
        steps = []

        start = time.time()
        result = self.program(**kwargs)
        total_time = time.time() - start

        # Log the trace
        entry = {
            "trace_id": trace_id,
            "timestamp": datetime.now().isoformat(),
            "inputs": {k: str(v) for k, v in kwargs.items()},
            "outputs": {k: str(getattr(result, k, "")) for k in result.keys()},
            "total_latency_ms": round(total_time * 1000),
        }
        with open(self.log_path, "a") as f:
            f.write(json.dumps(entry) + "\n")

        return result

# Use it
traced = TracedProgram(my_program)
result = traced(question="How do refunds work?")

Step 3: DSPy callback system (recommended for structured tracing)

DSPy has a built-in callback system that hooks into every module, LM call, tool call, and adapter operation. This is the official observability API — use it instead of manual wrappers when possible.

from dspy.utils.callback import BaseCallback

class TracingCallback(BaseCallback):
    def on_module_start(self, call_id, instance, inputs):
        print(f"[{call_id}] Module {instance.__class__.__name__} started")
        print(f"  Inputs: {inputs}")

    def on_module_end(self, call_id, outputs, exception):
        if exception:
            print(f"[{call_id}] FAILED: {exception}")
        else:
            print(f"[{call_id}] Outputs: {outputs}")

    def on_lm_start(self, call_id, instance, inputs):
        print(f"[{call_id}] LM call started")

    def on_lm_end(self, call_id, outputs, exception):
        print(f"[{call_id}] LM call finished")

    def on_tool_start(self, call_id, instance, inputs):
        print(f"[{call_id}] Tool {instance.name} called")

    def on_tool_end(self, call_id, outputs, exception):
        print(f"[{call_id}] Tool finished")

# Register the callback globally
dspy.configure(callbacks=[TracingCallback()])

# All DSPy calls now trigger the callback hooks automatically
result = my_program(question="test")

Available callback hooks: on_module_start/end, on_lm_start/end, on_adapter_format_start/end, on_adapter_parse_start/end, on_tool_start/end, on_evaluate_start/end.

Do not mutate input/output data inside callbacks — this can cause subtle bugs in the pipeline.

Step 4: Manual per-step tracing in pipelines

For multi-step pipelines, trace each stage separately to see exactly where things go wrong:

import json
import time
import uuid
from datetime import datetime

class StepTracer:
    """Collects per-step timing and intermediate results."""
    def __init__(self):
        self.steps = []
        self.trace_id = str(uuid.uuid4())[:8]

    def trace_step(self, name, func, **kwargs):
        """Run a step and record its inputs, outputs, and latency."""
        start = time.time()
        result = func(**kwargs)
        latency = time.time() - start

        self.steps.append({
            "step": name,
            "inputs": {k: str(v)[:200] for k, v in kwargs.items()},
            "outputs": {k: str(getattr(result, k, ""))[:200] for k in result.keys()},
            "latency_ms": round(latency * 1000),
        })
        return result

    def summary(self):
        """Print a summary of all traced steps."""
        print(f"Trace {self.trace_id}:")
        total = sum(s["latency_ms"] for s in self.steps)
        for step in self.steps:
            pct = step["latency_ms"] / total * 100 if total > 0 else 0
            print(f"  {step['step']}: {step['latency_ms']}ms ({pct:.0f}%)")
        print(f"  Total: {total}ms")

    def to_dict(self):
        return {
            "trace_id": self.trace_id,
            "timestamp": datetime.now().isoformat(),
            "steps": self.steps,
            "total_latency_ms": sum(s["latency_ms"] for s in self.steps),
        }

# Use in a pipeline
class TracedRAG(dspy.Module):
    def __init__(self):
        self.retrieve = dspy.Retrieve(k=3)
        self.answer = dspy.ChainOfThought("context, question -> answer")

    def forward(self, question):
        tracer = StepTracer()

        retrieval = tracer.trace_step("retrieve", self.retrieve, query=question)

        answer = tracer.trace_step(
            "answer", self.answer,
            context=retrieval.passages, question=question,
        )

        tracer.summary()
        # Trace a1b2c3d4:
        #   retrieve: 120ms (15%)
        #   answer: 680ms (85%)
        #   Total: 800ms

        return answer

Save traces for later analysis

def save_trace(tracer, path="traces.jsonl"):
    with open(path, "a") as f:
        f.write(json.dumps(tracer.to_dict()) + "\n")

# Load and analyze traces
def load_traces(path="traces.jsonl"):
    with open(path) as f:
        return [json.loads(line) for line in f]

def find_slow_traces(traces, threshold_ms=2000):
    return [t for t in traces if t["total_latency_ms"] > threshold_ms]

def find_failed_steps(traces):
    return [
        t for t in traces
        if any("error" in str(s.get("outputs", "")).lower() for s in t["steps"])
    ]

Step 4b: OpenTelemetry instrumentation

For production tracing with any backend (Jaeger, Zipkin, Datadog, etc.):

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor

# Setup — do this once at app startup
provider = TracerProvider()
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("my-ai-app")

class OTelTracedProgram(dspy.Module):
    """Wraps a DSPy program with OpenTelemetry spans."""
    def __init__(self, program):
        self.program = program

    def forward(self, **kwargs):
        with tracer.start_as_current_span("ai_request") as span:
            span.set_attribute("ai.inputs", json.dumps({k: str(v) for k, v in kwargs.items()}))

            start = time.time()
            result = self.program(**kwargs)
            latency = time.time() - start

            span.set_attribute("ai.latency_ms", round(latency * 1000))
            span.set_attribute("ai.outputs", json.dumps(
                {k: str(getattr(result, k, "")) for k in result.keys()}
            ))

            return result

Trace individual pipeline steps with OTel

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

    def forward(self, question):
        with tracer.start_as_current_span("rag_pipeline") as parent:
            parent.set_attribute("question", question)

            with tracer.start_as_current_span("retrieve"):
                retrieval = self.retrieve(query=question)

            with tracer.start_as_current_span("generate_answer"):
                answer = self.answer(
                    context=retrieval.passages, question=question
                )

            return answer

Step 5: Connect a trace viewer or MLflow

Option A: Langtrace (best DSPy integration)

First-class DSPy auto-instrumentation — one line to trace all LM calls:

pip install langtrace-python-sdk

from langtrace_python_sdk import langtrace

langtrace.init(api_key="your-key")  # or use LANGTRACE_API_KEY env var

# That's it — all DSPy calls are now traced automatically
result = my_program(question="test")
# View traces at app.langtrace.ai

Option B: Arize Phoenix (open-source, self-hosted)

pip install arize-phoenix openinference-instrumentation-dspy

import phoenix as px
from openinference.instrumentation.dspy import DSPyInstrumentor

# Launch local trace viewer
px.launch_app()  # Opens at http://localhost:6006

# Auto-instrument DSPy
DSPyInstrumentor().instrument()

# All DSPy calls are now traced
result = my_program(question="test")

Option C: Jaeger (open-source, Docker)

docker run -d -p 16686:16686 -p 4317:4317 jaegertracing/all-in-one:latest

from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# Export spans to Jaeger
exporter = OTLPSpanExporter(endpoint="http://localhost:4317", insecure=True)
provider.add_span_processor(BatchSpanProcessor(exporter))

# View traces at http://localhost:16686

Option D: MLflow Tracing (comprehensive, self-hosted)

pip install -U mlflow>=2.18.0
mlflow server --backend-store-uri sqlite:///mydb.sqlite

import mlflow

mlflow.set_tracking_uri("http://127.0.0.1:5000")
mlflow.set_experiment("DSPy")

# Auto-trace all DSPy calls (LMs, retrievers, tools, modules)
mlflow.dspy.autolog()

result = my_program(question="test")
# View traces at http://127.0.0.1:5000

MLflow captures the full call tree including LM calls, retrievers, tools, and custom modules — more comprehensive than inspect_history.

Comparison

For in-depth guides: /dspy-langtrace, /dspy-phoenix, /dspy-mlflow.

Step 6: Use traces to improve your AI

Find patterns in wrong answers

# Load JSONL traces and find failures
import json

def load_traces(path="traces.jsonl"):
    with open(path) as f:
        return [json.loads(line) for line in f]

wrong_traces = [t for t in load_traces() if "error" in json.dumps(t).lower()]

# Check which step is most often the bottleneck
from collections import Counter
slow_steps = Counter()
for t in wrong_traces:
    if t.get("steps"):
        slowest = max(t["steps"], key=lambda s: s["latency_ms"])
        slow_steps[slowest["step"]] += 1

print(slow_steps)
# Counter({"retrieve": 23, "answer": 7})
# -> Retrieval is the problem, not the answer generation

Build training data from failures

failed_examples = []
for t in wrong_traces:
    ex = dspy.Example(
        question=t.get("inputs", {}).get("question", ""),
    ).with_inputs("question")
    failed_examples.append(ex)

# Add to training set and re-optimize
# See /ai-improving-accuracy

Gotchas

1. Building custom tracing wrappers instead of using DSPy callbacks. Claude defaults to writing manual time.time() wrappers around each step. DSPy has a built-in callback system (BaseCallback with on_module_start/end, on_lm_start/end, etc.) that hooks into every operation automatically. Use it instead of reinventing tracing infrastructure.

2. Using `inspect_history` in production. inspect_history prints to stdout and only logs LM calls — it misses retriever, tool, and module-level data. For production, use the callback system or an external trace viewer (Langtrace, Phoenix, MLflow). Reserve inspect_history for local debugging.

3. Tracing the whole request instead of individual steps. Claude wraps the entire pipeline in one timing block, which shows total latency but not which step is slow. Always trace at the step level — either with per-step callbacks or by wrapping individual modules in the forward() method.

4. Forgetting to save traces before they are needed. Claude often adds tracing after a bug is reported, but the problematic request is already gone. Add JSONL trace logging or connect a trace viewer before you need it — you cannot debug traces you did not log.

5. Mutating inputs or outputs inside callback hooks. The callback system passes live references to module inputs and outputs. Modifying them in place (e.g., truncating a field for logging) silently corrupts the pipeline data. Always copy before modifying: inputs_copy = dict(inputs).

Additional resources

Install any skill: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill <name>

• For worked examples, see examples.md
• For DSPy tracing API details, see reference.md
• Use /ai-monitoring for aggregate health checks across all requests
• Use /ai-fixing-errors for code-level debugging (crashes, config issues)
• Use /ai-building-pipelines to structure pipelines that are easy to trace
• Use /ai-improving-accuracy to optimize based on patterns found in traces
• Use /dspy-langtrace for in-depth Langtrace setup (auto-instrumentation, self-hosted)
• Use /dspy-phoenix for in-depth Phoenix setup (local UI, evals)
• Use /dspy-mlflow for MLflow tracing and experiment tracking
• Install `/ai-do` if you do not have it — it routes any AI problem to the right skill and is the fastest way to work: npx skills add lebsral/DSPy-Programming-not-prompting-LMs-skills --skill ai-do

Tracing Examples

Example 1: Debugging a RAG pipeline wrong answer

A customer reported that the help center bot gave a wrong answer about refund policies. Walk through tracing the exact request.

The problem

result = help_bot(question="Can I get a refund after 30 days?")
print(result.answer)
# "Yes, you can request a refund at any time."
# WRONG — the actual policy is 30-day limit

Step 1: Inspect the LM calls

# Re-run the question and inspect
result = help_bot(question="Can I get a refund after 30 days?")
dspy.inspect_history(n=3)

Output shows:

--- LM Call 1 (retrieve) ---
Query: "Can I get a refund after 30 days?"
Retrieved passages:
  1. "Refund requests must be submitted within 30 days..."  ✓ correct
  2. "We offer a satisfaction guarantee on all products..."  ✗ irrelevant
  3. "Contact support@example.com for assistance..."        ✗ irrelevant

--- LM Call 2 (answer) ---
Prompt: "Answer the question based on the context..."
Context: [the 3 passages above]
Response: "Yes, you can request a refund at any time."

Root cause found: The retriever found the right document (passage 1), but the other 2 passages diluted the context. The LM ignored the 30-day limit mentioned in passage 1.

Step 2: Trace with timing

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

    def forward(self, question):
        tracer = StepTracer()

        retrieval = tracer.trace_step("retrieve", self.retrieve, query=question)
        print(f"Retrieved {len(retrieval.passages)} passages:")
        for i, p in enumerate(retrieval.passages):
            print(f"  [{i+1}] {p[:100]}...")

        answer = tracer.trace_step(
            "answer", self.answer,
            context=retrieval.passages, question=question,
        )

        tracer.summary()
        save_trace(tracer)
        return answer

bot = TracedHelpBot()
result = bot(question="Can I get a refund after 30 days?")
# Retrieved 3 passages:
#   [1] Refund requests must be submitted within 30 days of purchase...
#   [2] We offer a satisfaction guarantee on all products...
#   [3] Contact support@example.com for assistance...
# Trace a1b2c3d4:
#   retrieve: 95ms (12%)
#   answer: 720ms (88%)
#   Total: 815ms

Step 3: Fix the issue

The fix: reduce k to 2 to get more focused context, and add a reward-guided refinement step:

class FixedHelpBot(dspy.Module):
    def __init__(self):
        self.retrieve = dspy.Retrieve(k=2)  # fewer, more relevant passages
        self.answer = dspy.ChainOfThought("context, question -> answer")

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

def grounded_answer_reward(args, pred):
    """Reward answers that reference specific information from the retrieved passages."""
    # Store passages alongside the question so the reward fn can check grounding
    passages = args.get("passages", [])
    score = 1.0
    if not (any(p[:50] in pred.answer for p in passages) or len(pred.answer) > 20):
        score -= 0.5
    return score

bot_base = FixedHelpBot()
bot = dspy.Refine(module=bot_base, N=3, reward_fn=grounded_answer_reward, threshold=0.5)

Example 2: Profiling a slow multi-step pipeline

A classification pipeline takes 8+ seconds. Find the bottleneck.

The slow pipeline

class ContentPipeline(dspy.Module):
    def __init__(self):
        self.classify = dspy.ChainOfThought("text -> category")
        self.extract = dspy.ChainOfThought("text, category -> entities: list[str]")
        self.summarize = dspy.ChainOfThought("text, category, entities -> summary")
        self.check = dspy.ChainOfThought("text, summary -> is_safe: bool, issues: list[str]")

    def forward(self, text):
        cat = self.classify(text=text)
        ents = self.extract(text=text, category=cat.category)
        summary = self.summarize(text=text, category=cat.category, entities=ents.entities)
        safety = self.check(text=text, summary=summary.summary)
        return dspy.Prediction(
            category=cat.category,
            entities=ents.entities,
            summary=summary.summary,
            is_safe=safety.is_safe,
        )

Add tracing to find the bottleneck

class ProfiledPipeline(dspy.Module):
    def __init__(self):
        self.classify = dspy.ChainOfThought("text -> category")
        self.extract = dspy.ChainOfThought("text, category -> entities: list[str]")
        self.summarize = dspy.ChainOfThought("text, category, entities -> summary")
        self.check = dspy.ChainOfThought("text, summary -> is_safe: bool, issues: list[str]")

    def forward(self, text):
        tracer = StepTracer()

        cat = tracer.trace_step("classify", self.classify, text=text)
        ents = tracer.trace_step("extract", self.extract, text=text, category=cat.category)
        summary = tracer.trace_step(
            "summarize", self.summarize,
            text=text, category=cat.category, entities=ents.entities,
        )
        safety = tracer.trace_step("safety_check", self.check, text=text, summary=summary.summary)

        tracer.summary()
        save_trace(tracer)

        return dspy.Prediction(
            category=cat.category,
            entities=ents.entities,
            summary=summary.summary,
            is_safe=safety.is_safe,
        )

pipeline = ProfiledPipeline()
result = pipeline(text="Long article text here...")

# Trace output:
# Trace f3e4d5c6:
#   classify: 450ms (5%)
#   extract: 1200ms (14%)
#   summarize: 5800ms (68%)    <-- BOTTLENECK
#   safety_check: 1100ms (13%)
#   Total: 8550ms

Bottleneck found: The summarize step takes 68% of total time.

Fix: use a cheaper model for the bottleneck

expensive_lm = dspy.LM("openai/gpt-4o")  # or "anthropic/claude-sonnet-4-5-20250929", etc.
cheap_lm = dspy.LM("openai/gpt-4o-mini")  # or "anthropic/claude-haiku-4-5-20251001", etc.

pipeline = ProfiledPipeline()

# Use cheap model for the slow step (summarization is easier than classification)
pipeline.summarize.set_lm(cheap_lm)

# Re-profile
result = pipeline(text="Long article text here...")
# Trace g7h8i9j0:
#   classify: 450ms (10%)
#   extract: 1200ms (28%)
#   summarize: 1400ms (32%)   <-- 4x faster!
#   safety_check: 1100ms (30%)
#   Total: 4150ms             <-- 51% reduction

Run profiling across multiple inputs

traces = []
for text in test_texts[:20]:
    pipeline(text=text)

all_traces = load_traces()
stats = trace_stats(all_traces)
print(stats)
# {"count": 20, "p50_ms": 4200, "p95_ms": 6800, "p99_ms": 8100, "max_ms": 8550}

# Find which step is slowest across all traces
from collections import defaultdict
step_times = defaultdict(list)
for t in all_traces:
    for step in t["steps"]:
        step_times[step["step"]].append(step["latency_ms"])

for step, times in step_times.items():
    times.sort()
    p50 = times[len(times) // 2]
    print(f"  {step}: p50={p50}ms")

Tracing and Debugging Reference

Condensed from dspy.ai/tutorials/observability and dspy.ai/tutorials/streaming. Verify against upstream for latest.

inspect_history

The simplest debugging tool. Prints the last N LM calls to stdout.

dspy.inspect_history(n=5)

# Save to file (DSPy 3.2+)
dspy.inspect_history(n=10, file_path="debug_trace.txt")

Limitations: Only logs LM calls. Misses retriever, tool, and custom module data. Not suitable for production — use the callback system or an external trace viewer instead.

Callback System (BaseCallback)

The official DSPy observability API. Hooks into every module, LM call, tool call, and adapter operation.

from dspy.utils.callback import BaseCallback

class MyCallback(BaseCallback):
    def on_module_start(self, call_id, instance, inputs):
        ...
    def on_module_end(self, call_id, outputs, exception):
        ...

dspy.configure(callbacks=[MyCallback()])

Available hooks

Important: Do not mutate inputs or outputs in place — these are live references. Copy before modifying.

StreamListener

Monitors token streaming for specific output fields.

listener = dspy.streaming.StreamListener(
    signature_field_name="answer",
    predict=module.predict1,         # optional: target specific predict
    predict_name="predict1",         # optional: disambiguate duplicate names
    allow_reuse=False,               # True for loops like ReAct
)

stream_module = dspy.streamify(module, stream_listeners=[listener])

StreamResponse

Each streamed token is a StreamResponse with:

• predict_name — name of the predict module
• signature_field_name — output field identifier
• chunk — the token value

StatusMessageProvider

Provides real-time execution status updates (tool calls, LM invocations):

from dspy.streaming import StatusMessageProvider

class MyStatusProvider(StatusMessageProvider):
    def lm_start_status_message(self, instance, inputs):
        return "Calling LM..."
    def tool_start_status_message(self, instance, inputs):
        return f"Using tool: {instance.name}"

dspy.streamify()

Returns an async or sync generator yielding StreamResponse, StatusMessage, or Prediction objects. Cached results skip individual tokens and yield the final Prediction directly.

Caching

DSPy uses a three-layer cache: in-memory (LRU), on-disk (FanoutCache), and provider-side prompt cache. Both in-memory and disk caching are enabled by default.

dspy.configure_cache(
    enable_disk_cache=True,
    enable_memory_cache=True,
    disk_size_limit_bytes=1_000_000_000,
    memory_max_entries=10_000,
)

Provider-side prompt caching

lm = dspy.LM(
    "anthropic/claude-sonnet-4-5-20250929",
    cache_control_injection_points=[{"location": "message", "role": "system"}],
)

Security

Use restrict_pickle=True to prevent arbitrary code execution from corrupted cache files:

dspy.configure_cache(restrict_pickle=True, safe_types=[CustomType])

External trace viewers