Agent 实现方法论 From Scratch

可观测性与成本 v07

可观测性与成本 v07

v05（ch18）我们给产品装上了权限层：审批双向流、审计落库，每个破坏性动作都有据可查。但「查」的是谁批了什么。还有一个更基本的问题被一路搁置：这个 Agent 跑起来之后，它内部到底发生了什么？

这些问题，v05 的产品一个都答不上来：事件在 SSE 流上发完就没了，终端打印的日志混在一起分不清先后，done 上的 usage 只告诉你最后一轮用了多少 token，中间过程是黑盒。本章给产品加 v07：可观测性与成本：

数据落在产品的 trace 存储里，通过 GET /traces、GET /costs/<sessionId>、GET /failures 三个新端点暴露。ch18 的审批审计是「谁干的」，本章的观测是三件套：发生了什么、花了多少钱、坏在哪一类。

本章目标

读完本章并做完配套练习后，你应该能够：

概念与动机：看不见 Agent 内部发生了什么

先看一个「错误答案」：print 大法

假设我们没有观测层。v05 的代码在事件流里加了一行 console.log(event)，然后开始调：

tool_call_start { callId: 'call_1', name: 'add', args: { a: 'x', b: 2 } }
tool_result     { callId: 'call_1', ok: false, error: 'Error: add() expects numeric...' }
text_delta      { delta: 'add 工具执行失败了：...' }
done            { usage: { inputTokens: 52, outputTokens: 26 } }

单看一个会话，好像还行。但产品里跑着几十个会话（ch15 起的多会话），print 出来的行混在一起。两个会话的 tool_call_start 交错出现，你分不清 call_1 是哪个会话的；同一个会话里事件谁先谁后要靠肉眼排；想「查一下 10 分钟前那个报错的会话」？日志早被冲走了。更糟的是：事件流是瞬时而逝的。SSE 推给前端，前端渲染完，服务器内存里的缓冲是 ch16 之前的临时物，没有任何「这一轮完整发生了什么」的持久记录。

smolagents 的官方教程把这个问题讲得很直接：agent runs are complicated to debug，多步 Agent 的日志很快就会铺满整个控制台，而且大部分错误是「LLM 犯傻」类，模型下一轮自己就纠正了（Hugging Face · Inspecting runs with OpenTelemetry）。靠肉眼看一堆 print 去还原「哪一步错了、为什么错」，不是调试，是考古。

日志与 trace 的区别

我们的做法正是把两者合起来：事件行是 trace 的砖。每个事件先盖 seq（全局序号）+ ts（墙钟时间）变成结构化的一行，再按会话归进「这一轮对话」的 trace 里。ch15 的事件协议给了我们免费的礼物：事件本来就带 sessionId，天然可以分组；done 事件本来带 usage（ch15 起），token 有处可查。

参照：业界怎么做可观测性

我们借鉴的是「事件盖章 + 按运行聚合 + 派生指标」这个机制；实现（seq/ts、trace 结构、成本纯函数、三桶看板）是我们自己的。

结构化日志与 trace

事件行：给每个事件盖上 seq 与 ts

观测层做的第一件事非常朴素：每个协议事件进来时，盖两个戳：

// lib/trace.mjs
export function stampEvent(event, seq, ts = new Date().toISOString()) {
  return { ...event, seq, ts };   // 复制，不改原事件
}

关键设计：seq/ts 不进协议。 盖章副本是观测层自己的视图，转发给 SSE 缓冲的是原始事件，wire 上跑的还是 ch15 定稿的事件，一个字节没改。可观测性是一层包在外面的观察者，不是协议的一部分。

trace：一轮对话 = 一条 span

盖完章的事件按 sessionId 归组：一次运行（loop 的一轮对话）= 一条 trace。trace 的生命周期由三个事件界定：

事件作用
该会话的第一个事件打开 trace（起点 = 该事件的 ts）
done累计 usage；关闭 trace，结果 ok
error关闭 trace，结果 fail

trace 里除了盖章事件序列，还累计三样派生物：token（来自 done.usage）、耗时（起点 → 终点，毫秒）、结果（ok/fail）。多轮会话（同一 sessionId 多次 POST /chat）会产生多条 trace，一条一轮。

