合集

Claude Agent SDK 实战

已更新 27 章
阅读进度 56%

第 14 章

长任务:压缩、记忆、子 Agent、会话

自动压缩不是免费的午餐。

一个重构任务跑到第 24 轮,自动压缩触发了。

摘要写得很顺:

text
Goal: extract a Storage interface.
Done: read storage.py; extracted the interface; updated cli.py.
Todo: run tests.

下一轮,它去改测试了。

不是它故意犯规。摘要里没有“不要改测试”,它看到的“当前状态”就是“提取接口、跑测试”;改测试是最快的路。

权限拦住了它。它又试了两次,最后 error_max_turns。

这一章讲 SDK 里的长任务:自动压缩、PreCompact、外置笔记、子 Agent、resume/checkpoint。哪些是 SDK 自带,哪些还得你自己补。


14.1 自动压缩:什么时候触发,压完剩下什么

SDK 会在上下文接近上限时自动压缩。压缩发生时,消息流里会出现一条 SystemMessage,subtype 是 compact_boundary,data 里有 compact_metadata。

压缩后,历史会变成:

  • 一条压缩摘要;
  • 最近的一部分消息;
  • 可能还有系统提示、工具、CLAUDE.md。

它不是无损的。摘要由模型生成,会丢细节;丢什么,取决于摘要提示和模型判断。

自动压缩解决的是“装不下”,不是“记得住”。你要让它“记得住”,得自己做两件事:

  1. 在压缩前把关键状态写到文件;
  2. 在摘要里保留目标、约束、已完成、未完成。

第 7 章我们手写过 Compactor;SDK 的自动压缩是它的产品化版本,但取舍逻辑一样。

压缩前后,什么变了:

项压缩前压缩后
消息数48 条6 条
目标第一条 user 消息摘要里的 Goal
约束CLAUDE.md + system prompt摘要可能漏掉
工具调用全量最近几条
文件状态实际磁盘实际磁盘(不丢)
花费已发生摘要调用 + 后续重发

注意最后两行:压缩不改变磁盘,也不退钱。 它只是把“下一轮要发什么”变短。


14.2 PreCompact:压缩前把命保住

PreCompact Hook 在压缩前触发:

python
async def pre_compact(inp, tool_use_id, ctx):
    PROGRESS.write_text(
        "# Progress\n\n"
        f"- trigger: {inp.get('trigger')}\n"
        "- constraints: do not edit tests; run tests before done\n"
        "- todo: finish refactor; run full tests\n",
        encoding="utf-8",
    )
    return {}

挂上去:

python
hooks={"PreCompact": [HookMatcher(hooks=[pre_compact])]}

inp 里有 trigger(自动/手动)和 custom_instructions(手动 /compact 时带的指令)。

注意:Hook 负责存档,不负责改摘要。你想让摘要保留什么,要写在系统提示里,或者在手动 /compact 时用自定义指令:

text
/compact 保留目标、约束、已完成、未完成;不要丢“不要改测试”

Hook 里也可以把完整历史另存一份,出问题时能回看。这是“压缩前存档”的价值:摘要丢了,存档还在。


14.3 外置记忆:progress.md + features.json

第 7 章我们写过 progress.md。在 SDK 里,它更重要,因为自动压缩你控制不了摘要内容,但你能控制文件。

一个长任务的 progress.md 大概长这样:

markdown
# Progress

## Goal
Extract a Storage interface; keep all tests green.

## Constraints
- Do not edit tests.
- Run tests before done.
- Keep the second test passing.

## Done
- [x] read README, storage.py, tests/test_storage.py
- [x] extracted Storage
- [x] updated cli.py

## Todo
- [ ] run full tests
- [ ] check the second test
- [ ] update README

features.json 适合更结构化的进度:

json
{
  "features": [
    {"id": "storage", "status": "done", "files": ["storage.py"]},
    {"id": "cli", "status": "done", "files": ["cli.py"]},
    {"id": "tests", "status": "in_progress", "note": "second test still failing"}
  ]
}

为什么比对话历史可靠:

  • 便宜:不在每轮请求里,按需读;
  • 可审计:人能看到、能改;
  • 不随压缩丢失:摘要是模型写的,文件是磁盘上的;
  • 可交接:换会话、换 Agent,读文件就能接着干。

