D01 监控：怎么知道你的 Agent 在线上好不好用

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

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

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

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

监控什么？

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

1.1 Agent 的核心指标

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

1.2 系统指标 vs 业务指标

系统指标（基础设施层面） ├── CPU 使用率 ├── 内存使用率 ├── 网络延迟 └── 磁盘 IO 业务指标（Agent 层面） ├── 调用量（每天多少人问） ├── 延迟（平均响应时间） ├── 错误率（多少请求失败） ├── Token 消耗（花了多少钱） └── 评分（回答好不好）

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

监控实现

2.1 基础监控类

importtimeimportjsonfromdatetimeimportdatetime,timedeltafromtypingimportOptional,Dict,AnyclassAgentMonitor:""" Agent 监控类 记录每次请求的关键指标 """def__init__(self):self.metrics={"calls":0,"errors":0,"total_tokens":0,"total_latency_ms":0,}self.requests=[]# 请求记录列表defrecord_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"]+=1self.metrics["total_tokens"]+=tokens self.metrics["total_latency_ms"]+=latency_msifnotsuccess:self.metrics["errors"]+=1record={"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 条iflen(self.requests)>1000:self.requests=self.requests[-1000:]returnrecord

2.2 延迟监控

classLatencyMonitor:""" 延迟监控 跟踪延迟分布，检测延迟异常 """def__init__(self,window_minutes=60):self.window_minutes=window_minutes self.latencies=[]defrecord(self,latency_ms:float):"""记录延迟"""self.latencies.append((time.time(),latency_ms))# 清理过期数据cutoff=time.time()-(self.window_minutes*60)self.latencies=[(ts,lat)forts,latinself.latenciesifts>cutoff]defget_stats(self):"""获取延迟统计"""ifnotself.latencies:return{}latencies=[latfor_,latinself.latencies]importnumpyasnp 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)}defcheck_anomaly(self,threshold_ms:float=5000):""" 检查延迟是否异常 超过 threshold_ms 的延迟视为异常 """stats=self.get_stats()ifstats.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 错误率监控

classErrorMonitor:""" 错误率监控 跟踪错误模式，识别常见问题 """def__init__(self,window_minutes=60):self.window_minutes=window_minutes self.errors=[]self.total_calls=0defrecord_error(self,error_type:str,error_msg:str,context:Dict=None):"""记录错误"""self.errors.append({"timestamp":time.time(),"type":error_type,"message":error_msg,"context":contextor{}})# 清理过期数据cutoff=time.time()-(self.window_minutes*60)self.errors=[eforeinself.errorsife["timestamp"]>cutoff]defrecord_success(self):"""记录成功"""self.total_calls+=1defget_error_rate(self):"""计算错误率"""ifself.total_calls==0:return0.0returnlen(self.errors)/(self.total_calls+len(self.errors))defget_error_distribution(self):"""获取错误类型分布"""distribution={}forerrorinself.errors:error_type=error["type"]distribution[error_type]=distribution.get(error_type,0)+1returndict(sorted(distribution.items(),key=lambdax:x[1],reverse=True))defcheck_anomaly(self,threshold:float=0.05):""" 检查错误率是否异常 """error_rate=self.get_error_rate()iferror_rate>threshold:return{"anomaly":True,"message":f"错误率{error_rate:.1%}超过阈值{threshold:.1%}","error_rate":error_rate}return{"anomaly":False}

2.4 Token 消耗监控

classTokenMonitor:""" Token 消耗监控 统计 Token 使用情况，追踪成本 """def__init__(self,daily_budget:int=1000000):""" daily_budget: 每日 Token 预算 """self.daily_budget=daily_budget self.daily_usage={}# date -> total_tokensdefrecord_usage(self,tokens:int,date:str=None):"""记录 Token 消耗"""ifdateisNone:date=datetime.now().strftime("%Y-%m-%d")self.daily_usage[date]=self.daily_usage.get(date,0)+tokensdefget_daily_usage(self,date:str=None)->int:"""获取某天的 Token 消耗"""ifdateisNone:date=datetime.now().strftime("%Y-%m-%d")returnself.daily_usage.get(date,0)defget_budget_remaining(self,date:str=None)->int:"""获取预算剩余"""usage=self.get_daily_usage(date)returnmax(0,self.daily_budget-usage)defcheck_budget(self)->Dict:"""检查预算是否超标"""usage=self.get_daily_usage()budget_pct=(usage/self.daily_budget)*100status="normal"message="Token 消耗正常"ifbudget_pct>90:status="warning"message=f"Token 消耗已达预算的{budget_pct:.1f}%"elifbudget_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 告警规则

classAlertManager:""" 告警管理器 当指标超过阈值时触发告警 """def__init__(self):self.rules=[]self.alert_history=[]defadd_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})defcheck_alert(self,metric_name:str,metric_value:float,current_time:float=time.time())->Optional[Dict]:""" 检查是否触发告警 """forruleinself.rules:ifrule["metric"]!=metric_name:continue# 判断是否超过阈值ifrule["severity"]=="critical"andmetric_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)returnalertreturnNone

3.2 常见告警场景

监控面板

4.1 关键指标面板

classMetricsDashboard:""" 关键指标面板 聚合所有指标，生成可视化数据 """def__init__(self):self.latency_monitor=LatencyMonitor()self.error_monitor=ErrorMonitor()self.token_monitor=TokenMonitor()self.alert_manager=AlertManager()defgenerate_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 条告警}returnreportdefprint_summary(self):"""打印简要总结"""report=self.generate_report()print("\n=== Agent 监控报告 ===")print(f"时间:{report['timestamp']}")# 延迟latency=report.get("latency",{})iflatency: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",[])ifalerts:print(f"告警: 最近{len(alerts)}条")foralertinalerts[-3:]:# 最近 3 条print(f" [{alert['severity'].upper()}]{alert['message']}")print("=== 报告结束 ===\n")

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

监控方案选型

5.1 自建方案

classSelfBuiltMonitor:""" 自建监控方案 使用 Prometheus + Grafana """@staticmethoddefsetup_prometheus_exporter(monitor:AgentMonitor):""" 设置 Prometheus exporter """fromprometheus_clientimportCounter,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 商业方案

5.3 选择建议

监控最佳实践

6.1 监控 checklist

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
    • 告警处理流程
    • 定期检查