Observability

Comprehensive observability for agent systems: logging, metrics, tracing, alerting, and debugging.

Observability Pillars

Logging

Structured Logging

import structlog
import logging

# Configure structlog
structlog.configure(
    processors=[
        structlog.stdlib.filter_by_level,
        structlog.stdlib.add_logger_name,
        structlog.stdlib.add_log_level,
        structlog.stdlib.PositionalArgumentsFormatter(),
        structlog.processors.TimeStamper(fmt="iso"),
        structlog.processors.StackInfoRenderer(),
        structlog.processors.format_exc_info,
        structlog.processors.UnicodeDecoder(),
        structlog.processors.JSONRenderer()
    ],
    context_class=dict,
    logger_factory=structlog.stdlib.LoggerFactory(),
    wrapper_class=structlog.stdlib.BoundLogger,
    cache_logger_on_first_use=True,
)

logger = structlog.get_logger()

# Usage with context
logger = logger.bind(agent_id="agent-123", session_id="session-456")

async def process_request(request: Request):
    logger.info("request_received", 
                method=request.method,
                path=request.path,
                user_id=request.user.id)

    try:
        result = await process(request)
        logger.info("request_completed", 
                    duration_ms=duration,
                    status="success")
        return result
    except Exception as e:
        logger.exception("request_failed",
                         error=str(e),
                         error_type=type(e).__name__)
        raise

Log Levels & Sampling

# Dynamic log levels
import logging
from functools import wraps

def sampled_logging(sample_rate: float = 0.1):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            if random.random() < sample_rate:
                return await func(*args, **kwargs)
            return await func(*args, **kwargs)
        return wrapper
    return decorator

# Structured fields for querying
logger.info("tool_called",
    tool_name="search_web",
    tool_args={"query": "AI agents"},
    duration_ms=245,
    success=True,
    tokens_used=150,
    cost_usd=0.0003)

Log Aggregation (Loki)

# promtail-config.yaml
clients:
  - url: http://loki:3100/loki/api/v1/push

scrape_configs:
- job_name: agent-logs
  static_configs:
  - targets:
      - localhost
    labels:
      job: agent-service
      environment: production
      agent: agent-service
    labels:
      agent: agent-service
      environment: production

  pipeline_stages:
  - json:
      expressions:
        agent_id: agent_id
        session_id: session_id
        tool_name: tool_name
        duration_ms: duration_ms
        success: success
  - labels:
        agent_id:
        session_id:
        tool_name:
        success:

Metrics

Prometheus Metrics

from prometheus_client import Counter, Histogram, Gauge, Summary

# Counters
AGENT_REQUESTS_TOTAL = Counter(
    'agent_requests_total',
    'Total requests processed',
    ['agent', 'status', 'tool']
)

AGENT_TOOL_CALLS_TOTAL = Counter(
    'agent_tool_calls_total',
    'Total tool calls',
    ['agent', 'tool', 'status']
)

AGENT_ERRORS_TOTAL = Counter(
    'agent_errors_total',
    'Total errors',
    ['agent', 'error_type', 'tool']
)

# Histograms
AGENT_REQUEST_DURATION = Histogram(
    'agent_request_duration_seconds',
    'Request latency in seconds',
    ['agent', 'operation'],
    buckets=[0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10, 30, 60]
)

AGENT_TOKEN_USAGE = Histogram(
    'agent_tokens_total',
    'Tokens used',
    ['agent', 'type'],  # input, output, total
    buckets=[10, 50, 100, 500, 1000, 5000, 10000, 50000, 100000]
)

AGENT_COST_USD = Histogram(
    'agent_cost_usd',
    'Cost in USD',
    ['agent', 'operation'],
    buckets=[0.001, 0.005, 0.01, 0.05, 0.1, 0.5, 1, 5, 10, 50, 100]
)

# Gauges
AGENT_ACTIVE_SESSIONS = Gauge(
    'agent_active_sessions',
    'Currently active sessions',
    ['agent']
)

AGENT_QUEUE_DEPTH = Gauge(
    'agent_queue_depth',
    'Tasks waiting in queue',
    ['agent', 'queue']
)

AGENT_ACTIVE_CONNECTIONS = Gauge(
    'agent_active_connections',
    'Active WebSocket/HTTP connections',
    ['agent']
)

