Distributed Tracing for Multi-Agent Systems

When a single agent fails, debugging is straightforward — read the logs. When five agents collaborate on a task and the result is wrong, finding the failure point is like finding a needle in a haystack. Distributed tracing gives you a map: every decision, tool call, and handoff visualized in a single trace.

Why Multi-Agent Systems Need Tracing

Multi-agent systems introduce complexity that traditional logging can’t handle:

• Cross-agent communication — Agent A calls Agent B which calls Agent C
• Parallel execution — Multiple agents working simultaneously
• Tool call chains — Agent uses Tool A which triggers Tool B
• State propagation — Understanding how state flows between agents
• Error attribution — Knowing which agent caused a failure

Without tracing, you’re debugging blind.

OpenTelemetry for Agents

OpenTelemetry (OTel) is the industry standard for distributed tracing. It works perfectly for agent systems:

from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter

# Set up tracing
provider = TracerProvider()
processor = BatchSpanProcessor(OTLPSpanExporter(endpoint="http://localhost:4317"))
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)

tracer = trace.get_tracer("agent-system")

Instrumenting Agent Execution

Basic Agent Span

from opentelemetry.trace import StatusCode

class TracedAgent:
    def __init__(self, name: str, llm, tools):
        self.name = name
        self.llm = llm
        self.tools = tools
    
    def execute(self, task: str, parent_context=None) -> str:
        with tracer.start_as_current_span(
            "agent.execute",
            attributes={
                "agent.name": self.name,
                "agent.task": task,
            }
        ) as span:
            try:
                # Reasoning step
                with tracer.start_as_current_span("agent.reason") as reason_span:
                    plan = self._plan(task)
                    reason_span.set_attribute("agent.plan", str(plan))
                
                # Tool execution
                results = []
                for tool_call in plan.get("tool_calls", []):
                    result = self._execute_tool(tool_call)
                    results.append(result)
                
                # Synthesis
                with tracer.start_as_current_span("agent.synthesize") as synth_span:
                    response = self._synthesize(task, results)
                    synth_span.set_attribute("agent.response_length", len(response))
                
                span.set_status(StatusCode.OK)
                return response
            
            except Exception as e:
                span.set_status(StatusCode.ERROR, str(e))
                span.record_exception(e)
                raise

Tracing Tool Calls

class TracedToolExecutor:
    def execute(self, tool_name: str, arguments: dict) -> dict:
        with tracer.start_as_current_span(
            "tool.execute",
            attributes={
                "tool.name": tool_name,
                "tool.arguments": json.dumps(arguments)[:500],
            }
        ) as span:
            start_time = time.time()
            
            try:
                result = self._run_tool(tool_name, arguments)
                duration_ms = (time.time() - start_time) * 1000
                
                span.set_attribute("tool.result_length", len(str(result)))
                span.set_attribute("tool.duration_ms", duration_ms)
                span.set_attribute("tool.success", True)
                span.set_status(StatusCode.OK)
                
                return {"success": True, "result": result, "duration_ms": duration_ms}
            
            except Exception as e:
                duration_ms = (time.time() - start_time) * 1000
                span.set_attribute("tool.success", False)
                span.set_attribute("tool.error", str(e))
                span.set_status(StatusCode.ERROR, str(e))
                span.record_exception(e)
                
                return {"success": False, "error": str(e), "duration_ms": duration_ms}

Tracing Agent-to-Agent Handoffs

class TracedMultiAgentSystem:
    def __init__(self, agents: dict):
        self.agents = agents
    
    def orchestrate(self, task: str) -> str:
        with tracer.start_as_current_span(
            "orchestration.run",
            attributes={"orchestration.task": task}
        ) as span:
            current_agent = "supervisor"
            context = {"task": task}
            
            while current_agent != "done":
                with tracer.start_as_current_span(
                    f"agent.{current_agent}",
                    attributes={"agent.id": current_agent}
                ) as agent_span:
                    
                    # Execute agent
                    result = self.agents[current_agent].execute(
                        task=task,
                        context=context
                    )
                    
                    # Record handoff
                    next_agent = result.get("next_agent", "done")
                    
                    if next_agent != "done":
                        with tracer.start_as_current_span(
                            "agent.handoff",
                            attributes={
                                "handoff.from": current_agent,
                                "handoff.to": next_agent,
                                "handoff.context": json.dumps(result.get("context", {}))[:200]
                            }
                        ):
                            pass  # Handoff span records the transition
                    
                    agent_span.set_attribute("agent.next", next_agent)
                    current_agent = next_agent
                    context.update(result.get("context", {}))
            
            return context.get("final_answer", "")