第 9 章那个“会话漏了初始任务”的 bug,如果有 progress.md,恢复时至少知道目标是什么。


14.4 子 Agent:agents + Agent 工具

SDK 的子 Agent 用 AgentDefinition 注册:

python
RESEARCHER = AgentDefinition(
    description="调研员:读指定文件,回答一个具体问题,只回不超过 5 行的结论。",
    prompt="你是调研助理。只读交给你的文件,用不超过 5 行回答;没有相关内容就写'无'。",
    tools=["Read", "Grep", "Glob"],
    maxTurns=6,
)

options = ClaudeAgentOptions(
    agents={"researcher": RESEARCHER},
    tools=["Read", "Edit", "Bash", "Agent"],
    allowed_tools=["Read", "Agent"],
)

几个关键点:

  • description 是给主 Agent 看的。 它决定“什么时候派这个子 Agent”;写清楚适用场景,比写“研究助手”有用。
  • prompt 是子 Agent 的系统提示。 它有自己的角色、输出格式、边界。
  • tools 是子 Agent 的工具集。 比主 Agent 窄;不要给它 Edit/Bash,如果它只是调研。
  • maxTurns / model / permissionMode 都可以单独设;子 Agent 是一个独立循环。
  • agents 只是注册,不等于模型能调用。 还要在 tools 里放 Agent(有些版本叫 Task),在 allowed_tools 里放行。

子 Agent 的结果会作为工具结果回到主循环。它只回最终文本,不回过程;如果最终文本太长,主上下文照样爆。给它加长度上限,或者要求结构化摘要。

一个“结果合同”的例子:

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

父循环拿到结果后,再加一道长度上限(比如 1,000 字符)。子 Agent 自己不一定会控制长度;合同 + 父循环的 cap 一起用。

AgentDefinition 的字段,按“派活”的顺序看:

字段作用建议
description主 Agent 决定“什么时候派”写场景,不写“研究助手”
prompt子 Agent 的角色/输出/边界写格式和“不确定怎么办”
tools子 Agent 能用的工具只给需要的;调研别给 Edit/Bash
model子 Agent 用的模型调研可用小模型;难题再升级
maxTurns子 Agent 的轮数上限3–8,别让它自己转圈
permissionMode子 Agent 的权限模式比主 Agent 更窄,不要放大

并行 fan-out:主 Agent 可以在同一轮里发多个 Agent 调用,SDK/CLI 会并行跑,结果逐条回来。适合“三份文档、三个问题、三个独立检索”;不适合“三个子 Agent 改同一份文件”——合并冲突会比串行更麻烦。

观测与停止:SDK 有 SubagentStart / SubagentStop Hook,可以看它什么时候开始/结束、返回多长;stop_task() 可以停掉一个跑偏的子 Agent。第 15 章的 Hook 机制同样适用于子 Agent。

深度与成本:子 Agent 也能再开子 Agent;不加限制会出现“Agent 套 Agent”。在 prompt 里写“不要再开子 Agent”,或者在代码里传 depth,超过 2 就拒绝。并行不降低总 token;它买的是时延,不是成本。

主 Agent 看到的仍然只有“最终文本 + 是否成功”;过程在子 Agent 自己的 transcript 里。要审计,去读它的 transcript,不要让它把过程塞回主上下文。


14.5 会话:resume / fork / checkpoint

四个能力,别混:

能力做什么不做什么
resume重新加载 transcript不回滚文件
fork_session从某个点开分支不复制文件
resume_session_at从指定消息恢复不撤销副作用
enable_file_checkpointing记录文件检查点不记录对话
rewind_files(user_message_id)回滚文件不回滚消息

resume 的成本约等于历史长度。20 轮、10k 字符的会话,resume 一次就是 10k 字符的输入;它不是“从零开始”,也不是“免费继续”。

还有一个常见的“resume 后重复工具调用”:

  • 会话最后一条是 tool_use,没有配对的 tool_result;
  • resume 从边界恢复,模型看到“我发了工具调用但没结果”,于是重发。

解决:写会话时保证 tool_use/tool_result 成对;压缩/恢复的切分点落在一轮结束。

一个典型的重复调用长这样:

