AI员工可观测性实战：基于执行网关的日志采集与重放体系

1. 为什么“能跑”的AI员工系统，最后都卡在了“说不清”上

做AI员工系统的团队，几乎都会经历同一个阶段：Demo跑通那一刻，所有人都觉得这事成了。Agent能接需求、能调工具、能写文件、能发消息，流程串起来像模像样。可一旦把它放进真实业务里跑上两周，问题就全冒出来了——用户说“昨天那单它明明答应了，今天怎么不认账”，你打开系统一看，只有几行零散的print输出，连当时喂给模型的完整上下文都拼不出来。这时候你才意识到，AI员工和传统程序最大的区别，不是它会不会犯错，而是它犯错之后你根本不知道它当时“看到了什么、想了什么、做了什么”。

这就是可观测性要解决的问题。传统后端服务的可观测性有三大支柱：日志、指标、链路追踪。但AI员工系统多了一层更麻烦的东西——决策过程。一个普通接口调用失败，你看堆栈就知道哪一行崩了；可AI员工回答“我建议退款”，你没法从堆栈里看出它是因为检索到了哪条政策、参考了哪段历史对话、还是单纯被某句话带偏了。所以给AI员工加可观测性，不能照搬传统那套，得从“记日志”升级到“可重放”。

我先把结论摆在这：可观测性的终点不是让你看到更多日志，而是让你能在任意一个历史时刻，把当时的完整执行现场重新跑一遍。日志是原料，重放才是成品。这篇文章我会围绕“执行网关”这个核心组件，讲清楚怎么从零搭一套能记、能查、能重放的AI员工可观测体系，包括数据结构怎么设计、日志怎么采集、重放怎么做到幂等、以及我在实际项目里踩过的那些坑。适合正在做Agent系统、AI工作流平台、或者任何“模型驱动决策”类产品的同学参考，不管你现在是用Python手搓还是用现成框架，这套思路都能落地。

2. 整体设计思路：把AI员工的每次执行当成一次“可回放的会话”

2.1 核心矛盾：AI员工的执行现场是“有状态且不可逆”的

传统服务记日志，记的是“发生了什么”，比如user 123 login success。但AI员工的执行现场要复杂得多，它至少包含四层信息：输入层（用户说了什么、系统注入了什么上下文）、决策层（模型收到了什么prompt、返回了什么、中间调了哪些工具）、执行层（工具真实执行的结果、外部系统的返回）、状态层（这次执行前后，会话状态、记忆、变量发生了什么变化）。这四层里任何一层缺失，重放就会失真。

更麻烦的是，AI员工的执行往往是不可逆的。比如它调了一个“发送邮件”的工具，邮件真发出去了，你重放的时候不能真再发一次。所以重放机制必须区分“只读重放”和“副作用重放”，前者用于排查问题，后者用于回归测试，两者对日志的要求完全不同。我在设计时把这条作为第一原则：日志必须记录“决策依据”和“执行结果”，但重放时默认只回放决策依据，副作用操作走mock。

2.2 为什么选“执行网关”作为切入点

你可能会问，为什么不直接在Agent代码里埋点，非要搞一个网关？我试过直接埋点，结论是：埋点会污染业务逻辑，而且很难保证一致性。Agent的代码本来就在快速迭代，今天加个工具、明天改个prompt，你要求每个开发都在正确的位置打正确的日志，现实里根本做不到。执行网关的思路是：所有AI员工的执行请求，都必须经过一个统一的入口，由网关负责记录、转发、重放。业务代码只管写逻辑，可观测性由网关兜底。

这个网关的职责很明确：接收执行请求（包含会话ID、输入、上下文快照）、生成全局trace ID、按顺序记录每个步骤（模型调用、工具调用、状态变更）、把请求转发给真正的执行器、收集执行结果并落盘。它不关心Agent内部怎么实现，只关心“输入是什么、经过了哪些步骤、输出是什么”。这样一来，无论你用LangChain、AutoGPT还是自己手搓的循环，只要走网关，就自动获得可观测性。

2.3 日志结构设计：从“文本行”到“结构化事件流”

传统日志是文本行，grep一下还行，但要做重放就太弱了。我采用的是结构化事件流，每次执行生成一个trace，trace下挂多个span，每个span是一个原子事件。核心字段包括：

这个结构的好处是，重放时只需要按timestamp排序，依次把input_snapshot喂给对应的执行器，就能还原整个决策链。对于replayable=false的span，重放时直接返回记录的output_snapshot，不真正执行。这样既保证了重放的完整性，又避免了副作用。

2.4 存储选型：别一上来就上ELK，先想清楚查询模式

