第17章：高级特性速览 —— 把剩下的知识点补齐

一句话：把 Agent SDK 剩余的实用特性都学一遍，查漏补缺，让你对整个 SDK 有一个完整的认知。

本章目标

• 了解文件检查点机制，学会"后悔药"怎么用
• 掌握 TODO 追踪功能，让 Agent 自己管理任务进度
• 学会让 Agent 向用户提问，做到交互式协作
• 了解 Slash Commands 自定义命令的用法
• 了解插件系统的定位和当前状态
• 预览 TypeScript V2 新 API
• 掌握成本追踪和 Token 用量分析
• 理解 Stop Reasons 的含义和应对策略
• 学会利用 Prompt 缓存来省钱

前置知识

• 需要先看完第4章（query() 函数基础用法）
• 需要先看完第7章（流式输出，理解消息类型）
• 建议看完第10-11章（自定义工具 / MCP）

17.1 文件检查点（File Checkpointing）

Agent 改错文件了怎么办？

你让 Agent 帮你重构代码，它改了十几个文件，结果改完一运行 —— 炸了。你想回退到改之前的状态，怎么办？

如果你用了 Git，可以 git checkout . 一键还原。但如果这些文件还没提交呢？或者你不想依赖 Git 呢？

这就是文件检查点要解决的问题。

检查点 = 自动存档，随时可以读档

玩过单机游戏的都知道"存档"和"读档"的概念。文件检查点就是这个意思：

• 存档：SDK 在 Agent 修改文件之前，自动给文件拍一个"快照"
• 读档：你觉得 Agent 改得不好，可以随时回退到某个快照

比 Ctrl+Z 更强的地方在于：它能同时回退多个文件的修改，而且是跨多步操作的。

工作原理

当你启用文件检查点后：

1. Agent 每次使用 Write、Edit 或 NotebookEdit 工具修改文件时，SDK 会自动备份原文件
2. 消息流中的 user 类型消息会携带一个 checkpoint UUID（检查点标识符）
3. 你可以保存这个 UUID，之后用它来"读档" —— 把文件恢复到那个时间点的状态

代码示例：启用检查点并回退

Python 版本：

from claude_code_sdk import query, ClaudeCodeOptions

async def refactor_with_checkpoint():
    checkpoint_id = None
    session_id = None

    # 第一步：启用检查点，让 Agent 干活
    async for message in query(
        prompt="重构 src/ 目录下的代码，把所有的 var 改成 const/let",
        options=ClaudeCodeOptions(
            allowed_tools=["Read", "Edit", "Glob", "Grep"],
            enable_file_checkpointing=True,  # 开启检查点！
        )
    ):
        # 捕获检查点 UUID
        if message.type == "user":
            checkpoint_id = message.checkpoint_id
            session_id = message.session_id
            print(f"检查点已创建: {checkpoint_id}")

        if message.type == "result":
            print(f"Agent 完成了: {message.result_text}")

    # 第二步：如果不满意，回退！
    if checkpoint_id and session_id:
        print("不满意，正在回退...")
        # 恢复会话并回退文件
        async for message in query(
            prompt="",  # 不需要新指令
            options=ClaudeCodeOptions(
                session_id=session_id,
                enable_file_checkpointing=True,
            )
        ):
            if hasattr(message, 'client'):
                await message.client.rewind_files(checkpoint_id)
                print("文件已恢复到修改之前的状态！")
                break

TypeScript 版本：

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

async function refactorWithCheckpoint() {
  let checkpointId: string | undefined;
  let sessionId: string | undefined;

  // 启用检查点
  for await (const message of query({
    prompt: "重构 src/ 目录下的代码，把所有的 var 改成 const/let",
    options: {
      allowedTools: ["Read", "Edit", "Glob", "Grep"],
      enableFileCheckpointing: true, // 开启检查点
    }
  })) {
    if (message.type === "user") {
      checkpointId = message.checkpointId;
      sessionId = message.sessionId;
    }

    if (message.type === "result") {
      console.log("Agent 完成了:", message.resultText);
    }
  }

  // 不满意？回退！
  if (checkpointId && sessionId) {
    console.log("正在回退到检查点...");
    for await (const message of query({
      prompt: "",
      options: {
        sessionId,
        enableFileCheckpointing: true,
      }
    })) {
      await message.client?.rewindFiles(checkpointId);
      console.log("文件已恢复！");
      break;
    }
  }
}

实际应用场景

注意事项

• 检查点只备份被 Agent 工具修改过的文件，不是整个文件系统的快照
• 回退操作（rewind_files）必须在消息流还没结束之前调用，或者通过恢复会话来调用
• 文件检查点不替代 Git，它更像是一个"会话级别"的撤销功能
• 需要 SDK 版本较新才支持此功能

