引言 Agent系统的监控告警比传统应用复杂一个量级——除了需要监控CPU、内存、延迟等基础设施指标外，还需要监控Token消耗、工具成功率、幻觉率、安全违规率等Agent特有指标。一个没有完善监控的Agent系统就像盲飞——出了问题不知道哪里出了问题，没出问题不知道什么时候会出问题。
2026年，Prometheus + Grafana + AlertManager已成为Agent监控告警的事实标准，但Agent系统需要在此基础上构建专门的监控体系。
指标体系设计 四层指标架构 ┌─────────────────────────────────────────────────────┐ │ Layer 4: Business Metrics │ │ 用户满意度、任务完成率、对话质量评分 │ ├─────────────────────────────────────────────────────┤ │ Layer 3: Agent Metrics │ │ Token消耗、工具调用成功率、幻觉率、安全违规率 │ ├─────────────────────────────────────────────────────┤ │ Layer 2: Application Metrics │ │ 请求QPS、响应延迟、错误率、并发会话数 │ ├─────────────────────────────────────────────────────┤ │ Layer 1: Infrastructure Metrics │ │ CPU、内存、GPU利用率、磁盘IO、网络流量 │ └─────────────────────────────────────────────────────┘ Agent核心指标定义 from prometheus_client import Counter, Histogram, Gauge, Summary # ===== Layer 3: Agent特有指标 ===== # Token消耗 token_usage = Counter( "agent_token_total", "Total tokens consumed", ["tenant_id", "model", "type"] # type: input/output ) # 工具调用 tool_calls = Counter( "agent_tool_calls_total", "Total tool calls", ["tool_name", "status"] # status: success/failed/timeout ) tool_latency = Histogram( "agent_tool_latency_seconds", "Tool execution latency", ["tool_name"], buckets=[0.1, 0.5, 1, 5, 10, 30, 60] ) # Agent质量 hallucination_rate = Gauge( "agent_hallucination_rate", "Hallucination rate (rolling 1h)", ["model"] ) safety_violations = Counter( "agent_safety_violations_total", "Safety violations detected", ["type", "severity"] ) # 循环检测 cycle_detections = Counter( "agent_cycle_detections_total", "Cycle detections", ["cycle_type", "resolution"] # resolution: broken/escalated ) # 会话指标 active_sessions = Gauge( "agent_active_sessions", "Active sessions", ["tenant_id"] ) session_duration = Histogram( "agent_session_duration_seconds", "Session duration", buckets=[10, 30, 60, 120, 300, 600, 1200] ) # 路由决策 routing_decisions = Counter( "agent_routing_decisions_total", "Routing decisions", ["source_model", "target_model", "reason"] ) Prometheus配置 # prometheus.yml global: scrape_interval: 15s evaluation_interval: 15s rule_files: - "agent_alerts.yml" scrape_configs: # Agent应用指标 - job_name: "agent-service" metrics_path: /metrics static_configs: - targets: ["agent-service:9090"] labels: service: "agent" # LLM推理服务 - job_name: "llm-inference" metrics_path: /metrics static_configs: - targets: ["llm-inference:9090"] # 工具执行服务 - job_name: "tool-executor" metrics_path: /metrics static_configs: - targets: ["tool-executor-0:9090", "tool-executor-1:9090"] 告警规则 # agent_alerts.yml groups: - name: agent_infra_alerts rules: # 高错误率 - alert: AgentHighErrorRate expr: | sum(rate(agent_requests_total{status="error"}[5m])) by (service) / sum(rate(agent_requests_total[5m])) by (service) > 0.05 for: 2m labels: severity: critical team: agent-platform annotations: summary: "Agent error rate > 5%" description: "{{ $labels.service }} error rate is {{ $value | humanizePercentage }}" # P99延迟过高 - alert: AgentHighLatency expr: | histogram_quantile(0.99, rate(agent_request_duration_seconds_bucket[5m]) ) > 5 for: 5m labels: severity: warning annotations: summary: "Agent P99 latency > 5s" - name: agent_quality_alerts rules: # 幻觉率过高 - alert: AgentHighHallucination expr: agent_hallucination_rate > 0.05 for: 10m labels: severity: warning team: agent-quality annotations: summary: "Hallucination rate > 5% for {{ $labels.model }}" # 安全违规 - alert: AgentSafetyViolation expr: increase(agent_safety_violations_total[1h]) > 0 labels: severity: critical team: security annotations: summary: "Safety violation detected" # 循环检测频繁 - alert: AgentFrequentCycles expr: | increase(agent_cycle_detections_total[1h]) > 10 for: 5m labels: severity: warning annotations: summary: "Frequent cycle detections (>10/hour)" - name: agent_cost_alerts rules: # Token消耗异常 - alert: AgentTokenSpike expr: | rate(agent_token_total[5m]) > 2 * avg_over_time(rate(agent_token_total[5m])[1h:5m]) for: 10m labels: severity: warning team: agent-platform annotations: summary: "Token consumption spiked 2x above average" # 工具调用失败率 - alert: ToolFailureRate expr: | sum(rate(agent_tool_calls_total{status="failed"}[5m])) by (tool_name) / sum(rate(agent_tool_calls_total[5m])) by (tool_name) > 0.1 for: 5m labels: severity: warning annotations: summary: "Tool {{ $labels.tool_name }} failure rate > 10%" 告警路由与通知 class AlertRouter: """告警路由器""" ROUTING_RULES = { "critical": { "channels": ["pagerduty", "slack:#oncall", "sms"], "escalation_delay_min": 5, "escalation_target": "team-lead", }, "warning": { "channels": ["slack:#alerts"], "escalation_delay_min": 30, "escalation_target": "secondary-oncall", }, "info": { "channels": ["slack:#monitoring"], "escalation_delay_min": None, "escalation_target": None, } } async def handle_alert(self, alert: dict): """处理告警""" severity = alert["labels"]["severity"] rule = self.ROUTING_RULES[severity] # 告警去重 if await self._is_duplicate(alert): logger.debug(f"Duplicate alert suppressed: {alert['fingerprint']}") return # 告警分组 group_key = self._get_group_key(alert) group = await self._get_or_create_group(group_key) group.add_alert(alert) # 发送通知 for channel in rule["channels"]: await self._send_notification(channel, group) # 设置升级定时器 if rule["escalation_delay_min"]: asyncio.create_task( self._schedule_escalation( group, rule["escalation_delay_min"], rule["escalation_target"] ) ) 告警治理 class AlertGovernance: """告警治理——防止告警风暴""" def __init__(self): self.suppression_rules = [] self.rate_limits = {} def should_send(self, alert: dict) -> bool: """判断告警是否应该发送""" # 1. 维护窗口抑制 if self._in_maintenance_window(alert): return False # 2. 依赖抑制——如果上游告警活跃，抑制下游 if self._suppressed_by_dependency(alert): return False # 3. 频率限制——同一告警5分钟内只发一次 alert_key = alert["fingerprint"] if alert_key in self.rate_limits: last_sent = self.rate_limits[alert_key] if (datetime.now() - last_sent).total_seconds() < 300: return False # 4. 告警噪音评分 noise_score = self._calculate_noise_score(alert) if noise_score < 0.3: return False self.rate_limits[alert_key] = datetime.now() return True Grafana仪表板 { "dashboard": { "title": "Agent System Overview", "panels": [ { "title": "Request Rate & Error Rate", "targets": [ { "expr": "sum(rate(agent_requests_total[5m]))", "legendFormat": "QPS" }, { "expr": "sum(rate(agent_requests_total{status=\"error\"}[5m])) / sum(rate(agent_requests_total[5m]))", "legendFormat": "Error Rate" } ] }, { "title": "Token Consumption by Model", "targets": [ { "expr": "sum(rate(agent_token_total[5m])) by (model)", "legendFormat": "{{model}}" } ] }, { "title": "Tool Success Rate", "targets": [ { "expr": "sum(rate(agent_tool_calls_total{status=\"success\"}[5m])) by (tool_name) / sum(rate(agent_tool_calls_total[5m])) by (tool_name)", "legendFormat": "{{tool_name}}" } ] }, { "title": "Active Sessions & Concurrency", "targets": [ { "expr": "agent_active_sessions", "legendFormat": "{{tenant_id}}" } ] } ] } } SLI/SLO定义 # Agent系统SLO定义 slo: availability: target: 99.9% window: 30d query: | 1 - (sum(rate(agent_requests_total{status="error"}[5m])) / sum(rate(agent_requests_total[5m]))) latency_p99: target: 2000ms window: 7d query: | histogram_quantile(0.99, rate(agent_request_duration_seconds_bucket[5m])) quality_score: target: 0.85 window: 7d query: | avg(agent_response_quality_score) safety: target: 99.99% window: 30d query: | 1 - (increase(agent_safety_violations_total[30d]) / sum(increase(agent_requests_total[30d]))) 总结 Agent系统的监控告警需要覆盖从基础设施到业务质量的四个层次。指标设计要全面但不过载，告警规则要精准且有层次，通知路由要高效且不产生噪音。告警治理是长期工作——定期回顾告警有效性，淘汰无用告警，优化有用告警。
...