很多人一提到日志就想到ELK，但AI员工的可观测性查询模式和传统日志完全不同。传统日志是“按关键词搜”，AI员工是“按trace_id查完整链路”和“按会话ID查历史执行”。前者是点查，后者是范围查。我的选型是：热数据用SQLite或PostgreSQL，冷数据归档到对象存储。SQLite足够支撑单机每天几十万次执行的写入，查询用索引优化后毫秒级返回。等量级上来了再换ClickHouse，但别一开始就过度设计。

提示：如果你用PostgreSQL，建议把input_snapshot和output_snapshot存成jsonb，这样可以直接用SQL查“所有调用了退款工具且金额大于100的执行”，排查问题时非常方便。

3. 核心细节解析：日志采集、脱敏与重放的三道坎

3.1 日志采集：别让日志成为系统的性能瓶颈

采集这块我踩过最大的坑是同步写日志拖慢主流程。最开始我在网关里直接await db.insert(log)，结果模型调用本身200ms，写日志花了80ms，整体延迟涨了40%。后来改成异步队列+批量落盘：网关把日志事件推到内存队列，后台协程每100ms或每满100条批量写入。这样对主流程的影响降到1ms以内。

具体实现上，我用的是Python的asyncio.Queue加一个消费者协程。关键点是队列要有上限，防止日志堆积把内存打爆。我的配置是maxsize=10000，满了就丢弃低优先级的debug日志，保留error和model_call。另外，日志写入失败不能影响主流程，所有写操作包在try里，失败只打warning。

import asyncio import json from datetime import datetime class LogCollector: def __init__(self, db_pool, maxsize=10000): self.queue = asyncio.Queue(maxsize=maxsize) self.db_pool = db_pool self.batch_size = 100 self.flush_interval = 0.1 async def emit(self, event: dict): try: self.queue.put_nowait(event) except asyncio.QueueFull: # 队列满时降级，只保留关键事件 if event.get("event_type") in ("error", "model_call"): # 强制挤入，丢弃最老的 try: self.queue.get_nowait() self.queue.put_nowait(event) except Exception: pass async def consumer(self): batch = [] while True: try: event = await asyncio.wait_for( self.queue.get(), timeout=self.flush_interval ) batch.append(event) if len(batch) >= self.batch_size: await self._flush(batch) batch = [] except asyncio.TimeoutError: if batch: await self._flush(batch) batch = [] async def _flush(self, batch): try: async with self.db_pool.acquire() as conn: await conn.executemany( "INSERT INTO traces (...) VALUES (...)", batch ) except Exception as e: print(f"log flush failed: {e}")

3.2 脱敏：日志里不能出现的东西，比你想的多

AI员工的日志里会包含大量敏感信息：用户的原始输入、模型的完整输出、工具返回的业务数据。我见过最离谱的案例是有人把用户的身份证号直接写进了日志，结果日志文件被运维同学随手发到了群里。脱敏必须在采集阶段做，不能等到查询时再过滤，因为一旦落盘就有泄露风险。

我的脱敏策略分三层：第一层是字段级脱敏，对已知的敏感字段（手机号、身份证、银行卡、邮箱）用正则替换；第二层是模型输出脱敏，因为模型可能把敏感信息复述出来，所以对output_snapshot整体做一次敏感词扫描；第三层是工具返回脱敏，外部系统返回的数据往往包含业务敏感信息，需要在网关层做白名单过滤，只保留排查问题必需的字段。

注意：脱敏规则要可配置，不同业务线的敏感字段不一样。我建议把规则做成YAML文件，支持热加载，这样新增敏感字段不用重启服务。

3.3 重放：最难的不是“重跑”，而是“跑得一样”

重放的核心挑战是幂等性和确定性。幂等性好理解，副作用操作不能真执行。确定性就麻烦了：同样的输入，模型可能返回不同的结果，因为温度参数、模型版本、甚至服务端的负载均衡都会影响输出。我的做法是重放时强制使用记录中的模型输出，也就是说，重放不是重新调模型，而是用日志里记录的output_snapshot来驱动后续流程。这样重放的结果是确定的，用于排查“当时为什么走了这条分支”非常有效。

如果你确实需要重新调模型做回归测试，那就走另一条路：影子重放。把历史输入喂给新版本的Agent，对比新旧输出的差异，但不影响线上状态。这两种重放模式我都在网关里实现了，通过一个replay_mode参数切换。

class ReplayEngine: def __init__(self, trace_store, executor_registry): self.trace_store = trace_store self.executors = executor_registry async def replay(self, trace_id: str, mode: str = "deterministic"): spans = await self.trace_store.get_spans(trace_id) spans.sort(key=lambda s: s["timestamp"]) context = {} for span in spans: if span["event_type"] == "model_call": if mode == "deterministic": # 直接用记录的输出，不重新调模型 output = span["output_snapshot"] else: # 影子模式，重新调模型 output = await self._call_model(span["input_snapshot"]) context["last_model_output"] = output elif span["event_type"] == "tool_call": if span["replayable"]: executor = self.executors.get(span["metadata"]["tool_name"]) output = await executor.execute(span["input_snapshot"]) else: output = span["output_snapshot"] context["last_tool_output"] = output elif span["event_type"] == "state_change": context.update(span["output_snapshot"]) return context

