Prometheus

引言 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系统的监控告警需要覆盖从基础设施到业务质量的四个层次。指标设计要全面但不过载，告警规则要精准且有层次，通知路由要高效且不产生噪音。告警治理是长期工作——定期回顾告警有效性，淘汰无用告警，优化有用告警。 ...

监控分层架构 AI 应用监控需要三层视角，每层关注不同问题： ┌─────────────────────────────────────────────────┐ │ Layer 3: 模型质量监控 │ │ - 回答正确率 / 幻觉率 / 毒性检测 / 偏见 │ │ - 用户满意度反馈 │ ├─────────────────────────────────────────────────┤ │ Layer 2: 应用性能监控 │ │ - 首字延迟 (TTFT) / 完整延迟 / 吞吐量 │ │ - Token 消耗 / 成本 / 缓存命中率 │ ├─────────────────────────────────────────────────┤ │ Layer 1: 基础设施监控 │ │ - GPU 利用率 / 内存 / 网络 / 磁盘 │ │ - API 错误率 / 限流次数 / 超时 │ └─────────────────────────────────────────────────┘ 指标设计 核心指标清单 层级 指标 类型 告警阈值 模型 回答质量评分 Gauge < 0.8 模型 幻觉率 Gauge > 5% 模型 用户点赞率 Gauge < 70% 应用 TTFT (首字延迟) Histogram P99 > 2s 应用 完整延迟 Histogram P99 > 15s 应用 Token 消耗 Counter 日预算 120% 应用 缓存命中率 Gauge < 15% 基础设施 GPU 利用率 Gauge > 90% 基础设施 API 错误率 Gauge > 1% 基础设施 限流次数 Counter > 100/min Prometheus 指标定义 from prometheus_client import Counter, Histogram, Gauge, Summary # === 模型质量指标 === llm_quality_score = Gauge( "llm_quality_score", "Average answer quality score (0-1)", ["model", "task_type"] ) llm_hallucination_rate = Gauge( "llm_hallucination_rate", "Hallucination rate detected", ["model"] ) llm_user_feedback = Counter( "llm_user_feedback_total", "User feedback counts", ["model", "feedback"] # feedback: positive/negative ) # === 应用性能指标 === llm_ttft = Histogram( "llm_time_to_first_token_seconds", "Time to first token in seconds", ["model"], buckets=[0.1, 0.3, 0.5, 1, 2, 5, 10] ) llm_total_latency = Histogram( "llm_total_latency_seconds", "Total request latency in seconds", ["model"], buckets=[0.5, 1, 2, 5, 10, 20, 30, 60] ) llm_tokens_total = Counter( "llm_tokens_total", "Total tokens consumed", ["model", "direction"] # direction: input/output ) llm_cost_total = Counter( "llm_cost_usd_total", "Total cost in USD", ["model"] ) llm_cache_hit_rate = Gauge( "llm_cache_hit_rate", "Cache hit rate", ["cache_type"] # cache_type: exact/semantic ) # === 基础设施指标 === gpu_utilization = Gauge( "gpu_utilization_percent", "GPU utilization percentage", ["gpu_id", "node"] ) api_error_rate = Gauge( "llm_api_error_rate", "LLM API error rate", ["provider", "error_type"] ) 指标采集 中间件式采集 import time from contextlib import asynccontextmanager class LLMInstrumentation: def __init__(self, registry): self.registry = registry @asynccontextmanager async def trace_request(self, model: str, task_type: str): request_id = generate_id() start = time.monotonic() first_token_time = None async def on_first_token(): nonlocal first_token_time first_token_time = time.monotonic() try: yield {"request_id": request_id, "on_first_token": on_first_token} # 请求成功 elapsed = time.monotonic() - start llm_total_latency.labels(model=model).observe(elapsed) if first_token_time: ttft = first_token_time - start llm_ttft.labels(model=model).observe(ttft) except Exception as e: api_error_rate.labels( provider=model, error_type=type(e).__name__ ).inc() raise def record_tokens(self, model, input_tokens, output_tokens): llm_tokens_total.labels( model=model, direction="input" ).inc(input_tokens) llm_tokens_total.labels( model=model, direction="output" ).inc(output_tokens) # 成本计算 cost = self._calc_cost(model, input_tokens, output_tokens) llm_cost_total.labels(model=model).inc(cost) def _calc_cost(self, model, in_tok, out_tok): prices = { "gpt-4o": (0.0025, 0.01), "gpt-4o-mini": (0.00015, 0.0006), "claude-sonnet": (0.003, 0.015), } in_price, out_price = prices.get(model, (0, 0)) return (in_tok * in_price + out_tok * out_price) / 1000 使用方式 instrument = LLMInstrumentation(registry) async def chat_completion(messages, model="gpt-4o"): async with instrument.trace_request(model, "chat") as ctx: response = await llm_client.chat.completions.create( model=model, messages=messages, stream=True, ) async for chunk in response: if chunk.choices[0].delta.content: if not ctx["first_token_called"]: await ctx["on_first_token"]() ctx["first_token_called"] = True yield chunk.choices[0].delta.content instrument.record_tokens(model, input_count, output_count) 质量监控：LLM-as-a-Judge class QualityMonitor: def __init__(self, judge_model="gpt-4o", sample_rate=0.1): self.judge_model = judge_model self.sample_rate = sample_rate async def maybe_evaluate(self, query, response, context=None): """按采样率随机评估回答质量""" import random if random.random() > self.sample_rate: return score, issues = await self._judge(query, response, context) llm_quality_score.labels( model=self.judge_model, task_type="auto" ).set(score) if "hallucination" in issues: llm_hallucination_rate.labels( model=self.judge_model ).inc() async def _judge(self, query, response, context): prompt = f"""评估以下回答的质量。 问题：{query} 回答：{response} 参考资料：{context or '无'} 请检查： 1. 事实准确性（是否有幻觉） 2. 回答完整性 3. 逻辑一致性 输出 JSON：{{"score": 0.0-1.0, "issues": ["..."]}}""" result = await llm_client.chat.completions.create( model=self.judge_model, messages=[{"role": "user", "content": prompt}], response_format={"type": "json_object"}, ) data = json.loads(result.choices[0].message.content) return data["score"], data.get("issues", []) Grafana 仪表板 关键 Panel 配置 Panel 1: 延迟分布（P50/P90/P99） ...