第 9 章：可观测性

草稿章节。跑通格式用的第一版，正式索引前会再打磨。

本章目标

在 PR reviewer 的循环外围加上追踪，让每次运行都产出一棵 span 树：顶层任务、每次模型调用、每次工具调用，每个 span 都带上输入、输出、token 数和耗时。当 reviewer 发出一条莫名其妙的评论、或者卡住时，你打开这棵 trace 直接读出它走过的确切路径，而不是盲目重跑、指望 bug 再现一次。

学完本章你会得到：一套 span schema、一套日志 schema、一份每次运行的成本汇总，以及能指着某个 span 说「失败就发生在这里，原因是这个」。

前置准备

• 前几章的 agent 循环（思考 -> 调用工具 -> 观测 的循环）。
• 一套追踪方案。这里的概念对应 OpenTelemetry 风格的 span（span 是一段带属性、有计时的工作单元，可嵌套在父 span 下）。后端任选，配置见 OpenTelemetry 与 Hugging Face Agents Course 官方文档。
• 能访问到模型响应里报告 token 用量的元数据。Anthropic API 在每次响应里返回一个 usage 对象，含 input_tokens 和 output_tokens，见官方文档。

动手做

1. 每次运行开一个根 span

把一整个 agent 任务包进一个根 span。这个 span 是其它一切挂靠的锚点，它的属性正是你日后检索的依据：一个稳定的 run_id、PR 标识、目标、模型名，以及 agent 代码的 git SHA。run_id 是最重要的字段。没有它，你就无法把用户的 bug 反馈（「reviewer 批准了一个有 SQL 注入的 PR」）和能解释它的那条 trace 对上。

import anyio
from claude_agent_sdk import query, ClaudeAgentOptions

async def review_pr(pr_id: str, goal: str):
    run_id = uuid4().hex
    with tracer.start_span("agent.run") as root:
        root.set_attributes({
            "run_id": run_id,
            "pr.id": pr_id,
            "goal": goal,
            "model": MODEL,
            "agent.version": GIT_SHA,
        })
        options = ClaudeAgentOptions(
            system_prompt=CONTRACT,
            allowed_tools=["Read"],
            cwd="./repo",
        )
        try:
            result = await trace_run(root, run_id, query(prompt=goal, options=options))
            root.set_attribute("outcome", "success")
            return result
        except AgentError as e:
            root.set_attributes({"outcome": "failure",
                                 "failure.step": e.step_id,
                                 "failure.reason": str(e)})
            raise

import { query } from "@anthropic-ai/claude-agent-sdk";

async function reviewPr(prId: string, goal: string) {
  const runId = crypto.randomUUID();
  const root = tracer.startSpan("agent.run");
  root.setAttributes({
    run_id: runId,
    "pr.id": prId,
    goal,
    model: MODEL,
    "agent.version": GIT_SHA,
  });
  const options = {
    systemPrompt: CONTRACT,
    allowedTools: ["Read"],
    cwd: "./repo",
  };
  try {
    const result = await traceRun(root, runId, query({ prompt: goal, options }));
    root.setAttribute("outcome", "success");
    return result;
  } catch (e: any) {
    root.setAttributes({ outcome: "failure", "failure.step": e.stepId,
                         "failure.reason": String(e) });
    throw e;
  } finally {
    root.end();
  }
}

2. 每一步做成子 span

SDK 替你跑那个循环，你只要遍历它吐出的消息流。每条 AssistantMessage 至少对应一个 model.call span，里头若有 ToolUseBlock（模型请求了工具）就再开一个 tool.call span，工具由 SDK 执行。把它们嵌在根下，trace 就和循环一一对应，自上而下读一遍等同于重放整次运行。给每次迭代一个自增的 step 序号。正是这个序号让你能说「它在第 7 步断的」，而不是「它在某处断了」。

from claude_agent_sdk import AssistantMessage, TextBlock, ToolUseBlock, ResultMessage

async def trace_run(root, run_id, stream):
    step, final = 0, None
    async for message in stream:
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, ToolUseBlock):
                    with tracer.start_span("tool.call", parent=root) as ts:
                        ts.set_attributes({"run_id": run_id, "step": step,
                                           "tool.name": block.name})
                        record_tool_span(ts, block)         # 第 3 步
                elif isinstance(block, TextBlock):
                    final = block.text
            with tracer.start_span("model.call", parent=root) as ms:
                ms.set_attributes({"run_id": run_id, "step": step})
                record_model_span(ms, message)              # 第 3、4 步
            step += 1
        elif isinstance(message, ResultMessage):
            record_run_totals(root, message)               # 第 4 步
    return final