17.2 TODO 追踪（Todo Tracking）

Agent 也需要"待办清单"

想象你让 Agent 干一件大事，比如"帮我搭建一个完整的 REST API 项目"。这个任务包含很多步骤：

1. 初始化项目
2. 设计数据库 Schema
3. 写 CRUD 接口
4. 加上认证中间件
5. 写测试
6. 加上错误处理

Agent 怎么保证不遗漏？怎么让你知道它干到哪一步了？

答案就是 TODO 追踪。Agent 会自动创建一个待办清单，逐项完成，并实时更新状态。

TODO 的生命周期

每个 TODO 项都有一个状态：

pending（待处理）→ in_progress（进行中）→ completed（已完成）

当一组 TODO 全部完成后，整组会被自动清理掉。

什么时候会自动创建 TODO？

SDK 会在以下情况自动创建 TODO 列表：

• 复杂的多步骤任务：需要 3 个以上独立操作的任务
• 用户给了一个列表：比如"帮我做这 5 件事"
• 非简单操作：需要跟踪进度才好管理的任务
• 用户明确要求：比如"用 TODO 来跟踪进度"

简单任务（比如"帮我改一下这个变量名"）不会触发 TODO 创建，因为没必要。

监控 TODO 变化

TODO 的状态更新会出现在消息流中。你可以通过监听消息来跟踪进度：

TypeScript 版本：

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

for await (const message of query({
  prompt: "优化我的 React 应用性能，用 TODO 跟踪进度",
  options: {
    allowedTools: ["Read", "Edit", "Glob", "Grep", "Bash"],
  }
})) {
  // TODO 状态更新会出现在 assistant 类型的消息中
  if (message.type === "assistant") {
    // 检查是否有 TODO 相关的工具调用
    if (message.content) {
      for (const block of message.content) {
        if (block.type === "tool_use" && block.name === "TodoWrite") {
          console.log("TODO 状态更新:");
          console.log("  任务:", block.input.todos);
        }
      }
    }
  }

  if (message.type === "result") {
    console.log("全部完成！");
  }
}

构建一个进度追踪器

如果你想在自己的应用中展示 Agent 的任务进度，可以构建一个 TODO 追踪器：

TypeScript 版本：

class TodoTracker {
  private todos: Map<string, { content: string; status: string }> = new Map();

  update(todoData: any) {
    if (todoData.todos) {
      for (const todo of todoData.todos) {
        this.todos.set(todo.id, {
          content: todo.content,
          status: todo.status
        });
      }
    }
  }

  getProgress(): { total: number; completed: number; percentage: number } {
    const total = this.todos.size;
    const completed = [...this.todos.values()]
      .filter(t => t.status === "completed").length;
    return {
      total,
      completed,
      percentage: total > 0 ? Math.round((completed / total) * 100) : 0
    };
  }

  display() {
    console.log("\n=== 任务进度 ===");
    for (const [id, todo] of this.todos) {
      const icon = todo.status === "completed" ? "[x]"
                 : todo.status === "in_progress" ? "[~]"
                 : "[ ]";
      console.log(`${icon} ${todo.content}`);
    }
    const progress = this.getProgress();
    console.log(`\n进度: ${progress.completed}/${progress.total} (${progress.percentage}%)`);
  }
}

// 使用
const tracker = new TodoTracker();

for await (const message of query({
  prompt: "搭建一个 Express + TypeScript 项目，包含用户注册登录功能",
  options: {
    allowedTools: ["Read", "Write", "Edit", "Bash", "Glob"],
  }
})) {
  if (message.type === "assistant" && message.content) {
    for (const block of message.content) {
      if (block.type === "tool_use" && block.name === "TodoWrite") {
        tracker.update(block.input);
        tracker.display();
      }
    }
  }
}

运行后你可能看到这样的输出：

=== 任务进度 ===
[x] 初始化 TypeScript 项目配置
[~] 设计用户数据模型
[ ] 实现注册接口
[ ] 实现登录接口
[ ] 添加 JWT 认证中间件
[ ] 编写测试用例

进度: 1/6 (17%)

TODO 追踪的价值

1. 用户体验好：用户能看到 Agent 在做什么、做到哪了
2. 减少遗漏：Agent 自己也不会忘记该做的事
3. 可暂停恢复：结合 Session 恢复功能，可以中断后继续从上次的 TODO 进度接着做
4. 进度可视化：很容易做成进度条展示给用户

17.3 用户输入（User Input）

Agent 遇到不确定的事情，怎么办？

