Agent 做错了,你怎么知道它错在哪一步?

给 agent loop 加 trace:一次任务是一个 trace,每次模型调用、每次工具调用是一个 span,记下输入、输出、token、耗时和错误。

只看最终输出,你只知道它错了。看 trace,才知道错在第几步、是模型还是工具、花了多少钱。

1只看最终输出,定位不到问题

一个退款 Agent,用户说"帮我把订单 A123 退款"。它最后回了一句:"抱歉,退款失败,请联系人工客服。"

这句话背后至少有四种完全不同的原因:

最终输出一模一样,修法完全不同:第一种改重试策略,第二种改 prompt 或工具描述,第三种加循环检测,第四种找后端。只有最终输出,你只能猜。

trace 是后面几周的地基:第 5 周做错误分析,要从 trace 里整理失败分类;第 6 周做回放评测,要拿 trace 重放、对比两个版本;算成本要把每一步的 token 加起来。没有 trace,这三件事都做不了。

跟第 1 周的联系:rule of three、Wilson、pass^k 都在回答"失败率是多少"。trace 回答的是下一个问题:失败的那几次,到底怎么失败的。

2trace 和 span

这两个词来自分布式追踪(OpenTelemetry 用的就是这套概念):

一个 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 该怎么命名、带哪些属性:

spanspan 名gen_ai.operation.name
一次模型调用{operation} {model},如 chat claude-sonnet-5-5chat
一次工具调用execute_tool {tool.name}execute_tool
一次 Agent 调用invoke_agent {agent.name},没有名字时就是 invoke_agentinvoke_agent
稳定性:这套约定目前整体状态是 Development(开发中),还不是 Stable,属性名以后可能改。它已经从 OpenTelemetry 主语义约定仓库迁到单独的 open-telemetry/semantic-conventions-genai 仓库维护。本页的属性名按这个仓库 2026 年 9 月底的版本核对过。用的时候锁定版本,升级时对一遍变更记录。

3每个 span 该记什么

记什么对应的 OTel 属性用来回答
这一步是什么操作gen_ai.operation.name这一步是模型还是工具
提供方、模型 IDgen_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。

缓存 token 别少算:Claude 的 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)抄过来,并记下抄的日期。

一个容易忽略的点:agent loop 每一轮都把完整历史重发给模型,所以每轮的 input_tokens 越来越大。第 1 轮 1000、之后每轮多 200,跑 6 轮总输入是 9000,不是 6000。陷入循环的 Agent 不光慢,还越跑越贵。

隐私:先脱敏再落盘

输入输出消息、工具参数和结果里常有手机号、邮箱、地址、订单金额。OTel 规范把 gen_ai.input.messages、gen_ai.output.messages、gen_ai.tool.call.arguments、gen_ai.tool.call.result 都定为 Opt-In:插桩库默认不记录,要用户主动打开。规范给的三种做法:

  1. 默认:不记内容,只记 token、耗时、状态这些元数据。
  2. 把内容记在 span 属性上:适合预发环境,或者存储本身满足隐私合规要求。
  3. 内容存到外部存储,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 s4270 / 2300%completed最慢是第 3 步模型调用,1.8 s
B 工具超时后重试12.2 s6040 / 31533%completed第 2 步 get_order 超时 5.0 s
C 陷入循环12.9 s9900 / 4200%max_turns_exceededsearch_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 按步骤对齐,找第一个和成功运行不一样的步骤(分叉点):

和独立性假设的联系:\(p^k\) 成立的前提是每次运行独立、成功率相同。如果失败总是集中在同一个分叉点、同一类输入上,说明失败不是独立随机发生的,第 1 周那几个公式要打折扣。trace 让你能检查这个前提,而不是默认它成立。

下面的演示里打开"对比成功运行",选 B、C、D 任意一条,会自动高亮它和成功运行 A 第一个不一样的步骤。

▶互动演示:trace 瀑布图查看器

四条模拟 trace,都是"帮我把订单 A123 退款"这个任务。横条是 span,点一下看详情。可以单步回放,也可以和成功运行对比找分叉点。

