阅读时间：约 12 分钟
前置知识：Agent 开发基础（P01-P08 系列）

P01-P08 讲了怎么写 Agent。现在 Agent 写好了，准备上线。上线之后呢？

用户问了一个问题，Agent 返回了。但你不知道这个回答好不好。
用户说"这 AI 越来越智障了"，但你是第一个知道的。
某个功能突然变差了，你花了三天才定位到原因。

这些问题的共同点：没有监控。

监控什么？

Agent 监控盯的是业务行为：用户等了多久、答得对不对、花了多少 Token。服务器 CPU 是基础设施的事。

1.1 Agent 的核心指标

指标含义怎么算
调用量总共有多少人问记录每次请求
延迟用户等了多久记录请求到返回的时间
错误率多少请求失败失败数 / 总请求数
Token 消耗花了多少 Token统计请求和响应的 token
评分回答好不好人工或自动打分

📌 本章要点：Agent 监控不只是系统指标（CPU、内存），更重要的是业务指标（延迟、错误率、评分、Token 消耗）。

1.2 系统指标 vs 业务指标

系统指标（基础设施层面）
├── CPU 使用率
├── 内存使用率
├── 网络延迟
└── 磁盘 IO

业务指标（Agent 层面）
├── 调用量（每天多少人问）
├── 延迟（平均响应时间）
├── 错误率（多少请求失败）
├── Token 消耗（花了多少钱）
└── 评分（回答好不好）

Agent 监控要同时关注这两层。

监控实现

2.1 基础监控类

import time
import json
from datetime import datetime, timedelta
from typing import Optional, Dict, Any

class AgentMonitor:
    """
    Agent 监控类
    记录每次请求的关键指标
    """
    
    def __init__(self):
        self.metrics = {
            "calls": 0,
            "errors": 0,
            "total_tokens": 0,
            "total_latency_ms": 0,
        }
        self.requests = []  # 请求记录列表
    
    def record_request(
        self,
        query: str,
        response: str,
        latency_ms: float,
        tokens: int,
        success: bool = True,
        error_msg: Optional[str] = None
    ):
        """
        记录一次请求
        """
        now = datetime.now()
        
        self.metrics["calls"] += 1
        self.metrics["total_tokens"] += tokens
        self.metrics["total_latency_ms"] += latency_ms
        
        if not success:
            self.metrics["errors"] += 1
        
        record = {
            "timestamp": now.isoformat(),
            "query": query,
            "response_length": len(response),
            "latency_ms": latency_ms,
            "tokens": tokens,
            "success": success,
            "error_msg": error_msg
        }
        
        self.requests.append(record)
        
        # 只保留最近 1000 条
        if len(self.requests) > 1000:
            self.requests = self.requests[-1000:]
        
        return record

2.2 延迟监控

class LatencyMonitor:
    """
    延迟监控
    跟踪延迟分布，检测延迟异常
    """
    
    def __init__(self, window_minutes=60):
        self.window_minutes = window_minutes
        self.latencies = []
    
    def record(self, latency_ms: float):
        """记录延迟"""
        self.latencies.append((time.time(), latency_ms))
        
        # 清理过期数据
        cutoff = time.time() - (self.window_minutes * 60)
        self.latencies = [
            (ts, lat) for ts, lat in self.latencies 
            if ts > cutoff
        ]
    
    def get_stats(self):
        """获取延迟统计"""
        if not self.latencies:
            return {}
        
        latencies = [lat for _, lat in self.latencies]
        
        import numpy as np
        latencies_sorted = sorted(latencies)
        n = len(latencies_sorted)
        
        return {
            "count": n,
            "avg_ms": round(np.mean(latencies_sorted), 1),
            "p50_ms": round(latencies_sorted[int(n * 0.5)], 1),
            "p95_ms": round(latencies_sorted[int(n * 0.95)], 1),
            "p99_ms": round(latencies_sorted[min(int(n * 0.99), n - 1)], 1),
            "max_ms": round(max(latencies_sorted), 1)
        }
    
    def check_anomaly(self, threshold_ms: float = 5000):
        """
        检查延迟是否异常
        超过 threshold_ms 的延迟视为异常
        """
        stats = self.get_stats()
        
        if stats.get("p99_ms", 0) > threshold_ms:
            return {
                "anomaly": True,
                "message": f"p99 延迟 {stats['p99_ms']:.0f}ms 超过阈值 {threshold_ms}ms",
                "p99_ms": stats["p99_ms"]
            }
        
        return {"anomaly": False}

