Agent Performance Monitoring

Why This Matters

Without proper monitoring, you're flying blind. Performance monitoring enables you to detect issues before users complain, understand bottlenecks, measure quality, and optimize costs. In production AI systems, monitoring is not optional—it's essential for maintaining service reliability and user trust.

Real-World Analogy

Performance monitoring is like a car's dashboard. The speedometer (latency), fuel gauge (cost), temperature warning (error rate), and check engine light (health checks) all provide critical information. Without these instruments, you wouldn't know if you're about to run out of gas or if the engine is overheating until it's too late.

Monitoring Architecture

Metrics Collector Implementation

import time
import asyncio
import logging
from dataclasses import dataclass, field
from typing import Any, Callable
from collections import defaultdict
import statistics
import json

logger = logging.getLogger(__name__)

@dataclass
class MetricPoint:
    name: str
    value: float
    timestamp: float = field(default_factory=time.time)
    tags: dict[str, str] = field(default_factory=dict)
    metric_type: str = "gauge"

class MetricsCollector:
    def __init__(self, flush_interval: float = 10.0):
        self.metrics: list[MetricPoint] = []
        self.counters: dict[str, float] = defaultdict(float)
        self.gauges: dict[str, float] = {}
        self.histograms: dict[str, list[float]] = defaultdict(list)
        self.flush_interval = flush_interval
        self._running = False

    def increment(self, name: str, value: float = 1.0, tags: dict = None):
        key = f"{name}:{json.dumps(tags or {}, sort_keys=True)}"
        self.counters[key] += value
        self.metrics.append(MetricPoint(name, self.counters[key], tags=tags or {}, metric_type="counter"))

    def gauge(self, name: str, value: float, tags: dict = None):
        key = f"{name}:{json.dumps(tags or {}, sort_keys=True)}"
        self.gauges[key] = value
        self.metrics.append(MetricPoint(name, value, tags=tags or {}, metric_type="gauge"))

    def histogram(self, name: str, value: float, tags: dict = None):
        key = f"{name}:{json.dumps(tags or {}, sort_keys=True)}"
        self.histograms[key].append(value)
        self.metrics.append(MetricPoint(name, value, tags=tags or {}, metric_type="histogram"))

    def timer(self, name: str, tags: dict = None):
        return TimerContext(self, name, tags)

    def get_histogram_stats(self, name: str, tags: dict = None) -> dict:
        key = f"{name}:{json.dumps(tags or {}, sort_keys=True)}"
        values = self.histograms.get(key, [])
        if not values:
            return {}
        return {
            "count": len(values),
            "min": min(values),
            "max": max(values),
            "mean": statistics.mean(values),
            "median": statistics.median(values),
            "p95": sorted(values)[int(len(values) * 0.95)] if len(values) > 20 else max(values),
            "p99": sorted(values)[int(len(values) * 0.99)] if len(values) > 100 else max(values),
        }

    def flush(self) -> list[MetricPoint]:
        batch = self.metrics.copy()
        self.metrics.clear()
        return batch

    def get_summary(self) -> dict:
        return {
            "total_metrics": len(self.metrics),
            "counters": len(self.counters),
            "gauges": len(self.gauges),
            "histograms": len(self.histograms),
        }

class TimerContext:
    def __init__(self, collector: MetricsCollector, name: str, tags: dict = None):
        self.collector = collector
        self.name = name
        self.tags = tags
        self.start_time: float = 0

    async def __aenter__(self):
        self.start_time = time.time()
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        duration = time.time() - self.start_time
        self.collector.histogram(self.name, duration, self.tags)
        return False

    def __enter__(self):
        self.start_time = time.time()
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        duration = time.time() - self.start_time
        self.collector.histogram(self.name, duration, self.tags)
        return False

Distributed Tracing

import uuid
import time
from dataclasses import dataclass, field
from typing import Any, Optional
import json

@dataclass
class Span:
    trace_id: str
    span_id: str = field(default_factory=lambda: str(uuid.uuid4())[:8])
    parent_id: Optional[str] = None
    name: str = ""
    start_time: float = field(default_factory=time.time)
    end_time: float = 0.0
    attributes: dict[str, Any] = field(default_factory=dict)
    events: list[dict] = field(default_factory=list)
    status: str = "OK"

    @property
    def duration(self) -> float:
        return self.end_time - self.start_time

    def finish(self):
        self.end_time = time.time()

    def set_attribute(self, key: str, value: Any):
        self.attributes[key] = value

    def add_event(self, name: str, attributes: dict = None):
        self.events.append({
            "name": name,
            "timestamp": time.time(),
            "attributes": attributes or {},
        })

    def to_dict(self) -> dict:
        return {
            "trace_id": self.trace_id,
            "span_id": self.span_id,
            "parent_id": self.parent_id,
            "name": self.name,
            "start_time": self.start_time,
            "end_time": self.end_time,
            "duration": self.duration,
            "attributes": self.attributes,
            "events": self.events,
            "status": self.status,
        }