Custom Metrics Wrapper

class MetricsCollector:
    def __init__(self, agent_name: str):
        self.agent_name = agent_name

    @contextmanager
    def measure(self, operation: str, labels: dict = None):
        labels = labels or {}
        labels['agent'] = self.agent_name
        labels['operation'] = operation

        start = time.time()
        success = True
        try:
            yield
        except Exception as e:
            AGENT_ERRORS_TOTAL.labels(
                agent=self.agent_name,
                error_type=type(e).__name__,
                operation=operation
            ).inc()
            raise
        finally:
            duration = time.time() - start
            AGENT_REQUEST_DURATION.labels(**labels).observe(duration)
            AGENT_REQUESTS_TOTAL.labels(
                agent=self.agent_name,
                status='success' if success else 'error',
                operation=operation
            ).inc()

    def record_tokens(self, input_tokens: int, output_tokens: int, 
                      model: str = "gpt-4o"):
        AGENT_TOKEN_USAGE.labels(agent=self.agent_name, type='input').observe(input_tokens)
        AGENT_TOKEN_USAGE.labels(agent=self.agent_name, type='output').observe(output_tokens)
        AGENT_TOKEN_USAGE.labels(agent=self.agent_name, type='total').observe(input_tokens + output_tokens)

        # Cost calculation
        pricing = MODEL_PRICING.get(model, {"input": 0.01, "output": 0.03})
        cost = (input_tokens * pricing['input'] + output_tokens * pricing['output']) / 1000
        AGENT_COST_USD.labels(agent=self.agent_name, operation='llm').observe(cost)

Distributed Tracing

OpenTelemetry Setup

from opentelemetry import trace
from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export import BatchSpanProcessor
from opentelemetry.sdk.resources import Resource
from opentelemetry.instrumentation.openai import OpenAIInstrumentor
from opentelemetry.instrumentation.requests import RequestsInstrumentor

# Setup
resource = Resource.create({
    "service.name": "agent-service",
    "service.version": "1.2.3",
    "deployment.environment": "production"
})

provider = TracerProvider(resource=resource)
provider.add_span_processor(BatchSpanProcessor(
    OTLPSpanExporter(endpoint="http://tempo:4317", insecure=True)
))
trace.set_tracer_provider(provider)

# Auto-instrumentation
OpenAIInstrumentor().instrument()
RequestsInstrumentor().instrument()

# Get tracer
tracer = trace.get_tracer(__name__)

Custom Spans for Agent Operations

from opentelemetry.trace import SpanKind, Status, StatusCode

class TracedAgent:
    def __init__(self, name: str):
        self.name = name
        self.tracer = trace.get_tracer(__name__)

    async def run(self, input: str) -> str:
        with self.tracer.start_as_current_span(
            "agent.run",
            kind=SpanKind.SERVER,
            attributes={
                "agent.name": self.name,
                "input.length": len(input)
            }
        ) as span:
            try:
                # Planning
                with self.tracer.start_as_current_span("agent.plan") as plan_span:
                    plan = await self.plan(input)
                    plan_span.set_attribute("plan.steps", len(plan.steps))

                # Execution
                with self.tracer.start_as_current_span("agent.execute") as exec_span:
                    result = await self.execute(plan)
                    exec_span.set_attribute("result.length", len(result))

                span.set_status(Status(StatusCode.OK))
                return result
            except Exception as e:
                span.record_exception(e)
                span.set_status(Status(StatusCode.ERROR, str(e)))
                raise

    @contextmanager
    def tool_span(self, tool_name: str, args: dict):
        with self.tracer.start_as_current_span(
            f"tool.{tool_name}",
            kind=SpanKind.CLIENT,
            attributes={
                "tool.name": tool_name,
                "tool.args": json.dumps(args)[:1000]
            }
        ) as span:
            try:
                yield span
                span.set_status(Status(StatusCode.OK))
            except Exception as e:
                span.record_exception(e)
                span.set_status(Status(StatusCode.ERROR, str(e)))
                raise

Context Propagation

# Inject context into outgoing requests
from opentelemetry.propagate import inject, extract
from opentelemetry.propagators.textmap import TextMapPropagator

def inject_context(headers: dict) -> dict:
    inject(headers)
    return headers

