Distributed Tracing for AI Agents: OpenTelemetry, LangSmith, and Custom Telemetry

When a traditional microservice fails, you check the logs. When an AI agent fails, you need to reconstruct a reasoning chain that may span dozens of LLM calls, tool invocations, and conditional branches — across multiple services, each with their own latency profile and failure modes. Welcome to the world of AI agent observability.

Why Traditional Monitoring Falls Short

Standard APM tools (Datadog, New Relic, Grafana) were designed for deterministic systems. You instrument a function, measure its duration, count errors, and set alerts. AI agents break this model in several ways:

• Non-deterministic execution paths: The same input can produce different call chains depending on LLM output
• Semantic errors: The agent „succeeds“ (no exceptions) but produces wrong or harmful output
• Emergent behavior: Multi-agent systems exhibit behaviors not predictable from individual agent performance
• Cost as a first-class metric: Token usage is a critical operational signal, not just a billing concern

You need distributed tracing that captures the semantic content of agent execution, not just timing and error codes.

OpenTelemetry for AI Agents

OpenTelemetry (OTel) has become the standard for distributed tracing. While it wasn’t designed specifically for AI agents, its span-based model maps naturally to agent execution:

# otel_agent_tracer.py
from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.trace import SpanKind
import time
import json

# Configure OTel
provider = TracerProvider()
processor = BatchSpanProcessor(
    OTLPSpanExporter(endpoint="otel-collector:4317")
)
provider.add_span_processor(processor)
trace.set_tracer_provider(provider)
tracer = trace.get_tracer("ai-agent")

class TracedAgent:
    def __init__(self, agent, tracer):
        self.agent = agent
        self.tracer = tracer
    
    async def execute(self, user_input: str, session_id: str):
        with self.tracer.start_as_current_span(
            "agent.session",
            kind=SpanKind.SERVER,
            attributes={
                "session.id": session_id,
                "agent.model": self.agent.model_id,
                "agent.framework": self.agent.framework,
                "input.length": len(user_input)
            }
        ) as session_span:
            try:
                result = await self._run_agent_loop(user_input, session_span)
                session_span.set_attribute("output.length", len(result))
                session_span.set_attribute("status", "success")
                return result
            except Exception as e:
                session_span.set_attribute("status", "error")
                session_span.set_attribute("error.type", type(e).__name__)
                session_span.set_attribute("error.message", str(e))
                raise
    
    async def _run_agent_loop(self, user_input, parent_span):
        messages = [{"role": "user", "content": user_input}]
        total_tokens = 0
        turn_count = 0
        
        while turn_count < self.agent.max_turns:
            turn_count += 1
            
            with self.tracer.start_as_current_span(
                f"agent.turn.{turn_count}",
                attributes={"turn.number": turn_count}
            ) as turn_span:
                # LLM call
                with self.tracer.start_as_current_span(
                    "llm.call",
                    attributes={
                        "model": self.agent.model_id,
                        "messages.count": len(messages)
                    }
                ) as llm_span:
                    start = time.time()
                    response = await self.agent.llm.chat(messages)
                    latency = time.time() - start
                    
                    llm_span.set_attribute("tokens.prompt", response.usage.prompt_tokens)
                    llm_span.set_attribute("tokens.completion", response.usage.completion_tokens)
                    llm_span.set_attribute("tokens.total", response.usage.total_tokens)
                    llm_span.set_attribute("latency.ms", latency * 1000)
                    llm_span.set_attribute("finish_reason", response.finish_reason)
                    
                    total_tokens += response.usage.total_tokens
                
                # Tool calls
                if response.tool_calls:
                    for tool_call in response.tool_calls:
                        with self.tracer.start_as_current_span(
                            f"tool.call.{tool_call.name}",
                            attributes={
                                "tool.name": tool_call.name,
                                "tool.input": json.dumps(tool_call.arguments)[:1000]
                            }
                        ) as tool_span:
                            start = time.time()
                            tool_result = await self.agent.execute_tool(tool_call)
                            latency = time.time() - start
                            
                            tool_span.set_attribute("latency.ms", latency * 1000)
                            tool_span.set_attribute("output.length", len(str(tool_result)))
                            
                            if tool_result.error:
                                tool_span.set_attribute("error", True)
                                tool_span.set_attribute("error.message", tool_result.error)
                
                # Check for completion
                if not response.tool_calls:
                    parent_span.set_attribute("total_turns", turn_count)
                    parent_span.set_attribute("total_tokens", total_tokens)
                    return response.content
                
                messages.append(response.to_message())
        
        raise MaxTurnsExceeded(f"Agent exceeded {self.agent.max_turns} turns")

LangSmith: Purpose-Built for LLM Observability

While OpenTelemetry provides the infrastructure, LangSmith (from LangChain) offers AI-specific observability out of the box:

# langsmith_tracing.py
import os
from langsmith import traceable, Client

os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_PROJECT"] = "production-agent-v2"

client = Client()

@traceable(name="customer_support_agent", run_type="chain")
async def customer_support_agent(user_message: str, customer_id: str):
    """Full agent execution with automatic LangSmith tracing."""
    
    # Each @traceable function creates a span automatically
    context = await retrieve_customer_context(customer_id)
    intent = await classify_intent(user_message, context)
    
    if intent.category == "refund":
        result = await handle_refund_intent(user_message, context)
    elif intent.category == "technical":
        result = await handle_technical_intent(user_message, context)
    else:
        result = await handle_general_inquiry(user_message, context)
    
    # LangSmith automatically captures inputs, outputs, and latency
    # Plus you can add custom feedback
    return result