2.3 错误率监控

class ErrorMonitor:
    """
    错误率监控
    跟踪错误模式，识别常见问题
    """
    
    def __init__(self, window_minutes=60):
        self.window_minutes = window_minutes
        self.errors = []
        self.total_calls = 0
    
    def record_error(self, error_type: str, error_msg: str, context: Dict = None):
        """记录错误"""
        self.errors.append({
            "timestamp": time.time(),
            "type": error_type,
            "message": error_msg,
            "context": context or {}
        })
        
        # 清理过期数据
        cutoff = time.time() - (self.window_minutes * 60)
        self.errors = [e for e in self.errors if e["timestamp"] > cutoff]
    
    def record_success(self):
        """记录成功"""
        self.total_calls += 1
    
    def get_error_rate(self):
        """计算错误率"""
        if self.total_calls == 0:
            return 0.0
        
        return len(self.errors) / (self.total_calls + len(self.errors))
    
    def get_error_distribution(self):
        """获取错误类型分布"""
        distribution = {}
        for error in self.errors:
            error_type = error["type"]
            distribution[error_type] = distribution.get(error_type, 0) + 1
        
        return dict(sorted(distribution.items(), key=lambda x: x[1], reverse=True))
    
    def check_anomaly(self, threshold: float = 0.05):
        """
        检查错误率是否异常
        """
        error_rate = self.get_error_rate()
        
        if error_rate > threshold:
            return {
                "anomaly": True,
                "message": f"错误率 {error_rate:.1%} 超过阈值 {threshold:.1%}",
                "error_rate": error_rate
            }
        
        return {"anomaly": False}

2.4 Token 消耗监控

class TokenMonitor:
    """
    Token 消耗监控
    统计 Token 使用情况，追踪成本
    """
    
    def __init__(self, daily_budget: int = 1000000):
        """
        daily_budget: 每日 Token 预算
        """
        self.daily_budget = daily_budget
        self.daily_usage = {}  # date -> total_tokens
    
    def record_usage(self, tokens: int, date: str = None):
        """记录 Token 消耗"""
        if date is None:
            date = datetime.now().strftime("%Y-%m-%d")
        
        self.daily_usage[date] = self.daily_usage.get(date, 0) + tokens
    
    def get_daily_usage(self, date: str = None) -> int:
        """获取某天的 Token 消耗"""
        if date is None:
            date = datetime.now().strftime("%Y-%m-%d")
        return self.daily_usage.get(date, 0)
    
    def get_budget_remaining(self, date: str = None) -> int:
        """获取预算剩余"""
        usage = self.get_daily_usage(date)
        return max(0, self.daily_budget - usage)
    
    def check_budget(self) -> Dict:
        """检查预算是否超标"""
        usage = self.get_daily_usage()
        budget_pct = (usage / self.daily_budget) * 100
        
        status = "normal"
        message = "Token 消耗正常"
        
        if budget_pct > 90:
            status = "warning"
            message = f"Token 消耗已达预算的 {budget_pct:.1f}%"
        elif budget_pct > 100:
            status = "critical"
            message = f"Token 消耗已超预算 {budget_pct:.1f}%"
        
        return {
            "status": status,
            "usage": usage,
            "budget": self.daily_budget,
            "budget_pct": round(budget_pct, 1),
            "remaining": max(0, self.daily_budget - usage),
            "message": message
        }

告警设置

3.1 告警规则