你让 Agent "帮我搭建一个移动应用项目"，但 Agent 不知道你想用 React Native 还是 Flutter，不知道你需不需要后端，不知道你要上架 iOS 还是 Android。

这时候，Agent 有两个选择：

1. 猜一个 —— 可能猜错，做了一堆白费
2. 问你 —— 先搞清楚需求再动手

显然第二种更靠谱。User Input 机制就是让 Agent 能在工作过程中暂停下来，问你问题，等你回答了再继续。

AskUserQuestion 工具

SDK 内置了一个 AskUserQuestion 工具，Agent 可以通过它向用户提问。这个工具支持：

• 1-4 个问题
• 每个问题可以带 2-4 个选项（类似选择题）
• 用户可以选择选项，也可以自由输入

canUseTool 回调：核心机制

要启用用户输入功能，你需要在调用 query() 时传入一个 canUseTool 回调函数。当 Agent 想使用某个工具（包括 AskUserQuestion）时，这个回调会被触发。

Python 版本：

from claude_code_sdk import query, ClaudeCodeOptions

async def handle_tool(tool_name: str, tool_input: dict) -> str | bool:
    """处理 Agent 的工具使用请求"""

    if tool_name == "AskUserQuestion":
        # Agent 想问用户问题
        question = tool_input.get("question", "")
        options = tool_input.get("options", [])

        print(f"\nAgent 问你: {question}")
        if options:
            for i, opt in enumerate(options):
                print(f"  {i + 1}. {opt}")
            choice = input("请输入选项编号或自由输入: ")
        else:
            choice = input("请输入你的回答: ")

        return choice  # 把用户的回答返回给 Agent

    # 其他工具自动批准
    return True

# 使用
async for message in query(
    prompt="帮我搭建一个新的 Web 项目",
    options=ClaudeCodeOptions(
        allowed_tools=["Read", "Write", "Edit", "Bash", "Glob",
                       "AskUserQuestion"],  # 注意要包含这个工具
        can_use_tool=handle_tool,
    )
):
    if message.type == "result":
        print(f"项目搭建完成！\n{message.result_text}")

TypeScript 版本：

import { query } from "@anthropic-ai/claude-code";
import * as readline from "readline";

// 简单的命令行输入函数
function askUser(prompt: string): Promise<string> {
  const rl = readline.createInterface({
    input: process.stdin,
    output: process.stdout
  });
  return new Promise(resolve => {
    rl.question(prompt, answer => {
      rl.close();
      resolve(answer);
    });
  });
}

for await (const message of query({
  prompt: "帮我搭建一个新的 Web 项目",
  options: {
    allowedTools: ["Read", "Write", "Edit", "Bash", "Glob",
                   "AskUserQuestion"],
    canUseTool: async (toolName: string, toolInput: any) => {
      if (toolName === "AskUserQuestion") {
        const question = toolInput.question || "";
        const options = toolInput.options || [];

        console.log(`\nAgent 问你: ${question}`);
        if (options.length > 0) {
          options.forEach((opt: string, i: number) => {
            console.log(`  ${i + 1}. ${opt}`);
          });
        }

        const answer = await askUser("请输入你的回答: ");
        return answer; // 返回给 Agent
      }

      return true; // 其他工具自动批准
    }
  }
})) {
  if (message.type === "result") {
    console.log("项目搭建完成！");
  }
}

运行效果

Agent 问你: 你想用哪种前端框架？
  1. React (最流行，生态最丰富)
  2. Vue (上手简单，中文文档好)
  3. Svelte (性能极好，代码量少)
  4. Angular (企业级，功能全面)
请输入你的回答: 1

Agent 问你: 需要配置后端 API 吗？
  1. 是的，用 Express.js
  2. 是的，用 Fastify
  3. 不需要，只做前端
请输入你的回答: 3

... Agent 开始搭建 React 项目 ...

什么时候该让 Agent 问，什么时候该让它自己决定？

17.4 Slash Commands（自定义命令）

什么是 Slash Commands？

如果你用过 Slack 或 Discord，你一定见过 /command 这种斜杠命令。Claude Code 也支持类似的机制 —— 你可以定义自己的命令，用一个简短的斜杠指令触发一系列复杂的操作。

比如：

• /review —— 自动审查当前代码变更
• /deploy —— 自动执行部署流程
• /test —— 自动运行测试并分析失败原因

命令文件格式

每个命令是一个 .md 文件，放在项目的 .claude/commands/ 目录下。文件由两部分组成：

1. YAML 头部（frontmatter）：命令的元数据
2. 正文：命令的提示词模板

---
description: 审查代码变更，给出改进建议
argument-hint: [文件路径或 PR 编号]
allowed-tools: Read, Grep, Glob
---

请审查以下代码变更：

$ARGUMENTS