@traceable(name="intent_classifier", run_type="llm")
async def classify_intent(message: str, context: dict):
    """Automatically traced LLM call with full prompt/response capture."""
    response = await llm.chat([
        {"role": "system", "content": INTENT_PROMPT},
        {"role": "user", "content": f"Context: {context}nMessage: {message}"}
    ])
    return parse_intent(response)

# Add feedback programmatically or via LangSmith UI
def record_feedback(run_id: str, score: float, comment: str):
    client.create_feedback(
        run_id=run_id,
        key="accuracy",
        score=score,
        comment=comment
    )

LangSmith advantages for agent observability:

• Prompt versioning: Track which prompt version produced which output
• Dataset management: Curate evaluation sets from production traces
• Annotation queues: Route low-confidence outputs for human review
• Token analytics: Cost breakdown by agent, user, and time period

Custom Telemetry: Building What Doesn’t Exist

For production agent systems, you’ll often need custom telemetry that captures domain-specific signals:

# custom_telemetry.py
from dataclasses import dataclass, field
from typing import Any
from datetime import datetime
import asyncio

@dataclass
class AgentEvent:
    event_type: str  # "llm_call", "tool_call", "decision", "error", "human_intervention"
    timestamp: datetime
    session_id: str
    agent_id: str
    turn_number: int
    data: dict
    metrics: dict = field(default_factory=dict)

class AgentTelemetryCollector:
    def __init__(self, exporters: list):
        self.exporters = exporters
        self.event_buffer = []
    
    async def emit(self, event: AgentEvent):
        self.event_buffer.append(event)
        for exporter in self.exporters:
            await exporter.export(event)
    
    async def emit_decision(self, session_id: str, turn: int, 
                             decision: str, reasoning: str, confidence: float):
        """Track agent decision-making for auditability."""
        await self.emit(AgentEvent(
            event_type="decision",
            timestamp=datetime.utcnow(),
            session_id=session_id,
            agent_id="main",
            turn_number=turn,
            data={"decision": decision, "reasoning": reasoning},
            metrics={"confidence": confidence}
        ))
    
    async def emit_quality_signal(self, session_id: str, 
                                   signal_type: str, value: float):
        """Track quality metrics: hallucination score, relevance, safety."""
        await self.emit(AgentEvent(
            event_type="quality_signal",
            timestamp=datetime.utcnow(),
            session_id=session_id,
            agent_id="main",
            turn_number=0,
            data={"signal_type": signal_type},
            metrics={"value": value}
        ))

# Usage in agent loop
class ObservableAgent:
    def __init__(self, agent, telemetry: AgentTelemetryCollector):
        self.agent = agent
        self.telemetry = telemetry
    
    async def run(self, user_input: str, session_id: str):
        turn = 0
        while True:
            turn += 1
            
            # Track LLM call
            response = await self.agent.llm.chat(...)
            await self.telemetry.emit(AgentEvent(
                event_type="llm_call",
                timestamp=datetime.utcnow(),
                session_id=session_id,
                agent_id=self.agent.id,
                turn_number=turn,
                data={"model": self.agent.model_id},
                metrics={
                    "tokens": response.usage.total_tokens,
                    "latency_ms": response.latency * 1000
                }
            ))
            
            # Track decision
            if response.tool_calls:
                await self.telemetry.emit_decision(
                    session_id, turn,
                    decision=f"call_tool:{response.tool_calls[0].name}",
                    reasoning=response.reasoning,
                    confidence=response.confidence or 0.0
                )
            else:
                await self.telemetry.emit_decision(
                    session_id, turn,
                    decision="respond_to_user",
                    reasoning=response.reasoning,
                    confidence=response.confidence or 0.0
                )
                break

Building the Observability Dashboard

Combine traces, metrics, and logs into a unified agent observability dashboard:

# dashboard_metrics.py
# Key metrics to track for every production agent

AGENT_METRICS = {
    # Performance
    "agent.latency.p50": "Median response latency",
    "agent.latency.p99": "P99 response latency",
    "agent.turns.avg": "Average turns per session",
    "agent.turns.max": "Max turns observed",
    
    # Cost
    "agent.tokens.per_session": "Tokens consumed per session",
    "agent.cost.per_session": "USD cost per session",
    "agent.cost.per_task": "USD cost per completed task",
    
    # Quality
    "agent.task_completion_rate": "Tasks completed without escalation",
    "agent.hallucination_rate": "Detected hallucinations per 100 turns",
    "agent.tool_error_rate": "Tool call failures per 100 calls",
    
    # Reliability
    "agent.error_rate": "Sessions ending in error",
    "agent.timeout_rate": "Sessions exceeding max duration",
    "agent.retry_rate": "LLM calls requiring retry",
    
    # Safety
    "agent.safety_flag_rate": "Safety filter triggers per 100 sessions",
    "agent.human_intervention_rate": "Sessions requiring human takeover",
}

Key Takeaways

1. Use OpenTelemetry for infrastructure-level tracing — it’s vendor-neutral and integrates with your existing observability stack.
2. Add LangSmith or similar for AI-specific observability — prompt tracking, dataset management, and annotation workflows are purpose-built for LLM systems.
3. Build custom telemetry for domain signals — decision tracking, quality scores, and business KPIs that generic tools don’t capture.
4. Track cost as a first-class metric — token usage is both an operational signal and a business constraint.
5. Make traces actionable — every trace should link to the prompt version, model config, and tool definitions that produced it.

In 2026, the difference between teams that scale agents successfully and teams that don’t isn’t model quality — it’s observability maturity. You can’t improve what you can’t see.