trace 生成流水线

flowchart LR
  LOOP["lib/loop.mjs<br/>emit(事件流)<br/>text_delta / tool_call_start<br/>tool_result / done / error"]
  REC["lib/trace.mjs · createTraceRecorder<br/>stamp: +seq +ts<br/>append 入当前 trace<br/>done → 累计 usage"]
  STORE[("trace 存储<br/>finished traces<br/>t1 · t2 · t3 …")]
  VIEWS["界面层查询<br/>GET /traces<br/>GET /costs/:sessionId<br/>GET /failures"]
  WIRE["SSE 事件流<br/>原始事件，协议不动"]

  LOOP -->|onEvent| REC
  REC -->|盖章副本| STORE
  REC -->|原始事件| WIRE
  STORE --> VIEWS

分层红线在这里继续成立：loop 不知道有人在看。它只是照常 emit；createTraceRecorder 包在 onEvent 外面，服务端（lib/server.mjs）把「盖章 → 存 trace → 转发原始事件到 SSE 缓冲」串成一条链。把 server 和 cli 删掉，core 照常跑、照常测，观测是外围的事。

成本核算与 prompt cache 经济学

token 花到哪了：done.usage

ch15 设计事件协议时，给 done 带上了 usage: { inputTokens, outputTokens }，当时是为了让前端展示「本轮用量」。v07 让它成为成本核算的数据源：trace 累计的 usage，就是账单的输入。

成本核算 = 两个纯函数的事：

// lib/cost.mjs
// 单次调用成本：命中缓存的输入按折扣价，其余输入按全价，输出按输出价
export function costOfCall(usage, prices) {
  const input = usage.inputTokens ?? 0;
  const output = usage.outputTokens ?? 0;
  const cached = Math.min(usage.cachedTokens ?? 0, input);
  return (input - cached) * prices.inputPerMillion / 1_000_000
       + cached * prices.cachedInputPerMillion / 1_000_000
       + output * prices.outputPerMillion / 1_000_000;
}

单价模型（美元 / 百万 token，DEFAULT_PRICES）：输入 $3、输出 $15、缓存命中输入 $0.3。输出比输入贵 5 倍（生成比读贵），缓存命中只有输入价的 10%。

prompt cache：命中的前缀按 1 折计费

cachedTokens 从哪来？真实供应商在用量详情里报告：OpenAI 的 usage.prompt_tokens_details.cached_tokens（OpenAI · Prompt caching）、Anthropic 的 usage.cache_read_input_tokens。我们的 lib/provider.mjs 在解析响应时把它读出来放进 usage.cachedTokens，随 done 事件流进 trace，协议不加字段之外的东西，done.usage 只是多了个可选键。

为什么会有缓存命中？prompt cache 的机制是「精确前缀复用」：同一个 prompt 的开头部分，如果和之前某次请求完全一致，供应商就不用重新计算这部分的注意力，省算力，所以打折。OpenAI 对缓存命中的输入 token 收约 10% 的价格（90% 折扣）；要命中，前缀必须逐字节相同（OpenAI · Prompt caching）。

这直接导出 Agent 的一条工程铁律：稳定内容放前面，变化内容放后面。

build-your-own-agent 的 §7 用一个小例子量化了这个收益：同一个 Agent 循环，带缓存比不带缓存省 79%（build-your-own-agent §7）。我们 demo 里两轮调用各命中 30 个缓存 token，每轮省 30 × ($3 − $0.3) / 1e6 = $0.000081。数字小，机制是真的。

会话累计账单

单次调用的成本只回答「这一下多少钱」。产品要回答的是「这个会话花了多少」：sessionBill 把一次会话的所有调用（每个 done 的 usage = 一次调用）逐行列出、累计成 totals：

flowchart TD
  DONE1["done #1 usage<br/>{inputTokens, outputTokens, cachedTokens}"]
  DONE2["done #2 usage<br/>{inputTokens, outputTokens, cachedTokens}"]
  SPLIT["costOfCall 拆分计费<br/>未命中输入 × 输入价<br/>命中输入 × 缓存价<br/>输出 × 输出价"]
  SAVE["cacheSavings<br/>命中 token × (输入价 − 缓存价)"]
  BILL[("会话账单<br/>lines: 每次调用一行<br/>totals: 全量累计")]

  DONE1 --> SPLIT
  DONE2 --> SPLIT
  SPLIT --> BILL
  SPLIT --> SAVE
  SAVE --> BILL