审查要点：
1. 代码质量和可读性
2. 潜在的 Bug 和安全问题
3. 性能隐患
4. 测试覆盖度

请给出具体的改进建议，包括代码示例。

YAML 头部字段说明

目录结构

命令可以按分类放在子目录中：

.claude/
  commands/
    review.md          # /review 命令
    deploy.md          # /deploy 命令
    test/
      unit.md          # /test:unit 命令
      e2e.md           # /test:e2e 命令
    docs/
      generate.md      # /docs:generate 命令

子目录中的命令用冒号 : 来分隔路径，比如 /test:unit。

示例：创建一个 /review 命令

在 .claude/commands/review.md 中写入：

---
description: 审查 Git 暂存区的代码变更
allowed-tools: Read, Grep, Glob, Bash
---

请执行以下步骤来审查代码：

1. 运行 `git diff --staged` 查看暂存区的变更
2. 逐文件分析变更内容
3. 检查以下几个维度：
   - 代码逻辑是否正确
   - 是否有明显的 Bug
   - 命名是否清晰
   - 是否有重复代码
   - 错误处理是否完善
4. 给出总体评价和具体改进建议

输出格式：
- 总体评分（1-10）
- 每个文件的问题列表
- 改进建议（附代码示例）

然后在 Claude Code 中输入 /review，就会自动执行这个审查流程。

示例：创建一个 /commit 命令

---
description: 自动生成 commit message 并提交
allowed-tools: Bash, Read, Grep
---

请执行以下步骤：

1. 运行 `git diff --staged` 查看变更
2. 分析变更内容，生成符合 Conventional Commits 规范的 commit message
3. commit message 格式：`type(scope): description`
   - type: feat/fix/docs/style/refactor/test/chore
   - scope: 变更涉及的模块
   - description: 简短描述
4. 用 `git commit -m "生成的消息"` 提交

如果没有暂存的变更，提示用户先 `git add`。

命令中的变量

命令模板支持 $ARGUMENTS 变量，它会被替换为用户在命令后面输入的内容：

/review src/utils.ts

在这个例子中，$ARGUMENTS 就是 src/utils.ts。

17.5 Plugins（插件系统）

插件是什么？

插件是 Claude Code SDK 的一个深度扩展机制。如果说 MCP 是给 Agent 加"新工具"，Slash Commands 是给用户加"快捷方式"，那么插件就是给整个 SDK 加"新能力"。

插件 vs MCP vs Skills

当前状态

截至目前，插件系统还在持续演进中。官方文档提供了基础的插件 API，但生态还不算成熟。如果你需要扩展 SDK 功能，建议优先考虑：

1. MCP 工具 —— 给 Agent 添加新能力（第11章已讲）
2. Slash Commands —— 定义常用操作的快捷方式（本章 17.4 已讲）
3. Hooks —— 在特定事件前后执行自定义逻辑（比如工具调用前后）
4. 插件 —— 上面三个都满足不了时，再考虑插件

插件的基本结构

插件本质上是一个符合特定接口的模块，可以在 SDK 初始化时注册：

// 插件的基本形态（概念示例）
const myPlugin = {
  name: "my-custom-plugin",
  version: "1.0.0",

  // 在 SDK 初始化时调用
  initialize(context) {
    console.log("插件已加载");
  },

  // 在消息处理时调用
  onMessage(message) {
    // 自定义消息处理逻辑
  },

  // 在 SDK 关闭时调用
  cleanup() {
    console.log("插件已卸载");
  }
};

由于插件 API 还在演进中，具体的接口可能会变化。建议关注官方文档的更新。

17.6 TypeScript V2 预览版

为什么要有 V2？

在前面的章节中，我们一直在用 query() 这个函数。它是一个异步生成器（async generator），用 for await...of 来遍历消息流。这种方式能用，但有几个不太爽的地方：

1. 没有"会话"概念 —— 每次调用 query() 都像是在打一个新电话，想继续上次的对话得自己管 sessionId
2. 生命周期管理麻烦 —— 什么时候开始、什么时候结束、资源怎么释放，都得自己处理
3. 多轮对话写起来别扭 —— 每次都要写一个 for await 循环

V2 API 引入了 Session（会话）的概念，让这些事情变得更自然。

V2 的核心 API

V2 提供了三个核心函数：

import {
  unstable_v2_createSession,   // 创建会话
  unstable_v2_resumeSession,   // 恢复会话
  unstable_v2_prompt,          // 快速单轮调用（不需要会话）
} from "@anthropic-ai/claude-code";

注意函数名前面的 unstable_v2_ 前缀 —— 这说明 API 还不稳定，将来可能会变。但核心思路不会变。