3.4 上下文快照：重放的“燃料”

没有上下文快照，重放就是空谈。但快照不能太大，否则存储成本爆炸。我的策略是增量快照+全量锚点：每10次执行做一次全量快照，中间的执行只记录增量变化。重放时先加载最近的全量锚点，再依次应用增量。这样存储成本降低80%以上，重放速度也快很多。

快照的内容包括：会话历史（最近N轮对话）、记忆向量（如果有）、系统变量、用户画像。其中记忆向量比较特殊，它通常是高维数组，直接存JSON会很大。我的做法是存向量ID+版本号，重放时从向量库按ID取回。如果向量库里的数据已经过期，就降级为“无记忆重放”，并在日志里标记memory_missing=true。

4. 实操过程：从零搭一套可重放的执行网关

4.1 环境准备与依赖安装

我假设你用的是Python 3.10+，数据库用PostgreSQL 14+（SQLite也行，但并发写入弱一些）。核心依赖就三个：asyncpg用于异步数据库操作，pydantic用于数据结构校验，fastapi用于网关的HTTP接口（如果你需要跨语言调用）。安装命令：

pip install asyncpg pydantic fastapi uvicorn

数据库建表SQL我简化了一下，核心就两张表：traces存执行元信息，spans存每个步骤的详情。索引建在trace_id、session_id、timestamp上，查询时走索引。

CREATE TABLE traces ( trace_id VARCHAR(64) PRIMARY KEY, session_id VARCHAR(64) NOT NULL, agent_id VARCHAR(64) NOT NULL, start_time BIGINT NOT NULL, end_time BIGINT, status VARCHAR(16) DEFAULT 'running', metadata JSONB ); CREATE INDEX idx_traces_session ON traces(session_id); CREATE INDEX idx_traces_time ON traces(start_time); CREATE TABLE spans ( span_id VARCHAR(64) PRIMARY KEY, trace_id VARCHAR(64) NOT NULL, parent_span_id VARCHAR(64), event_type VARCHAR(32) NOT NULL, timestamp BIGINT NOT NULL, input_snapshot JSONB, output_snapshot JSONB, metadata JSONB, replayable BOOLEAN DEFAULT TRUE ); CREATE INDEX idx_spans_trace ON spans(trace_id, timestamp);

4.2 网关核心逻辑：拦截、记录、转发

网关的核心是一个中间件，它拦截所有执行请求，生成trace_id，然后按顺序记录每个步骤。这里的关键是步骤的边界怎么定。我的做法是：Agent每调用一次模型或工具，就产生一个span。网关不关心Agent内部怎么循环，只关心“你调了什么、传了什么、返回了什么”。

from fastapi import FastAPI, Request import uuid, time app = FastAPI() collector = LogCollector(db_pool) @app.post("/execute") async def execute(request: Request): body = await request.json() trace_id = str(uuid.uuid4()) session_id = body["session_id"] agent_id = body["agent_id"] # 记录trace开始 await collector.emit({ "trace_id": trace_id, "event_type": "trace_start", "timestamp": int(time.time() * 1000), "input_snapshot": body, "metadata": {"agent_id": agent_id, "session_id": session_id} }) # 转发给真正的执行器 executor = get_executor(agent_id) try: result = await executor.run(body, trace_id=trace_id) status = "success" except Exception as e: result = {"error": str(e)} status = "failed" # 记录trace结束 await collector.emit({ "trace_id": trace_id, "event_type": "trace_end", "timestamp": int(time.time() * 1000), "output_snapshot": result, "metadata": {"status": status} }) return {"trace_id": trace_id, "result": result}

执行器内部调模型和工具时，通过一个span_context把日志推给collector。这里我用了一个装饰器来简化埋点：

def traced(event_type: str, replayable: bool = True): def decorator(func): async def wrapper(*args, trace_id: str, **kwargs): span_id = str(uuid.uuid4()) input_snapshot = {"args": str(args), "kwargs": str(kwargs)} start = int(time.time() * 1000) try: output = await func(*args, **kwargs) await collector.emit({ "span_id": span_id, "trace_id": trace_id, "event_type": event_type, "timestamp": start, "input_snapshot": input_snapshot, "output_snapshot": output, "replayable": replayable, "metadata": {"duration": int(time.time()*1000) - start} }) return output except Exception as e: await collector.emit({ "span_id": span_id, "trace_id": trace_id, "event_type": "error", "timestamp": start, "input_snapshot": input_snapshot, "output_snapshot": {"error": str(e)}, "replayable": False, "metadata": {"duration": int(time.time()*1000) - start} }) raise return wrapper return decorator

