Claude Agent SDK 实战

崩了怎么办，脏了怎么办。

先看两条时间线。都是我真遇到过的。

时间线 A：跑到第 15 轮，进程被 Ctrl-C。

turn 1  读 README
turn 2  读 tests/test_core.py
...
turn 14 跑测试，3 个失败
turn 15 正在读 core.py
        ^C

终端一关，前面 14 轮读过的文件、跑过的测试、试过的方案，全没了。我只有一屏滚动的输出，连“上一轮改了什么”都说不清。

时间线 B：让主 Agent 去调研一个库。

主历史:  1 条任务
  └─ 子任务：调研 libx v2 的破坏性变更
       ├─ 搜了 12 页
       ├─ 读了 8 个文档
       └─ 把 3 万 token 的搜索结果塞回主历史
主历史:  40 条消息，后面全乱了

它不是忘了目标，是主上下文被调研垃圾淹了。你问它“接下来改哪”，它开始复述搜索结果。

这两个问题，一个要“记住”，一个要“隔离”。这一章把“子 Agent”讲完整：它是什么、怎么并行、怎么分工、权限和深度怎么管。

8.1 会话是一棵只追加的树

会话的本质不是“聊天记录”，是“状态”。

模型没有记忆；它每一轮能看到的，只有你重新发给它的 messages。所以“恢复会话”不是让模型想起什么，而是把之前的消息重新加载出来，再发一遍。

最小的一版，一行一条 JSON，只追加：

@dataclass
class SessionLog:
    path: Path

    def append(self, message: dict) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        with self.path.open("a", encoding="utf-8") as f:
            f.write(json.dumps(message, ensure_ascii=False) + "\n")

    def load(self) -> list[dict]:
        if not self.path.exists():
            return []
        return [
            json.loads(line)
            for line in self.path.read_text(encoding="utf-8").splitlines()
            if line.strip()
        ]

    def fork(self, new_path: Path) -> "SessionLog":
        new_path.parent.mkdir(parents=True, exist_ok=True)
        new_path.write_text(self.path.read_text(encoding="utf-8"), encoding="utf-8")
        return SessionLog(new_path)

三个动作：

• append：每产生一条消息，就落盘。不要等任务结束再写；崩的时候你等不到。
• load：进程重启后，把消息读回来，继续发。
• fork：复制一份，做“另一条分支”。这里只是整份复制；真实的 fork 要记录分支点，SDK 里是 fork_session。

为什么只追加？因为简单、可审计、不会写坏。你不需要“更新第 3 条消息”，只需要“在后面加一条”。出了问题，把 JSONL 打开，一眼能看到发生了什么。

一个最小的会话文件大概长这样：

{"role": "user", "content": "fix the failing test"}
{"role": "assistant", "content": [{"type": "tool_use", "id": "t1", "name": "read_file", "input": {"path": "README.md"}}]}
{"role": "user", "content": [{"type": "tool_result", "tool_use_id": "t1", "content": "..."}]}

第一行必须是 user。第 9 章会讲一个真实的装配 bug：我一开始漏了这条，恢复时模型看到一串工具调用，却不知道用户要干什么。

8.2 resume 到底恢复什么，不恢复什么

很多人以为 resume 是“让模型想起之前的事”。不是。

resume 做的是：把 transcript 重新加载，重新发一遍。它恢复的是消息，不是世界。

所以 resume 之后：

• token 会涨。 你不是从零开始，是把历史又发了一遍；
• 副作用不会自动回滚。 已经发出去的退款、已经改掉的文件，还在；
• 进程状态不会回来。 正在跑的命令、临时变量，都没了。

正确的心态是：resume 是“读档”，不是“时光倒流”。 读档之后，先检查世界有没有变，再决定下一步。

还有一笔账：resume 的输入成本，约等于历史字符数。20 轮、10k 字符的会话，resume 一次就是 10k 字符的输入；它不是“从零开始”，也不是“免费继续”。

8.3 fork：从某个点开一条分支

有时候你不想“接着跑”，你想“从某个点试另一条路”。

比如：

