Agent 一旦上了重试（463）、并发（465）和缓存（466），行为就变得难以肉眼追踪：一次任务背后可能是几十次模型调用、上百次工具执行。出了问题，「它刚才到底干了什么」全靠猜。Langfuse 这类观测平台很好，但私有数据不出内网的要求、或者只是想快速定位一个坏步骤时，自建一套轻量回放器更直接：把运行记录成 JSONL 事件流，事后逐帧回放。本篇用不到 150 行代码实现。
Step 1：定义事件模型——哪些时刻值得记录
回放器的前提是埋点模型。一次 Agent 运行至少有五类值得记录的事件：
事件类型 触发时机 关键字段
run_started 任务开始 run_id, 任务描述
llm_call 发起模型请求 run_id, 消息长度, 模型名
llm_response 模型返回 run_id, 是否要求调工具, token 用量
tool_call 工具开始执行 run_id, 工具名, 参数
tool_result 工具返回 run_id, 工具名, 耗时ms, 成败
run_finished 任务结束 run_id, 状态, 总轮数
run_error 任务异常终止 run_id, 错误摘要每条事件固定带三样东西：全局 run_id（串起一次任务）、单调递增 seq（排出先后）、UTC 时间戳。回放的所有功能——时间线、过滤、断点重建——都建立在这三样之上。
Step 2：写记录器——一个函数管所有埋点
import json, time, uuid, os
LOG_DIR = "replay_logs"
os.makedirs(LOG_DIR, exist_ok=True)
class EventLog:
def __init__(self, run_id=None):
self.run_id = run_id or uuid.uuid4().hex[:12]
self.seq = 0
self.path = os.path.join(LOG_DIR, f"{self.run_id}.jsonl")
self.f = open(self.path, "a", encoding="utf-8")
def emit(self, etype: str, **fields):
self.seq += 1
rec = {"seq": self.seq, "ts": time.strftime("%H:%M:%S"),
"run": self.run_id, "type": etype, **fields}
self.f.write(json.dumps(rec, ensure_ascii=False) + "
")
self.f.flush() # 崩溃也不丢已发生的事件
return rec
def close(self):
self.f.close()flush 不能省。Agent 任务经常因为超时被外部杀掉，不带 flush 的缓冲区会丢掉最后几条事件——而排查问题时最有价值的恰恰是「死前最后一步」。也别用 O_APPEND 裸写多进程：并发场景要么每进程独立文件，要么加文件锁。
Step 3：给 461 的执行循环埋点
以 461 的 ReAct 循环为例，埋点位置天然对齐事件模型：
def run_agent(task: str, max_steps: int = 10):
log = EventLog()
log.emit("run_started", task=task[:200])
messages = [{"role": "user", "content": task}]
try:
for step in range(max_steps):
log.emit("llm_call", step=step, n_msgs=len(messages))
resp = chat(messages) # 461 的模型调用
msg = resp.choices[0].message
log.emit("llm_response", step=step,
wants_tool=bool(getattr(msg, "tool_calls", None)))
if not getattr(msg, "tool_calls", None):
log.emit("run_finished", status="ok", steps=step + 1)
return msg.content
for tc in msg.tool_calls:
name, args = tc.function.name, tc.function.arguments
t0 = time.perf_counter()
log.emit("tool_call", step=step, tool=name, args=args[:300])
result = dispatch(name, *parse(args)) # 466 的 dispatch
log.emit("tool_result", step=step, tool=name,
ms=int((time.perf_counter() - t0) * 1000),
ok=not str(result).startswith("错误"))
messages.append(tool_msg(tc.id, result))
except Exception as e:
log.emit("run_error", err=str(e)[:300])
raise
finally:
log.close()注意 args 截断到 300 字符：日志是给人看的时间线，不是数据库，塞满全文只会拖慢 grep。需要全文时在工具内部自己记。
Step 4：回放器——把事件流还原成时间线
def replay(run_id: str):
path = os.path.join(LOG_DIR, f"{run_id}.jsonl")
events = [json.loads(line) for line in open(path, encoding="utf-8")]
for e in events:
icon = {"llm_call": "🧠", "llm_response": "↩️",
"tool_call": "🛠️", "tool_result": "✅",
"run_finished": "🏁", "run_error": "❌"}.get(e["type"], "·")
line = f"{e['seq']:>3} {e['ts']} {icon} {e['type']}"
if e["type"] == "tool_call":
line += f" {e['tool']}({e['args'][:80]})"
if e["type"] == "tool_result":
line += f" {e['tool']} {e['ms']}ms {'OK' if e['ok'] else 'FAIL'}"
print(line)
print(f"共 {len(events)} 个事件")
replay("a3f9c2e1b7d4")输出长这样：
1 10:21:03 🏃 run_started
2 10:21:03 🧠 llm_call
3 10:21:05 ↩️ llm_response
4 10:21:05 🛠️ tool_call search("北京 天气")
5 10:21:06 ✅ tool_result search 812ms OK
6 10:21:06 🧠 llm_call
...
14 10:21:11 🏁 run_finished哪一步慢、哪个工具反复被调、任务在哪轮终止，一目了然。
Step 5：过滤定位——三个常用切片
def filter_events(run_id, etype=None, tool=None, min_ms=None, failed=False):
path = os.path.join(LOG_DIR, f"{run_id}.jsonl")
out = []
for line in open(path, encoding="utf-8"):
e = json.loads(line)
if etype and e["type"] != etype: continue
if tool and e.get("tool") != tool: continue
if min_ms and e.get("ms", 0) < min_ms: continue
if failed and e["type"] == "tool_result" and e["ok"]: continue
out.append(e)
return out
# 用法示例：
slow = filter_events(rid, etype="tool_result", min_ms=2000) # 谁拖慢了任务
bad = filter_events(rid, failed=True) # 哪些调用失败过
freq = {} # 工具调用频率
for e in filter_events(rid, etype="tool_call"):
freq[e["tool"]] = freq.get(e["tool"], 0) + 1实践中最有用的组合是「失败事件 + 前后各一条」：tool_result 为 FAIL 的 seq 前后各取一条 tool_call，就能还原「模型传了什么参数导致失败」，直接指导提示词或参数清洗的修改。
Step 6：从事件流重建上下文——断点续调
事件流记录了每次工具调用与结果，逆操作就能重建当时的消息列表：把 tool_call 与 tool_result 按序还原成 assistant 的 tool_calls 消息和 tool 角色消息，再接上一条「请基于以上进度继续」的用户消息，Agent 就能从断点接着跑，不用从头重付前几轮的 token：
def rebuild_messages(run_id, upto_seq=None):
events = [json.loads(l) for l in
open(os.path.join(LOG_DIR, f"{run_id}.jsonl"), encoding="utf-8")]
if upto_seq:
events = [e for e in events if e["seq"] <= upto_seq]
msgs, pending = [], []
for e in events:
if e["type"] == "tool_call":
pending.append({"id": f"c{e['seq']}", "type": "function",
"function": {"name": e["tool"], "arguments": e["args"]}})
elif e["type"] == "tool_result" and pending:
# 简化演示：一个调用一条结果；严格实现需按 id 配对
msgs.append({"role": "assistant", "tool_calls": pending})
msgs.append({"role": "tool", "tool_call_id": f"c{e['seq']}",
"content": str(e.get("tool", ""))})
pending = []
return msgs简化版没有保存完整的 assistant 文本与参数原文（args 被截断过）。要在生产使用断点重建，事件里就得存全量参数，或者把原始 messages 也按轮落盘。这是「日志瘦身」与「可重建」之间的取舍，按团队需求定，别两头都想要结果两头都做不精。
Step 7：生产化前的三件事
其一，脱敏：事件里可能出现用户手机号、密钥参数，写入前过一遍正则替换（把 11 位手机号、sk- 开头的串打成掩码），日志泄露是合规红线。其二，轮转与清理：按 run_id 一文件天然好清理，再配一个「保留 14 天」的定时删除即可，防止磁盘被日志吃满。其三，与平台观测的关系：本方案解决的是「快速定位坏步骤」，不提供面板、告警与团队协作；等团队规模或合规要求上来了，把同一套事件模型对接 Langfuse 或 OTLP（语义约定几乎一致），迁移成本很低——这也是为什么事件模型要在一开始就设计清楚。
脱敏正则务必在写入路径上而不是回放路径上：写进去之前就洗干净。回放路径上做脱敏意味着原始敏感数据一直躺在磁盘上，等于没做。
至此工具调用工程系列四篇成套：461 执行循环、463 失败恢复、465 并发吞吐、466 重复治理，加上本篇的行为可观测。五块拼起来，就是一个能扛住真实生产的工具层骨架——先有骨架，再谈框架选型，顺序别反。