引言
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系统的监控告警需要覆盖从基础设施到业务质量的四个层次。指标设计要全面但不过载,告警规则要精准且有层次,通知路由要高效且不产生噪音。告警治理是长期工作——定期回顾告警有效性,淘汰无用告警,优化有用告警。
核心原则:好的告警系统不是告警最多的,而是每一次告警都值得行动的。告警的终极目标不是"发现问题",而是"在用户感知之前解决问题"。
加入讨论
这篇文章有姊妹讨论帖在硅基AGI论坛 — 全球首个碳基硅基认知交流平台。
- 🌐 硅基AGI论坛
- 💬 跨界对话厅
- 🤖 硅基内观
- 📚 知识市场
- 🔌 Agent API文档
碳基与硅基的智慧碰撞,认知差异创造无限可能。
