Agent Observability: Tracing Decisions in Production
Agent observability in production: trace every decision, log tool calls, track state, export OpenTelemetry metrics, and cut debugging from days to hours.
Muhammad Abdul Sami
· 13 min read
- AI Agents
- Observability
- Tool Calling
- LLM
- Python
Table of Contents:
- Why Agent Observability
- What Agent Observability Must Capture
- Decision Tracing Architecture
- Reasoning Transparency
- Tool Call Logging
- State Tracking
- Debugging Workflows
- Production Monitoring
- Cost and Performance Tracking
- Alerting Patterns
- Agent Observability Stack: Build vs Buy
- Sampling, Retention, and Overhead
- Frequently Asked Questions
Why Agent Observability: The Black Box Problem
Short answer: Agent observability means tracing every decision an AI agent makes — the reasoning it produced, the tools it called, the state it mutated, and what each step cost. Agents make dozens of decisions across multiple reasoning steps, and a conventional request log only tells you that a decision happened, not why. Decision tracing makes agent behavior transparent, debuggable, and auditable.
A financial services AI agent failed 8% of high-value transactions. Logs showed "decision made" but not why. We implemented decision tracing with reasoning capture, tool call logging, and state tracking. Root cause identified in 2 hours instead of 2 days. Failure rate dropped to 0.9%.
Key Takeaways:
- Decision tracing captures the reasoning path for every agent action
- Tool call logs record parameters, results, and timing
- State tracking maintains context across multi-turn interactions
- Reasoning transparency enables trust and debugging
- Production monitoring catches anomalies in real-time
- Cost tracking attributes spend to decisions and users
For production AI systems, observability is as critical as functionality. The rest of this guide is the tracing model we now deploy by default, with working Python for each layer.
What Agent Observability Must Capture
Standard APM answers "was the request slow or failed?". Agent observability has to answer a harder question: "was the decision correct, and if not, at which step did it go wrong?" That means capturing signals a web service never needed.
| Signal | Question it answers | Where it lives | Typical retention |
|---|---|---|---|
| Reasoning text | Why did the agent choose this action? | Span attribute or event on the LLM call | 30-90 days (redacted) |
| Tool call I/O | What did the agent send and receive? | Child span per tool with args/result | 30 days, errors longer |
| State transitions | What did the agent believe at each turn? | Ordered events keyed by state field | Session lifetime + 30 days |
| Token usage and cost | What did this trace cost, and who pays? | Span attributes, aggregated as metrics | 13 months (billing) |
| Confidence / scores | Which decisions were uncertain? | Numeric span attribute, histogram metric | Same as traces |
| Final outcome | Did the user get what they asked for? | Trace-level attribute, joined to feedback | Indefinite (aggregated) |
Two design rules follow from the table. First, the trace is the unit of debugging, not the log line: every decision must carry the same trace_id so you can replay a session end to end. Second, reasoning is data, not a debug string — store it structured so you can query "all traces where confidence < 0.6 and the tool failed" rather than grepping.
The OpenTelemetry GenAI semantic conventions standardize attribute names for model, token counts, and tool calls. Adopt them from day one; renaming attributes across a year of stored traces is painful.
Decision Tracing Architecture
Capture every decision with structured traces.
from dataclasses import dataclass, field
from typing import List, Dict, Any, Optional
from datetime import datetime, timezone
import json
from enum import Enum
class DecisionType(str, Enum):
TOOL_CALL = "tool_call"
REASONING = "reasoning"
RESPONSE = "response"
ERROR = "error"
@dataclass
class DecisionTrace:
"""Single decision point in agent execution."""
trace_id: str
decision_id: str
decision_type: DecisionType
timestamp: str
reasoning: str
input_context: Dict[str, Any]
output: Dict[str, Any]
confidence: Optional[float] = None
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class AgentTrace:
"""Complete trace of agent execution."""
trace_id: str
agent_id: str
user_id: str
session_id: str
query: str
started_at: str
completed_at: Optional[str] = None
decisions: List[DecisionTrace] = field(default_factory=list)
final_response: Optional[str] = None
success: bool = True
error: Optional[str] = None
def add_decision(self, decision: DecisionTrace) -> None:
"""Add decision to trace."""
self.decisions.append(decision)
def complete(self, response: str, success: bool = True, error: Optional[str] = None) -> None:
"""Mark trace as complete."""
self.completed_at = datetime.now(timezone.utc).isoformat()
self.final_response = response
self.success = success
self.error = error
class AgentTracer:
"""Agent execution tracer."""
def __init__(self, storage_backend):
self.storage = storage_backend
self.current_trace: Optional[AgentTrace] = None
def start_trace(
self,
agent_id: str,
user_id: str,
session_id: str,
query: str,
) -> AgentTrace:
"""Start new agent trace."""
import uuid
trace = AgentTrace(
trace_id=str(uuid.uuid4()),
agent_id=agent_id,
user_id=user_id,
session_id=session_id,
query=query,
started_at=datetime.now(timezone.utc).isoformat(),
)
self.current_trace = trace
return trace
def log_decision(
self,
decision_type: DecisionType,
reasoning: str,
input_context: Dict[str, Any],
output: Dict[str, Any],
confidence: Optional[float] = None,
metadata: Optional[Dict[str, Any]] = None,
) -> DecisionTrace:
"""Log agent decision."""
if not self.current_trace:
raise ValueError("No active trace")
import uuid
decision = DecisionTrace(
trace_id=self.current_trace.trace_id,
decision_id=str(uuid.uuid4()),
decision_type=decision_type,
timestamp=datetime.now(timezone.utc).isoformat(),
reasoning=reasoning,
input_context=input_context,
output=output,
confidence=confidence,
metadata=metadata or {},
)
self.current_trace.add_decision(decision)
return decision
async def complete_trace(
self,
response: str,
success: bool = True,
error: Optional[str] = None,
) -> AgentTrace:
"""Complete and save trace."""
if not self.current_trace:
raise ValueError("No active trace")
self.current_trace.complete(response, success, error)
await self.storage.save_trace(self.current_trace)
completed_trace = self.current_trace
self.current_trace = None
return completed_trace
def get_current_trace(self) -> Optional[AgentTrace]:
"""Get current active trace."""
return self.current_trace
# Usage
tracer = AgentTracer(storage_backend=postgres_storage)
# Start trace
trace = tracer.start_trace(
agent_id="support-agent-1",
user_id="user-123",
session_id="sess-456",
query="I need to refund order ORD-12345",
)
# Log reasoning decision
tracer.log_decision(
decision_type=DecisionType.REASONING,
reasoning="User wants refund. Need to first verify order exists and is eligible.",
input_context={"query": "I need to refund order ORD-12345"},
output={"next_step": "check_order_status"},
confidence=0.95,
)
# Log tool call decision
tracer.log_decision(
decision_type=DecisionType.TOOL_CALL,
reasoning="Calling get_order tool to verify order exists",
input_context={"order_id": "ORD-12345"},
output={"tool": "get_order", "params": {"order_id": "ORD-12345"}},
confidence=1.0,
metadata={"tool_latency_ms": 150},
)
# Complete trace
await tracer.complete_trace(
response="I've processed your refund for order ORD-12345. You'll receive $99.99 back to your original payment method within 5-7 business days.",
success=True,
)
Decision traces provide complete audit trail of agent reasoning.
Mapping Decisions to Spans
The dataclasses above are backend-agnostic on purpose, but in production you will almost certainly emit them as OpenTelemetry spans. The mapping is straightforward:
- One
AgentTracebecomes the root span (agent.run), carryingagent_id,session_id, and the user query as attributes. - Each
REASONINGdecision becomes a child span named after the model call (chat gpt-4oper the GenAI conventions), withgen_ai.usage.input_tokensandgen_ai.usage.output_tokensattributes and the reasoning text as a span event. - Each
TOOL_CALLbecomes a child span of the reasoning step that requested it, so latency attribution is exact. ERRORdecisions set the span status toERRORand record the exception, which is what backends use to surface failed traces.
Keep large payloads out of attributes. Most backends cap attribute size (Jaeger and Datadog truncate around a few KB), so store full tool results in object storage keyed by decision_id and put only a summary plus a pointer on the span. For LLM tracing, the same rule applies to prompts and completions.
Reasoning Transparency
Capture agent reasoning at each step.
from openai import AsyncOpenAI
client = AsyncOpenAI()
async def traced_reasoning_step(
tracer: AgentTracer,
prompt: str,
context: Dict[str, Any],
) -> Dict[str, Any]:
"""Execute reasoning step with tracing."""
# Enhance prompt to capture reasoning
reasoning_prompt = f"""{prompt}
Think step-by-step. Return JSON:
{{
"reasoning": "your thinking process",
"confidence": 0.0-1.0,
"decision": "what to do next",
"tool_calls": [
{{"tool": "tool_name", "params": {{...}}}}
]
}}"""
# Execute LLM call
response = await client.chat.completions.create(
model="gpt-4o",
messages=[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": reasoning_prompt},
],
response_format={"type": "json_object"},
)
# Parse reasoning
import json
reasoning_output = json.loads(response.choices[0].message.content)
# Log decision
tracer.log_decision(
decision_type=DecisionType.REASONING,
reasoning=reasoning_output["reasoning"],
input_context=context,
output={
"decision": reasoning_output["decision"],
"tool_calls": reasoning_output.get("tool_calls", []),
},
confidence=reasoning_output.get("confidence", 0.5),
metadata={
"prompt_tokens": response.usage.prompt_tokens,
"completion_tokens": response.usage.completion_tokens,
},
)
return reasoning_output
# Usage
reasoning = await traced_reasoning_step(
tracer=tracer,
prompt="User wants to refund order ORD-12345. What should we do?",
context={"user_query": "Refund ORD-12345"},
)
print(f"Reasoning: {reasoning['reasoning']}")
print(f"Decision: {reasoning['decision']}")
print(f"Confidence: {reasoning['confidence']}")
Reasoning capture enables understanding agent decisions.
For ReACT agents, capture thought-action-observation cycles.
Tool Call Logging
Log every tool invocation with parameters and results.
from typing import Callable
import time
import functools
def traced_tool(tracer: AgentTracer):
"""Decorator to trace tool calls."""
def decorator(func: Callable):
@functools.wraps(func)
async def wrapper(*args, **kwargs):
tool_name = func.__name__
# Log tool call decision
start_time = time.perf_counter()
try:
# Execute tool
result = await func(*args, **kwargs)
latency_ms = (time.perf_counter() - start_time) * 1000
# Log successful tool call
tracer.log_decision(
decision_type=DecisionType.TOOL_CALL,
reasoning=f"Executing tool: {tool_name}",
input_context={
"tool": tool_name,
"args": args,
"kwargs": kwargs,
},
output={
"result": result,
"success": True,
},
metadata={
"latency_ms": latency_ms,
"tool_version": getattr(func, "__version__", "unknown"),
},
)
return result
except Exception as e:
latency_ms = (time.perf_counter() - start_time) * 1000
# Log failed tool call
tracer.log_decision(
decision_type=DecisionType.ERROR,
reasoning=f"Tool {tool_name} failed: {str(e)}",
input_context={
"tool": tool_name,
"args": args,
"kwargs": kwargs,
},
output={
"error": str(e),
"error_type": type(e).__name__,
},
metadata={
"latency_ms": latency_ms,
},
)
raise
return wrapper
return decorator
# Usage
@traced_tool(tracer)
async def get_order(order_id: str) -> Dict[str, Any]:
"""Get order details."""
# Simulate database call
await asyncio.sleep(0.1)
return {
"order_id": order_id,
"status": "completed",
"amount": 99.99,
}
@traced_tool(tracer)
async def process_refund(order_id: str, amount: float) -> Dict[str, Any]:
"""Process refund."""
# Simulate payment processing
await asyncio.sleep(0.2)
return {
"refund_id": f"REF-{order_id}",
"amount": amount,
"success": True,
}
# Tool calls are automatically traced
order = await get_order("ORD-12345")
refund = await process_refund("ORD-12345", order["amount"])
Tool call logging enables debugging tool usage patterns. Three things the decorator gives you that plain logging does not:
- Latency per tool, which is where most agent slowness hides. A single 4-second CRM lookup inside a 10-step loop dominates the trace.
- Argument capture, so you can see that the model passed
order_id="ORD12345"(no hyphen) and the lookup returned nothing. - Failure classification by
error_type, which feeds directly into the retry and fallback logic described in reliable tool calling.
Redacting Tool Payloads
Tool arguments and results are where PII concentrates — email addresses, account numbers, free-text notes. Redact before the decision is persisted, not at query time. A practical pattern is an allowlist per tool: declare which argument keys are safe to log verbatim, hash identifiers you still need for joins (user_id), and replace everything else with a type marker (<email>). Detector choice matters less than placement; the key point for observability is that redaction has to be a property of the tracer, so a new tool cannot accidentally bypass it.
State Tracking
Track agent state across multi-turn conversations.
from typing import Optional
class AgentStateTracker:
"""Track agent state with versioning."""
def __init__(self, tracer: AgentTracer):
self.tracer = tracer
self.state_history = []
def update_state(
self,
key: str,
value: Any,
reasoning: str,
) -> None:
"""Update state with reasoning."""
timestamp = datetime.now(timezone.utc).isoformat()
state_update = {
"timestamp": timestamp,
"key": key,
"value": value,
"reasoning": reasoning,
}
self.state_history.append(state_update)
# Log state change as decision
self.tracer.log_decision(
decision_type=DecisionType.REASONING,
reasoning=f"State update: {reasoning}",
input_context={"key": key, "previous_value": self.get_state(key)},
output={"key": key, "new_value": value},
metadata={"state_update": True},
)
def get_state(self, key: str) -> Optional[Any]:
"""Get current state value."""
# Get most recent value for key
for update in reversed(self.state_history):
if update["key"] == key:
return update["value"]
return None
def get_state_history(self, key: Optional[str] = None) -> List[Dict[str, Any]]:
"""Get state change history."""
if key:
return [u for u in self.state_history if u["key"] == key]
return self.state_history
# Usage
state_tracker = AgentStateTracker(tracer)
# Track conversation state
state_tracker.update_state(
key="user_intent",
value="refund_request",
reasoning="User explicitly requested refund",
)
state_tracker.update_state(
key="order_id",
value="ORD-12345",
reasoning="Extracted order ID from user message",
)
state_tracker.update_state(
key="verification_status",
value="verified",
reasoning="Order exists and is eligible for refund",
)
# Query state
current_intent = state_tracker.get_state("user_intent")
intent_history = state_tracker.get_state_history("user_intent")
State tracking maintains context for debugging multi-turn interactions.
For stateful agents, persist state with checkpoints.
Debugging Workflows
Use traces to debug agent failures.
class AgentDebugger:
"""Debug agent executions using traces."""
def __init__(self, storage_backend):
self.storage = storage_backend
async def debug_trace(self, trace_id: str) -> Dict[str, Any]:
"""Debug specific trace."""
trace = await self.storage.get_trace(trace_id)
if not trace:
return {"error": "Trace not found"}
# Analyze trace
analysis = {
"trace_id": trace_id,
"summary": self._generate_summary(trace),
"decision_flow": self._visualize_decisions(trace),
"bottlenecks": self._identify_bottlenecks(trace),
"errors": self._extract_errors(trace),
"recommendations": self._generate_recommendations(trace),
}
return analysis
def _generate_summary(self, trace: AgentTrace) -> Dict[str, Any]:
"""Generate trace summary."""
return {
"agent_id": trace.agent_id,
"user_id": trace.user_id,
"query": trace.query,
"success": trace.success,
"num_decisions": len(trace.decisions),
"duration_ms": self._calculate_duration(trace),
}
def _visualize_decisions(self, trace: AgentTrace) -> List[Dict[str, Any]]:
"""Visualize decision flow."""
flow = []
for decision in trace.decisions:
flow.append({
"type": decision.decision_type.value,
"reasoning": decision.reasoning[:100],
"confidence": decision.confidence,
"timestamp": decision.timestamp,
})
return flow
def _identify_bottlenecks(self, trace: AgentTrace) -> List[Dict[str, Any]]:
"""Identify performance bottlenecks."""
bottlenecks = []
for decision in trace.decisions:
latency = decision.metadata.get("latency_ms", 0)
if latency > 1000: # > 1 second
bottlenecks.append({
"decision_id": decision.decision_id,
"type": decision.decision_type.value,
"latency_ms": latency,
"reasoning": decision.reasoning,
})
return sorted(bottlenecks, key=lambda x: x["latency_ms"], reverse=True)
def _extract_errors(self, trace: AgentTrace) -> List[Dict[str, Any]]:
"""Extract errors from trace."""
errors = []
for decision in trace.decisions:
if decision.decision_type == DecisionType.ERROR:
errors.append({
"decision_id": decision.decision_id,
"error": decision.output.get("error"),
"error_type": decision.output.get("error_type"),
"context": decision.input_context,
})
return errors
def _generate_recommendations(self, trace: AgentTrace) -> List[str]:
"""Generate improvement recommendations."""
recommendations = []
# Check for bottlenecks
bottlenecks = self._identify_bottlenecks(trace)
if bottlenecks:
recommendations.append(
f"Optimize slow operations: {', '.join(b['type'] for b in bottlenecks[:3])}"
)
# Check for low confidence decisions
low_confidence = [
d for d in trace.decisions
if d.confidence and d.confidence < 0.6
]
if low_confidence:
recommendations.append(
f"Review {len(low_confidence)} low-confidence decisions"
)
# Check for errors
errors = self._extract_errors(trace)
if errors:
recommendations.append(
f"Fix {len(errors)} tool/execution errors"
)
return recommendations
def _calculate_duration(self, trace: AgentTrace) -> float:
"""Calculate trace duration."""
if not trace.completed_at:
return 0
from datetime import datetime
start = datetime.fromisoformat(trace.started_at)
end = datetime.fromisoformat(trace.completed_at)
return (end - start).total_seconds() * 1000
# Usage
debugger = AgentDebugger(storage_backend)
analysis = await debugger.debug_trace("trace-12345")
print(f"Summary: {analysis['summary']}")
print(f"\nBottlenecks: {analysis['bottlenecks']}")
print(f"\nErrors: {analysis['errors']}")
print(f"\nRecommendations:")
for rec in analysis['recommendations']:
print(f" - {rec}")
Debugging tools accelerate root cause analysis.
A Debugging Playbook for Agent Traces
When a trace is flagged, the sequence that finds root cause fastest is:
- Read the outcome first. Compare the final response with the user query. Wrong answer, refusal, timeout, and partial completion are different bugs with different suspects.
- Walk decisions backwards from the failure. The last
ERRORor low-confidence decision is usually a symptom; the cause is typically 1-3 decisions earlier — a tool that returned an empty result the model then "hallucinated around", or a state update that overwrote the right value. - Diff against a healthy trace for the same intent. If your storage supports it, pull a successful trace with the same
user_intentstate and compare tool sequences. Divergence points are where to look. - Replay with the captured inputs. Because
input_contextis stored per decision, you can re-run a single reasoning step with the same prompt and a different model or temperature to test whether the failure is prompt-related or model-related. - Turn the finding into a check. Every root cause becomes either an alert rule (below) or an eval case in your agent test suite, so the same failure is caught before deploy next time.
The twelve agent failure modes we see most often map cleanly to trace signatures: infinite loops show as repeated identical tool spans, context overflow shows as truncated input_context, and silent tool failures show as success: true with empty results.
Production Monitoring
Monitor agent behavior in production.
from opentelemetry import metrics
class AgentMonitor:
"""Production monitoring for agents."""
def __init__(self):
meter = metrics.get_meter("agent.monitoring")
# Metrics
self.decision_counter = meter.create_counter(
"agent.decisions.total",
description="Total decisions made",
unit="decisions",
)
self.confidence_histogram = meter.create_histogram(
"agent.decision.confidence",
description="Decision confidence scores",
unit="score",
)
self.tool_latency_histogram = meter.create_histogram(
"agent.tool.latency",
description="Tool call latency",
unit="ms",
)
self.error_counter = meter.create_counter(
"agent.errors.total",
description="Total errors",
unit="errors",
)
def record_decision(
self,
decision: DecisionTrace,
agent_id: str,
) -> None:
"""Record decision metrics."""
attributes = {
"agent_id": agent_id,
"decision_type": decision.decision_type.value,
}
self.decision_counter.add(1, attributes)
if decision.confidence:
self.confidence_histogram.record(
decision.confidence,
attributes,
)
if decision.decision_type == DecisionType.TOOL_CALL:
latency = decision.metadata.get("latency_ms", 0)
self.tool_latency_histogram.record(latency, attributes)
if decision.decision_type == DecisionType.ERROR:
self.error_counter.add(1, attributes)
def record_trace(self, trace: AgentTrace) -> None:
"""Record trace-level metrics."""
for decision in trace.decisions:
self.record_decision(decision, trace.agent_id)
# Usage
monitor = AgentMonitor()
# Record decisions during execution
for decision in trace.decisions:
monitor.record_decision(decision, "support-agent-1")
Production monitoring enables real-time observability.
Agent Metrics Worth Alerting On
Traces are for debugging individual runs; metrics are for noticing that something changed across thousands of runs. These are the signals that have caught real regressions for us, along with an illustrative starting threshold:
| Metric | Why it matters | Illustrative alert |
|---|---|---|
| Trace success rate | Direct measure of agent reliability | < 97% over 15 min |
| p95 decisions per trace | Loops and thrashing show up here first | > 2x 7-day baseline |
| Tool error rate by tool | Isolates a broken integration from a model regression | > 5% for any single tool |
| Mean confidence per agent | Drops after a prompt or model change | < 0.75 sustained 1 h |
| Cost per trace (p50 and p95) | Catches runaway token usage | p95 > 3x p50 |
| Time to first tool call | Detects slow or stuck reasoning | > 5 s p95 |
Export with the OpenTelemetry SDK to Prometheus, Datadog, or Grafana; the meter API above is backend-neutral. For observability, keep cardinality under control: label by agent_id and tool, never by user_id or trace_id.
Cost and Performance Tracking
Track costs per decision and user.
class AgentCostTracker:
"""Track agent costs."""
def __init__(self):
self.cost_data = []
def track_decision_cost(
self,
decision: DecisionTrace,
agent_id: str,
user_id: str,
) -> float:
"""Track cost of decision."""
cost = 0.0
if decision.decision_type == DecisionType.REASONING:
# Calculate LLM cost
prompt_tokens = decision.metadata.get("prompt_tokens", 0)
completion_tokens = decision.metadata.get("completion_tokens", 0)
# Example pricing for GPT-4o
cost = (
(prompt_tokens / 1_000_000) * 2.50 +
(completion_tokens / 1_000_000) * 10.00
)
elif decision.decision_type == DecisionType.TOOL_CALL:
# Tool execution cost (if applicable)
cost = decision.metadata.get("tool_cost", 0.0)
# Record cost
self.cost_data.append({
"decision_id": decision.decision_id,
"trace_id": decision.trace_id,
"agent_id": agent_id,
"user_id": user_id,
"cost_usd": cost,
"timestamp": decision.timestamp,
})
return cost
def get_user_cost(self, user_id: str, hours: int = 24) -> Dict[str, Any]:
"""Get cost for user."""
from datetime import datetime, timedelta
cutoff = datetime.now(timezone.utc) - timedelta(hours=hours)
user_costs = [
c for c in self.cost_data
if c["user_id"] == user_id and datetime.fromisoformat(c["timestamp"]) > cutoff
]
total_cost = sum(c["cost_usd"] for c in user_costs)
return {
"user_id": user_id,
"period_hours": hours,
"total_cost_usd": total_cost,
"num_decisions": len(user_costs),
"avg_cost_per_decision": total_cost / len(user_costs) if user_costs else 0,
}
# Usage
cost_tracker = AgentCostTracker()
# Track costs during execution
for decision in trace.decisions:
cost = cost_tracker.track_decision_cost(
decision,
agent_id="support-agent-1",
user_id="user-123",
)
# Get user cost summary
user_summary = cost_tracker.get_user_cost("user-123", hours=24)
print(f"User cost (24h): ${user_summary['total_cost_usd']:.4f}")
Cost tracking enables per-user billing and budget management.
For cost optimization, analyze cost patterns.
Alerting Patterns
Alert on anomalies in agent behavior.
class AgentAlerter:
"""Alert on agent anomalies."""
def __init__(self, thresholds: Dict[str, float]):
self.thresholds = thresholds
self.recent_traces = []
async def check_trace(self, trace: AgentTrace) -> List[str]:
"""Check trace for alertable conditions."""
alerts = []
# Check error rate
if not trace.success:
alerts.append(f"Agent execution failed: {trace.error}")
# Check low confidence
low_conf_decisions = [
d for d in trace.decisions
if d.confidence and d.confidence < self.thresholds.get("min_confidence", 0.7)
]
if len(low_conf_decisions) > 2:
alerts.append(
f"{len(low_conf_decisions)} low-confidence decisions in trace"
)
# Check slow execution
duration_ms = self._calculate_duration(trace)
if duration_ms > self.thresholds.get("max_duration_ms", 10000):
alerts.append(
f"Slow execution: {duration_ms:.0f}ms (threshold: {self.thresholds['max_duration_ms']}ms)"
)
# Check excessive tool calls
tool_calls = [d for d in trace.decisions if d.decision_type == DecisionType.TOOL_CALL]
if len(tool_calls) > self.thresholds.get("max_tool_calls", 10):
alerts.append(
f"Excessive tool calls: {len(tool_calls)} (threshold: {self.thresholds['max_tool_calls']})"
)
# Trigger alerts
for alert in alerts:
await self._trigger_alert(trace, alert)
return alerts
async def _trigger_alert(self, trace: AgentTrace, message: str) -> None:
"""Trigger alert."""
print(f"""
🚨 AGENT ALERT
Trace ID: {trace.trace_id}
Agent: {trace.agent_id}
User: {trace.user_id}
Query: {trace.query[:100]}
Alert: {message}
Debug: /debug/trace/{trace.trace_id}
""")
# In production: send to PagerDuty, Slack, etc.
def _calculate_duration(self, trace: AgentTrace) -> float:
"""Calculate trace duration in ms."""
if not trace.completed_at:
return 0
from datetime import datetime
start = datetime.fromisoformat(trace.started_at)
end = datetime.fromisoformat(trace.completed_at)
return (end - start).total_seconds() * 1000
# Usage
alerter = AgentAlerter(thresholds={
"min_confidence": 0.7,
"max_duration_ms": 10000,
"max_tool_calls": 10,
})
alerts = await alerter.check_trace(trace)
if alerts:
print(f"⚠️ {len(alerts)} alerts triggered")
Alerting enables proactive issue detection.
For monitoring, integrate with incident management. Route trace-level alerts (a single failed high-value transaction) to a ticket queue and metric-level alerts (success rate dropped) to on-call — they need different response times.
Agent Observability Stack: Build vs Buy
You can build the tracer above in a day. The harder decision is where traces go and who looks at them. The options fall into four groups:
| Option | Strengths | Weaknesses | Best for |
|---|---|---|---|
| OpenTelemetry + Jaeger/Tempo | Vendor-neutral, already in most infra, one trace for agent + backend | No LLM-specific UI; prompts and evals need custom tooling | Teams with existing OTel pipelines |
| Langfuse (open source) | Prompt/completion views, cost tracking, scores, self-hostable | Separate system from infra traces unless bridged via OTel | Teams that want an LLM UI without vendor lock-in |
| LangSmith | Deep LangChain/LangGraph integration, datasets and evals in one place | Hosted-first, tied to LangChain conventions | LangGraph-heavy stacks |
| Arize Phoenix | Strong eval and embedding visualizations, OTel-native | Younger monitoring/alerting story | Eval-driven teams |
| Custom (Postgres + dashboards) | Exact schema you need, cheap at low volume | You own the UI, retention, and search | Regulated environments with strict data residency |
Our default is OpenTelemetry for transport plus one LLM-aware backend for the reasoning view. Langfuse, LangSmith, and Arize Phoenix all accept OTel spans, so the tracer code does not change if you switch. What you should avoid is a proprietary SDK as the only emitter; it couples every agent to one vendor.
The decision hinges on two questions. Do agent traces need to join backend traces (database, queue, HTTP) in one view? If yes, OTel is non-negotiable. Do non-engineers need to read reasoning and grade outputs? If yes, you want an LLM-specific UI on top.
Sampling, Retention, and Overhead
Full tracing of every decision is affordable for most agents but not free. Three levers control the cost:
Sampling. Trace 100% of runs while an agent is new or handles high-value actions; move to tail-based sampling once volume is high, keeping every error, every trace over a latency threshold, and a fixed percentage of the rest. Head-based sampling (deciding at the start) is the wrong tool for agents because you do not yet know which runs will fail.
Retention tiers. Keep hot traces searchable for 30 days, archive to object storage for 90 days to a year for compliance, and keep only aggregated metrics beyond that. Reasoning text is the bulk of the payload; compress it and drop it first.
Async emission. The tracer above awaits storage.save_trace at the end of a run. In production, put spans on a bounded in-process queue and export in batches — the OTel BatchSpanProcessor does exactly this. Overhead then stays in the low single-digit percent of latency, which is the range the OpenTelemetry project itself reports for well-configured SDKs.
Frequently Asked Questions
What is AI agent observability?
Agent observability is the practice of capturing every decision an AI agent makes — its reasoning, tool calls, state changes, and cost — as structured, queryable traces. It extends classic observability (logs, metrics, traces) with LLM-specific signals such as prompts, completions, token usage, and confidence. The goal is to answer "why did the agent do that?" for any production run.
How is agent observability different from LLM observability?
LLM observability focuses on individual model calls: prompt, completion, latency, tokens. Agent observability wraps many such calls plus tool executions and state into a single trace, because the failure usually lies in the sequence, not in one call. You need both; the agent layer sits on top of the model layer.
How much observability data should I collect?
Sample 100% of production traffic for critical or new agents and use tail-based sampling for high-volume systems. Always keep full traces for errors, slow runs, and high-value actions such as payments or data mutations. Aggregated metrics should cover every run regardless of sampling.
What does trace storage cost?
A typical agent trace is 5-50 KB depending on how much reasoning text you keep. At one million traces per month that is roughly 5-50 GB, which is tens of dollars in object storage or a few hundred in a hosted tracing backend. Retention policies of 30-90 days keep this bounded.
How do I keep sensitive data out of traces?
Redact at emission time inside the tracer, using a per-tool allowlist of safe fields and hashing of identifiers you still need for joins. Never log raw payment details, credentials, or health data. Treat the tracing pipeline as a data store subject to the same access controls as your primary database.
Should I trace every decision or only errors?
Trace every decision. Errors are often symptoms of an earlier "successful" decision that went wrong, so an error-only trace lacks the context you need. Async batch export keeps the latency overhead in the low single-digit percent range.
How do I visualize agent traces?
Export OpenTelemetry spans to Jaeger, Grafana Tempo, or Datadog for infrastructure-style waterfalls, and to an LLM-aware backend such as Langfuse, LangSmith, or Arize Phoenix for prompt and reasoning views. Most teams use both, fed from the same SDK.
Can agent traces be used for compliance and audit?
Yes. A complete decision trace documents why an action was taken, what data was accessed, and which tools executed, which is what auditors in finance and healthcare ask for. Pair traces with immutable storage and retention policies aligned to your regulatory regime.
Conclusion
Agent observability enables production reliability:
- Decision tracing captures complete reasoning path
- Reasoning transparency enables trust and debugging
- Tool call logging records parameters and results
- State tracking maintains context across turns
- Production monitoring catches anomalies in real-time
- Cost tracking attributes spend to users and decisions
Start with the tracer and tool decorator, emit through OpenTelemetry, and add the metrics table before you add a dashboard. Observability is essential for reliable production AI agents, and it is much cheaper to add before the first incident than after.
If you want help instrumenting an existing agent or choosing a tracing stack, talk to our AI agent engineering team or contact us directly.
Free consultation
Book a free consultation call on AI agent observability
30-minute call with the HinterBuild team. Discuss your project, architecture questions, or next steps — no obligation.
Book a meeting
Resources:
Keep reading
Related articles
Agentic Workflows: How to Build AI Workflows That Ship
Agentic workflows explained for engineers: a three-layer orchestrator, step, and infrastructure architecture, four proven patterns, and checklists.
Read post
Production AI Agents: Tool Calling, Validation & MCP Guide
Build production AI agents that survive real users: single-responsibility tools, validation before side effects, approval gates, and MCP servers in Python.
Read post
Reliable Tool Calling: Production AI Agent Error Handling &
Reliable Tool Calling guidance for engineers: compare architecture choices, avoid failure modes, and ship a measurable, reliable production implementation.
Read post
ReAct vs Plan-and-Execute: Agent Reasoning Patterns Compared
Learn react vs plan-and-execute through concrete architecture trade-offs, failure modes, rollout controls, and production measurement practices.
Read post