快速单轮调用：unstable_v2_prompt

如果你只是想快速问一个问题，不需要会话管理，可以用 unstable_v2_prompt()：

import { unstable_v2_prompt } from "@anthropic-ai/claude-code";

const result = await unstable_v2_prompt("1 + 1 等于几？");
console.log(result); // "2"

对比 V1 的写法：

// V1：需要写一个循环
for await (const message of query({ prompt: "1 + 1 等于几？" })) {
  if (message.type === "result") {
    console.log(message.resultText);
  }
}

// V2：一行搞定
const result = await unstable_v2_prompt("1 + 1 等于几？");

是不是简洁多了？

会话模式：createSession + send + stream

对于多轮对话场景，V2 的写法更自然：

import { unstable_v2_createSession } from "@anthropic-ai/claude-code";

// 创建一个会话（注意 await using 语法，自动清理资源）
await using session = unstable_v2_createSession({
  allowedTools: ["Read", "Edit", "Glob", "Bash"],
});

// 第一轮对话
session.send("帮我看看当前项目的目录结构");
for await (const event of session.stream()) {
  if (event.type === "text") {
    process.stdout.write(event.text);
  }
}

// 第二轮对话（自动继承上下文！）
session.send("src 目录里最大的文件是哪个？");
for await (const event of session.stream()) {
  if (event.type === "text") {
    process.stdout.write(event.text);
  }
}

// 第三轮对话
session.send("帮我重构一下那个文件");
for await (const event of session.stream()) {
  if (event.type === "text") {
    process.stdout.write(event.text);
  }
}
// session 离开作用域时自动清理

V1 vs V2 对比

恢复会话

如果你的应用崩溃了或者需要重启，可以用 unstable_v2_resumeSession() 恢复之前的会话：

import {
  unstable_v2_createSession,
  unstable_v2_resumeSession,
} from "@anthropic-ai/claude-code";

// 创建会话并保存 ID
const session = unstable_v2_createSession({
  allowedTools: ["Read", "Edit"],
});

session.send("帮我分析项目");
for await (const event of session.stream()) {
  // 处理事件...
}

// 保存会话 ID（比如存到数据库）
const sessionId = session.id;
await session[Symbol.asyncDispose]();

// ... 之后恢复 ...
const restored = unstable_v2_resumeSession(sessionId, {
  allowedTools: ["Read", "Edit"],
});

restored.send("继续上次的分析");
for await (const event of restored.stream()) {
  // 处理事件...
}

什么时候该用 V2？

• 个人项目、实验性项目 —— 放心用，体验更好
• 生产项目 —— 建议暂时还是用 V1 的 query()，等 V2 稳定后再迁移
• Python 项目 —— 目前只能用 V1

17.7 成本追踪（Cost Tracking）详解

用 Agent 也是要花钱的

每次 Agent 处理请求，都会消耗 Token。Token 就是钱。如果你在做一个商业产品，必须精确追踪每次调用的成本，否则月底账单会吓你一跳。

Token 使用量的四个维度

每条消息的 usage 字段包含四个关键数据：

message.usage = {
    "input_tokens": 1500,                # 输入 Token 数
    "output_tokens": 800,                # 输出 Token 数
    "cache_creation_input_tokens": 500,   # 缓存创建 Token 数
    "cache_read_input_tokens": 1000,      # 缓存命中 Token 数
}

它们分别是什么意思？

去重：相同 ID 的消息只算一次

这是一个容易踩的坑。在消息流中，你可能会看到多条消息的 id 是相同的：

assistant (text)      { id: "msg_1", usage: { output_tokens: 100 } }
assistant (tool_use)  { id: "msg_1", usage: { output_tokens: 100 } }
assistant (tool_use)  { id: "msg_1", usage: { output_tokens: 100 } }
assistant (text)      { id: "msg_2", usage: { output_tokens: 98 } }

上面这 4 条消息，msg_1 出现了 3 次，但 Token 只能算 1 次！因为它们属于同一个 API 调用的不同部分（文本 + 多个工具调用）。

正确的做法是按 message ID 去重：

seen_ids = set()
total_input = 0
total_output = 0

async for message in query(prompt="...", options=options):
    if message.type == "assistant" and hasattr(message, "usage"):
        msg_id = message.id
        if msg_id not in seen_ids:
            seen_ids.add(msg_id)
            total_input += message.usage.get("input_tokens", 0)
            total_output += message.usage.get("output_tokens", 0)

print(f"总输入 Token: {total_input}")
print(f"总输出 Token: {total_output}")

最终结果中的总成本

在 result 类型的消息中，SDK 会直接给你一个总成本：

