agent-health-monitoring

Agent Health Monitoring & Alerting

Overview

Production multi-agent systems fail silently. An agent that stops responding, returns empty results, or enters an infinite loop can degrade an entire workflow without triggering traditional infrastructure alerts. This skill covers how to build comprehensive health monitoring, metrics collection, and alerting for AI agent fleets.

---

Core Concepts

Agent Vital Signs

Metric	What It Measures	Why It Matters
Response Rate	% of agent invocations that return a result	Dropping rate indicates crashes or context overflows
Latency (P50/P95/P99)	Time from invocation to response	Spikes indicate context bloat or degraded model performance
Error Rate	% of invocations with errors/tool failures	Rising rate indicates systemic issues
Step Count	Number of reasoning steps per task	Unbounded growth indicates looping behavior
Tool Call Success Rate	% of tool calls that succeed	Drop indicates broken integrations or rate limiting
Token Consumption	Tokens used per agent run	Budget anomalies indicate runaway agents
Context Utilization	% of context window used	High utilization risks truncation and quality loss
Hallucination Score	Confidence calibration or factuality checks	Degrading accuracy undermines trust

Alert Severity Levels

Level	Color	Response Time	Examples
P0 (Critical)	🔴 Red	< 5 min	Agent completely down, data loss, security breach
P1 (High)	🟠 Orange	< 15 min	Error rate > 20%, latency 5x baseline
P2 (Medium)	🟡 Yellow	< 1 hour	Error rate > 5%, slow degradation
P3 (Low)	🔵 Blue	< 24 hours	Single agent underperforming, minor drift

---

Step-by-Step Implementation

Step 1: Instrument Every Agent

Wrap every agent invocation with telemetry:

class MonitoredAgent:
    """Agent wrapper that collects metrics on every invocation."""
    
    def __init__(self, agent, agent_name: str, metrics_client):
        self.agent = agent
        self.agent_name = agent_name
        self.metrics = metrics_client
    
    async def run(self, task: str) -> str:
        start_time = time.time()
        step_count = 0
        token_usage = 0
        
        try:
            result = await self.agent.run(task)
            
            # Collect metrics
            duration = time.time() - start_time
            self.metrics.timing(f"agent.{self.agent_name}.latency", duration)
            self.metrics.increment(f"agent.{self.agent_name}.invocations")
            self.metrics.increment(f"agent.{self.agent_name}.success")
            self.metrics.gauge(f"agent.{self.agent_name}.steps", step_count)
            
            return result
            
        except Exception as e:
            duration = time.time() - start_time
            self.metrics.increment(f"agent.{self.agent_name}.errors")
            self.metrics.timing(f"agent.{self.agent_name}.error_latency", duration)
            raise