• 同一组失败测试，想试两种修法；
• 一个方案改到一半，想回到分叉点，换一种实现；
• 想保留主线，另开一条“实验分支”。

fork 就是干这个的：

main = SessionLog(Path("session.jsonl"))
experiment = main.fork(Path("session-experiment.jsonl"))

教学版只是整份复制；真实 SDK 里是 fork_session / resume_session_at，可以指定从哪条消息分叉。

fork 的价值不是“多一份文件”，是不污染主线。实验失败，把分支删了；主线还在。

8.4 子 Agent：另一个循环，只交回结论

子 Agent 不是“函数调用”，是“另一个循环”。

它有自己的 messages、自己的工具、自己的预算；跑完之后，只把最终文本交给父循环：

def run_subagent(client, task, system, *, model, tools=None, run_tool=None, max_steps=6):
    messages = [{"role": "user", "content": task}]
    for _ in range(max_steps):
        reply = client.messages.create(
            model=model, max_tokens=2048, system=system,
            tools=tools or [], messages=messages,
        )
        messages.append({"role": "assistant", "content": [b.model_dump() for b in reply.content]})
        calls = [b for b in reply.content if b.type == "tool_use"]
        if not calls:
            text = "".join(b.text for b in reply.content if b.type == "text")
            return SubagentResult(text, messages)
        results = [
            {"type": "tool_result", "tool_use_id": c.id, "content": run_tool(c.name, c.input)}
            for c in calls
        ]
        messages.append({"role": "user", "content": results})
    return SubagentResult("stopped: max steps", messages)

关键点：

• 独立的历史。 子 Agent 读了多少文件、搜了多少网页，都在它自己的 messages 里；父循环看不到。
• 只回最终文本。 父循环只拿到一段结论，不拿到过程。
• 独立的预算。 max_steps、模型、工具集都可以和父循环不同。
• 可并行。 多个子 Agent 可以同时跑；父循环等结果。

agent_session.py 跑出来是这样：

session: 3 messages after resume
fork:    3 messages
subagent: 结论：这个库的 v2 默认开启缓存；迁移时先关缓存，再改 API。
main history: 1 message(s), unchanged

注意最后一行：主历史还是 1 条。子 Agent 跑完，主历史没被污染。

父循环看不到什么：

这不是“信息丢失”，是“隔离的代价”。你想看过程，去读子 Agent 的 transcript；不要让它把过程塞回主历史。

8.5 并行：三个调研员，一个主循环

子 Agent 最有用的时候，是“多个独立子问题”。

agent_parallel.py 起三个 researcher，每个读一份文档、回一条结论：

cd code
python3 -m tiny.agent_parallel

输出：

[subagent] researcher-1: A 文档：v2 默认开启缓存，迁移时先关缓存。
[subagent] researcher-2: B 文档：v2 删除了 v1 的同步 API，需要改调用点。
[subagent] researcher-3: C 文档：v2 的错误码从 4xx 改成了 5xx，重试逻辑要改。
wall: 0.20s (serial ≈ 0.60s)
main history: 1 message(s), unchanged

三个子 Agent 并行跑，墙钟时间约等于最慢的那个（0.20s），不是三个相加（0.60s）。主历史还是 1 条。

但并行不是免费的：

• 总 token 更高。 三个子 Agent 各自读文档、各自推理；你买的是时延，不是成本。
• 结果要合并。 父循环拿到三段结论，还是要自己汇总、去重、判断冲突。
• 任务要独立。 三个子 Agent 如果改同一份文件，合并冲突会比串行更麻烦。

所以并行的前提是：子任务彼此独立，且结果能合并。 调研、检索、多文件摘要适合；改代码不适合。

8.6 分工合同：别只说“你去调研一下”

子 Agent 的效果，很大程度上取决于你怎么“派活”。

一个可用的分工合同，至少写清六件事：

一个例子：

RESEARCHER = AgentDefinition(
    description="调研员：读指定文件，回答一个具体问题。",
    prompt=(
        "只回三段：结论一行；依据最多三行；不确定的地方写'不确定'。"
        "总长不超过 500 字；不要粘贴原文。"
    ),
    tools=["Read", "Grep", "Glob"],
    maxTurns=6,
)