class AlertManager:
    """
    告警管理器
    当指标超过阈值时触发告警
    """
    
    def __init__(self):
        self.rules = []
        self.alert_history = []
    
    def add_rule(
        self,
        name: str,
        metric: str,
        threshold: float,
        duration_minutes: int = 5,
        severity: str = "warning"  # "info", "warning", "critical"
    ):
        """添加告警规则"""
        self.rules.append({
            "name": name,
            "metric": metric,
            "threshold": threshold,
            "duration_minutes": duration_minutes,
            "severity": severity
        })
    
    def check_alert(
        self,
        metric_name: str,
        metric_value: float,
        current_time: float = time.time()
    ) -> Optional[Dict]:
        """
        检查是否触发告警
        """
        for rule in self.rules:
            if rule["metric"] != metric_name:
                continue
            
            # 判断是否超过阈值
            if rule["severity"] == "critical" and metric_value > rule["threshold"]:
                alert = {
                    "rule_name": rule["name"],
                    "metric": metric_name,
                    "value": metric_value,
                    "threshold": rule["threshold"],
                    "severity": rule["severity"],
                    "timestamp": current_time,
                    "message": f"[{rule['severity'].upper()}] {metric_name} 达到 {metric_value}，超过阈值 {rule['threshold']}"
                }
                self.alert_history.append(alert)
                return alert
        
        return None

3.2 常见告警场景

场景阈值严重程度处理方式
错误率 > 5%错误率 0.05警告查看错误日志
错误率 > 10%错误率 0.10严重立即回滚
延迟 > 10s延迟 10000ms警告查看系统资源
延迟 > 30s延迟 30000ms严重检查服务状态
Token 消耗 > 90% 预算百分比 0.90警告优化 Prompt
Token 消耗 > 100% 预算百分比 1.00严重紧急停止

监控面板

4.1 关键指标面板

class MetricsDashboard:
    """
    关键指标面板
    聚合所有指标，生成可视化数据
    """
    
    def __init__(self):
        self.latency_monitor = LatencyMonitor()
        self.error_monitor = ErrorMonitor()
        self.token_monitor = TokenMonitor()
        self.alert_manager = AlertManager()
    
    def generate_report(self) -> Dict:
        """生成监控报告"""
        report = {
            "timestamp": datetime.now().isoformat(),
            "latency": self.latency_monitor.get_stats(),
            "errors": {
                "rate": round(self.error_monitor.get_error_rate(), 3),
                "distribution": self.error_monitor.get_error_distribution()
            },
            "tokens": self.token_monitor.check_budget(),
            "alerts": self.alert_manager.alert_history[-10:]  # 最近 10 条告警
        }
        
        return report
    
    def print_summary(self):
        """打印简要总结"""
        report = self.generate_report()
        
        print("\n=== Agent 监控报告 ===")
        print(f"时间: {report['timestamp']}")
        
        # 延迟
        latency = report.get("latency", {})
        if latency:
            print(f"延迟: 平均 {latency.get('avg_ms', 0):.0f}ms, P95 {latency.get('p95_ms', 0):.0f}ms, P99 {latency.get('p99_ms', 0):.0f}ms")
        
        # 错误率
        error_rate = report.get("errors", {}).get("rate", 0)
        print(f"错误率: {error_rate:.1%}")
        
        # Token 消耗
        tokens = report.get("tokens", {})
        print(f"Token: 已用 {tokens.get('usage', 0)} / 预算 {tokens.get('budget', 0)} ({tokens.get('budget_pct', 0):.1f}%)")
        
        # 告警
        alerts = report.get("alerts", [])
        if alerts:
            print(f"告警: 最近 {len(alerts)} 条")
            for alert in alerts[-3:]:  # 最近 3 条
                print(f"  [{alert['severity'].upper()}] {alert['message']}")
        
        print("=== 报告结束 ===\n")

📌 本章要点：监控面板聚合延迟、错误率、Token 消耗等指标。定期生成报告，发现问题及时处理。

监控方案选型

5.1 自建方案