Adding Semantic Attributes

Define consistent attributes across your traces:

# Semantic conventions for agent traces
AGENT_ATTRIBUTES = {
    "agent.name": "Name of the agent",
    "agent.task": "Input task/query",
    "agent.plan": "Agent's execution plan",
    "agent.response_length": "Length of agent response",
    "tool.name": "Name of tool called",
    "tool.arguments": "Tool input arguments",
    "tool.result_length": "Length of tool result",
    "tool.success": "Whether tool call succeeded",
    "tool.duration_ms": "Tool execution time",
    "handoff.from": "Source agent",
    "handoff.to": "Destination agent",
    "handoff.context": "Context passed between agents",
    "error.type": "Error class name",
    "error.message": "Error message",
}

Visualizing Traces

With Jaeger

# Export to Jaeger
from opentelemetry.exporter.jaeger.thrift import JaegerExporter

jaeger_exporter = JaegerExporter(
    agent_host_name="localhost",
    agent_port=6831,
)

provider.add_span_processor(BatchSpanProcessor(jaeger_exporter))

With LangSmith

import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-api-key"

# LangSmith automatically traces LangChain/LangGraph calls

Custom Dashboard

from fastapi import FastAPI
from opentelemetry.trace import format_span_id

app = FastAPI()

@app.get("/traces/{trace_id}")
async def get_trace(trace_id: str):
    """Get all spans for a trace."""
    spans = await query_spans(trace_id)
    
    # Build timeline
    timeline = []
    for span in sorted(spans, key=lambda s: s.start_time):
        timeline.append({
            "id": format_span_id(span.span_id),
            "name": span.name,
            "start_ms": (span.start_time - spans[0].start_time) / 1e6,
            "duration_ms": span.duration / 1e6,
            "attributes": dict(span.attributes),
            "parent": format_span_id(span.parent.span_id) if span.parent else None,
        })
    
    return {"trace_id": trace_id, "spans": timeline}

Analyzing Agent Performance

Use traces to identify bottlenecks:

class AgentPerformanceAnalyzer:
    def analyze_trace(self, trace_id: str) -> dict:
        spans = self._get_spans(trace_id)
        
        # Find slowest operations
        tool_spans = [s for s in spans if s.name.startswith("tool.")]
        tool_spans.sort(key=lambda s: s.duration, reverse=True)
        
        # Calculate agent time distribution
        agent_spans = [s for s in spans if s.name.startswith("agent.")]
        agent_times = {}
        for span in agent_spans:
            agent_name = span.attributes.get("agent.name", "unknown")
            agent_times[agent_name] = agent_times.get(agent_name, 0) + span.duration
        
        # Find error patterns
        error_spans = [s for s in spans if s.status == StatusCode.ERROR]
        
        return {
            "total_duration_ms": sum(s.duration for s in spans) / 1e6,
            "slowest_tools": [
                {"name": s.attributes.get("tool.name"), "duration_ms": s.duration / 1e6}
                for s in tool_spans[:5]
            ],
            "agent_time_distribution": {
                k: v / 1e6 for k, v in agent_times.items()
            },
            "error_count": len(error_spans),
            "span_count": len(spans),
        }

Sampling Strategies

In production, you can’t trace everything:

from opentelemetry.sdk.trace.sampling import TraceIdRatioBased

# Sample 10% of traces in production
sampler = TraceIdRatioBased(0.1)

# Always trace errors
# (Use custom sampler for this)

provider = TracerProvider(sampler=sampler)

Conclusion

Distributed tracing transforms multi-agent debugging from guesswork into science. With OpenTelemetry, you get vendor-neutral instrumentation that works across frameworks. Start by instrumenting agent execution and tool calls, add semantic attributes for consistency, export to Jaeger or LangSmith for visualization, and use trace analysis to identify performance bottlenecks.