账单里每行还带一个 savedByCache，缓存省下的钱。这个数字是「cache 经济学」的直接体现：优化 prompt 结构（稳定前缀前置）能让它变大。

失败模式看板

成本账回答「花了多少」，看板回答「坏在哪一类」。ch13 我们给评估写过失败模式分类（planning / tool / efficiency）。「失败不能只知道『失败了』，要知道『为什么失败』」这个思路，原样搬到 trace 上：

桶触发条件含义该桶 trace 的 result
error出现 error 事件本轮崩溃（provider 挂了、密钥错了、响应畸形），用户什么都没拿到，最严重fail
tool出现 tool_result 且 ok === false工具调用失败但循环存活（ch08 的 error-as-data），模型可能已经恢复，是瑕疵不是事故ok
budget总 token 数超过配置的 tokenBudget成本问题，运行完全正确也可能超（ch13 的「超预算也算失败」，token 版）ok

分类用显式优先级（和 ch13 的 classifier 同一个设计）：第一条命中的规则胜出，一条 trace 只进一个桶，避免同一个失败被重复计数，也让「先修哪个」有依据：崩溃 > 工具失败 > 烧钱。

flowchart TD
  T["一条 trace"]
  T --> E{"有 error 事件？"}
  E -->|是| ER["error<br/>崩溃"]
  E -->|否| TF{"有 tool_result<br/>ok:false ？"}
  TF -->|是| TO["tool<br/>工具失败，循环存活"]
  TF -->|否| B{"总 token 数<br/>> tokenBudget ？"}
  B -->|是| BU["budget<br/>超预算"]
  B -->|否| OK["ok<br/>不进看板"]

看板 buildFailboard 把存储里所有 trace 过一遍分类器，产出 { total, ok, byKind, failures }：计数之外，每条失败还带证明事件（error 事件 / 失败的 tool_result / done 事件）和一句人话 message，运维不用点开原始日志就能知道「s-err 是 provider HTTP 500，s-budget 是 700 token 超了 600 的预算」。

手写实现：把观测层装进产品

一、trace 记录器（lib/trace.mjs）

三个纯函数 + 一个状态机。纯函数保证可测：给固定输入必然得到固定输出（ts 由调用方传入，测试不依赖时钟）。

// 盖戳（事件行）
export function stampEvent(event, seq, ts = new Date().toISOString()) {
  return { ...event, seq, ts };
}

// done 的 usage 累计进计数器（缺字段按 0）
export function accumulateUsage(base, event) {
  const u = event?.usage ?? {};
  return {
    inputTokens: (base.inputTokens ?? 0) + (u.inputTokens ?? 0),
    outputTokens: (base.outputTokens ?? 0) + (u.outputTokens ?? 0),
    cachedTokens: (base.cachedTokens ?? 0) + (u.cachedTokens ?? 0),
  };
}

// 关闭一条 trace：耗时、结果、最终答案（最后一个 text_delta）
export function finalizeTrace({ sessionId, startedAt, events, usage }, { endedAt, result }) {
  const finalText = [...events].reverse().find((e) => e.type === 'text_delta')?.delta ?? '';
  return { sessionId, startedAt, endedAt,
           durationMs: Date.parse(endedAt) - Date.parse(startedAt),
           events, usage, result, finalText };
}

状态机 createTraceRecorder 把这些组装起来，并且只包在 onEvent 外面，这是它最重要的性质：