for await (const message of query({ prompt: "..." })) {
  if (message.type === "result") {
    // 总成本（美元）
    console.log(`总成本: $${message.total_cost_usd}`);

    // 按模型分别统计
    if (message.modelUsage) {
      for (const [model, usage] of Object.entries(message.modelUsage)) {
        console.log(`模型 ${model}:`);
        console.log(`  输入 Token: ${usage.inputTokens}`);
        console.log(`  输出 Token: ${usage.outputTokens}`);
        console.log(`  费用: $${usage.costUSD}`);
      }
    }
  }
}

modelUsage：多模型场景下的按模型统计

如果你使用了子 Agent（subagent），主 Agent 和子 Agent 可能使用不同的模型。modelUsage 字段会按模型分别统计：

模型 claude-sonnet-4-20250514:
  输入 Token: 5000
  输出 Token: 2000
  费用: $0.031

模型 claude-haiku-3-20250514:
  输入 Token: 3000
  输出 Token: 1500
  费用: $0.005

构建一个成本看板

Python 完整示例：

from claude_code_sdk import query, ClaudeCodeOptions

class CostTracker:
    def __init__(self):
        self.seen_ids = set()
        self.total_input_tokens = 0
        self.total_output_tokens = 0
        self.total_cache_create = 0
        self.total_cache_read = 0
        self.total_cost_usd = 0.0

    def record(self, message):
        """记录消息的 Token 使用量（自动去重）"""
        if not hasattr(message, "usage") or not message.usage:
            return

        msg_id = getattr(message, "id", None)
        if msg_id and msg_id in self.seen_ids:
            return  # 已经记录过了，跳过
        if msg_id:
            self.seen_ids.add(msg_id)

        usage = message.usage
        self.total_input_tokens += usage.get("input_tokens", 0)
        self.total_output_tokens += usage.get("output_tokens", 0)
        self.total_cache_create += usage.get("cache_creation_input_tokens", 0)
        self.total_cache_read += usage.get("cache_read_input_tokens", 0)

    def record_result(self, message):
        """记录最终结果中的成本信息"""
        if hasattr(message, "total_cost_usd"):
            self.total_cost_usd = message.total_cost_usd

    def report(self):
        """输出成本报告"""
        print("\n=== 成本报告 ===")
        print(f"输入 Token:       {self.total_input_tokens:>10,}")
        print(f"输出 Token:       {self.total_output_tokens:>10,}")
        print(f"缓存创建 Token:   {self.total_cache_create:>10,}")
        print(f"缓存命中 Token:   {self.total_cache_read:>10,}")
        print(f"总 Token:         {self.total_tokens:>10,}")
        print(f"总成本:           ${self.total_cost_usd:.4f}")

    @property
    def total_tokens(self):
        return (self.total_input_tokens + self.total_output_tokens
                + self.total_cache_create + self.total_cache_read)


# 使用
tracker = CostTracker()

async for message in query(
    prompt="分析当前项目的架构并给出改进建议",
    options=ClaudeCodeOptions(
        allowed_tools=["Read", "Glob", "Grep", "Bash"]
    )
):
    if message.type == "assistant":
        tracker.record(message)

    if message.type == "result":
        tracker.record_result(message)
        print(message.result_text)

tracker.report()

运行后的输出：

... Agent 的分析结果 ...

=== 成本报告 ===
输入 Token:            8,500
输出 Token:            3,200
缓存创建 Token:        2,000
缓存命中 Token:        4,500
总 Token:             18,200
总成本:           $0.0524

17.8 Stop Reasons 详解

为什么 Agent 停下来了？

Agent 执行完毕后，result 消息中有一个 stop_reason 字段，告诉你 Agent 停下来的原因。这个信息很重要，因为不同的停止原因意味着不同的后续处理策略。

所有的 Stop Reasons

代码示例：处理不同的 Stop Reason

Python 版本：

from claude_code_sdk import query, ClaudeCodeOptions

async for message in query(
    prompt="帮我生成一份详细的代码审查报告",
    options=ClaudeCodeOptions(
        allowed_tools=["Read", "Glob", "Grep"],
    )
):
    if message.type == "result":
        stop_reason = message.stop_reason

        if stop_reason == "end_turn":
            # 正常完成
            print("任务完成！")
            print(message.result_text)

        elif stop_reason == "max_tokens":
            # Token 不够了
            print("警告：输出被截断了，结果可能不完整")
            print("建议：增大 max_tokens 或把任务拆小一点")
            print("已生成的部分：")
            print(message.result_text)

        elif stop_reason == "refusal":
            # Agent 拒绝执行
            print("Agent 拒绝了这个请求")
            print(f"原因：{message.result_text}")

        elif stop_reason == "interrupt":
            # 被中断了
            print("任务被中断")
            print("已完成的部分：")
            print(message.result_text)

