agent-observability

# Agent Observability Monitor AI agent behavior with logs, traces, metrics, and cost telemetry. This skill covers the full observability stack for LLM-powered applications: from raw Prometheus counters to Grafana dashboards, OpenTelemetry tracing, structured logging, cost tracking, SLO definition, and PII redaction. --- ## When to Use Apply this skill whenever you operate: - **Autonomous AI agents** that make multi-step tool calls (e.g., coding agents, support agents, data-pipeline agents). - **LLM-backed APIs** serving chat completions, summarisation, or classification behind a REST or gRPC gateway. - **RAG pipelines** where a retriever fetches context from a vector store before prompting a model. - **Multi-agent orchestrations** (crew-style or graph-based) where several agents collaborate on a single task. - **Batch inference jobs** that process thousands of prompts against a model endpoint. Key signals that you need this skill: 1. You cannot answer "what is p95 latency for agent responses this week?" 2. You have no per-request cost attribution. 3. Debugging a bad agent response requires grepping raw application logs. 4. You have no alerting on token-usage spikes or elevated error rates. --- ## Core Metrics Define these metrics at the application layer. All examples use the Prometheus client library naming conventions. ### Latency ```python from prometheus_client import Histogram # Total end-to-end latency for a full agent turn (user prompt -> final response) AGENT_LATENCY = Histogram( "agent_request_duration_seconds", "End-to-end latency of an agent request", labelnames=["agent_name", "model", "status"], buckets=(0.25, 0.5, 1, 2, 5, 10, 30, 60, 120), ) # Latency of a single LLM API call (one completion request) LLM_CALL_LATENCY = Histogram( "llm_call_duration_seconds", "Latency of an individual LLM API call", labelnames=["model", "provider", "stream"], buckets=(0.1, 0.25, 0.5, 1, 2, 5, 10, 30), ) # Latency of tool/function calls executed by the agent TOOL_CALL_LATENCY = Histogram( "agent_tool_call_duration_seconds", "Latency of a tool call executed by the agent", labelnames=["tool_name", "agent_name", "status"], buckets=(0.05, 0.1, 0.25, 0.5, 1, 2, 5, 10), ) ``` ### Token Usage ```python from prometheus_client import Counter, Histogram PROMPT_TOKENS = Counter( "llm_prompt_tokens_total", "Total prompt tokens sent to the model", labelnames=["model", "agent_name"], ) COMPLETION_TOKENS = Counter( "llm_completion_tokens_total", "Total completion tokens received from the model", labelnames=["model", "agent_name"], ) CACHED_TOKENS = Counter( "llm_cached_tokens_total", "Prompt tokens served from KV-cache (provider-reported)", labelnames=["model", "agent_name"], ) TOKENS_PER_REQUEST = Histogram( "llm_tokens_per_request", "Total tokens (prompt + completion) per request", labelnames=["model", "agent_name"], buckets=(100, 500, 1000, 2000, 4000, 8000, 16000, 32000, 64000, 128000), ) ``` ### Cost ```python from prometheus_client import Counter LLM_COST = Counter( "llm_cost_dollars_total", "Estimated cost in USD for LLM usage", labelnames=["model", "agent_name", "cost_type"], # cost_type: prompt | completion ) ``` ### Tool Calls ```python from prometheus_client import Counter TOOL_CALLS_TOTAL = Counter( "agent_tool_calls_total", "Total tool calls made by agents", labelnames=["tool_name", "agent_name", "status"], # status: success | error | timeout ) ``` ### Errors and Retries ```python from prometheus_client import Counter, Gauge LLM_ERRORS = Counter( "llm_errors_total", "Errors returned by the LLM provider", labelnames=["model", "provider", "error_type"], # error_type: rate_limit | timeout | 5xx | auth ) LLM_RETRIES = Counter( "llm_retries_total", "Retried LLM API calls", labelnames=["model", "provider", "retry_reason"], ) AGENT_ACTIVE_REQUESTS = Gauge( "agent_active_requests", "Number of agent requests currently in flight", labelnames=["agent_name"], ) ``` --- ## OpenTelemetry Integration Use the OpenTelemetry Python SDK to create traces that capture every step of an agent turn: the top-level request, each LLM call, each tool execution, and retrieval operations. ### Setup ```python # otel_setup.py from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import BatchSpanProcessor from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter from opentelemetry.sdk.resources import Resource def init_tracing(service_name: str, otlp_endpoint: str = "http://localhost:4317"): resource = Resource.create({ "service.name": service_name, "service.version": "1.0.0", "deployment.environment": "production", }) provider = TracerProvider(resource=resource) exporter = OTLPSpanExporter(endpoint=otlp_endpoint, insecure=True) provider.add_span_processor(BatchSpanProcessor(exporter)) trace.set_tracer_provider(provider) return trace.get_tracer(service_name) ``` ### Tracing LLM Calls ```python # llm_tracing.py import time from opentelemetry import trace from opentelemetry.trace import StatusCode tracer = trace.get_tracer("agent.llm") def traced_llm_call(client, messages, model="gpt-4o", **kwargs): """Wrap an LLM completion call with a full OpenTelemetry span.""" with tracer.start_as_current_span("llm.chat_completion") as span: span.set_attribute("llm.model", model) span.set_attribute("llm.provider", "openai") span.set_attribute("llm.message_count", len(messages)) span.set_attribute("llm.temperature", kwargs.get("temperature", 1.0)) span.set_attribute("llm.max_tokens", kwargs.get("max_tokens", 0)) start = time.perf_counter() try: response = client.chat.completions.create( model=model, messages=messages, **kwargs ) elapsed = time.perf_counter() - start usage = response.usage span.set_attribute("llm.prompt_tokens", usage.prompt_tokens) span.set_attribute("llm.completion_tokens", usage.completion_tokens) span.set_attribute("llm.total_tokens", usage.total_tokens) span.set_attribute("llm.duration_seconds", elapsed) span.set_attribute("llm.finish_reason", response.choices[0].finish_reason) span.set_status(StatusCode.OK) # Update Prometheus counters PROMPT_TOKENS.labels(model=model, agent_name="default").inc(usage.prompt_tokens) COMPLETION_TOKENS.labels(model=model, agent_name="default").inc(usage.completion_tokens) LLM_CALL_LATENCY.labels(model=model, provider="openai", stream="false").observe(elapsed) return response except Exception as exc: elapsed = time.perf_counter() - start span.set_status(StatusCode.ERROR, str(exc)) span.record_exception(exc) LLM_ERRORS.labels(model=model, provider="openai", error_type=type(exc).__name__).inc() raise ``` ### Tracing Tool Execution ```python # tool_tracing.py import functools from opentelemetry import trace from opentelemetry.trace import StatusCode tracer = trace.get_tracer("agent.tools") def traced_tool(tool_name: str): """Decorator that wraps a tool function with an OTel span and Prometheus metrics.""" def decorator(func): @functools.wraps(func) def wrapper(*args, **kwargs): with tracer.start_as_current_span(f"tool.{tool_name}") as span: span.set_attribute("tool.name", tool_name) span.set_attribute("tool.args_count", len(args) + len(kwargs)) import time start = time.perf_counter() try: result = func(*args, **kwargs) elapsed = time.perf_counter() - start span.set_attribute("tool.duration_seconds", elapsed) span.set_status(StatusCode.OK) TOOL_CALLS_TOTAL.labels( tool_name=tool_name, agent_name="default", status="success" ).inc() TOOL_CALL_LATENCY.labels( tool_name=tool_name, agent_name="default", status="success" ).observe(elapsed) return result except Exception as exc: elapsed = time.perf_counter() - start span.set_status(StatusCode.ERROR, str(exc)) span.record_exception(exc) TOOL_CALLS_TOTAL.labels( tool_name=tool_name, agent_name="default", status="error" ).inc() TOOL_CALL_LATENCY.labels( tool_name=tool_name, agent_name="default", status="error" ).observe(elapsed) raise return wrapper return decorator # Usage @traced_tool("web_search") def web_search(query: str) -> str: # ... tool implementation ... pass @traced_tool("sql_query") def sql_query(statement: str) -> list: # ... tool implementation ... pass ``` ### Propagating Trace Context Across Services ```python # context_propagation.py from opentelemetry import context from opentelemetry.propagate import inject, extract import httpx def call_downstream_service(url: str, payload: dict) -> dict: """Propagate the current trace context to a downstream HTTP service.""" headers = {} inject(headers) # injects traceparent + tracestate headers response = httpx.post(url, json=payload, headers=headers) response.raise_for_status() return response.json() def extract_context_from_request(request_headers: dict): """Extract trace context from incoming request headers (for the receiving service).""" ctx = extract(request_headers) token = context.attach(ctx) return token # call context.detach(token) when done ``` --- ## Structured Logging Emit JSON logs for every agent action so they can be ingested by Loki, Elasticsearch, or Datadog. ### Python Logging Configuration ```python # logging_config.py import logging import json import sys from datetime import datetime, timezone class AgentJSONFormatter(logging.Formatter): """Structured JSON formatter for agent logs.""" def format(self, record: logging.LogRecord) -> str: log_entry = { "timestamp": datetime.now(timezone.utc).isoformat(), "level": record.levelname, "logger": record.name, "message": record.getMessage(), "module": record.module, "function": record.funcName, "line": record.lineno, } # Merge any extra fields attached to the record for key in ("trace_id", "span_id", "agent_name", "model", "tool_name", "request_id", "user_id", "prompt_tokens", "completion_tokens", "cost_usd", "duration_seconds", "status", "error_type"): value = getattr(record, key, None) if value is not None: log_entry[key] = value if record.exc_info and record.exc_info[0] is not None: log_entry["exception"] = self.formatException(record.exc_info) return json.dumps(log_entry, default=str) def configure_logging(level: str = "INFO"): handler = logging.StreamHandler(sys.stdout) handler.setFormatter(AgentJSONFormatter()) root = logging.getLogger() root.setLevel(getattr(logging, level)) root.handlers = [handler] # Suppress noisy libraries logging.getLogger("httpx").setLevel(logging.WARNING) logging.getLogger("opentelemetry").setLevel(logging.WARNING) ``` ### Logging Agent Actions ```python # agent_logging.py import logging from opentelemetry import trace logger = logging.getLogger("agent") def log_llm_call(model: str, prompt_tokens: int, completion_tokens: int, duration: float, cost: float, status: str = "ok"): span = trace.get_current_span() ctx = span.get_span_context() if span else None logger.info( "LLM call completed", extra={ "trace_id": format(ctx.trace_id, "032x") if ctx else None, "span_id": format(ctx.span_id, "016x") if ctx else None, "model": model, "prompt_tokens": prompt_tokens, "completion_tokens": completion_tokens, "duration_seconds": round(duration, 3), "cost_usd": round(cost, 6), "status": status, "agent_name": "default", }, )