export function createTraceRecorder({ forward } = {}) {
  const running = new Map();  // sessionId -> 正在跑的 trace
  const finished = [];        // 已完成的 trace，旧的在前
  let seq = 0, traceCount = 0;

  return {
    instrument(event) {
      const stamped = stampEvent(event, seq++);            // ① 盖章
      let trace = running.get(event.sessionId);
      if (!trace) {                                        // ② 首个事件开 trace
        trace = { sessionId: event.sessionId, startedAt: stamped.ts,
                  events: [], usage: { inputTokens: 0, outputTokens: 0, cachedTokens: 0 } };
        running.set(event.sessionId, trace);
      }
      trace.events.push(stamped);                          // ③ 入列
      if (event.type === 'done') trace.usage = accumulateUsage(trace.usage, event); // ④ 累计
      if (event.type === 'done' || event.type === 'error') {                        // ⑤ 关闭
        const t = finalizeTrace(trace, { endedAt: stamped.ts,
          result: event.type === 'done' ? 'ok' : 'fail' });
        finished.push({ id: `t${++traceCount}`, ...t });
        running.delete(event.sessionId);
      }
      forward?.(event);                                    // ⑥ 原始事件继续走
      return stamped;                                      //    盖章副本留给自己
    },
    traces: () => finished.slice(),
    tracesOf: (sessionId) => finished.filter((t) => t.sessionId === sessionId),
  };
}

服务端把它串进事件链（lib/server.mjs）：

const traces = createTraceRecorder({ forward: (event) => buffer.append(event.sessionId, event) });
// ...
runAgent(session.messages, { ..., onEvent: traces.instrument, ... });

二、成本核算（lib/cost.mjs）

纯函数三件套（costOfCall / cacheSavings / sessionBill，代码见上文与 lib/cost.mjs）。服务端 GET /costs/<sessionId> 把该会话的每条 trace 里的 done 事件（每次调用一次）喂给 sessionBill，得到「每次调用一行 + 会话累计」，再把所有 trace 的 totals 汇总成会话总账。单价从哪来？lib/config.mjs 新增三个环境变量（INPUT_PRICE_PER_M / OUTPUT_PRICE_PER_M / CACHED_INPUT_PRICE_PER_M，默认 3 / 15 / 0.3），单价是配置，不是硬编码，换模型改配置即可。

三、失败看板（lib/failboard.mjs）

classifyTrace（优先级 error > tool > budget）+ buildFailboard（聚合 + 明细）。预算 tokenBudget 也是配置（TOKEN_BUDGET，默认 1000）。GET /failures 直接返回 buildFailboard(traces.traces(), { tokenBudget: config.tokenBudget })。

四、跑起来看三件套

demo 用确定性 mock 模型跑五个场景（正常两轮 / 工具失败 / provider 崩溃 / 超预算），先看 wire 事件，再看三个观测视图。demo 把 TOKEN_BUDGET 收紧到 600，让超预算场景（700 token）落进 budget 桶，默认 1000 不变，仅演示收紧。

先看一个会话的 wire 事件（协议原样，没有 seq/ts，那是观测视图）：

--- session s-ok: POST /chat { text: "现在几点了？" } ---
  [1] tool_call_start get_time({})
  [2] tool_result ok=true result="2026-08-11T14:26:29.496Z"
  [3] text_delta "查询完成，当前时间是 2026-08-11T14:26:29.496Z。"
  [4] done usage={"inputTokens":48,"outputTokens":24,"cachedTokens":30}

视图一 · GET /traces：同样的四个事件，观测层把它们变成盖章事件行，并算好派生物：

  trace t1 · session=s-ok result=ok
  events=4 duration=1ms tokens(in=48 out=24 cached=30)
  [00] 2026-08-11T14:26:29.496Z tool_call_start get_time({})
  [01] 2026-08-11T14:26:29.496Z tool_result ok=true result="2026-08-11T14:26:29.496Z"
  [02] 2026-08-11T14:26:29.497Z text_delta "查询完成，当前时间是 2026-08-11T14:26:29.496Z。"
  [03] 2026-08-11T14:26:29.497Z done usage={"inputTokens":48,"outputTokens":24,"cachedTokens":30}
  finalText: "查询完成，当前时间是 2026-08-11T14:26:29.496Z。"

视图二 · GET /costs/s-ok：s-ok 跑了两轮（两次 POST /chat），账单累计两轮四行：

  prices: input $3.000000/M · output $15.000000/M · cached input $0.300000/M
  run t1 (ok) — 1 model call(s)
    call 1: in=48 (cached 30) out=24 → $0.000423  [cache saved $0.000081]
  run t2 (ok) — 1 model call(s)
    call 1: in=48 (cached 30) out=24 → $0.000423  [cache saved $0.000081]
  session totals: $0.000846 · tokens in=96 out=48 cached=60 · cache saved $0.000162