class Tracer:
    def __init__(self, service_name: str = "agent"):
        self.service_name = service_name
        self.spans: list[Span] = []
        self.active_span: Optional[Span] = None

    def start_trace(self, name: str, attributes: dict = None) -> Span:
        trace_id = str(uuid.uuid4())
        span = Span(trace_id=trace_id, name=name, attributes=attributes or {})
        self.spans.append(span)
        self.active_span = span
        return span

    def start_span(self, name: str, parent: Span = None, attributes: dict = None) -> Span:
        parent = parent or self.active_span
        span = Span(
            trace_id=parent.trace_id if parent else str(uuid.uuid4()),
            parent_id=parent.span_id if parent else None,
            name=name,
            attributes=attributes or {},
        )
        self.spans.append(span)
        self.active_span = span
        return span

    def finish_span(self, span: Span):
        span.finish()
        if self.active_span and self.active_span.span_id == span.span_id:
            self.active_span = None

    def get_trace(self, trace_id: str) -> list[Span]:
        return [s for s in self.spans if s.trace_id == trace_id]

    def export_trace(self, trace_id: str) -> dict:
        spans = self.get_trace(trace_id)
        return {
            "trace_id": trace_id,
            "spans": [s.to_dict() for s in spans],
            "total_duration": max(s.end_time for s in spans) - min(s.start_time for s in spans) if spans else 0,
        }

Alert Manager

import asyncio
import time
from dataclasses import dataclass, field
from typing import Any, Callable, Coroutine
from enum import Enum
import logging

logger = logging.getLogger(__name__)

class AlertSeverity(Enum):
    CRITICAL = "critical"
    WARNING = "warning"
    INFO = "info"

@dataclass
class AlertRule:
    name: str
    condition: Callable[[], bool]
    severity: AlertSeverity
    message: str = ""
    cooldown: float = 300.0
    last_triggered: float = 0.0
    enabled: bool = True

@dataclass
class Alert:
    rule_name: str
    severity: AlertSeverity
    message: str
    timestamp: float = field(default_factory=time.time)
    resolved: bool = False

class AlertManager:
    def __init__(self):
        self.rules: list[AlertRule] = []
        self.alerts: list[Alert] = []
        self.handlers: dict[AlertSeverity, list[Callable]] = {
            sev: [] for sev in AlertSeverity
        }

    def add_rule(self, rule: AlertRule):
        self.rules.append(rule)

    def register_handler(self, severity: AlertSeverity, handler: Callable[[Alert], Coroutine]):
        self.handlers[severity].append(handler)

    async def evaluate(self):
        for rule in self.rules:
            if not rule.enabled:
                continue
            if time.time() - rule.last_triggered < rule.cooldown:
                continue
            try:
                if rule.condition():
                    alert = Alert(
                        rule_name=rule.name,
                        severity=rule.severity,
                        message=rule.message,
                    )
                    self.alerts.append(alert)
                    rule.last_triggered = time.time()
                    await self._notify_handlers(alert)
            except Exception as e:
                logger.error(f"Error evaluating rule {rule.name}: {e}")

    async def _notify_handlers(self, alert: Alert):
        handlers = self.handlers.get(alert.severity, [])
        for handler in handlers:
            try:
                await handler(alert)
            except Exception as e:
                logger.error(f"Error in alert handler: {e}")

    def get_active_alerts(self) -> list[Alert]:
        return [a for a in self.alerts if not a.resolved]

    def resolve_alert(self, rule_name: str):
        for alert in reversed(self.alerts):
            if alert.rule_name == rule_name and not alert.resolved:
                alert.resolved = True
                break

Mathematical Foundation

Latency Percentile Calculation:

Where:

— k-th percentile latency
— Sorted latency values
— Total number of observations

Error Rate Calculation:

SLA Compliance:

Apdex Score (Application Performance Index):

Where:

— Requests with response time ≤ T
— Requests with response time ≤ 4T
— Target response time

Mean Time Between Failures:

Performance Considerations

Component	Latency Impact	Cost Impact	Accuracy Impact
Metrics Collection	+0.1-1ms per metric	+Storage	N/A
Distributed Tracing	+1-5ms per trace	+Storage	N/A
Logging	+0.5-2ms per log	+Storage	N/A
Alerting	+10-50ms evaluation	+Compute	N/A

Security Considerations

Metric data: Contains operational data, protect from unauthorized access
Trace data: May contain sensitive request details, implement sampling
Log data: Mask sensitive information (PII, credentials)
Alert routing: Ensure alerts go to authorized channels only
Access control: Restrict monitoring dashboard access

Interview Questions

1. What are the four golden signals of monitoring?

Answer: The four golden signals (from Google SRE): 1) Latency — Time to serve a request, 2) Traffic — Demand placed on the system, 3) Errors — Rate of failed requests, 4) Saturation — How "full" the system is. For agents, add: 5) Quality — Task completion rate, 6) Cost — Token/API usage. These provide a comprehensive view: latency and errors measure user experience, traffic and saturation measure capacity, quality measures effectiveness, cost measures efficiency.

2. How do you implement distributed tracing for multi-agent systems?

