Mental model: traces, not logs
Most developers start by adding print statements or basic logger calls inside their agent loop. That gives you events, but events alone do not let you answer "why did the agent call the payment API three times in one session?" For that you need a trace: a tree of causally linked spans. Borrow the OpenTelemetry mental model. Each agent execution is a root span (the trace). Each ReAct iteration is a child span. Each tool call within that iteration is a grandchild span. Every span carries a trace_id, a span_id, and a parent_span_id. With that tree structure you can reconstruct the full decision path for any session, filter down to the exact iteration where reasoning went sideways, and diff two sessions that produced different outcomes.
The data you need to capture at each span: (1) the observation or user input that triggered this step, (2) the model's reasoning output verbatim (do not summarize it; you need the actual chain-of-thought), (3) every tool call with its exact arguments, (4) the raw tool result, (5) the model's action decision, and (6) metadata: agent_id, session_id, user_id, model name, prompt token count, completion token count, latency in ms, and a criticality flag if the action touches sensitive systems. Omitting any of these creates blind spots that surface during incidents at the worst time.
Real-world scenario: catching an agent that loops on error
Imagine a customer support agent that can look up orders, issue refunds, and escalate to humans. In production you start seeing complaints that certain sessions take five minutes before the agent gives up. You pull the trace for one of those sessions and find a pattern: the agent called get_order_status, got a 503 from your internal API, then reasoned "the order might be delayed, let me check shipment" and called get_shipment_status, which also 503'd, then re-tried get_order_status again. No hard retry limit was enforced at the tool layer, and the model's reasoning kept finding new justifications to keep trying. Without the per-span tool-call argument log you would only see "agent spent 5 minutes" and have no idea which tool was the culprit. With the trace you see the tool call pattern immediately and add a circuit breaker. This is the core value proposition: the audit trail makes the implicit visible.
Tradeoffs vs alternative approaches
You have three main options for where to store traces. First, write structured JSON to stdout and let a log aggregator (Fluentd, Vector, Fluent Bit) ship it to a backend (Loki, Elasticsearch, CloudWatch Logs). This is low friction, works everywhere, and decouples the agent from the observability stack. Downside: you give up native span relationships and rely on trace_id/span_id conventions you enforce yourself. Second, use an OpenTelemetry SDK and export spans directly to an OTLP collector (Jaeger, Tempo, Honeycomb, Datadog). You get native trace visualization, waterfall views, and built-in parent/child linking. Downside: adds SDK setup complexity and the OTel Python SDK has non-trivial overhead if you instrument every LLM call synchronously. Third, use a purpose-built LLM observability platform like LangSmith, Arize Phoenix, or Helicone. These understand LLM-specific concepts (prompt, completion, token cost) out of the box and provide agent-specific dashboards. Downside: vendor lock-in and another credential to manage. For most production teams the right answer starts at option 1 for quick time-to-value and migrates to option 2 or 3 once you have enough trace volume to justify the investment.
What changes at scale
At 10 concurrent users, synchronous logging to a local file or database works fine. At 10,000 users, synchronous writes will add measurable latency to your agent loop, especially if each ReAct iteration emits 4-6 spans. The fix is to push spans onto an in-process async queue (asyncio.Queue or a thread-safe queue) and have a background worker batch-ship them to the backend. Size the queue and set a max-wait to handle bursts without unbounded memory growth. At 10 million agent sessions per day, trace volume becomes a storage and cost problem. You need a tiered retention policy: keep full verbose traces for 7 days, downsample to summary-level traces for 90 days, keep only anomalous sessions indefinitely. Define "anomalous" ahead of time: sessions with more than N tool calls, sessions that triggered a refund, sessions where a safety classifier flagged the output, or sessions with total cost above a threshold. Build that filtering into your pipeline before you need it.
Alerting and behavioral drift detection
Audit trails are reactive by themselves. Add a proactive layer by computing rolling metrics over your trace stream: average tool calls per session, p95 latency per tool, error rate per tool, token spend per session, and frequency of safety classifier flags. Set thresholds and alert when any metric exceeds its baseline by more than two standard deviations. This catches behavioral drift, which is the pattern where your agent gradually starts behaving differently because of upstream model updates or slow changes in the distribution of incoming queries, without any single catastrophic failure to trigger a human review. Pipe these metrics to whatever your team already uses (Datadog, Grafana, PagerDuty) so agent health sits alongside your other production services rather than in a separate, unmonitored dashboard that nobody checks.
Key Takeaways
- Capture the full causal chain: observation, reasoning, tool call, result, and final action per step.
- Use structured, append-only trace records with a span model so you can reconstruct any agent session.
- Emit traces asynchronously to avoid blocking agent execution; buffer locally if the sink is unavailable.
- Set anomaly thresholds on tool-call frequency, token spend, and error rates to catch behavioral drift early.
Pro tips
- Store the raw model output, not a parsed or summarized version. You will need the exact text when debugging hallucinated tool arguments or when a safety review requires the verbatim chain-of-thought.
- Assign a criticality level to each tool at registration time (read-only, state-mutating, irreversible) and include that level on every tool-call span. Downstream alerting and human review queues can then filter purely on criticality without reading the payload.
- Trace IDs must propagate across service boundaries. If your agent calls an internal microservice, pass the trace_id as an HTTP header (X-Trace-Id). Without this you cannot correlate an agent decision with the downstream effect it caused in another system.
- Sampling is fine for high-volume LLM call spans, but never sample tool-call spans that touch external state. A sampled read is harmless to miss; a sampled write that caused a problem and was dropped from the trace is a compliance liability.
Common pitfalls
- Mistake: Logging only the final agent output, skipping intermediate reasoning steps. Fix: Emit a span for each ReAct iteration's thought block; without it you cannot reconstruct why the agent chose a particular tool.
- Mistake: Writing trace records synchronously inside the agent loop. Fix: Push to an in-process async queue and ship in a background task to avoid adding latency proportional to your logging backend's p99.
- Mistake: Using mutable log records that can be overwritten after the fact. Fix: Treat each span as append-only; never update a span, emit a new one with a reference to the original span_id if you need a correction.
- Mistake: Storing PII (user messages, email addresses) verbatim in traces shipped to third-party observability platforms. Fix: Scrub or tokenize PII fields before emission; keep a mapping in your own secure store if re-identification is ever needed.
When to use stdout logging vs OTel SDK vs purpose-built LLM observability
| Option | Use when | Avoid when |
|---|---|---|
| Structured stdout + log aggregator (Loki, Elasticsearch) | Early prototype, existing log pipeline, team already on ELK/Grafana stack. | You need native waterfall trace views or automatic token-cost dashboards without building them yourself. |
| OpenTelemetry SDK + OTLP collector (Jaeger, Tempo, Honeycomb) | Polyglot system, multiple services sharing one trace, team already uses OTel for non-AI services. | You want LLM-specific concepts (prompt, completion, cost) out of the box; OTel has no standard semantic convention for those yet. |
| Purpose-built LLM platform (LangSmith, Arize Phoenix, Helicone) | Team wants prompt-level replay, cost attribution per session, and agent-step visualization with minimal setup. | Strict data residency requirements prevent sending prompts/completions to a third-party SaaS. |
| Custom append-only audit DB (Postgres + JSONB, BigQuery) | Compliance mandate requires immutable records you control, long retention (years), and complex forensic queries. | You need real-time dashboards; query latency on raw JSONB at scale requires extra indexing work. |
Code Example
# opentelemetry-sdk 1.24, openai 1.30
import uuid, time, json
from dataclasses import dataclass, field, asdict
from typing import Any
@dataclass
class AgentSpan:
trace_id: str
span_id: str = field(default_factory=lambda: uuid.uuid4().hex)
parent_span_id: str | None = None
event_type: str = "" # thought | tool_call | tool_result | final_action
agent_id: str = ""
session_id: str = ""
timestamp_ms: int = field(default_factory=lambda: int(time.time() * 1000))
payload: dict = field(default_factory=dict)
def emit(self):
# Replace print with your log shipper (stdout -> Vector -> Loki, etc.)
print(json.dumps(asdict(self)))
# Usage inside a ReAct loop iteration
trace_id = uuid.uuid4().hex
span = AgentSpan(
trace_id=trace_id,
event_type="tool_call",
agent_id="research-agent-v2",
session_id="user-abc-session-1",
payload={"tool": "web_search", "args": {"query": "GDPR fines 2024"}}
)
span.emit()How this code works
This code establishes a standardized system for logging an AI agent's actions, creating an audit trail crucial for monitoring, debugging, and ensuring agent safety. It defines an AgentSpan structure to capture detailed information about each decision or event an agent makes, such as a "tool_call" or "final_action." By recording these individual steps with context, the system provides a clear, traceable history of the agent's behavior, which is essential for understanding its operational flow and identifying any issues. The emit method then serializes this structured data into JSON and outputs it, typically to a dedicated logging service.
The AgentSpan dataclass is a template for these event records. Key fields like trace_id help link related actions across a longer agent task, while span_id uniquely identifies each specific event. Notice the use of default_factory for fields like span_id, timestamp_ms, and payload. This subtle but important design choice ensures that a fresh, unique value (like a new UUID or the current time) or an empty dictionary is generated every time a new AgentSpan is created, preventing unexpected sharing of mutable default objects between different span instances, a common pitfall. The payload field offers flexible storage for any contextual data relevant to the specific event.
Production-grade example
Async trace queue, background batch shipper, retry with backoff, token logging, and error capture per span.
# opentelemetry-sdk 1.24, openai 1.30, tenacity 8.3
import asyncio, os, time, uuid, logging, json
from dataclasses import dataclass, field, asdict
from typing import Any
from openai import AsyncOpenAI, APIError, RateLimitError
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
logging.basicConfig(level=logging.INFO, format='{"level": "%(levelname)s", "msg": %(message)s}')
log = logging.getLogger("agent.audit")
TRACE_QUEUE: asyncio.Queue = asyncio.Queue(maxsize=10_000)
@dataclass
class AgentSpan:
trace_id: str
span_id: str = field(default_factory=lambda: uuid.uuid4().hex)
parent_span_id: str | None = None
event_type: str = ""
agent_id: str = os.getenv("AGENT_ID", "unknown")
session_id: str = ""
timestamp_ms: int = field(default_factory=lambda: int(time.time() * 1000))
payload: dict = field(default_factory=dict)
prompt_tokens: int = 0
completion_tokens: int = 0
latency_ms: int = 0
error: str | None = None
async def emit(self):
try:
TRACE_QUEUE.put_nowait(asdict(self))
except asyncio.QueueFull:
log.warning(json.dumps({"warn": "trace_queue_full", "dropped_span": self.span_id}))
async def trace_shipper():
"""Background task: batch-ship spans every 500ms or when 50 accumulate."""
batch = []
while True:
deadline = time.monotonic() + 0.5
while time.monotonic() < deadline and len(batch) < 50:
try:
span = TRACE_QUEUE.get_nowait()
batch.append(span)
except asyncio.QueueEmpty:
await asyncio.sleep(0.05)
if batch:
# Replace with real sink: POST to Loki, write to BigQuery, etc.
log.info(json.dumps({"batch_size": len(batch), "spans": batch}))
batch = []
client = AsyncOpenAI(api_key=os.environ["OPENAI_API_KEY"])
@retry(
retry=retry_if_exception_type((RateLimitError, APIError)),
wait=wait_exponential(multiplier=1, min=2, max=30),
stop=stop_after_attempt(4),
)
async def call_model_with_trace(
messages: list[dict],
trace_id: str,
parent_span_id: str | None,
session_id: str,
) -> tuple[str, AgentSpan]:
span = AgentSpan(
trace_id=trace_id,
parent_span_id=parent_span_id,
event_type="llm_call",
session_id=session_id,
)
t0 = time.monotonic()
try:
response = await asyncio.wait_for(
client.chat.completions.create(
model="gpt-4o",
messages=messages,
timeout=30,
),
timeout=35,
)
span.latency_ms = int((time.monotonic() - t0) * 1000)
span.prompt_tokens = response.usage.prompt_tokens
span.completion_tokens = response.usage.completion_tokens
content = response.choices[0].message.content or ""
span.payload = {"model": "gpt-4o", "output_preview": content[:200]}
await span.emit()
return content, span
except Exception as exc:
span.error = type(exc).__name__ + ": " + str(exc)[:200]
span.latency_ms = int((time.monotonic() - t0) * 1000)
await span.emit()
raise
async def main():
asyncio.create_task(trace_shipper())
trace_id = uuid.uuid4().hex
session_id = "demo-session-001"
messages = [{"role": "user", "content": "Summarize GDPR Article 17 in plain English."}]
result, span = await call_model_with_trace(messages, trace_id, None, session_id)
print(result)
await asyncio.sleep(1) # let shipper flush
if __name__ == "__main__":
asyncio.run(main())How this code works
This code establishes a robust audit trail for an AI agent's decisions, specifically when interacting with a Large Language Model (LLM). Its job is to capture detailed information about each LLM call—like prompt/completion tokens, latency, and any errors—and log it in a structured way. This allows for crucial monitoring of agent safety, performance, and behavior, enabling teams to review and understand why an agent acted in a certain way.
The system uses an AgentSpan dataclass to define the structure of each audit record, containing a trace_id to link related events and a unique span_id. When an AgentSpan is ready, its emit method asynchronously pushes it to a TRACE_QUEUE (an asyncio.Queue), preventing the main agent logic from blocking. A separate trace_shipper background task then efficiently pulls these spans, batches them, and logs them to a placeholder sink, ready for shipment to a real-world logging system. The call_model_with_trace function wraps the LLM call, measuring latency_ms, recording token usage, and importantly, capturing error details if the call fails. A subtle but critical feature is the @retry decorator from tenacity, which makes the agent resilient by automatically reattempting LLM calls that encounter RateLimitError or APIError with an exponential backoff.
Practice & master
Try the exercise, check your understanding, then mark this lesson mastered to track your path to pro.
Exercise
Build a minimal agent trace collector for a two-step ReAct loop. Your agent should emit a structured span for each step: one for the model's reasoning call and one for the (simulated) tool call. Print the complete trace as a JSON array at the end of the session, ordered by timestamp.
import uuid, time, json
from dataclasses import dataclass, field, asdict
from typing import Any
TRACE: list[dict] = []
@dataclass
class Span:
trace_id: str
span_id: str = field(default_factory=lambda: uuid.uuid4().hex)
parent_span_id: str | None = None
event_type: str = "" # "thought" | "tool_call" | "tool_result"
payload: dict = field(default_factory=dict)
timestamp_ms: int = field(default_factory=lambda: int(time.time() * 1000))
def record(self):
# TODO: append asdict(self) to TRACE
pass
def simulate_react_loop(user_query: str):
trace_id = uuid.uuid4().hex
# Step 1: model reasons about the query
thought_span = Span(
trace_id=trace_id,
event_type="thought",
# TODO: fill payload with {"input": user_query, "reasoning": "I should search for this"}
)
# TODO: record the span
# Step 2: model decides to call a tool
tool_span = Span(
trace_id=trace_id,
parent_span_id=thought_span.span_id,
event_type="tool_call",
# TODO: fill payload with {"tool": "web_search", "args": {"query": user_query}}
)
# TODO: record the span
return TRACE
if __name__ == "__main__":
trace = simulate_react_loop("What is the capital of France?")
print(json.dumps(sorted(trace, key=lambda s: s["timestamp_ms"]), indent=2))Quick check
Why should trace spans be treated as append-only records rather than mutable log entries?
An agent emits 6 spans per ReAct iteration and handles 500 concurrent sessions. What is the primary risk of synchronous span writes inside the agent loop?
You need to correlate an agent's tool call with a side effect it caused in a downstream microservice. What must both systems share for the correlation to work?