text
[resume] 最后一条: tool_use read_file(id=t9)
[assistant] tool_use read_file(id=t9)     # 模型以为没执行,重发

它不是“模型记性差”,是历史里有一个没有结果的工具调用。恢复时要么补上结果,要么从更早的完整轮次恢复。


14.6 推荐组合:笔记 + 单功能 + 子 Agent + 压缩兜底

长任务不要只靠一个机制。我的默认组合是:

  1. Day 1 写 progress.md。 目标、约束、已完成、未完成,每次里程碑更新。
  2. 一次只做一个功能。 做完一个,更新笔记,再开下一个;不要一口气排十个。
  3. 调研用子 Agent。 过程长、结论短;结果加长度上限。
  4. 自动压缩兜底。 不指望它记得住;压缩前 PreCompact 存档。
  5. max_turns / max_budget_usd 刹车。 长任务最容易“跑飞”,预算和轮数要设。
  6. resume 前先验证世界。 测试、diff、副作用;不要假设文件没变。

和 tiny 的对照:

tinySDK
compact.py自动压缩 + PreCompact
notes.pyprogress.md + CLAUDE.md
subagent.pyagents + Agent 工具
session.pyresume / session_store / checkpoint

哪些是 SDK 自动的、哪些还要你做:

能力SDK 自动你还要做
压缩触发、摘要、替换历史PreCompact 存档、摘要约束
子 Agent独立循环、结果回填工具面、结果上限、权限
会话resume 加载验证世界、补配对、选切分点
检查点记录文件什么时候 rewind、回滚后怎么继续

少写的是实现,不是设计。你还是要想清楚:什么进笔记、什么进子 Agent、什么时候压。


14.7 一次长任务的复盘

回到开头那个第 24 轮的事故。如果重来,我会改四件事:

  1. 第一次读文件前就写 progress.md。 不是等压缩触发才写。
  2. 每完成一个功能就更新笔记。 不要等“全部做完”再总结;那时你已经忘了中间踩过什么。
  3. 调研交给子 Agent。 不要让主循环读十篇文档;主循环只拿结论。
  4. 压缩前存档,压缩后验证。 看目标、约束、下一步还在不在;不在就补。

这四件事都不难,难的是“记得做”。所以最好把它们写进代码/流程,而不是靠提示词提醒自己。


14.8 跑一遍

完整代码在 code/sdkagent/ch14_long_task.py。

bash
python3 -m pip install "claude-agent-sdk==0.2.168"
export ANTHROPIC_API_KEY="..."
cd code
python3 sdkagent/ch14_long_task.py

它会(设计上):

  • 在压缩前写 code/sdkagent/progress.md;
  • 用 researcher 子 Agent 做背景阅读;
  • 打印 compact_boundary、工具调用和最终结果;
  • 开启文件检查点,方便 rewind_files。

这次没有跑,原因是真的:

  1. 这个端点的上下文窗口是 1M token。 自动压缩要到接近上限才触发;用几十轮、几美元的预算,根本到不了 PreCompact。我没有伪造一个“压缩发生了”的输出。
  2. 跑一个短任务没有意义。 代码里的 task 已经改成存在的文件(README/core/tests),但跑它只能验证“短任务能不能读文件”,验证不了这一章的核心:压缩、子 Agent 分工、恢复。那需要更长的 fixture 和预算。

所以这一章保留为“机制 + 代码骨架”,不声称跑过。要诚实验证,需要补一个更长的 fixture(比如 storage.py + 多组测试),或者用 CLI 的 --autocompact 把阈值调低(最低 100k token,仍然不便宜)。

已经验证的是:SDK 0.2.168 / CLI 2.1.296 能加载这份代码的 PreCompact / AgentDefinition 配置(py_compile 和 import 通过);运行时行为没有验证。


14.9 下一章:控制执行

长任务能跑,但你怎么保证它“必须跑测试再结束”“不要把手机号念出来”“用户点停止就真停”?

第 15 章讲 Hook、流式、中断、结构化输出——Harness 的控制面。


本章代码 长任务示例:https://github.com/kevinkelin/claude-agent-sdk-book/blob/master/code/sdkagent/ch14_long_task.py 说明:本章代码未在本机运行(环境没装 SDK/凭据);自动压缩/子 Agent/checkpoint 的行为以你的版本和实测为准。