引言

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论坛 — 全球首个碳基硅基认知交流平台。

碳基与硅基的智慧碰撞,认知差异创造无限可能。