class SelfBuiltMonitor:
    """
    自建监控方案
    使用 Prometheus + Grafana
    """
    
    @staticmethod
    def setup_prometheus_exporter(monitor: AgentMonitor):
        """
        设置 Prometheus exporter
        """
        from prometheus_client import Counter, Histogram, Gauge
        
        # 调用量
        call_counter = Counter(
            'agent_calls_total',
            'Total agent calls',
            ['status']  # 'success', 'error'
        )
        
        # 延迟
        latency_histogram = Histogram(
            'agent_latency_seconds',
            'Agent latency in seconds',
            ['endpoint']
        )
        
        # Token 消耗
        token_gauge = Gauge(
            'agent_tokens_total',
            'Total tokens consumed'
        )
        
        return {
            "call_counter": call_counter,
            "latency_histogram": latency_histogram,
            "token_gauge": token_gauge
        }

5.2 商业方案

方案特点适用场景
Prometheus + Grafana免费，开源，可定制自建运维团队
Datadog功能丰富，UI 好预算充足，追求效率
New Relic全链路追踪复杂分布式系统
Sentry错误追踪快速定位错误

5.3 选择建议

因素自建方案商业方案
成本免费（但有人力成本）按用量收费
功能需要自己开发开箱即用
定制性高低
维护成本高低
适合团队有运维团队小团队或初创

监控最佳实践

6.1 监控 checklist

检查项频率说明
调用量每小时查看调用趋势
错误率每小时发现异常错误
延迟分布每天检查 P95/P99 延迟
Token 消耗每天控制成本
告警处理实时及时响应

6.2 告警处理流程

告警触发
    ↓
判断严重程度
    ↓
├── 信息级 → 记录日志，下次处理
├── 警告级 → 15 分钟内处理
└── 严重级 → 立即处理
    ↓
记录处理过程和结果
    ↓
复盘告警，优化规则

📌 本章要点：建立监控 checklist，告警按严重程度分级处理。每次告警后复盘，优化监控规则。

总结

  1. Agent 监控的核心是业务指标：延迟、错误率、评分、Token 消耗。系统指标（CPU、内存）是辅助。
  2. 同时关注系统层和业务层。CPU 90% 是问题，但延迟 10s 更严重。
  3. 告警要分级处理。信息级、警告级、严重级，不同级别不同处理方式。
  4. 自建方案和商业方案各有优劣。小团队用商业方案，大团队用自建方案。
  5. 建立监控 checklist。定期检查指标，及时发现问题。

🤔 思考一下：你的 Agent 上线后，多久没检查过监控了？

思维导图

  • 监控
    • 核心指标
      • 调用量
      • 延迟
      • 错误率
      • Token 消耗
      • 评分
    • 系统指标 vs 业务指标
      • 系统层：CPU、内存、网络
      • 业务层：延迟、错误率、评分、Token
    • 监控实现
      • 延迟监控
      • 错误率监控
      • Token 消耗监控
    • 告警设置
      • 告警规则
      • 告警场景
      • 告警处理流程
    • 监控面板
      • 关键指标面板
      • 定期生成报告
    • 监控方案选型
      • 自建方案（Prometheus + Grafana）
      • 商业方案（Datadog, New Relic, Sentry）
    • 最佳实践
      • 监控 checklist
      • 告警处理流程
      • 定期检查
# python # 人工智能 # 深度学习
Logo
CSDN-OPC开发者社区

这里是“一人公司”的成长家园。我们提供从产品曝光、技术变现到法律财税的全栈内容，并连接云服务、办公空间等稀缺资源，助你专注创造，无忧运营。

更多推荐

cover

从零构建 Java 智能体 RAG 系统：Milvus 向量数据库实战指南

avatar CSDN-OPC开发者社区
cover

DeepSeek-Harness-入门指南

avatar CSDN-OPC开发者社区
cover

提升办公效率：OpenClaw 本地自动化 AI 工具搭建实战教程

avatar CSDN-OPC开发者社区