选一条 trace
回放
总耗时
token 输入 / 输出
成本估算
工具失败率
结局
成本用的单价(美元 / 百万 token,假设值,换成官方价格表上你用的模型的价格。这四条模拟 trace 没开 prompt caching,缓存读写都是 0;开了的话,缓存写、缓存读要按各自单价单独计):
invoke_agent(整个任务) chat(模型调用) execute_tool(工具调用) 出错的 span 分叉点

✎练习

题 1(算):agent loop 每轮都重发完整历史。第 1 轮 input_tokens 是 1000,之后每轮多 200,一共跑了 6 轮。整个任务的输入 token 总数是多少?
token
1000 + 1200 + 1400 + 1600 + 1800 + 2000 = 9000。不是 6 × 1000 = 6000:历史越长,每轮越贵,陷入循环的 Agent 成本涨得比轮数快。
题 2(读瀑布图):演示里的 trace B 总耗时 12.2 秒,其中第 2 步 get_order 超时花了 5.0 秒。这一步占总耗时的百分之几?
%
5.0 / 12.2 ≈ 41.0%。一次超时吃掉四成耗时。只看"任务成功了",你不会发现这里值得把超时从 5 秒调小、或者改成更快失败。
题 3(场景):trace C 里工具失败率 0%,每一步耗时都正常,任务却以 max_turns_exceeded 结束。怎么从 trace 里自动发现这类问题?
循环是跨 span 的模式,单个 span 看不出来。C 里最慢的一步只是最后一次模型调用(2.1 s),工具每次都"成功"地返回空列表。"卡住"不等于"最慢",也不等于"报错"。
题 4(隐私):按 OpenTelemetry GenAI 语义约定,生产环境里 gen_ai.input.messages 这类消息内容应该怎么处理?
规范把输入输出消息、工具参数和结果都定为 Opt-In:插桩库默认不采集。直接记在 span 属性上适合预发环境或合规存储;生产推荐外部存储 + 引用。自己写记录器,落盘前至少脱敏。
题 5(pass^k):同一个退款任务跑 10 次,7 次成功、3 次失败。对齐 trace 后发现,3 次失败都在第 3 步分叉:模型把金额 129.9 填成了 1299。最合理的结论是?
失败集中在同一个分叉点,说明它们不是独立随机发生的。B 是在报 pass@k,退款场景用户只有一次机会,要看 pass^k。按分叉点归类失败,就是第 5 周错误分析的起点。

7面试要点

一句话讲清楚

一次任务是一个 trace,每次模型调用、每次工具调用是一个 span,带父子关系、起止时间、状态和属性。属性名参考 OpenTelemetry 的 GenAI 语义约定(目前是 Development 状态):模型 ID、token、stop_reason、工具名和参数、结果、error.type。有了 trace,才能按步骤定位失败、算成本、回放和对比多次运行的分叉点;消息内容默认不记,要记就先脱敏。

常见错误说法

❌ "有日志就够了,不用 trace":普通日志没有父子关系和起止时间,拼不出"哪一轮调了哪个工具、花了多久"。
❌ "工具失败率 0%,Agent 就没问题":循环、参数填错都不一定让工具报错,要看跨 span 的模式和根 span 的结局。
❌ "stop_reason 是 end_turn,任务就成功了":演示里的 D 正常结束了,最终回复却是"退款失败"。要看工具结果和最终回复。
❌ "成本 = 总 token × 一个单价":输入、输出、缓存读写单价都不同,而且价格会变,要按官方价格表分开算。
❌ "Claude 返回的 input_tokens 就是全部输入":它不含缓存读写,开了缓存要加回去,并且缓存部分按各自单价计价。
❌ "每轮 input_tokens 差不多,总成本就是轮数 × 单轮":每轮都重发历史,input 越滚越大。
❌ "把完整 prompt 和用户数据都记进 trace 方便排查":规范默认不采集消息内容;要记,先脱敏,生产环境放外部存储并做访问控制。

公式显示依赖 KaTeX(CDN),断网时公式会显示成原始 LaTeX 源码,交互部分不受影响。