阅读时间约 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 基础监控类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:boolTrue,error_msg:Optional[str]None): 记录一次请求 nowdatetime.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.requestsself.requests[-1000:]returnrecord2.2 延迟监控classLatencyMonitor: 延迟监控 跟踪延迟分布检测延迟异常 def__init__(self,window_minutes60):self.window_minuteswindow_minutes self.latencies[]defrecord(self,latency_ms:float):记录延迟self.latencies.append((time.time(),latency_ms))# 清理过期数据cutofftime.time()-(self.window_minutes*60)self.latencies[(ts,lat)forts,latinself.latenciesiftscutoff]defget_stats(self):获取延迟统计ifnotself.latencies:return{}latencies[latfor_,latinself.latencies]importnumpyasnp latencies_sortedsorted(latencies)nlen(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:float5000): 检查延迟是否异常 超过 threshold_ms 的延迟视为异常 statsself.get_stats()ifstats.get(p99_ms,0)threshold_ms:return{anomaly:True,message:fp99 延迟{stats[p99_ms]:.0f}ms 超过阈值{threshold_ms}ms,p99_ms:stats[p99_ms]}return{anomaly:False}2.3 错误率监控classErrorMonitor: 错误率监控 跟踪错误模式识别常见问题 def__init__(self,window_minutes60):self.window_minuteswindow_minutes self.errors[]self.total_calls0defrecord_error(self,error_type:str,error_msg:str,context:DictNone):记录错误self.errors.append({timestamp:time.time(),type:error_type,message:error_msg,context:contextor{}})# 清理过期数据cutofftime.time()-(self.window_minutes*60)self.errors[eforeinself.errorsife[timestamp]cutoff]defrecord_success(self):记录成功self.total_calls1defget_error_rate(self):计算错误率ifself.total_calls0:return0.0returnlen(self.errors)/(self.total_callslen(self.errors))defget_error_distribution(self):获取错误类型分布distribution{}forerrorinself.errors:error_typeerror[type]distribution[error_type]distribution.get(error_type,0)1returndict(sorted(distribution.items(),keylambdax:x[1],reverseTrue))defcheck_anomaly(self,threshold:float0.05): 检查错误率是否异常 error_rateself.get_error_rate()iferror_ratethreshold: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:int1000000): daily_budget: 每日 Token 预算 self.daily_budgetdaily_budget self.daily_usage{}# date - total_tokensdefrecord_usage(self,tokens:int,date:strNone):记录 Token 消耗ifdateisNone:datedatetime.now().strftime(%Y-%m-%d)self.daily_usage[date]self.daily_usage.get(date,0)tokensdefget_daily_usage(self,date:strNone)-int:获取某天的 Token 消耗ifdateisNone:datedatetime.now().strftime(%Y-%m-%d)returnself.daily_usage.get(date,0)defget_budget_remaining(self,date:strNone)-int:获取预算剩余usageself.get_daily_usage(date)returnmax(0,self.daily_budget-usage)defcheck_budget(self)-Dict:检查预算是否超标usageself.get_daily_usage()budget_pct(usage/self.daily_budget)*100statusnormalmessageToken 消耗正常ifbudget_pct90:statuswarningmessagefToken 消耗已达预算的{budget_pct:.1f}%elifbudget_pct100:statuscriticalmessagefToken 消耗已超预算{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:int5,severity:strwarning# 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:floattime.time())-Optional[Dict]: 检查是否触发告警 forruleinself.rules:ifrule[metric]!metric_name:continue# 判断是否超过阈值ifrule[severity]criticalandmetric_valuerule[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)returnalertreturnNone3.2 常见告警场景场景阈值严重程度处理方式错误率 5%错误率 0.05警告查看错误日志错误率 10%错误率 0.10严重立即回滚延迟 10s延迟 10000ms警告查看系统资源延迟 30s延迟 30000ms严重检查服务状态Token 消耗 90% 预算百分比 0.90警告优化 PromptToken 消耗 100% 预算百分比 1.00严重紧急停止监控面板4.1 关键指标面板classMetricsDashboard: 关键指标面板 聚合所有指标生成可视化数据 def__init__(self):self.latency_monitorLatencyMonitor()self.error_monitorErrorMonitor()self.token_monitorTokenMonitor()self.alert_managerAlertManager()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):打印简要总结reportself.generate_report()print(\n Agent 监控报告 )print(f时间:{report[timestamp]})# 延迟latencyreport.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_ratereport.get(errors,{}).get(rate,0)print(f错误率:{error_rate:.1%})# Token 消耗tokensreport.get(tokens,{})print(fToken: 已用{tokens.get(usage,0)}/ 预算{tokens.get(budget,0)}({tokens.get(budget_pct,0):.1f}%))# 告警alertsreport.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_counterCounter(agent_calls_total,Total agent calls,[status]# success, error)# 延迟latency_histogramHistogram(agent_latency_seconds,Agent latency in seconds,[endpoint])# Token 消耗token_gaugeGauge(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告警按严重程度分级处理。每次告警后复盘优化监控规则。总结Agent 监控的核心是业务指标延迟、错误率、评分、Token 消耗。系统指标CPU、内存是辅助。同时关注系统层和业务层。CPU 90% 是问题但延迟 10s 更严重。告警要分级处理。信息级、警告级、严重级不同级别不同处理方式。自建方案和商业方案各有优劣。小团队用商业方案大团队用自建方案。建立监控 checklist。定期检查指标及时发现问题。思考一下你的 Agent 上线后多久没检查过监控了思维导图监控核心指标调用量延迟错误率Token 消耗评分系统指标 vs 业务指标系统层CPU、内存、网络业务层延迟、错误率、评分、Token监控实现延迟监控错误率监控Token 消耗监控告警设置告警规则告警场景告警处理流程监控面板关键指标面板定期生成报告监控方案选型自建方案Prometheus Grafana商业方案Datadog, New Relic, Sentry最佳实践监控 checklist告警处理流程定期检查