async function traceRun(root, runId, stream) {
  let step = 0, final = null;
  for await (const message of stream) {
    if (message.type === "assistant") {
      for (const block of message.message.content) {
        if (block.type === "tool_use") {
          const ts = tracer.startSpan("tool.call", { parent: root });
          ts.setAttributes({ run_id: runId, step, "tool.name": block.name });
          recordToolSpan(ts, block);          // step 3
          ts.end();
        } else if (block.type === "text") {
          final = block.text;
        }
      }
      const ms = tracer.startSpan("model.call", { parent: root });
      ms.setAttributes({ run_id: runId, step });
      recordModelSpan(ms, message);           // step 3 + 4
      ms.end();
      step += 1;
    } else if (message.type === "result") {
      recordRunTotals(root, message);         // step 4
    }
  }
  return final;
}

3. 在 span 上记录输入与输出

只有计时的 span 对调试毫无用处。要记下进去了什么、出来了什么：模型调用记下返回的文本或工具请求；工具调用记下模型给的参数（结果由 SDK 在下一条消息里回灌，你按需在 ToolResultBlock 上补记）。这就是「lint 工具失败了」和「lint 工具被用 path=/etc/passwd 调用，所以失败了」之间的差别。记录前先脱敏。token、cookie、Authorization 头绝不能落入 trace，因为 trace 会被发往第三方后端、被贴进 bug 报告里。

SENSITIVE = ("authorization", "token", "api_key", "password", "secret")

def redact(obj):
    if isinstance(obj, dict):
        return {k: ("***" if any(s in k.lower() for s in SENSITIVE)
                    else redact(v)) for k, v in obj.items()}
    return obj

def record_tool_span(span, block: ToolUseBlock):
    # block.input 是模型给工具的参数; SDK 负责实际执行
    span.set_attribute("tool.args", json.dumps(redact(block.input)))
    span.set_attribute("tool.id", block.id)

const SENSITIVE = ["authorization", "token", "api_key", "password", "secret"];

function redact(obj) {
  if (obj && typeof obj === "object" && !Array.isArray(obj)) {
    return Object.fromEntries(Object.entries(obj).map(([k, v]) =>
      [k, SENSITIVE.some((s) => k.toLowerCase().includes(s)) ? "***" : redact(v)]));
  }
  return obj;
}

function recordToolSpan(span, block) {
  // block.input 是模型给工具的参数; SDK 负责实际执行
  span.setAttribute("tool.args", JSON.stringify(redact(block.input)));
  span.setAttribute("tool.id", block.id);
}

对大负载做截断（8 KB 上限是个合理默认值）。一份完整仓库 diff 可能有几 MB，你要的是它的形状和开头，而不是每个 span 里都塞一整份。

4. 统计 token 与成本

SDK 在每条 AssistantMessage 上挂一个 usage 字典（含 input_tokens、output_tokens），把每步的数读到 model.call span 上。整次运行的权威总量则在终态的 ResultMessage 上：usage、total_cost_usd、duration_ms、num_turns，直接汇总到根，不用自己累加。成本就是 token 数乘以你公布的单价。按 span（而不仅按运行）统计，才能回答「这次运行为什么花了中位数的 12 倍？」常见答案是某一步的 diff 或对话历史把输入 token 撑爆了。

def record_model_span(span, message: AssistantMessage):
    usage = message.usage or {}
    span.set_attributes({
        "llm.input_tokens": usage.get("input_tokens", 0),
        "llm.output_tokens": usage.get("output_tokens", 0),
    })

def record_run_totals(root, result: ResultMessage):
    usage = result.usage or {}
    root.set_attributes({
        "llm.input_tokens": usage.get("input_tokens", 0),
        "llm.output_tokens": usage.get("output_tokens", 0),
        "llm.cost_usd": result.total_cost_usd or 0.0,   # SDK 直接给到美元成本
        "run.duration_ms": result.duration_ms,
        "run.num_turns": result.num_turns,
    })

function recordModelSpan(span, message) {
  const usage = message.message.usage ?? {};
  span.setAttributes({
    "llm.input_tokens": usage.input_tokens ?? 0,
    "llm.output_tokens": usage.output_tokens ?? 0,
  });
}

