监控与运维Agent上线后怎么运维包括日志和指标和告警上个月有个朋友跟我说他的客服Agent上线一周就出事了。用户反馈回复变慢有时候还返回空消息。他打开服务器一看日志全是print打印的散装文本翻了半天也没找到哪条请求出了问题。最后发现是某个工具API偶发超时Agent静默吞掉了异常继续跑用户那边就收到空回复。这事我太熟了。Agent上线容易运维难跟传统Web服务完全不是一个量级。传统服务出了问题你看HTTP状态码就行Agent出问题你得扒调用过程看模型哪一步生成歪了、工具哪一次超时了、token是不是烧超了。这篇就把Agent上线后的运维体系从头搭一遍包括结构化日志、关键指标、告警规则以及用LangSmith做可观测性。做完你能有一套可以直接抄走的监控代码。Agent运维为什么难传统服务监控看几个数就行。QPS、响应时间、错误率三板斧。Agent多了几层你没法绕过的东西。第一层是模型调用。模型不是确定性的同样的输入跑两遍结果可能不一样。你没法靠重放请求来复现问题。而且模型调用慢动辄几秒十几秒中间还可能调好几次用户等的是整条流程跑完的时间。第二层是工具调用。Agent可能调搜索、调数据库、调第三方API。每个工具都有自己的失败模式超时、返回格式变了、接口限流。工具出问题Agent的表现就跟着出问题但日志里可能只看到模型最终返回了一个奇怪的答案。第三层是成本。每次调用都在花钱token消耗和响应质量直接挂钩。token烧多了可能是prompt没优化好也可能是模型在那反复调工具兜圈子。没有监控你根本不知道钱花哪了。日志体系设计先说日志。print打印的文本日志上线就是灾难。查询没法过滤没法聚合排查问题靠肉眼翻。生产环境必须用结构化日志。结构化日志每条都是一个JSON对象字段固定机器可读。Python里用structlog或者标准库的logging配JSON formatter都行。我习惯用structlog配置简单输出干净。每条日志至少要带这几个字段。timestamp是时间戳level是日志级别event是事件名request_id是请求追踪IDuser_id是用户标识还有业务相关的字段比如model_name、token_usage、tool_name。request_id特别重要一次用户请求可能触发多次模型调用和工具调用全靠request_id串起来排查时按这个ID过滤整条执行过程一清二楚。调用过程怎么串。用户请求进来时生成一个request_id存到contextvar里。后面所有的日志、模型调用、工具调用都从contextvar里取这个ID带上。这样同一个请求的所有日志都有同一个request_id过滤一下就能看到完整的执行过程。关键监控指标日志是事后排查用的指标是实时看大盘用的。Agent运维要盯几个核心指标。响应时间。从用户发请求到收到完整回复的时间。这个要分位数看p50、p95、p99。平均值会被快请求拉低掩盖长尾问题。p99飙高说明有少量请求特别慢通常是个别工具超时或者模型生成卡住了。Token消耗。每次模型调用花了多少token分prompt token和completion token。按天、按用户、按场景聚合。这个直接关系到成本必须实时盯。成功率。请求成功返回有效结果的比率。注意Agent的成功不好定义。HTTP 200不代表成功模型可能返回了空内容或者胡说八道。我的做法是看用户是否在短时间内重新问了同样的问题如果重问了大概率是上一次没答好算一次隐式失败。工具调用次数。一次请求里Agent调了几次工具。正常情况一两次就够了。如果某个请求工具调用了七八次要么是Agent在那兜圈子要么是工具返回结果有问题模型一直在重试。这个指标异常通常意味着prompt或者工具设计有问题。完整代码Agent监控系统下面这套代码是一个完整的Agent监控系统包含结构化日志、请求追踪、指标采集和告警。可以直接跑。 Agent监控系统 包含结构化日志、请求追踪、指标采集、告警规则 依赖: pip install structlog importstructlogimporttimeimportuuidimportjsonfromcontextvarsimportContextVarfromdataclassesimportdataclass,fieldfromcollectionsimportdefaultdictfromdatetimeimportdatetimefromtypingimportOptional# 请求追踪 # 用contextvar存request_id同一次请求的所有日志共享同一个IDrequest_id_var:ContextVar[str]ContextVar(request_id,default)user_id_var:ContextVar[str]ContextVar(user_id,default)# 结构化日志配置 # structlog配置输出JSON格式自动带上request_id和user_iddefadd_context(logger,method_name,event_dict):日志处理器把contextvar里的request_id和user_id塞进每条日志ridrequest_id_var.get()# 从上下文取请求IDuiduser_id_var.get()# 从上下文取用户IDifrid:event_dict[request_id]ridifuid:event_dict[user_id]uidreturnevent_dict structlog.configure(processors[structlog.stdlib.add_log_level,add_context,# 自动注入追踪IDstructlog.processors.TimeStamper(fmtiso),structlog.processors.JSONRenderer(),# 输出JSON格式],wrapper_classstructlog.stdlib.BoundLogger,)loggerstructlog.get_logger()# 指标采集器 # 收集响应时间、token消耗、成功率、工具调用次数dataclassclassMetricStore:内存指标存储生产环境换Prometheusresponse_times:listfield(default_factorylist)# 响应时间列表token_usage:dictfield(default_factorylambda:{prompt:0,completion:0})success_count:int0# 成功次数fail_count:int0# 失败次数tool_calls:dictfield(default_factorylambda:defaultdict(int))defrecord_response_time(self,seconds:float):记录单次请求的响应时间self.response_times.append(seconds)# 只保留最近1000条防止内存膨胀iflen(self.response_times)1000:self.response_timesself.response_times[-1000:]defrecord_tokens(self,prompt:int,completion:int):记录token消耗分输入和输出self.token_usage[prompt]prompt self.token_usage[completion]completiondefrecord_result(self,success:bool):记录请求成功或失败ifsuccess:self.success_count1else:self.fail_count1defrecord_tool_call(self,tool_name:str):记录工具被调用的次数self.tool_calls[tool_name]1defget_percentile(self,p:float)-float:计算响应时间的分位数p0.95就是p95ifnotself.response_times:return0.0sorted_timessorted(self.response_times)indexint(len(sorted_times)*p)returnsorted_times[min(index,len(sorted_times)-1)]defsummary(self)-dict:汇总当前指标用于展示和告警判断totalself.success_countself.fail_countreturn{p50_response:round(self.get_percentile(0.50),2),p95_response:round(self.get_percentile(0.95),2),p99_response:round(self.get_percentile(0.99),2),success_rate:round(self.success_count/total,4)iftotal0else0,total_tokens:self.token_usage[prompt]self.token_usage[completion],tool_calls:dict(self.tool_calls),}metricsMetricStore()# 告警规则 # 简单的阈值告警生产环境接Alertmanager或钉钉机器人classAlertChecker:检查指标是否触发告警阈值defcheck(self,metric_summary:dict)-list:检查所有规则返回触发的告警列表alerts[]# 规则1p95响应时间超过10秒ifmetric_summary[p95_response]10:alerts.append(fP95响应时间{metric_summary[p95_response]}s 超过阈值10s)# 规则2成功率低于90%且有数据时才判断if0metric_summary[success_rate]0.9:alerts.append(f成功率{metric_summary[success_rate]:.1%}低于阈值90%)# 规则3单个工具调用次数异常多可能Agent在兜圈子fortool_name,countinmetric_summary[tool_calls].items():ifcount50:alerts.append(f工具{tool_name}调用{count}次疑似异常)returnalerts alert_checkerAlertChecker()# 监控装饰器 # 套在Agent的入口函数上自动采集日志和指标defmonitored(func):装饰器自动记录请求追踪、响应时间、成功率和tokendefwrapper(*args,**kwargs):# 生成请求追踪ID截短到8位方便看ridstr(uuid.uuid4())[:8]request_id_var.set(rid)# 从参数里提取user_id假设第一个参数是user_idifargsandisinstance(args[0],str):user_id_var.set(args[0])logger.info(agent_request_start,input_previewstr(kwargs)[:200])start_timetime.time()try:resultfunc(*args,**kwargs)elapsedtime.time()-start_time# 记录成功metrics.record_response_time(elapsed)metrics.record_result(successTrue)logger.info(agent_request_end,elapsedround(elapsed,2),statussuccess,)returnresultexceptExceptionase:elapsedtime.time()-start_time metrics.record_response_time(elapsed)metrics.record_result(successFalse)# 错误日志带上异常类型和消息方便分类排查logger.error(agent_request_error,elapsedround(elapsed,2),errorstr(e),error_typetype(e).__name__,)raisereturnwrapper# LangSmith可观测性 # 设环境变量即可LangChain调用自动上报defsetup_langsmith(api_key:str,project:stragent-prod): 配置LangSmith追踪 设完之后所有LangChain调用自动上报业务代码不用改 生产环境建议采样别全量开 importos os.environ[LANGCHAIN_TRACING_V2]trueos.environ[LANGCHAIN_API_KEY]api_key os.environ[LANGCHAIN_PROJECT]project# 使用示例 # 模拟一个Agent调用展示监控效果monitoreddefrun_agent(user_id:str,question:str)-str:模拟Agent处理用户问题logger.info(agent_processing,questionquestion[:100])# 模拟工具调用metrics.record_tool_call(search)logger.info(tool_called,toolsearch,query模拟搜索)time.sleep(0.5)# 模拟模型调用和token消耗metrics.record_tokens(prompt150,completion80)logger.info(llm_called,modelgpt-4o-mini,prompt_tokens150,completion_tokens80)# 模拟处理结果answerf关于「{question}」的回答returnanswer# 运行和查看指标 if__name____main__:# 配置LangSmith可选有key才开# setup_langsmith(ls__your_key, agent-prod)# 跑几个请求看看效果run_agent(user_001,怎么退货)run_agent(user_002,订单状态查询)run_agent(user_003,退款流程是什么)# 打印指标汇总summarymetrics.summary()print(\n 指标汇总 )print(json.dumps(summary,indent2,ensure_asciiFalse))# 检查告警alertsalert_checker.check(summary)ifalerts:print(\n 告警 )forainalerts:print(f [告警]{a})else:print(\n 无告警 )效果验证跑起来你会看到几样东西。控制台输出JSON格式的日志每条都带request_id、时间戳、事件名、业务字段。三次请求的日志按request_id区分得清清楚楚。最后打印的指标汇总能看到p50、p95、p99响应时间成功率token总量工具调用次数。告警检查没有触发说明一切正常。判断成功的标准很简单。日志里每条都是合法JSON能用日志系统直接过滤。指标汇总的数字合理响应时间在预期范围内。告警部分没有误报。常见的报错有两个。一个是structlog没装pip install structlog就行。另一个是contextvar在异步场景下需要注意asyncio里每个task有独立的contextrequest_id不会串但如果你用了线程池contextvar不会自动传递到子线程需要手动copy_context。踩坑记录第一个坑日志里记了完整的用户输入。有一次排查问题日志里把用户的完整问题包括手机号全打出来了。后来做数据脱敏审查被查出来整改了一轮。教训是日志里记输入预览就行截断到前200字符敏感字段单独脱敏处理。第二个坑告警阈值设得太敏感。一开始我把p95响应时间告警阈值设成5秒Agent正常跑也就3到4秒稍微慢一点就告警。大半夜被钉钉机器人吵醒爬起来一看啥事没有。后来调到10秒同时加了连续触发3次才告警的逻辑误报少多了。告警阈值一定要跑一周真实数据再定拍脑袋设的值不是太高就是太低。第三个坑LangSmith的trace量太大。线上全量开LangSmith追踪一天几万次调用免费额度几天就吃完了。后来改成采样上报只采10%的请求加所有错误请求。正常请求采够量看趋势就行错误请求一个都不能漏。延伸与判断这套监控体系同样适用于其他场景。任何有多次内部调用的服务都能用比如微服务编排、数据处理流水线。核心思路都一样结构化日志加请求追踪加指标采集加告警。局限性也有。内存指标存储不持久重启就没了生产环境得换Prometheus加Grafana。成功率指标的定义比较粗糙靠用户重问来判断隐式失败不够精确。更准确的做法是加一个输出质量评估环节用小模型或者规则检查答案质量。我的建议是先搭结构化日志和请求追踪这两个最基础也最值。指标和告警可以后面逐步加。LangSmith在开发阶段就接上生产环境采样上报。别等出事了才想起来加监控那时候你连问题出在哪都不知道。下一篇聊成本控制Agent跑起来token哗哗烧怎么把账单压下来。