Agent 做错了,你怎么知道它错在哪一步?
给 agent loop 加 trace:一次任务是一个 trace,每次模型调用、每次工具调用是一个 span,记下输入、输出、token、耗时和错误。
只看最终输出,你只知道它错了。看 trace,才知道错在第几步、是模型还是工具、花了多少钱。
1只看最终输出,定位不到问题
一个退款 Agent,用户说"帮我把订单 A123 退款"。它最后回了一句:"抱歉,退款失败,请联系人工客服。"
这句话背后至少有四种完全不同的原因:
- 查订单的接口超时了,模型没有重试;
- 模型把金额 129.9 读成了 1299,退款接口拒绝;
- 模型一直调用搜索工具、一直搜不到,最后被轮数上限截断;
- 退款接口本身挂了。
最终输出一模一样,修法完全不同:第一种改重试策略,第二种改 prompt 或工具描述,第三种加循环检测,第四种找后端。只有最终输出,你只能猜。
跟第 1 周的联系:rule of three、Wilson、pass^k 都在回答"失败率是多少"。trace 回答的是下一个问题:失败的那几次,到底怎么失败的。
2trace 和 span
这两个词来自分布式追踪(OpenTelemetry 用的就是这套概念):
- trace:一次完整的任务。用户发来一个请求,Agent 从开始跑到结束,就是一个 trace,有一个
trace_id。 - span:trace 里的一步。每个 span 有自己的
span_id、父 span(parent_id)、开始和结束时间、状态(成功 / 出错)和一组属性。
一个 agent loop 的 trace 通常长这样:
invoke_agent ← 根 span:整个任务
├── chat claude-sonnet-5-5 ← 第 1 轮模型调用,决定调 get_order
├── execute_tool get_order ← 工具调用
├── chat claude-sonnet-5-5 ← 第 2 轮,决定调 refund
├── execute_tool refund
└── chat claude-sonnet-5-5 ← 第 3 轮,stop_reason = end_turn,结束
把每个 span 按开始时间画成横条,就是瀑布图(waterfall):横轴是时间,一眼能看出哪一步最慢、哪一步出错、一共走了几轮。下面的互动演示就是一个瀑布图查看器。
OpenTelemetry 的 GenAI 语义约定
属性叫什么名字,不用自己发明。OpenTelemetry 有一套 GenAI 语义约定,规定了模型调用、工具调用、Agent 调用的 span 该怎么命名、带哪些属性:
| span | span 名 | gen_ai.operation.name |
|---|---|---|
| 一次模型调用 | {operation} {model},如 chat claude-sonnet-5-5 | chat |
| 一次工具调用 | execute_tool {tool.name} | execute_tool |
| 一次 Agent 调用 | invoke_agent {agent.name},没有名字时就是 invoke_agent | invoke_agent |
open-telemetry/semantic-conventions-genai 仓库维护。本页的属性名按这个仓库 2026 年 9 月底的版本核对过。用的时候锁定版本,升级时对一遍变更记录。
3每个 span 该记什么
| 记什么 | 对应的 OTel 属性 | 用来回答 |
|---|---|---|
| 这一步是什么操作 | gen_ai.operation.name | 这一步是模型还是工具 |
| 提供方、模型 ID | gen_ai.provider.name、gen_ai.request.model、gen_ai.response.model | 换模型前后对比;回放时锁定版本 |
| token 用量 | gen_ai.usage.input_tokens(含缓存部分)、gen_ai.usage.output_tokens、gen_ai.usage.cache_read.input_tokens、gen_ai.usage.cache_write.input_tokens(后者旧名 gen_ai.usage.cache_creation.input_tokens,改名还没发版,只能锁 commit) | 成本 |
| 停止原因 | gen_ai.response.finish_reasons | 正常结束、要调工具,还是被 max_tokens 截断 |
| 输入 / 输出消息 | gen_ai.input.messages、gen_ai.output.messages | 模型"看到了什么、说了什么" |
| 工具名、参数、结果 | gen_ai.tool.name、gen_ai.tool.call.arguments、gen_ai.tool.call.result | 是参数填错了,还是工具坏了 |
| 错误 | error.type(这个属性是 Stable)+ span 状态 | 失败率、失败分类 |
| 延迟 | span 自带的开始、结束时间,不是属性 | 慢在哪一步 |
用 Claude API 时,这些值都在响应里:response.usage.input_tokens / output_tokens、response.stop_reason(end_turn、tool_use、max_tokens 等)、response.model、response.id。
usage.input_tokens 只是没命中缓存的那部分;开了 prompt caching,还有 cache_creation_input_tokens(写缓存)和 cache_read_input_tokens(读缓存)。OTel 规范的 Anthropic 部分要求 gen_ai.usage.input_tokens = input_tokens + cache_read_input_tokens + cache_creation_input_tokens,并把后两项分别记到 gen_ai.usage.cache_read.input_tokens 和 gen_ai.usage.cache_write.input_tokens。直接把 Claude 的 input_tokens 写进去,开缓存时输入会少算一大截。成本怎么从 token 算
$$\text{成本} = \sum_{\text{模型 span}} \frac{\text{未缓存输入} \times p_{\text{输入}} + \text{缓存写} \times p_{\text{缓存写}} + \text{缓存读} \times p_{\text{缓存读}} + \text{输出} \times p_{\text{输出}}}{10^6}$$未缓存输入 = Claude 的 usage.input_tokens = OTel 的 input_tokens − 缓存读 − 缓存写。单价的单位是"美元 / 百万 token",不同模型、输入和输出、缓存写和缓存读都不一样(缓存写比普通输入贵,缓存读便宜得多),而且会调整。不要把价格写死在代码里,放进配置,从官方价格表(Claude 的在 platform.claude.com/docs/en/about-claude/pricing)抄过来,并记下抄的日期。
隐私:先脱敏再落盘
输入输出消息、工具参数和结果里常有手机号、邮箱、地址、订单金额。OTel 规范把 gen_ai.input.messages、gen_ai.output.messages、gen_ai.tool.call.arguments、gen_ai.tool.call.result 都定为 Opt-In:插桩库默认不记录,要用户主动打开。规范给的三种做法:
- 默认:不记内容,只记 token、耗时、状态这些元数据。
- 把内容记在 span 属性上:适合预发环境,或者存储本身满足隐私合规要求。
- 内容存到外部存储,span 上只记引用:生产环境推荐,可以单独做访问控制。
自己写记录器时,至少在写盘前过一遍脱敏函数(下面的代码就这么做)。注意:脱敏的是落盘的 trace,模型本身拿到的还是原文。
4最小实现:50 行 JSONL 记录器
不用框架。一个 span 就是一个 with 块:进入时记开始时间和父 span,退出时记结束时间,往 JSONL 文件追加一行。父子关系用一个栈维护:栈顶就是当前的父 span。
import json, re, time, uuid
from contextlib import contextmanager
EMAIL = re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+")
PHONE = re.compile(r"1\d{10}")
def redact(value):
"""脱敏:落盘前把邮箱、手机号换掉。"""
if isinstance(value, str):
return PHONE.sub("<phone>", EMAIL.sub("<email>", value))
if isinstance(value, dict):
return {k: redact(v) for k, v in value.items()}
if isinstance(value, list):
return [redact(v) for v in value]
return value
class Tracer:
"""一次任务 = 一个 trace;每个 span 结束时往 JSONL 追加一行。"""
def __init__(self, path, clock=time.time):
self.path, self.clock = path, clock
self.trace_id = uuid.uuid4().hex
self._stack = [] # 当前打开的 span_id,栈顶就是父 span
@contextmanager
def span(self, name, **attrs):
span = {"trace_id": self.trace_id, "span_id": uuid.uuid4().hex[:16],
"parent_id": self._stack[-1] if self._stack else None,
"name": name, "start": self.clock(), "status": "ok", "attrs": attrs}
self._stack.append(span["span_id"])
try:
yield span
except Exception as e:
span["status"] = "error"
span["attrs"]["error.type"] = type(e).__name__
raise
finally:
self._stack.pop()
span["end"] = self.clock()
record = {**span, "attrs": redact(span["attrs"])} # 只脱敏内容,ID 和时间不动
with open(self.path, "a", encoding="utf-8") as f:
f.write(json.dumps(record, ensure_ascii=False) + "\n")
然后把它插进 agent loop。call_model 是对模型客户端的一层适配(真实实现里包一下 client.messages.create,把 usage、stop_reason、tool_use 块取出来),tools 是工具名到函数的字典:
def run_agent(task, call_model, tools, tracer, model_id, max_turns=6):
messages = [{"role": "user", "content": task}]
with tracer.span("invoke_agent", **{"gen_ai.operation.name": "invoke_agent"}) as root:
for turn in range(1, max_turns + 1):
with tracer.span(f"chat {model_id}", **{"gen_ai.operation.name": "chat",
"gen_ai.provider.name": "anthropic", "gen_ai.request.model": model_id}) as s:
resp = call_model(messages)
u = resp["usage"] # Claude 的 input_tokens 不含缓存部分,OTel 口径要加回去
c_read, c_write = u.get("cache_read_input_tokens") or 0, u.get("cache_creation_input_tokens") or 0
s["attrs"].update({"turn": turn, "gen_ai.input.messages": messages,
"gen_ai.output.messages": resp["content"],
"gen_ai.usage.input_tokens": u["input_tokens"] + c_read + c_write,
"gen_ai.usage.cache_read.input_tokens": c_read,
"gen_ai.usage.cache_write.input_tokens": c_write,
"gen_ai.usage.output_tokens": u["output_tokens"],
"gen_ai.response.finish_reasons": [resp["stop_reason"]]})
messages = messages + [{"role": "assistant", "content": resp["content"]}]
stop = resp["stop_reason"]
if stop != "tool_use": # max_tokens、refusal 等也会停,不能都算 completed
root["attrs"]["agent.outcome"] = "completed" if stop == "end_turn" else stop
return resp["content"]
results = []
for call in resp["tool_calls"]:
with tracer.span(f"execute_tool {call['name']}", **{
"gen_ai.operation.name": "execute_tool", "gen_ai.tool.name": call["name"],
"gen_ai.tool.call.arguments": call["input"]}) as t:
try:
out, is_error = tools[call["name"]](**call["input"]), False
except Exception as e: # 工具报错不能让 loop 崩,要把错误还给模型
out, is_error = f"{type(e).__name__}: {e}", True
t["status"], t["attrs"]["error.type"] = "error", type(e).__name__
t["attrs"]["gen_ai.tool.call.result"] = out
results.append({"tool_use_id": call["id"], "content": out, "is_error": is_error})
messages = messages + [{"role": "user", "content": results}]
root["status"], root["attrs"]["error.type"] = "error", "max_turns_exceeded"
return None
几个细节:① 工具报错要在 span 里标 error,但不能让 loop 崩,错误信息要作为 tool_result 还给模型,让它有机会重试。② 超过 max_turns 时根 span 标 max_turns_exceeded,否则"被截断"和"正常结束"在 trace 里分不出来;模型因 max_tokens、refusal 等原因停下时,结局记原始 stop_reason,只有 end_turn 才算 completed。③ clock 可以注入,测试时用假时钟,trace 就是确定的。④ 消息结构做了简化,规范里的消息格式是 role + parts;agent.outcome、turn 是自定义属性,不在规范里。⑤ 脱敏只作用于 attrs,不碰 ID:十六进制的 trace_id 也可能凑出"1 开头的 11 位数字",被手机号正则误伤。⑥ 真要上线,把 Tracer 换成 OpenTelemetry SDK 加一个 exporter:属性名基本不用改,但结构化值(消息、参数、结果)要按规范序列化,token 要按上面的规则换算。
我用一个不调 API 的模拟 loop 跑了一遍(脚本化的模型回复 + 假时钟,第一次调 get_order 超时 5 秒)。输出的 JSONL 每行一个 span,第 2 行是出错的工具调用:
{"trace_id": "e8cb…", "span_id": "462e6ba7da404028", "parent_id": "1430c8db05554f45",
"name": "execute_tool get_order", "start": 1001.6, "status": "error",
"attrs": {"gen_ai.operation.name": "execute_tool", "gen_ai.tool.name": "get_order",
"gen_ai.tool.call.arguments": {"order_id": "A123"}, "error.type": "TimeoutError",
"gen_ai.tool.call.result": "TimeoutError: orders-api 5s timeout"}, "end": 1006.6}
span 是在结束时写盘的,所以根 span 排在文件最后一行。分析时要按 start 重新排序。工具返回的邮箱 zhang.san@example.com 在文件里是 <email>,脱敏生效。
5从 trace 算出四个数
trace 落成 JSONL 之后,分析就是几行 Python。再传一个价格配置文件(从官方价格表抄的单价),顺便按上面的公式算出成本:
import json, sys
from collections import Counter
spans = [json.loads(line) for line in open(sys.argv[1], encoding="utf-8")]
spans.sort(key=lambda s: (s["start"], s["parent_id"] is not None)) # 同一时刻开始时父 span 排前面
a = lambda s, k, d=None: s["attrs"].get(k, d)
dur = lambda s: s["end"] - s["start"]
root = next(s for s in spans if s["parent_id"] is None)
steps = [s for s in spans if s is not root]
for i, s in enumerate(steps, 1): # 1. 每步耗时
print(f'{i:2} {dur(s):5.2f}s {s["status"]:5} {s["name"]}')
tin = sum(a(s, "gen_ai.usage.input_tokens", 0) for s in steps) # 2. 总 token(输入含缓存)
tout = sum(a(s, "gen_ai.usage.output_tokens", 0) for s in steps)
tools = [s for s in steps if a(s, "gen_ai.operation.name") == "execute_tool"]
fail = sum(s["status"] == "error" for s in tools) / len(tools) if tools else 0.0 # 3. 工具失败率
calls = Counter((a(s, "gen_ai.tool.name"), json.dumps(a(s, "gen_ai.tool.call.arguments"),
ensure_ascii=False, sort_keys=True)) for s in tools)
slow = max(steps, key=dur) # 4. 卡在哪一步
print(f"总耗时 {dur(root):.1f}s | token 输入 {tin} 输出 {tout} | 工具失败率 {fail:.0%} | "
f"结局 {a(root, 'agent.outcome') or a(root, 'error.type')}")
print(f"最慢一步: 第 {steps.index(slow) + 1} 步 {slow['name']} {dur(slow):.1f}s | "
f"同参数调用 ≥3 次: {[k for k, n in calls.items() if n >= 3]}")
def cost(s, price): # price 单位:美元 / 百万 token,从官方价格表抄进配置文件
c_read = a(s, "gen_ai.usage.cache_read.input_tokens", 0)
c_write = a(s, "gen_ai.usage.cache_write.input_tokens", 0)
uncached = a(s, "gen_ai.usage.input_tokens", 0) - c_read - c_write # = Claude 的 usage.input_tokens
return (uncached * price["input"] + c_write * price["cache_write"]
+ c_read * price["cache_read"] + a(s, "gen_ai.usage.output_tokens", 0) * price["output"]) / 1e6
if len(sys.argv) > 2: # 5. 成本:python analyze.py trace.jsonl prices.json
price = json.load(open(sys.argv[2], encoding="utf-8"))
print(f"成本 ${sum(cost(s, price) for s in steps):.4f}")
三个模拟任务跑出来的结果(和下面互动演示里的 A、B、C 是同一组数据):
| 任务 | 总耗时 | 输入 / 输出 token | 工具失败率 | 结局 | 卡在哪 |
|---|---|---|---|---|---|
| A 一次成功 | 5.5 s | 4270 / 230 | 0% | completed | 最慢是第 3 步模型调用,1.8 s |
| B 工具超时后重试 | 12.2 s | 6040 / 315 | 33% | completed | 第 2 步 get_order 超时 5.0 s |
| C 陷入循环 | 12.9 s | 9900 / 420 | 0% | max_turns_exceeded | search_orders 同参数调了 6 次 |
注意 C:工具失败率 0%,每一步耗时都正常,任务却失败了。工具每次都"成功"返回了空列表,模型每次都换汤不换药地再搜一遍。看单个 span 看不出问题,要看跨 span 的模式:"同一个工具、同样的参数,调了 3 次以上"。"卡住"不等于"最慢"。
第 1 周的 Wilson 区间在这里也用得上:40 次工具调用里 6 次出错,失败率点估计 15%,Wilson 95% 区间是 [7.1%, 29.1%]。trace 攒得不够多,失败率也只是个粗估。而且这个区间默认 40 次相互独立;同一条 trace 里的重试、循环会扎堆出错,更稳妥的是按 trace 算失败率,n 取 trace 数而不是工具调用数(见第 1 周「独立性假设」)。
6测开视角:flaky 失败、多次运行对比、pass^k
flaky 失败可以复盘了
传统 UI 自动化里,flaky 用例最难的是"重跑就过了,没法复现"。Agent 更严重:同一个输入,模型每次采样都可能走不同的路。有了 trace,失败的那一次已经被完整记下来,不需要复现:打开 trace,看它在哪一步做了和平时不一样的决定。
同一任务多次运行,对比 trace 找分叉点
第 1 周讲 pass^k:同一个任务跑 k 次,要求每次都对。假设跑 10 次,7 次成功、3 次失败。只看结果,你只有一个数:70%。把 10 条 trace 按步骤对齐,找第一个和成功运行不一样的步骤(分叉点):
- 如果 3 次失败都在同一步分叉(比如第 3 步模型都把金额 129.9 填成 1299),这是一个稳定可复现的缺陷,不是随机噪声。修掉它,pass^k 会整体上一个台阶。
- 如果 3 次失败分叉点各不相同,那是多个独立的小问题,要分别归类。第 5 周的错误分析就是从这里开始。
下面的演示里打开"对比成功运行",选 B、C、D 任意一条,会自动高亮它和成功运行 A 第一个不一样的步骤。
▶互动演示:trace 瀑布图查看器
四条模拟 trace,都是"帮我把订单 A123 退款"这个任务。横条是 span,点一下看详情。可以单步回放,也可以和成功运行对比找分叉点。
✎练习
7面试要点
一句话讲清楚
一次任务是一个 trace,每次模型调用、每次工具调用是一个 span,带父子关系、起止时间、状态和属性。属性名参考 OpenTelemetry 的 GenAI 语义约定(目前是 Development 状态):模型 ID、token、stop_reason、工具名和参数、结果、error.type。有了 trace,才能按步骤定位失败、算成本、回放和对比多次运行的分叉点;消息内容默认不记,要记就先脱敏。
常见错误说法
❌ "工具失败率 0%,Agent 就没问题":循环、参数填错都不一定让工具报错,要看跨 span 的模式和根 span 的结局。
❌ "stop_reason 是 end_turn,任务就成功了":演示里的 D 正常结束了,最终回复却是"退款失败"。要看工具结果和最终回复。
❌ "成本 = 总 token × 一个单价":输入、输出、缓存读写单价都不同,而且价格会变,要按官方价格表分开算。
❌ "Claude 返回的 input_tokens 就是全部输入":它不含缓存读写,开了缓存要加回去,并且缓存部分按各自单价计价。
❌ "每轮 input_tokens 差不多,总成本就是轮数 × 单轮":每轮都重发历史,input 越滚越大。
❌ "把完整 prompt 和用户数据都记进 trace 方便排查":规范默认不采集消息内容;要记,先脱敏,生产环境放外部存储并做访问控制。
公式显示依赖 KaTeX(CDN),断网时公式会显示成原始 LaTeX 源码,交互部分不受影响。