1. 为什么“AI员工”光记日志远远不够1.1 从“能跑”到“敢用”的那道坎过去一年我陆续帮几个团队把内部流程里的重复性工作交给 AI 员工来做——自动处理工单、自动回复客户咨询、自动整理数据报表。刚开始大家都很兴奋觉得效率翻倍。但真正上线跑了两周之后问题就来了某天早上运营同事跑过来问“昨天那批客户咨询为什么有三条回复的内容完全不对”我打开系统一看日志里只有一行task completed至于它当时看到了什么、调用了哪个工具、中间经过了哪些判断全都没有记录。那一刻我才意识到AI 员工和传统程序最大的区别在于它的行为是不确定的。传统程序出 bug你可以复现AI 员工出问题你连它当时“在想什么”都不知道。这就是“可观测性”要解决的核心问题。很多人把可观测性等同于“记日志”其实差得远。日志只是最基础的一层真正完整的可观测性要能回答三个问题发生了什么日志、为什么发生追踪与上下文、能不能重来一遍可重放。对于 AI 员工这种带推理、带工具调用、带外部依赖的系统来说第三点尤其关键——因为它的输出往往不可复现你只有把当时的完整执行上下文保存下来才能在事后“重放”那一次执行定位到底是提示词的问题、工具返回的问题还是模型本身抽风。我写这篇东西是想把过去大半年踩过的坑、试过的方案、最后沉淀下来的架构讲清楚。适合两类人看一类是正在做 AI 员工/智能体系统、被线上问题折磨的工程师另一类是刚开始接触可观测性、想知道怎么把它落到 AI 场景里的同学。不需要你有多深的分布式追踪背景我会尽量用大白话把原理和实操都讲透。1.2 一个真实的翻车现场先讲个具体案例方便后面展开。我们有个 AI 员工负责处理售后工单流程大概是读取工单内容 → 判断问题类型 → 查询订单系统 → 生成回复 → 必要时转人工。有一次客户投诉说“明明我的订单已经发货了AI 却回复说没找到订单”。我去查日志发现只有一条order_query的调用记录返回结果是空。但问题是这个订单在系统里明明存在。后来我们花了整整一个下午才定位到AI 员工在调用订单查询工具时把订单号里的字母O识别成了数字0。这个错误发生在模型生成工具参数的那一步而我们的日志只记录了“调用了工具”没有记录“调用时传了什么参数”。如果当时有完整的执行追踪把每一步的输入输出都存下来这个问题五分钟就能定位。更理想的情况是如果支持重放我可以直接拿当时那次执行的上下文重新跑一遍改一下参数验证逻辑立刻验证修复是否有效。这个案例说明了一件事AI 员工的可观测性必须覆盖“输入 → 推理 → 工具调用 → 输出”的完整链路而且每一步都要能回放。下面我就按这个思路从整体设计讲到具体实现。2. 整体设计三层可观测性架构怎么搭2.1 先想清楚要观测什么动手之前我建议先列一张“观测清单”。AI 员工的执行链路通常包含这几个环节每个环节都有对应的观测需求执行环节要观测的内容典型问题接收输入用户原始输入、会话上下文、系统提示词输入被截断、上下文丢失模型推理完整提示词、模型返回、token 消耗、耗时提示词注入、模型幻觉工具调用工具名、入参、返回值、耗时、异常参数错误、超时、返回异常结果输出最终回复、后处理逻辑、输出格式格式错误、内容不当状态流转任务状态变化、重试次数、转人工触发死循环、状态卡死这张表是我踩坑之后总结的一开始我只记了“工具调用”和“结果输出”结果模型推理那一段成了黑盒出了问题根本没法查。后来补上完整提示词记录才发现很多问题其实出在提示词拼接环节——比如上下文太长被截断导致模型“看不到”关键信息。注意记录完整提示词会涉及用户隐私和敏感数据生产环境一定要做脱敏处理或者只记录哈希值和长度需要排查时再通过安全通道获取原文。2.2 三层架构的分工我把整个可观测性体系分成三层每层职责清晰互不干扰第一层是采集层负责在各个执行节点埋点把原始数据抓出来。这一层的关键是“无侵入”——不能因为加了观测就拖慢主流程。我的做法是在执行网关里统一埋点所有 AI 员工的执行都经过网关网关负责在关键节点打点业务代码基本不用改。第二层是存储与索引层负责把采集到的数据存下来并且能快速检索。这里有个选型问题日志、追踪、指标要不要分开存我的经验是执行链路数据用结构化存储比如 JSON 文档或宽表指标数据用时序库两者通过 trace_id 关联。不要试图用一套系统解决所有问题否则查询会非常痛苦。第三层是重放与查询层负责把一次执行完整还原出来支持“重新跑一遍”。这一层是 AI 员工可观测性区别于传统系统的核心。传统系统重放靠的是“输入 代码版本”AI 员工重放还需要“模型版本 提示词版本 工具版本 随机种子”缺一不可。2.3 为什么选“执行网关”作为埋点中心市面上埋点方式主要有三种SDK 埋点、代理埋点、网关埋点。我最后选了网关埋点原因很实际SDK 埋点需要在每个 AI 员工里引入依赖改造成本高而且不同团队用的框架不一样维护起来很累。代理埋点比如在模型调用层做拦截能覆盖模型调用但覆盖不了工具调用和业务逻辑。网关埋点把所有执行请求统一收口不管底层用什么框架、调什么模型都经过网关埋点一次到位。具体做法是AI 员工的每次任务执行都向执行网关注册一个trace网关生成全局唯一的trace_id然后在每个关键节点模型调用前后、工具调用前后、输出前后记录结构化事件。这些事件按时间顺序串起来就是一次完整的执行链路。# 执行网关埋点的简化示意 class ExecutionGateway: def execute(self, task): trace_id generate_trace_id() self.record(trace_id, task_start, { input: task.input, context: task.context, prompt_version: task.prompt_version, model_version: task.model_version }) try: result self.run_pipeline(task, trace_id) self.record(trace_id, task_end, {result: result}) return result except Exception as e: self.record(trace_id, task_error, {error: str(e)}) raise这段代码只是示意实际实现要考虑异步、批量写入、失败降级等问题。核心思想是埋点逻辑集中在网关业务代码只负责传上下文。3. 核心细节日志、追踪、重放各自怎么落地3.1 日志不是越多越好而是要“结构化”很多人一提到记日志就是print或者logger.info(something happened)。这种日志在 AI 员工场景里基本没用因为你需要的是能检索、能聚合、能关联的结构化数据。我的做法是统一日志格式每条日志都是一个 JSON 对象包含固定字段{ trace_id: abc-123, span_id: span-001, timestamp: 2024-01-15T10:23:45.123Z, event_type: tool_call, tool_name: order_query, input: {order_id: O12345}, output: {status: not_found}, duration_ms: 234, status: success }这样存下来之后你可以按trace_id把一次执行的所有事件串起来也可以按tool_name统计某个工具的失败率还可以按duration_ms找出慢调用。结构化的价值在于它让日志从“给人看的文本”变成了“给系统查的数据”。实操心得字段命名一定要统一比如时间戳统一用 ISO 8601状态统一用success/failure/timeout不要一会儿ok一会儿success。我一开始没注意后来做聚合查询时被坑得很惨。3.2 追踪链路把一次执行串成一条线日志是点追踪是线。AI 员工的一次执行往往涉及多次模型调用和工具调用这些调用之间有先后依赖关系必须用trace_id和span_id把它们串起来。具体做法是网关生成trace_id后每次模型调用或工具调用都生成一个子span_id并记录parent_span_id。这样就能还原出一棵树trace: abc-123 ├── span: task_start ├── span: model_call_1 (判断问题类型) │ └── span: tool_call_1 (查询订单) ├── span: model_call_2 (生成回复) └── span: task_end有了这棵树排查问题时就能快速定位是哪一步出了岔子。比如前面那个订单号识别错误的问题一看tool_call_1的入参是O12345而不是012345立刻就知道是模型生成参数时出的错。这里有个细节要注意异步调用和并行调用的 span 关系要处理好。AI 员工有时候会并行调用多个工具这时候不能简单用父子关系要用“兄弟关系”加时间戳来还原并行结构。我试过用简单的顺序记录结果并行调用的耗时统计完全不对后来改成记录start_time和end_time用时间轴来还原才准确。3.3 可重放最难但最值钱的一层可重放是 AI 员工可观测性的“圣杯”。它的核心思想是把一次执行的所有输入、上下文、版本信息都存下来需要时能原样重新执行一遍。要实现重放必须记录这几类信息输入数据用户原始输入、会话历史、系统提示词版本信息模型版本、提示词版本、工具版本、代码版本执行参数温度、top_p、随机种子等外部依赖快照工具调用返回的数据因为外部系统可能已经变了最后一点特别关键。比如你重放一次订单查询如果直接调实时接口返回的可能是最新状态而不是当时的状态。所以工具调用的返回值必须快照存储重放时用快照而不是实时调用。# 重放时的工具调用拦截 class ReplayToolWrapper: def __init__(self, original_tool, snapshot_store): self.original_tool original_tool self.snapshot_store snapshot_store def call(self, trace_id, tool_name, params): if self.is_replaying: # 重放模式从快照读取 return self.snapshot_store.get(trace_id, tool_name, params) else: # 正常模式实时调用并快照 result self.original_tool.call(params) self.snapshot_store.save(trace_id, tool_name, params, result) return result这个设计的好处是重放时不会对外部系统产生副作用也不会因为外部数据变化导致重放结果不一致。注意重放不是万能的。如果 AI 员工的行为依赖实时数据比如股票价格重放只能还原“当时看到的数据”不能还原“如果当时数据不同会怎样”。这一点要在使用重放时心里有数。4. 实操过程从零搭一套可观测性体系4.1 第一步定义事件模型和存储结构动手写代码之前先把事件模型定下来。我建议用“事件表 快照表”两张表事件表存执行链路的所有事件字段包括trace_id、span_id、parent_span_id、event_type、timestamp、payloadJSON、duration_ms、status。快照表存工具调用的输入输出快照字段包括trace_id、tool_name、params_hash、params、result、timestamp。存储选型上如果数据量不大每天百万级事件以内用 PostgreSQL 的 JSONB 字段就够用查询也方便。数据量再大可以考虑 ClickHouse 或者 Elasticsearch。我一开始用的是 MongoDB后来发现聚合查询不如 PostgreSQL 顺手就换回来了。CREATE TABLE execution_events ( trace_id VARCHAR(64) NOT NULL, span_id VARCHAR(64) NOT NULL, parent_span_id VARCHAR(64), event_type VARCHAR(32) NOT NULL, timestamp TIMESTAMPTZ NOT NULL, payload JSONB, duration_ms INTEGER, status VARCHAR(16), PRIMARY KEY (trace_id, span_id) ); CREATE INDEX idx_trace ON execution_events(trace_id); CREATE INDEX idx_type_time ON execution_events(event_type, timestamp);索引很关键。我一开始只建了主键索引查询慢得离谱后来补上trace_id和event_type timestamp的索引查询速度从几秒降到几十毫秒。4.2 第二步在执行网关里埋点网关埋点的核心是“切面”。不管 AI 员工内部怎么实现网关在几个关键切面统一打点任务开始记录输入、上下文、版本信息模型调用前记录完整提示词模型调用后记录模型返回、token 消耗、耗时工具调用前记录工具名和入参工具调用后记录返回值和耗时任务结束记录最终输出和总耗时def run_pipeline(self, task, trace_id): # 模型调用埋点 self.record(trace_id, model_call_start, { prompt: task.prompt, model: task.model_version }) model_result self.model.call(task.prompt) self.record(trace_id, model_call_end, { output: model_result.text, tokens: model_result.usage, duration_ms: model_result.duration }) # 工具调用埋点 for tool_call in model_result.tool_calls: self.record(trace_id, tool_call_start, { tool: tool_call.name, params: tool_call.params }) tool_result self.tools.call(tool_call.name, tool_call.params) self.record(trace_id, tool_call_end, { result: tool_result, duration_ms: tool_result.duration })埋点写入要用异步批量方式不能阻塞主流程。我的做法是先把事件写到内存队列后台线程批量刷到数据库。这样即使数据库短暂不可用也不会影响 AI 员工的正常执行。4.3 第三步实现重放引擎重放引擎的核心是“拦截 替换”。正常执行时工具调用走实时接口并快照重放时工具调用走快照读取。class ReplayEngine: def replay(self, trace_id): # 加载原始执行的所有事件 events self.load_events(trace_id) # 提取输入和版本信息 task_start find_event(events, task_start) input_data task_start.payload[input] versions { model: task_start.payload[model_version], prompt: task_start.payload[prompt_version] } # 用快照模式重新执行 with self.snapshot_mode(trace_id): result self.gateway.execute( input_data, versionsversions, replayTrue ) return result重放的时候要注意模型调用本身可能不完全可复现即使温度设为 0不同批次的推理结果也可能有细微差异。所以重放的价值不在于“得到完全一样的结果”而在于“验证修复逻辑是否生效”。比如你改了参数校验逻辑重放时就能看到新的校验是否能拦住之前的错误参数。4.4 第四步搭一个查询面板数据存下来之后得有个地方能查。我用的是最简单的方案一个 Web 页面输入trace_id就能看到完整执行链路每个节点可以展开看详情。面板要支持几个核心功能按 trace_id 查询还原单次执行按时间范围 事件类型查询找特定类型的问题按工具名统计看哪个工具失败率高一键重放选中某次执行点击重放这个面板不需要多漂亮但一定要快。我见过太多团队花大力气做可视化结果查询慢得要死最后没人用。查询性能比界面美观重要十倍。5. 常见问题与排查技巧实录5.1 日志写入拖慢主流程怎么办这是最常见的问题。一开始我直接在业务代码里同步写数据库结果 AI 员工的响应时间从 2 秒涨到 5 秒。后来改成异步批量写入响应时间基本没受影响。具体做法用内存队列缓冲事件后台线程每 100 毫秒或每 100 条批量写入一次。队列满了就丢弃低优先级事件比如调试日志保证核心事件不丢。class AsyncEventWriter: def __init__(self, batch_size100, flush_interval0.1): self.queue Queue(maxsize10000) self.batch_size batch_size self.flush_interval flush_interval self.worker Thread(targetself._flush_loop, daemonTrue) self.worker.start() def write(self, event): try: self.queue.put_nowait(event) except QueueFull: # 队列满时丢弃调试级别事件 if event.level ! debug: self.queue.put(event, timeout0.1)实操心得队列大小要压测过再定。我一开始设了 1000结果高峰期丢了不少事件后来改成 10000 才稳住。但也不能无限大否则内存会爆。5.2 重放时结果不一致怎么排查重放结果不一致通常有三个原因原因表现排查方法模型版本不一致输出风格或内容差异大检查重放时用的模型版本是否和原始一致外部数据变化工具返回结果不同确认是否走了快照模式随机性细微差异检查温度、种子参数是否一致我遇到最多的是第二种重放时忘了走快照直接调了实时接口结果外部数据已经变了。解决办法是在重放引擎里强制拦截所有工具调用不允许走实时接口。5.3 敏感数据怎么处理完整记录提示词和工具返回值必然会涉及敏感数据。我的做法是三层处理第一层字段级脱敏。手机号、身份证号、银行卡号等敏感字段在写入前替换成掩码。第二层访问控制。查询面板需要权限普通开发只能看脱敏后的数据只有特定角色能申请查看原文。第三层定期清理。原始数据保留 30 天之后只保留脱敏版本和统计指标。注意脱敏逻辑要放在网关层统一做不能指望每个业务自己处理。我见过有团队让业务代码自己脱敏结果漏了好几个字段出了安全事故。5.4 常见问题速查表问题可能原因解决方向查不到某次执行的日志trace_id 生成失败或写入丢失检查网关埋点是否覆盖所有入口日志时间戳乱序异步写入导致用事件产生时间而非写入时间排序重放报错找不到快照快照未保存或已清理检查快照保存逻辑和保留策略查询面板加载慢索引缺失或数据量过大补索引考虑冷热数据分离工具调用耗时统计不准并行调用未正确处理用 start/end 时间戳而非顺序累加6. 几个我踩过的坑和最后的体会6.1 不要一开始就追求大而全我最初的设计想把日志、指标、追踪、重放全做进去结果做了两个月还没上线。后来砍掉指标和复杂追踪先做“结构化日志 基础重放”两周就上线了。上线之后根据实际需求再逐步补反而效率更高。可观测性体系是长出来的不是设计出来的。6.2 重放的价值在“验证”不在“复现”很多人对重放的期待是“完全复现当时的结果”但实际上由于模型的随机性完全复现很难。重放真正的价值是当你修复了一个 bug可以用历史数据验证修复是否有效。比如你加了一个参数校验规则重放历史执行看有多少次会被新规则拦住这比写单元测试更贴近真实场景。6.3 存储成本要提前算完整记录提示词和工具返回值数据量增长很快。我算过一笔账一次执行平均产生 5KB 数据每天 10 万次执行就是 500MB一个月 15GB。如果保留一年就是 180GB。这还不算索引和备份。所以保留策略一定要提前定原始数据保留 30 天聚合指标长期保留这样成本可控。6.4 最后分享一个小技巧如果你刚开始做不知道从哪下手可以先从“记录完整提示词”开始。这是投入最小、收益最大的一步。很多 AI 员工的问题光看提示词就能定位一大半。等这一步跑顺了再逐步加工具调用记录、追踪链路、重放能力。一步一步来比一上来就搞大架构靠谱得多。这套东西我们跑了半年多线上问题的平均定位时间从原来的两三个小时降到了二十分钟以内。最明显的变化是以前出了问题大家互相甩锅现在直接甩 trace_id谁的问题一目了然。这大概就是可观测性最实在的价值。