第 12 章：成本与延迟

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

本章目标

做一个把同样的活干得更省钱、更快的 PR 审查 agent 版本。读完本章，你会有：一个按步骤挑选模型的路由器、一段被缓存的提示前缀、一次对改动文件的并发扇出、以及一个一拿到结论就立刻收尾的停止条件。你还会有一份成本核算脚手架，把每次审查实际花了多少打印出来，让下一步优化由数字驱动，而不是凭感觉。

想清楚一件事就行。每次模型调用都有两种成本：钱（输入加输出 token，按模型计价）和时间（一次往返，它主导了用户感知到的延迟）。随手写的 agent 把这两样当免费的，量一大就得为此付钱。生产里的 agent 会把它们当成一笔要省着花的预算。

前置准备

• 前几章已经跑通、并带有 token 与耗时日志的 agent 主循环。如果你现在还看不到每次调用的输入 token、输出 token、墙钟耗时，先把这件事做好。
• 浏览器里开着服务商的定价页。把价格当作随时查阅的输入，而不是要背下来的常量。模型与费率会变，当前数字请看官方定价页。
• 一组固定的评测集：十到二十个结果已知的真实 PR。它是你衡量每次改动的那把尺子。

动手做

1. 建立基线

不先量一遍，就无从优化。在固定 PR 集上跑一遍，每次审查记四个数：输入 token、输出 token、墙钟秒数、模型调用次数（工具调用轮数）。求和再相除，得到每次审查的均值。把它放进一张能反复重跑的小表里，每改一项就重跑。Microsoft 的生产课讲的是同一件事：每次运行的 token 成本和延迟，就是让你能看清 agent 内部、而不是只盯着一个黑盒的指标。

import time
import anyio
from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

# 成本核算脚手架。账单不用自己算价格：ResultMessage 直接给你
# 这次跑的总成本、轮数与 token 用量（usage 的具体字段名以官方为准）。
async def review_with_cost(pr_id, diff, options):
    start = time.monotonic()
    async for message in query(prompt=f"审查这段 diff：\n{diff}", options=options):
        if isinstance(message, ResultMessage):   # 终态：成本与用量都在这里
            elapsed = time.monotonic() - start
            print(f"review={pr_id} turns={message.num_turns} "
                  f"cost=${message.total_cost_usd:.4f} secs={elapsed:.1f} "
                  f"usage={message.usage}")

anyio.run(review_with_cost, pr_id, diff, options)

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

// 成本核算脚手架。账单不用自己算价格：result 消息直接给你
// 这次跑的总成本、轮数与 token 用量。
async function reviewWithCost(prId, diff, options) {
  const start = Date.now();
  for await (const message of query({ prompt: `审查这段 diff：\n${diff}`, options })) {
    if (message.type === "result" && message.subtype === "success") {
      const secs = (Date.now() - start) / 1000;
      console.log(`review=${prId} turns=${message.num_turns} ` +
        `cost=$${message.total_cost_usd.toFixed(4)} secs=${secs.toFixed(1)}`,
        message.usage);
    }
  }
}

2. 给模型分层

不是每一步都需要最强的模型。最终的审查判断用强模型，因为漏掉一个真 bug 代价很高；机械性的子任务用更便宜、更快的模型：判断改了哪些文件、概括一段 diff、决定某个文件值不值得读。这就是生产课里的「路由模型」模式：按复杂度路由，把大模型留给真正需要推理的地方。省下来的钱会累加，因为便宜模型那几步每次审查通常要跑很多次，而强模型那一步只跑一次。

from claude_agent_sdk import ClaudeAgentOptions

# 路由表：哪一步用哪个模型。model 接受别名 "haiku"/"sonnet"/"opus"。
MODEL_FOR = {
    "triage_changed_files":  "haiku",   # 便宜，跑一次
    "summarize_hunk":        "haiku",   # 便宜，逐文件跑
    "is_file_worth_reading": "haiku",   # 便宜，逐文件跑
    "final_review_verdict":  "opus",    # 强，跑一次
    "security_deep_dive":    "opus",    # 强，仅在 triage 标出风险时
}

def options_for(step):
    return ClaudeAgentOptions(model=MODEL_FOR[step], allowed_tools=["Read"])