每轮 $0.000423 的组成：未命中输入 18 × $3 + 命中输入 30 × $0.3 + 输出 24 × $15，除以一百万。缓存省了 $0.000081/轮——这就是「命中部分按 1 折计费」落在账单上的样子。

视图三 · GET /failures：五个场景归桶，一眼看出今晚坏在哪：

  total=5 ok=2 byKind={"error":1,"tool":1,"budget":1}
  ✗ t3 · s-toolfail · kind=tool · result=ok
    events: tool_result
    message: Error: add() expects numeric a and b, got a="x", b=2
  ✗ t4 · s-err · kind=error · result=fail
    events: error
    message: provider HTTP 500: {"error":{"message":"upstream provider timeout (simulated)"}}
  ✗ t5 · s-budget · kind=budget · result=ok
    events: done
    message: total tokens 700 exceed budget 600

注意三桶的语义差别：t3 和 t5 的 result 都是 ok（循环跑完了），但一个工具坏了、一个烧钱超预算；只有 t4 是 fail（provider 崩溃，用户什么都没拿到）。看板把「崩溃 / 工具失败 / 超预算」分开计数，正是为了回答「先修什么」。

常见坑与失败模式

坑一：trace 过大。 事件行是「每个事件一行」。一个 10 轮的工具循环，加上每个 tool_result 里可能很大的工具输出（读文件、搜索结果），一条 trace 就能到几十 KB。全量存、全量查（GET /traces 返回全部事件）在小 demo 里没问题，生产里要么截断大 payload（tool 输出只记前 N 字符）、要么把事件行与派生物分开存（行存冷存储、指标存热存储）、要么给 trace 设上限。我们 v07 选择「全量 + 内存」，教学优先；真实产品的第一条经验就是给 trace 设体积上限。

坑二：时间与顺序口径。 三个具体陷阱：① ts 是观测层看到事件的时间，不是事件发生的时间。loop 里两个事件同毫秒发出时 ts 可能相同，durationMs 会是 0；要精确计时，得在事件源头打点（loop 里记 performance.now()），观测层只负责聚合。② seq 是全局序号，跨会话可比较先后；单会话内若有人按 ts 排序，同毫秒的事件顺序会「乱」，排序键应该是 seq，不是 ts。③ 多进程/多实例部署时，全局 seq 需要换成 (实例 id, 本地 seq) 或时间戳加序号，否则两个实例各数各的，全局序就崩了。

坑三：成本口径，中间轮次的 token 不可见。 这是本章最诚实的一个坑：ch15 协议只在 done 上带 usage，且只带最后一轮 LLM 调用的用量。一个「查时间」的会话要两轮（round 1 调工具、round 2 出答案），但 trace 里只有 round 2 的 usage（demo 里 in=48 out=24），round 1 的 42/18 在协议里根本没有出现过。按 trace 核算的成本是「协议可见用量」的下界，不是精确账单。要精确到每次调用，三条路：a) 在 provider 层按调用记账（响应里本来就有 usage，只是协议没传输）；b) 流式响应时用 stream_options.include_usage 让最后一块增量自带 usage（OpenAI 的做法）；c) 扩展协议给每次 LLM 调用一个事件。记账的精度取决于协议能带出多少数据。这是设计事件协议时就要想好的事（ch15 埋的伏笔，本章兑现一半）。

坑四：把观测逻辑塞进 core。 在 loop 里直接写 console.log、直接算成本、直接归桶，core 从此认识「可观测性」，删掉观测模块 core 就变样。红线不变：loop 只做一件事（emit 协议事件）；盖戳、聚合、记账、归桶全部在 onEvent 外面（recorder）和纯函数里（cost/failboard）。判断标准还是那句话：把 server 和 cli 删掉，loop 还能不能跑、能不能测？

小结

下一章（ch21）做 MCP 接入 v08：手写 JSON-RPC 2.0 over stdio，把外部工具接进注册表，可观测层正好用来回答「MCP 工具调用为什么失败、花了多少」。先把本章练习做完：亲手实现事件行 → trace → 成本 → 看板这四步。

延伸阅读

完成阅读，去做练习 →