4.3 重放接口：一键还原历史执行

重放接口接收trace_id和mode，返回重放后的最终状态。对于排查问题，我通常用deterministic模式，快速定位是哪一步的输入导致了错误输出。对于回归测试，用shadow模式，对比新旧Agent的行为差异。

@app.post("/replay/{trace_id}") async def replay(trace_id: str, mode: str = "deterministic"): engine = ReplayEngine(trace_store, executor_registry) result = await engine.replay(trace_id, mode=mode) return {"trace_id": trace_id, "mode": mode, "result": result}

实测下来，一次包含20个span的trace，重放耗时在50ms以内，比重新跑一遍Agent快了两个数量级。这对于线上问题排查来说，体验提升是巨大的——以前要复现一个bug得折腾半小时，现在输入trace_id，3秒看到完整决策链。

4.4 查询面板：让非技术人员也能看懂

日志和重放能力有了，但你不能指望产品经理去写SQL。我搭了一个简单的查询面板，支持按会话ID、时间范围、状态筛选，点进去能看到trace的完整时间线，每个span可以展开看输入输出。面板用FastAPI的Jinja2模板渲染，前端就一个HTML文件加几行JS，不依赖任何前端框架。

面板的核心功能是时间线视图：横轴是时间，每个span是一个节点，模型调用用蓝色、工具调用用绿色、错误用红色。点击节点展开详情，支持一键重放。这个面板上线后，客服团队自己就能查“用户投诉的那次对话到底发生了什么”，不用再找研发。

5. 常见问题与排查技巧实录

5.1 日志丢失：异步队列的“最后一公里”

异步队列最大的风险是进程退出时队列里还有没落盘的日志。我遇到过服务被kill -9，最后200条日志全丢了，偏偏那200条里就有出错的关键信息。解决办法是注册atexit钩子和信号处理，在进程退出前强制flush队列。另外，对于error级别的日志，我改成同步写入，牺牲一点性能换可靠性。

import atexit, signal def graceful_shutdown(): loop = asyncio.get_event_loop() loop.run_until_complete(collector.flush_all()) atexit.register(graceful_shutdown) signal.signal(signal.SIGTERM, lambda *_: graceful_shutdown())

5.2 重放结果不一致：时间戳和随机数在作怪

重放时如果Agent内部用了datetime.now()或random.random()，结果就会和原始执行不一致。我的做法是在网关层注入虚拟时钟和固定随机种子，重放时用记录的时间戳和种子，保证确定性。具体实现是在input_snapshot里带上virtual_time和random_seed，执行器从上下文里取，而不是直接调系统函数。

5.3 存储膨胀：日志保留策略要提前定

AI员工的日志量比传统服务大得多，因为每次执行都包含完整的prompt和输出。我算过一笔账：单次执行平均产生5KB日志，每天1万次执行就是50MB，一个月1.5GB。如果不加控制，半年就把数据库撑爆了。我的策略是分级保留：error和model_call保留90天，tool_call保留30天，debug日志保留7天。归档时把冷数据导出到对象存储，数据库只留索引。

5.4 常见问题速查表

5.5 独家避坑技巧

第一个技巧：给每个span加一个checksum，对input_snapshot做哈希。重放时校验checksum，如果不一致说明日志被篡改或损坏，直接报错而不是静默重放。这个在排查“为什么重放结果不对”时特别有用。

第二个技巧：重放时开启“差异对比”模式，把重放过程中每个span的输出和原始输出做diff，高亮不一致的地方。我靠这个功能定位过好几次“模型版本升级导致行为变化”的问题。

第三个技巧：别把日志当数据库用。我见过有人把业务状态直接存在日志里，查询时从日志里拼状态，结果日志一归档业务就挂了。日志就是日志，业务状态该存哪存哪，两者不要混。

6. 后续可以怎么扩展

这套网关跑稳定之后，我陆续加了一些扩展。一个是实时告警：对error span做实时统计，5分钟内错误率超过阈值就发通知。另一个是执行回放录制：把重放过程录成视频或GIF，用于团队复盘和新人培训。还有一个比较有意思的是反事实重放：修改历史trace里的某个span输入，看后续决策会怎么变，用于分析“如果当时检索到的是另一条政策，结果会不会不同”。

如果你也在做AI员工系统，我的建议是可观测性别等到出问题才做。前期多花两天把网关搭好，后期能省下几十个小时的排查时间。而且这套东西一旦跑起来，你会发现它不只是排查工具，更是理解Agent行为的显微镜——很多你以为是模型“抽风”的问题，重放一遍就发现是上下文注入错了或者工具返回格式变了。看得见，才管得住。