// 路由表：哪一步用哪个模型。model 接受别名 "haiku"/"sonnet"/"opus"。
const MODEL_FOR = {
  triage_changed_files:  "haiku",   // 便宜，跑一次
  summarize_hunk:        "haiku",   // 便宜，逐文件跑
  is_file_worth_reading: "haiku",   // 便宜，逐文件跑
  final_review_verdict:  "opus",    // 强，跑一次
  security_deep_dive:    "opus",    // 强，仅在 triage 标出风险时
};

function optionsFor(step) {
  return { model: MODEL_FOR[step], allowedTools: ["Read"] };
}

取舍在于：便宜模型做 triage 可能误判某文件低风险而跳过。防住这点的办法是让 triage 偏向升级（拿不准就交给强模型），并对便宜模型标过风险的任何文件保留强模型的安全复查。

3. 缓存稳定前缀

你的系统提示、工具 schema、审查清单在每次调用里都一模一样。不开缓存，你就要在每次审查的每一轮里都按全价重新处理它们。提示缓存让你一次性付一笔写入费，之后在缓存有效期内以很深的折扣读回同一段前缀。用 Agent SDK 时你不必手动标 cache_control 断点：把稳定内容放进 system_prompt，CLI 会替你缓存这段前缀，每个 PR 各异的 diff 走 prompt。

要点只有一条：稳定的放 system_prompt，每个 PR 变的放 prompt。把易变内容混进 system_prompt，前缀就会失配，永远命中不了。命中与否，看 ResultMessage.usage 里的缓存读字段是否非零（具体字段名以官方为准）。

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

options = ClaudeAgentOptions(
    system_prompt=REVIEW_RUBRIC,   # 稳定前缀：CLI 缓存它
    allowed_tools=["Read"],
    model="opus",
)

async for message in query(
    prompt=f"审查这段 diff：\n{pr_diff}",   # 每个 PR 都在变
    options=options,
):
    if isinstance(message, ResultMessage):
        print(message.usage)   # 看缓存读字段是否非零，确认命中

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

const options = {
  systemPrompt: REVIEW_RUBRIC,   // 稳定前缀：CLI 缓存它
  allowedTools: ["Read"],
  model: "opus",
};

for await (const message of query({
  prompt: `审查这段 diff：\n${prDiff}`,   // 每个 PR 都在变
  options,
})) {
  if (message.type === "result" && message.subtype === "success") {
    console.log(message.usage);   // 看缓存读字段是否非零，确认命中
  }
}

两点运维提醒。默认缓存有效期较短（5 分钟，每次命中刷新）；对那些不是每隔几分钟就复用一次的前缀，存在一个 "ttl": "1h" 选项，代价是更高的写入费。还有一个最小可缓存长度（当前的中大型模型量级在 ~1k token 左右）：低于它，缓存会被静默跳过。验证是否命中就看 cache_read_input_tokens 是否非零。准确的阈值与字段请看提示缓存文档。

4. 批量处理独立的工作

如果你要审查十个彼此无依赖的文件，不要做十次串行往返。串行调用的延迟是每次往返之和；并发调用的延迟由最慢的那个决定。把逐文件的概括与风险检查并发扇出去，再汇总它们交给唯一一次最终判断。对多文件 PR 来说，这一招对墙钟时间的提升最大。

import asyncio
from claude_agent_sdk import query, ClaudeAgentOptions, AssistantMessage, TextBlock

async def summarize_and_flag(f):
    opts = ClaudeAgentOptions(model="haiku", allowed_tools=["Read"])
    out = ""
    async for message in query(prompt=f"概括并标注风险：{f}", options=opts):
        if isinstance(message, AssistantMessage):
            for block in message.content:
                if isinstance(block, TextBlock):
                    out += block.text
    return out

async def review_pr(files):
    # 把便宜、独立的逐文件工作并发扇出：每个文件一次 query()
    findings = await asyncio.gather(*[summarize_and_flag(f) for f in files])  # haiku，并行
    # 一次强模型调用看到全部汇总上下文
    return await final_verdict(findings)               # opus，一次

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

async function summarizeAndFlag(f) {
  const opts = { model: "haiku", allowedTools: ["Read"] };
  let out = "";
  for await (const message of query({ prompt: `概括并标注风险：${f}`, options: opts })) {
    if (message.type === "assistant") {
      for (const block of message.message.content) {
        if (block.type === "text") out += block.text;
      }
    }
  }
  return out;
}

async function reviewPr(files) {
  // 把便宜、独立的逐文件工作并发扇出：每个文件一次 query()
  const findings = await Promise.all(files.map(summarizeAndFlag));  // haiku，并行
  // 一次强模型调用看到全部汇总上下文
  return finalVerdict(findings);                       // opus，一次
}