description 是父循环唯一能看到的东西。它决定“派不派”；prompt 是子 Agent 自己的系统提示，决定“怎么干”。

结果合同要写“输出什么、多长、不确定怎么办”。否则子 Agent 会给你一篇论文，主上下文还是爆。

8.7 权限与深度：子 Agent 不是“权限放大器”

子 Agent 有自己的工具集，但这不等于“它可以绕过父循环的权限”。

三个原则：

第一，工具集要窄。 调研子 Agent 只给 Read/Grep/Glob；不要给 Edit/Bash，除非它真的需要。子 Agent 的 tools 是父循环授予的，不是它自己抢的。

第二，权限不会自动继承。 子 Agent 有自己的 permissionMode 和工具面；父循环的 can_use_tool 不一定覆盖它。第 14 章会讲 SDK 里怎么给子 Agent 设权限。

第三，限制转包深度。 子 Agent 也能再开子 Agent；不加深度限制，就会出现“Agent 套 Agent 套 Agent”，成本和时延都失控。可以在 prompt 里写“不要再开子 Agent”，或者在代码里传 depth，超过 2 就拒绝。

SDK 里还有两个观测点：

• SubagentStart / SubagentStop Hook：看它什么时候开始/结束、返回多长；
• stop_task()：停掉一个跑偏的子 Agent。

第 14 章会把这两个接上。

8.8 一次真实的污染

时间线 B 里，如果我让子 Agent 把“搜索结果原文”回填给主循环，主历史会从 1 条涨到 40 条：

差别不在“子 Agent 跑得好不好”，在你让它交回什么。

我的做法是：

1. 子 Agent 的 system 里明确写“只回一段结论 + 三条依据”；
2. 父循环对结果做长度上限（比如 1,000 字符）；
3. 子 Agent 的过程写进它自己的 transcript，需要时再查；
4. 父循环拿到结论后，必须自己验证（跑测试/看 diff），不能把结论当事实。

第 4 条最重要。子 Agent 说“这个库可以升级”，不代表真的可以；它只是另一个会犯错的循环。

8.9 什么时候不该开子 Agent

不是所有任务都值得开子 Agent。开一个，就多一个循环、多一份 token、多一套权限。

一句话：子 Agent 适合“查”，不适合“改”。

8.10 恢复清单

进程崩了、上下文脏了，先按这六条走：

• 会话文件在不在？最后一条是什么？
• 初始任务写进会话了吗？（第 9 章那个 bug）
• progress.md 在不在？做到哪、还差什么？
• 文件检查点有没有？改坏的代码能不能回滚？
• resume 之后，先验证世界有没有变（测试、diff、副作用）。
• 主历史脏了的话，能不能 fork 一条干净的分支？

六条里过不了三条，别急着 resume；先把状态找回来。

8.11 跑一遍

cd code
python3 -m tiny.agent_session
python3 -m tiny.agent_parallel

两个 demo 都不调用模型，用脚本演示会话恢复、子 Agent 隔离和并行。

agent_session.py 会生成两个文件，想清理就删掉：

rm -f code/tiny/session.jsonl code/tiny/session-fork.jsonl

真实子 Agent 还要处理权限、结果大小和转包深度；第 14 章会把 SDK 的 agents / Agent / SubagentStart / stop_task 接上。

本章代码 会话日志：https://github.com/kevinkelin/claude-agent-sdk-book/blob/master/code/tiny/session.py 子 Agent：https://github.com/kevinkelin/claude-agent-sdk-book/blob/master/code/tiny/subagent.py 会话/子 Agent 演示：https://github.com/kevinkelin/claude-agent-sdk-book/blob/master/code/tiny/agent_session.py 并行子 Agent：https://github.com/kevinkelin/claude-agent-sdk-book/blob/master/code/tiny/agent_parallel.py 说明：两个 demo 都不调用模型，用脚本演示会话恢复、隔离和并行；真实子 Agent 还要处理权限、结果大小和转包深度。