def extract_context(headers: dict) -> Context:
    return extract(headers)

# In HTTP client
async def make_request(url: str, data: dict, headers: dict = None):
    headers = headers or {}
    inject(headers)  # Inject trace context

    async with httpx.AsyncClient() as client:
        response = await client.post(url, json=data, headers=headers)
        return response

Alerting

Alert Rules (Prometheus)

# alerts.yml
groups:
- name: agent-alerts
  rules:
  # High error rate
  - alert: AgentHighErrorRate
    expr: |
      sum(rate(agent_errors_total[5m])) by (agent) 
      / sum(rate(agent_requests_total[5m])) by (agent) > 0.05
    for: 2m
    labels:
      severity: critical
    annotations:
      summary: "High error rate for {{ $labels.agent }}"
      description: "Error rate is {{ $value | humanizePercentage }}"

  # High latency
  - alert: AgentHighLatency
    expr: |
      histogram_quantile(0.95, 
        sum(rate(agent_request_duration_seconds_bucket[5m])) by (agent, le)
      ) > 10
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "High latency for {{ $labels.agent }}"
      description: "P95 latency is {{ $value }}s"

  # Queue backlog
  - alert: AgentQueueBacklog
    expr: |
      agent_queue_depth > 1000
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "Queue backlog for {{ $labels.agent }}"

  # High cost
  - alert: AgentHighCost
    expr: |
      sum(rate(agent_cost_usd_sum[1h])) by (agent) > 10
    for: 10m
    labels:
      severity: warning
    annotations:
      summary: "High cost for {{ $labels.agent }}"
      description: "Hourly cost is ${{ $value }}"

  # Low success rate
  - alert: AgentLowSuccessRate
    expr: |
      sum(rate(agent_requests_total{status="success"}[5m])) by (agent)
      / sum(rate(agent_requests_total[5m])) by (agent) < 0.95
    for: 5m
    labels:
      severity: critical
    annotations:
      summary: "Low success rate for {{ $labels.agent }}"

Alertmanager Config

# alertmanager.yml
global:
  resolve_timeout: 5m
  smtp_smarthost: 'smtp.gmail.com:587'
  smtp_from: 'alerts@example.com'
  smtp_auth_username: 'alerts@example.com'
  smtp_auth_password: '${SMTP_PASSWORD}'

route:
  group_by: ['alertname', 'agent']
  group_wait: 30s
  group_interval: 5m
  repeat_interval: 4h
  receiver: 'default'
  routes:
  - match:
      severity: critical
    receiver: 'critical'
    continue: true
  - match:
      severity: warning
    receiver: 'warning'

receivers:
- name: 'default'
  email_configs:
  - to: 'team@example.com'
    send_resolved: true
- name: 'critical'
  pagerduty_configs:
  - service_key: '${PAGERDUTY_KEY}'
    severity: critical
- name: 'warning'
  slack_configs:
  - api_url: '${SLACK_WEBHOOK}'
    channel: '#alerts-warning'
    send_resolved: true
    title: '{{ .GroupLabels.alertname }}'
    text: '{{ range .Alerts }}{{ .Annotations.summary }}\n{{ end }}'

Debugging & Profiling

Profiling

# PySpy for CPU profiling
# pip install py-spy
# py-spy record -o profile.svg --pid <PID>

# Or programmatically
import pyinstrument

profiler = pyinstrument.Profiler()
profiler.start()

# Your agent code here
result = await agent.run(input)

profiler.stop()
profiler.print()
profiler.write_html("profile.html")

Memory Profiling

# Memory profiling with memory_profiler
from memory_profiler import profile

@profile
async def memory_intensive_function():
    # Your code
    pass

# Or with objgraph
import objgraph

def count_objects():
    print(f"Total objects: {objgraph.count()}")
    print("Top types:", objgraph.most_common_types(limit=10))
    objgraph.show_most_common_types(limit=10)

Memory Leak Detection

import tracemalloc

tracemalloc.start()

# ... your agent code ...

snapshot1 = tracemalloc.take_snapshot()
# ... run agent ...
snapshot2 = tracemalloc.take_snapshot()

top_stats = snapshot2.compare_to(snapshot1, 'lineno')
for stat in top_stats[:10]:
    print(stat)