扇出时要留意限流：用有界并发（在 gather 外套一个信号量）能防止你撞上服务商的每分钟上限。这一点我们在部署章里收紧。

5. 提前终止并限制轮数

一旦 agent 拿到足够信息可以给出结论，就停。这里有两种浪费预算的失败模式：agent 继续去读它根本不需要的文件；agent 陷入循环（反复读、反复想）却不收敛。设一个显式的停止条件和一个硬性的工具调用轮数上限。生产课点名了死循环这个问题；清晰的终止条件就是解法。

from claude_agent_sdk import query, ClaudeAgentOptions, ResultMessage

options = ClaudeAgentOptions(
    system_prompt=REVIEW_RUBRIC,
    allowed_tools=["Read"],
    max_turns=6,           # 轮数硬上限：迷糊的循环到顶就停
    max_budget_usd=0.50,   # 预算上限：超了返回 error_max_budget_usd
)

async for message in query(prompt=f"审查这段 diff：\n{pr_diff}", options=options):
    if isinstance(message, ResultMessage):
        if message.subtype == "error_max_budget_usd":
            # 触顶仍未收敛：标记交人工
            handle_capped(message, reason="预算上限")
        # 完成：拿到结论自然收尾，不灌水

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

const options = {
  systemPrompt: REVIEW_RUBRIC,
  allowedTools: ["Read"],
  maxTurns: 6,         // 轮数硬上限：迷糊的循环到顶就停
  maxBudgetUsd: 0.50,  // 预算上限：超了返回 error_max_budget_usd
};

for await (const message of query({ prompt: `审查这段 diff：\n${prDiff}`, options })) {
  if (message.type === "result") {
    if (message.subtype === "error_max_budget_usd") {
      // 触顶仍未收敛：标记交人工
      handleCapped(message, "预算上限");
    }
    // 完成：拿到结论自然收尾，不灌水
  }
}

习得「精打细算」小满不再每步都掏顶配模型，琐碎的活派给便宜模型，能复用的前缀缓存起来，独立的文件并发着读，一拿到结论就停，每次审查花了多少 token、多少秒它都记得清清楚楚。

如何验证

每改一项就重跑基线集并对比那张表。确认每次审查的 token 与墙钟下降，且强模型的调用次数大致稳定在一次。最关键的是，确认在一组留出的 PR 上审查质量没有退化：一个会漏掉真 bug 的便宜 agent 不是更便宜，而是坏了。对缓存而言，在一次审查的第二次调用上断言 cache_read_input_tokens > 0；如果它是零，说明你的前缀在变，或者短于最小长度。

习得「拿数字说话」你现在能在一组留出的 PR 上确认省下来的钱没拿质量换，每改一项就重跑基线表，看 token 和延迟有没有真降、缓存有没有命中、有没有漏掉真 bug。

原理

这四招各管一头，而且能叠加用，互不干扰。分层降低每一步的单价。缓存降低重复前缀的单价。批处理不改变总 token，只降墙钟。提前终止减少步数。按上面的方式做，它们都不会用质量换省钱，因为每一个砍掉的，都是本来就对结论没用的工作。

小结

成本与延迟是设计约束，不是事后才想的事。先测一个基线，把便宜的活交给便宜的模型，把不变的前缀缓存住，把独立的步骤并行，一拿到结论就停。每改一项都重新测一遍，下一步改什么看数字，别凭感觉。

常见坑

• 缓存了每次都在变的内容（缓存块里塞了时间戳或每个 PR 各异的上下文），结果哈希永远对不上，永远命中不了。
• 因为顶配模型已经接好了，就拿它去做琐碎分类。
• 一边抠 token，一边悄悄把审查质量做差了；务必每次重查留出集。
• 无界并发扇出撞上服务商限流，把一次提速变成一墙的 429。
• 把价格当事实写死在代码或正文里；费率会变，请去定价页查。

小满学会了省着花，甚至会主动说：「这活不值得用大模型，我用小的就行。」它开始像个会过日子的人。账房，亮了。

0123456789101112131415

刚点亮 账房 · 地图已点亮 13 / 16

你决定了，带它出门，见见真实的世界。下一站：门外天光。

来源

1. Anthropic 文档：提示缓存 · official
2. Microsoft：AI Agents in Production（成本管理） · official