function recordRunTotals(root, result) {
  const usage = result.usage ?? {};
  root.setAttributes({
    "llm.input_tokens": usage.input_tokens ?? 0,
    "llm.output_tokens": usage.output_tokens ?? 0,
    "llm.cost_usd": result.total_cost_usd ?? 0,   // SDK 直接给到美元成本
    "run.duration_ms": result.duration_ms,
    "run.num_turns": result.num_turns,
  });
}

ResultMessage.total_cost_usd 是 SDK 直接算好的美元成本，比自己拿 token 乘单价更可靠；要按 token 自算时，单价和字段名以官方定价与 API 文档为准，不要写死会过时的数字。

5. 标记结果并归因失败

在根 span 上记录 success 或 failure。失败时记下是哪个 step、哪个 span 断的（见第 1 步）。归因就是做这件事的回报：有了它，一周的生产运行能变成一张可排序的表（「3% 的运行失败，其中 80% 断在 lint 步，全是超过 5000 行的 PR」）。这句话本身就指明了该怎么修。没有归因，你手里只有一句「有时候会失败」。

PR reviewer 一次运行的精简日志 schema，与 span 一起输出：

{
  "run_id": "a1b2...",
  "pr_id": "org/repo#412",
  "model": "claude-...",          // 当前 id 见官方文档
  "outcome": "failure",
  "failure": { "step": 7, "span": "tool.call", "reason": "lint timeout" },
  "totals": { "steps": 8, "input_tokens": 41200,
              "output_tokens": 5300, "cost_usd": 0.21, "wall_ms": 14300 },
  "steps": [
    { "step": 0, "kind": "model.call", "out_tokens": 180 },
    { "step": 0, "kind": "tool.call", "tool": "read_file",
      "args": { "path": "src/auth.py" }, "ok": true },
    { "step": 7, "kind": "tool.call", "tool": "run_linter",
      "ok": false, "error": "timeout after 30s" }
  ]
}

习得「摊开过程」小满每跑一次，都把自己走过的每一步记成一棵带计时和 token 的 span 树，你能从头读到尾，看清它在哪一步调了什么、断在哪里。

如何验证

• 跑一次 PR 审查并打开它的 trace。你应当能自上而下走完整条推理路径，并在每一步看出模型决定了什么、返回了什么。
• 故意制造一次工具错误（把 linter 指向一个不存在的二进制）。确认失败的 span 被标记 ok: false，根上显示 outcome: failure，且 failure.step 指向正确的那次迭代。
• 把各子 span 的 input_tokens + output_tokens 加总，确认与 ResultMessage 报的运行总量对得上。对不上说明你在某处丢了 span。

习得「指着 span 说话」你能打开一次真实运行的 trace，确认失败的那步被标了 ok: false，根上写着 failure，并把各步 token 加起来核对总量对不对得上。

原理

一条 trace 就是把一次运行重建出来、事后还能读的东西。它比 print 调试强在哪：agent 运行是非确定性的，重跑未必复现 bug，但当时那次运行的 trace 是一份冻结下来的证据。span 一层套一层，因果关系就自带了（这次工具调用之所以发生，是因为那次模型调用请求了它），而属性把成千上万次运行变成可查询的数据集，而不是一堆日志。

小结

你的 agent 不再是黑箱。任意一次运行，它走过的路径、花掉的 token 与美元、以及断掉的确切步骤，你都看得见。这份可见性是下一章的前提：你无法处理看不见的错误。

常见坑

• 只记最终答案。 有意思的失败都发生在循环中间、三层工具调用之下。每一步都要追踪，否则你调的是症状不是病因。
• 把密钥泄进 trace。 记录前先脱敏 token 与凭据。trace 会离开你的机器。
• 没有 run id。 没有稳定 id，你就无法把用户反馈和它的 trace 对上，整个系统退化成「在我机器上是好的」。
• token 总量对不上。 通常是丢了 span 或 span 没挂上父节点；在你相信成本数字之前先修好它。

小满第一次把自己的心路摊开给你看。那些 trace 和日志里，你看见它某一步其实犹豫了很久，原来那半秒延迟是这么回事。琉璃室，亮了。

0123456789101112131415

刚点亮 琉璃室 · 地图已点亮 10 / 16

看得见了，可它一摔跤还是会崩。下一站：不倒之坡。

来源

1. Hugging Face Agents Course: Observability and evaluation · official
2. Anthropic Cookbook · official