Tutorial 9: Production Observability for Agents

Agents in production fail in ways you didn't see in development. Observability is how you find and fix what you didn't anticipate.

What You'll Build

A production observability setup: tracing every agent step, alerting on anomalies, dashboards showing health.

Step 1: Trace Every Run (15 min)

def agent(task, trace_id=None):
    trace_id = trace_id or str(uuid4())
    
    log_event("agent_start", {
        "trace_id": trace_id,
        "task": task,
        "timestamp": datetime.now(),
    })
    
    try:
        # ... agent loop ...
        log_event("agent_complete", {
            "trace_id": trace_id,
            "result": result,
            "steps": len(trace),
        })
    except Exception as e:
        log_event("agent_failed", {
            "trace_id": trace_id,
            "error": str(e),
        })
        raise

Every run has a trace ID. Find it later.

Step 2: Trace Each Step (15 min)

def execute_step(trace_id, step_num, tool, input):
    log_event("step_start", {
        "trace_id": trace_id,
        "step": step_num,
        "tool": tool,
        "input": input,
    })
    
    start = time.time()
    
    try:
        result = TOOL_FN[tool](**input)
        log_event("step_complete", {
            "trace_id": trace_id,
            "step": step_num,
            "result_size": len(str(result)),
            "latency_ms": (time.time() - start) * 1000,
        })
        return result
    except Exception as e:
        log_event("step_failed", {
            "trace_id": trace_id,
            "step": step_num,
            "error": str(e),
        })
        raise

Every step's inputs, outputs, errors, latency.

Step 3: Structured Logging (15 min)

Don't log strings. Log structured data:

def log_event(event_type, attributes):
    logger.info(json.dumps({
        "event": event_type,
        "service": "agent",
        "timestamp": datetime.now().isoformat(),
        **attributes,
    }))

Structured logs query well. Strings don't.

Step 4: Send to Observability Platform (varies)

For production:

• Datadog APM: out-of-the-box distributed tracing

• Honeycomb: strong for high-cardinality events

• OpenTelemetry + Tempo/Jaeger: open source

Each call becomes a span; spans nest into traces.

from opentelemetry import trace

tracer = trace.get_tracer(__name__)

def execute_step(tool, input):
    with tracer.start_as_current_span(f"step.{tool}") as span:
        span.set_attribute("tool", tool)
        span.set_attribute("input.summary", summarize(input))
        
        try:
            return TOOL_FN[tool](**input)
        except Exception as e:
            span.record_exception(e)
            raise

Step 5: Build the Dashboard (varies)

Key metrics for an agent:

Real-time:
- Active agents (concurrent runs)
- Average task duration
- Step count distribution
- Tool usage frequency

24-hour:
- Total tasks
- Success rate
- Failure rate by cause
- Cost
- Median latency

Trends (7-day):
- Pass rate over time
- Cost trend
- Top failing tasks

Visible. Updated. Watched.

Step 6: Alert on Anomalies (15 min)

Set alerts:

- name: Agent failure rate elevated
  condition: failure_rate > 10% for 5 minutes
  action: Page oncall

- name: Agent cost spike
  condition: hourly_cost > 3x average
  action: Notify team

- name: Agent latency degraded
  condition: p95_latency > 30s for 10 minutes
  action: Notify team

Alerts before issues become incidents.

Step 7: Capture Failure Traces (15 min)

def on_agent_failure(trace_id, error, full_trace):
    # Save the entire trace for debugging
    save_failure({
        "trace_id": trace_id,
        "error": str(error),
        "full_trace": full_trace,
        "timestamp": datetime.now(),
    })

Failures get rich detail. Investigate later.

Step 8: User-Visible Trace ID (10 min)

When something goes wrong:

# In the API response
if error:
    return {
        "error": "Something went wrong",
        "trace_id": trace_id,
    }

User reports the trace ID; support looks up exact details.

Step 9: Sample for Quality Review (15 min)

import random

def maybe_sample(trace):
    # Sample 1% for human review
    if random.random() < 0.01:
        queue_for_review(trace)

Random sampling catches issues automated checks miss.

Step 10: Replay Failures (varies)

For debugging:

def replay_failure(trace_id):
    failure = load_failure(trace_id)
    
    # Re-run with same input
    new_trace = agent(failure["task"])
    
    # Compare
    return compare_traces(failure["full_trace"], new_trace)

Replay catches whether you've actually fixed something.

What You Just Did

You have production-grade observability. Every agent action is traced; failures are captured; anomalies alert; debugging is tractable.

Common Failure Modes

Logs as strings. Hard to query. Always structured.

No alerts. Issue runs for hours; nobody notices.

No trace ID. User report → can't find the run.

Sample size 0. Pure automated metrics; misses qualitative issues.

Slow log ingestion. Real-time issues invisible until tomorrow.