Answer: Use OpenTelemetry or Jaeger: 1) Generate trace ID at entry point, 2) Propagate trace context across agent calls (HTTP headers or message metadata), 3) Create spans for each agent operation, 4) Record timing, attributes, and events, 5) Export to trace store, 6) Visualize in Jaeger/Zipkin. Key: ensure trace context propagation across async boundaries, message queues, and external API calls. Use span links for parallel operations. Implement sampling to control trace volume.

3. What is the difference between metrics, logs, and traces?

Answer: Metrics are numerical measurements over time (latency, error rate, throughput)—efficient for aggregation and alerting. Logs are discrete events with context (error messages, debug info)—useful for debugging specific issues. Traces follow a request through the system—essential for understanding distributed workflows. Together they form the three pillars of observability. Correlate them: use trace IDs to link logs to traces, and metric labels to identify which traces to examine.

4. How do you design effective alerting rules?

Answer: Principles: 1) Alert on symptoms, not causes (latency high, not CPU high), 2) Use multi-window alerting (short window for detection, long for confirmation), 3) Set appropriate thresholds based on baseline, 4) Include runbooks with actionable steps, 5) Implement cooldown periods to prevent alert storms, 6) Use severity levels (critical/warning/info), 7) Test alerts regularly, 8) Monitor alert fatigue. For agents: alert on error rate spikes, latency degradation, quality drops, and cost anomalies.

5. How would you implement cost monitoring for LLM-based agents?

Answer: Track: 1) Token usage — Input/output tokens per request, 2) Model costs — Price per token for each model, 3) Cost per request — Total cost divided by requests, 4) Cost by feature — Which features consume most tokens, 5) Budget alerts — Warn when approaching spending limits. Implementation: instrument LLM calls to log token counts, aggregate by feature/model/time, calculate costs using pricing data, set up alerts for anomalies. Use caching to reduce redundant calls and token optimization to minimize waste.

6. What is Apdex and why is it useful for agent monitoring?

Answer: Apdex (Application Performance Index) converts response time measurements into a uniform score between 0 and 1. It classifies responses as satisfied (fast), tolerating (acceptable), or failed (slow). Formula: (satisfied + tolerating/2) / total. Useful because: 1) Single number summarizes performance, 2) User-centric (based on experience, not technical metrics), 3) Easy to track over time, 4) Comparable across services. For agents: set target response time based on use case (e.g., 1s for chat, 5s for analysis).

7. How do you implement health checks for agent systems?

Answer: Health checks verify the system is functioning correctly: 1) Liveness — Is the agent running? Check process status, memory usage, 2) Readiness — Can it accept requests? Check dependencies (LLM API, databases), 3) Deep health — Is it functioning correctly? Run synthetic tests. Implement: HTTP endpoints (/health, /ready), periodic self-tests, dependency health aggregation. Use circuit breakers to detect degraded dependencies. Return detailed health status for debugging. Alert on health check failures.

8. How would you perform capacity planning for agent systems?

Answer: Steps: 1) Baseline — Measure current resource usage and performance, 2) Projection — Forecast growth based on usage trends, 3) Model — Estimate capacity needs using queuing theory, 4) Test — Load test to validate assumptions, 5) Plan — Define scaling policies and thresholds. For agents: consider token generation rate limits, API rate limits, concurrent request capacity, and cost constraints. Use queuing theory (M/M/1, M/M/c models) to predict latency under load. Implement auto-scaling based on queue depth and latency metrics.

Common Pitfalls

Pitfall	Solution
Alert fatigue from too many alerts	Use severity levels and alert on symptoms
Missing trace context in async calls	Propagate context through message metadata
High monitoring overhead	Use sampling and batch exports
No baseline for comparison	Establish performance baselines in staging
Ignoring cost metrics	Track token usage and costs per feature
Incomplete health checks	Include deep health checks with synthetic tests
Monitoring without action	Include runbooks and escalation procedures
Siloed observability	Correlate metrics, logs, and traces

KnowledgeCheck

What are the four golden signals of monitoring?
- a) CPU, Memory, Disk, Network
- b) Latency, Traffic, Errors, Saturation
- c) Logs, Metrics, Traces, Alerts
- d) Request, Response, Error, Timeout
What is the purpose of distributed tracing?
- a) To track individual function calls
- b) To follow a request across multiple services/agents
- c) To measure CPU usage
- d) To log all application events
How does Apdex score performance?
- a) As a percentage of successful requests
- b) As a score between 0 and 1 based on response time
- c) As the average response time
- d) As the number of errors per request
What should health checks verify?
- a) Only that the process is running
- b) Liveness, readiness, and deep functionality
- c) Only that the database is accessible
- d) Only that the API is responding
Why is cost monitoring important for LLM agents?
- a) To reduce code complexity
- b) To track token usage and prevent overspending
- c) To increase response quality
- d) To simplify deployment
What is the benefit of correlating metrics, logs, and traces?
- a) Reduces storage requirements
- b) Enables faster debugging by linking related data
- c) Simplifies the monitoring stack
- d) Reduces alert noise

Answers: 1-b, 2-b, 3-b, 4-b, 5-b, 6-b

Agent Performance Monitoring: Metrics, Logging, Tracing & Alerting