重试策略

当遇到非正常停止时，你可以设计重试策略：

import asyncio
from claude_code_sdk import query, ClaudeCodeOptions

async def robust_query(prompt: str, max_retries: int = 3):
    """带重试的查询函数"""
    for attempt in range(max_retries):
        result_text = ""
        stop_reason = None

        async for message in query(
            prompt=prompt,
            options=ClaudeCodeOptions(
                allowed_tools=["Read", "Glob", "Grep"],
                max_turns=50,
            )
        ):
            if message.type == "result":
                result_text = message.result_text
                stop_reason = message.stop_reason

        if stop_reason == "end_turn":
            return result_text  # 成功

        if stop_reason == "max_tokens":
            print(f"第 {attempt + 1} 次尝试：输出被截断，准备重试...")
            prompt = f"继续上次未完成的任务。上次的部分输出：{result_text[:500]}..."
            continue

        if stop_reason == "refusal":
            print("Agent 拒绝执行，不再重试")
            return None

    print(f"重试 {max_retries} 次后仍未成功")
    return result_text  # 返回最后一次的部分结果

17.9 Prompt 缓存优化

缓存是怎么回事？

每次你调用 Agent，SDK 都会把 system prompt（系统提示词）和工具定义发给模型。如果你连续多次调用，这些内容每次都一样，但每次都要重新"读"一遍 —— 这就是浪费。

Prompt 缓存就是用来解决这个问题的：第一次调用时把不变的内容缓存起来，后续调用直接复用缓存，不用重新处理。

缓存的三种 Token

在 Token 使用量中，有两个跟缓存有关的字段：

1. cache_creation_input_tokens：第一次请求时，把系统提示词和工具定义写入缓存所消耗的 Token。这个费用比普通输入 Token 稍贵。
2. cache_read_input_tokens：后续请求时，从缓存中读取内容所消耗的 Token。这个费用比普通输入 Token 便宜很多（大约只有 10%）。

直观的费用对比

假设你的系统提示词 + 工具定义总共 5000 个 Token，连续调用 10 次：

没有缓存：

每次请求: 5000 input_tokens × $0.003/1K = $0.015
10 次总计: $0.15

有缓存：

第 1 次: 5000 cache_creation_tokens × $0.00375/1K = $0.01875
第 2-10 次: 5000 cache_read_tokens × $0.0003/1K × 9 = $0.0135
10 次总计: $0.03225

省了 78%！ 调用次数越多，省得越多。

怎么最大化缓存命中？

缓存命中的前提是前缀完全一致。也就是说，你的系统提示词和工具定义必须跟上次完全一样，缓存才能生效。

最佳实践：

# 好的做法：保持 options 不变
COMMON_OPTIONS = ClaudeCodeOptions(
    allowed_tools=["Read", "Edit", "Glob", "Grep", "Bash"],
    system_prompt="你是一个代码审查助手，专注于 Python 代码质量。",
)

# 每次调用用相同的 options
async for msg in query(prompt="审查 file1.py", options=COMMON_OPTIONS):
    ...

# 缓存命中！因为 system_prompt 和 tools 没变
async for msg in query(prompt="审查 file2.py", options=COMMON_OPTIONS):
    ...

# 缓存命中！
async for msg in query(prompt="审查 file3.py", options=COMMON_OPTIONS):
    ...

# 不好的做法：每次都改 options
for file in files:
    async for msg in query(
        prompt=f"审查 {file}",
        options=ClaudeCodeOptions(
            allowed_tools=["Read", "Edit", "Glob", "Grep", "Bash"],
            # 每次都不一样的 system_prompt → 缓存失效！
            system_prompt=f"你正在审查 {file}，请专注于这个文件。",
        )
    ):
        ...

监控缓存效果

你可以在成本追踪中观察缓存的效果：

async for message in query(prompt="...", options=options):
    if message.type == "assistant" and hasattr(message, "usage"):
        usage = message.usage
        cache_create = usage.get("cache_creation_input_tokens", 0)
        cache_read = usage.get("cache_read_input_tokens", 0)
        regular_input = usage.get("input_tokens", 0)

        if cache_read > 0:
            hit_rate = cache_read / (cache_read + regular_input) * 100
            print(f"缓存命中率: {hit_rate:.1f}%")
        elif cache_create > 0:
            print(f"首次请求，正在创建缓存 ({cache_create} tokens)")

动手练习

练习1：实现文件检查点保护

目标： 让 Agent 修改代码，但可以一键回退。

步骤：

