agent_observability v1.0.0

name: "agent_observability" description: "Instruments AI agent runs with structured traces, token accounting, and failure forensics using OpenTelemetry GenAI conventions. Triggers when an agent moves from prototype to service, shows flaky failures with no reproducer, needs cost attribution per user or feature, or adopts Langfuse, LangSmith, Phoenix, or similar tracing."

Agent Observability

Purpose

Turn an agent from a black box into a replayable record. One user-facing request becomes one root trace with nested spans for every model call, tool execution, retrieval step, and guard check. Each span records what the step saw, what it chose, and what it cost, so failures are debuggable and spend is attributable.

Workflow

1. Inventory the run. Map the agent's loop: model calls, tool executions, retrieval steps, verification and guard checks, sub-agent delegations. Note providers, models, prompt versions, and any existing APM wiring.
2. Set the schema. Emit OpenTelemetry spans in the GenAI gen_ai.* attribute namespace so the data stays portable across backends. One trace per request, one span per step. Attribute details live in references/span-attributes.md.
3. Capture tokens and cost. Record input/output token counts on every model span, plus cache-read and reasoning tokens where the provider reports them. Derive dollars from a versioned price table, never from hardcoded rates. Roll spend up to trace, session, user, and agent.
4. Protect prompts. Content capture (prompt and response bodies) is opt-in. Redact or mask PII at write time, set a retention window, and take the OTel spec's warning seriously: message content routinely contains PII.
5. Wire quality signals. Run LLM-as-judge evaluators on a small sample of production traces, asynchronously off the hot path. Capture user feedback (thumbs, edits, escalation to a human) and tie it to the trace that produced the answer.
6. Establish baselines. Track cost per request, latency percentiles, and judge scores as rolling baselines. Alert on drift against the baseline, not on fixed thresholds that page on normal variance.
7. Alert with context. Every alert carries a trace link: error-rate spikes, cost anomalies, token overruns, judge-score drops.
8. Debug from traces. When a run fails, walk the span tree with references/failure-forensics.md.

Output Contract

• Every span carries: name, status, model and request metadata, token counts, latency, and trace/span IDs that link sub-agent calls back to the parent trace.
• Every trace carries: total tokens, total derived cost, step count, an outcome signal, and a prompt/config fingerprint (hash of the system prompt, tool schema versions, and model ID).
• Dashboards split cost, latency, and quality by agent, user, feature, and step type.

Operating Rules

• Instrument before the first incident, not after. Trace every run in dev; in production use head-based sampling plus 100 percent tail sampling for errors and outliers.
• Log inputs and outcomes, not just errors. A trace without inputs cannot be replayed; a trace without an outcome signal cannot become an eval.
• Attribute regressions with fingerprints. Without a hash of the prompt and config that produced a run, a quality drop is unassignable.
• Watch convergence, not just totals. Turns per run and repeated-action detection catch wandering agents before the bill does.
• Reconcile derived cost against the provider invoice. When the numbers disagree, find the gap; unreported cache and reasoning tokens are the usual suspects.
• Never report a number you did not measure. Token counts come from the provider's usage payload; dollars come from the versioned price table.

Worked example

Stdlib-only tracer an agent can drop into a project. Run it with python3, get a JSON trace:

python
import hashlib, json, time, uuid
from contextlib import contextmanager

class Tracer:
    def __init__(self):
        self.spans = []
        self._stack = []

    @staticmethod
    def _attr_name(key):
        # gen_ai_operation_name -> gen_ai.operation.name (matches the OTel spec)
        if key.startswith("gen_ai_"):
            return "gen_ai." + key[7:].replace("_", ".")
        return key

    @contextmanager
    def span(self, name, **attrs):
        attrs = {Tracer._attr_name(k): v for k, v in attrs.items()}
        span = {"span_id": uuid.uuid4().hex[:16], "name": name,
                "parent_id": self._stack[-1]["span_id"] if self._stack else None,
                "start": time.time(), "attrs": attrs}
        self._stack.append(span)
        try:
            yield span
        except Exception as e:
            span["status"] = {"code": "ERROR", "message": str(e)}
            raise
        else:
            span.setdefault("status", {"code": "OK"})
        finally:
            span["end"] = time.time()
            span["duration_s"] = round(span["end"] - span["start"], 4)
            self._stack.pop()
            self.spans.append(span)

def fingerprint(*parts):
    return hashlib.sha256("|".join(parts).encode()).hexdigest()[:12]

def agent_run(tracer, model, price_table, goal):
    trace_id = uuid.uuid4().hex
    total_cost = 0.0
    total_in, total_out = 0, 0
    with tracer.span("invoke_agent", gen_ai_operation_name="invoke_agent",
                     gen_ai_request_model=model,
                     gen_ai_agent_id="billing-refund-v2",
                     prompt_fp=fingerprint("sys-prompt-v3", "tool-schema-v7", model)):
        with tracer.span("chat", gen_ai_operation_name="chat",
                         gen_ai_request_model=model) as s:
            # In real code these come from the provider's usage payload.
            in_tok, out_tok = 1420, 280
            cost = in_tok * price_table["input"] + out_tok * price_table["output"]
            s["attrs"].update({"gen_ai.usage.input_tokens": in_tok,
                               "gen_ai.usage.output_tokens": out_tok,
                               "cost_usd": round(cost, 6)})
            total_cost += cost
            total_in += in_tok
            total_out += out_tok
        with tracer.span("execute_tool get_weather",
                         gen_ai_operation_name="execute_tool",
                         gen_ai_tool_name="get_weather"):
            pass  # Tool runs here. Redact args and results at write time.
    return {"trace_id": trace_id, "goal": goal,
            "total_cost_usd": round(total_cost, 6),
            "total_input_tokens": total_in, "total_output_tokens": total_out,
            "spans": tracer.spans}

if __name__ == "__main__":
    # Placeholder rates. Load real ones from a versioned price table.
    price = {"input": 3.0 / 1_000_000, "output": 15.0 / 1_000_000}
    t = Tracer()
    trace = agent_run(t, "claude-3-7-sonnet-20250219", price, "check refund status")
    with open("/tmp/agent_trace.json", "w") as f:
        json.dump(trace, f, indent=2)
    print("trace_id", trace["trace_id"], "cost_usd", trace["total_cost_usd"])