Log Analysis Queries

Loki Queries (LogQL)

# Error rate by agent
sum by (agent) (rate({job="agent-service"} |= "ERROR" [5m]))

# Latency percentiles
histogram_quantile(0.95, sum(rate(agent_request_duration_seconds_bucket[5m])) by (le, agent))

# Tool usage
sum by (tool) (rate(agent_tool_calls_total[5m]))

# Error patterns
{job="agent-service"} |= "ERROR" | json | level="error" | tool_name != ""

# Slow queries
{job="agent-service"} | json | duration_ms > 5000

# Specific tool errors
{job="agent-service"} | json | tool_name="search_web" | level="error"

# Session tracking
{job="agent-service"} | json | session_id="abc123"

# Rate limited
{job="agent-service"} | json | status_code="429"

Dashboards (Grafana)

Agent Overview Dashboard

{
  "dashboard": {
    "title": "Agent Service Overview",
    "panels": [
      {
        "title": "Request Rate",
        "type": "graph",
        "targets": [{
          "expr": "sum(rate(agent_requests_total[5m])) by (agent)",
          "legendFormat": "{{agent}}"
        }]
      },
      {
        "title": "Error Rate",
        "targets": [{
          "expr": "sum(rate(agent_errors_total[5m])) by (agent) / sum(rate(agent_requests_total[5m])) by (agent)",
          "legendFormat": "{{agent}}"
        }]
      },
      {
        "title": "Latency (P50, P95, P99)",
        "targets": [
          {"expr": "histogram_quantile(0.50, sum(rate(agent_request_duration_seconds_bucket[5m])) by (le, agent))", "legendFormat": "P50 {{agent}}"},
          {"expr": "histogram_quantile(0.95, sum(rate(agent_request_duration_seconds_bucket[5m])) by (le, agent))", "legendFormat": "P95 {{agent}}"},
          {"expr": "histogram_quantile(0.99, sum(rate(agent_request_duration_seconds_bucket[5m])) by (le, agent))", "legendFormat": "P99 {{agent}}"}
        ]
      },
      {
        "title": "Token Usage & Cost",
        "targets": [
          {"expr": "sum(rate(agent_tokens_total{type=\"total\"}[5m])) by (agent)", "legendFormat": "Tokens/sec {{agent}}"},
          {"expr": "sum(rate(agent_cost_usd_sum[1h])) by (agent)", "legendFormat": "Cost/hr {{agent}}"}
        ]
      },
      {
        "title": "Queue Depth",
        "targets": [{
          "expr": "agent_queue_depth",
          "legendFormat": "{{agent}}"
        }]
      },
      {
        "title": "Tool Usage",
        "targets": [{
          "expr": "sum(rate(agent_tool_calls_total[5m])) by (tool, agent)",
          "legendFormat": "{{agent}} - {{tool}}"
        }]
      }
    ]
  }
}

Related Notes

• Deployment Architecture — Deployment patterns
• Scaling Strategies — Scaling strategies
• Reliability — Circuit breakers, retries
• Security — Security
• Cost Optimization — Cost optimization
• Compliance — Compliance
• Deployment Architecture — Deployment
• Observability — DevOps observability
• CI CD Best Practices — CI/CD
• Infrastructure as Code — IaC
• Supply Chain Security — Supply chain security
• Reliability — Reliability
• Cost Optimization — Cost optimization
• Compliance — Compliance
• Research Directions — Observability research

Sources: Prometheus, Grafana, Loki, Tempo, Jaeger, OpenTelemetry, OpenTelemetry Python, PyInstrument, py-spy, memory_profiler, objgraph, tracemalloc, Prometheus Client, Grafana, Loki, Tempo, Jaeger, Zipkin, Zipkin, Zipkin, Alertmanager, PagerDuty, Opsgenie, Grafana OnCall, Prometheus Alerting, Prometheus Recording Rules, Prometheus Federation, Thanos, Cortex, Mimir, VictoriaMetrics, Grafana Loki, Grafana Tempo, Grafana Pyroscope, Grafana Faro, Grafana Beyla, eBPF, bpftrace, perf, perfetto, Sentry, Datadog, New Relic, Honeycomb, Lightstep, Elastic APM, Stackdriver.