1. 创建一个测试文件，写入一些简单的代码
2. 启用 enable_file_checkpointing=True，让 Agent 修改这个文件
3. 保存检查点 UUID
4. 验证文件确实被修改了
5. 调用 rewind_files() 回退
6. 验证文件恢复到了原始状态

骨架代码：

from claude_code_sdk import query, ClaudeCodeOptions

async def checkpoint_demo():
    # 第一步：让 Agent 修改文件
    checkpoint_id = None

    async for message in query(
        prompt="把 test.py 文件中的所有 print 语句改成 logging.info",
        options=ClaudeCodeOptions(
            allowed_tools=["Read", "Edit"],
            enable_file_checkpointing=True,
        )
    ):
        if message.type == "user":
            checkpoint_id = message.checkpoint_id
            print(f"保存检查点: {checkpoint_id}")
        # ...

    # 第二步：检查文件是否被修改
    # ...

    # 第三步：回退到检查点
    # ...

练习2：构建带 TODO 追踪的项目初始化工具

目标： 让 Agent 搭建一个项目，并通过 TODO 追踪显示进度。

步骤：

1. 给 Agent 一个复杂的任务（比如"创建一个 Flask REST API 项目，包含用户注册、登录、个人信息管理"）
2. 在 prompt 中明确要求"用 TODO 列表跟踪进度"
3. 监听消息流中的 TODO 更新
4. 在终端中实时显示任务进度条

提示：

# 进度条显示函数
def show_progress(completed, total):
    bar_length = 30
    filled = int(bar_length * completed / total)
    bar = "█" * filled + "░" * (bar_length - filled)
    print(f"\r[{bar}] {completed}/{total} ({completed*100//total}%)", end="")

练习3：体验 V2 API（TypeScript）

目标： 用 V2 API 实现一个简单的多轮对话。

步骤：

1. 安装最新版的 @anthropic-ai/claude-code
2. 用 unstable_v2_createSession() 创建会话
3. 实现至少 3 轮对话，每轮对话基于上一轮的上下文
4. 对比 V1 的 query() 写法，感受区别

骨架代码：

import { unstable_v2_createSession } from "@anthropic-ai/claude-code";

async function main() {
  await using session = unstable_v2_createSession({
    allowedTools: ["Read", "Glob"],
  });

  // 第一轮
  session.send("当前目录有哪些文件？");
  for await (const event of session.stream()) {
    // 处理...
  }

  // 第二轮（自动继承上下文）
  session.send("最大的文件是哪个？有多少行？");
  for await (const event of session.stream()) {
    // 处理...
  }

  // 第三轮
  session.send("总结一下这个项目是做什么的");
  for await (const event of session.stream()) {
    // 处理...
  }
}

main();

本章小结

这一章我们快速浏览了 Agent SDK 的九个高级特性，来回顾一下：

1. 文件检查点 —— Agent 修改文件前自动存档，不满意随时回退。就像游戏的存档/读档。

2. TODO 追踪 —— Agent 自动创建待办清单，逐项完成并更新状态。让复杂任务有条不紊。

3. 用户输入 —— Agent 不确定的时候可以问你，通过 AskUserQuestion 工具和 canUseTool 回调实现。

4. Slash Commands —— 自定义命令快捷方式，用 .md 文件定义，一个斜杠就能触发复杂操作。

5. 插件系统 —— 深度扩展 SDK 的机制，目前还在演进中，日常使用优先考虑 MCP 和 Slash Commands。

6. TypeScript V2 预览 —— 新的 Session-based API，写起来更自然，但目前还不稳定。

7. 成本追踪 —— 精确追踪 Token 使用量和费用，注意按 message ID 去重。total_cost_usd 和 modelUsage 让费用一目了然。

8. Stop Reasons —— 了解 Agent 为什么停下来：end_turn（正常完成）、max_tokens（输出截断）、refusal（拒绝执行）、interrupt（被中断）。不同原因对应不同的处理策略。

9. Prompt 缓存 —— 保持 system prompt 和工具定义不变，让缓存生效，连续调用可以省 70-80% 的输入 Token 费用。

这些特性中，成本追踪和Stop Reasons 是每个生产项目必须关注的；文件检查点和TODO 追踪能大幅提升用户体验；用户输入让 Agent 更智能地与人协作；Prompt 缓存能帮你省钱。

至此，Claude Code Agent SDK 的核心知识点我们已经全部覆盖了。你已经具备了用 SDK 构建各种 Agent 应用的理论基础。

下一章预告

恭喜你！理论部分全部学完了！从下一章开始，我们进入实战篇。

第18章我们会从零开始，用前面学到的所有知识，构建一个完整的实际项目。不再是零散的代码片段，而是一个真正能跑起来、能解决实际问题的 Agent 应用。准备好了吗？