learn claude code S04 Subagent 详解笔记
s04 引入了一个新工具task——不是自己去探索,而是派一个"子代理"去探索,子代理做完回来只汇报结果。父代理 (Parent Agent) 子代理 (Subagent)| messages=[...已有上下文] | | messages=[] ← 空白开始 || | 派发 | || prompt="调研所有..." | | 执行 bash/read/edit || description="调
S04 Subagent 详解笔记
基于
s04_subagent.py源码逐行分析。
一、问题:复杂探索任务会把上下文撑爆
s03 的 Agent 有了 todo 工具,能自己做计划、追进度,多步任务不容易跑偏了。但还有一种场景 todo 解决不了:
你给 Agent 发布一个复杂问题:“帮我调研一下,这个仓库里所有 Python 文件用到了哪些第三方库?列一张表,标注每个库在哪些文件里用到、分别是什么用途。”
Agent 开始干活:
第 1 轮: bash: grep -r "import" --include="*.py" . ← 输出 2000 行
第 2 轮: bash: grep -r "from" --include="*.py" . ← 输出 3000 行
第 3 轮: read_file: requirements.txt ← 500 行
第 4 轮: read_file: setup.py ← 800 行
第 5 轮: bash: pip list ← 150 行
第 6 轮: read_file: src/main.py ← 600 行
第 7 轮: read_file: src/utils.py ← 400 行
...以此类推,messages 里堆积了数万行输出
到第 15 轮时,messages 上下文已经臃肿不堪。原始问题"调研所有 Python 文件的第三方库"沉在上下文的底部,被几万行 grep 输出和文件内容压着。模型的注意力已经难以聚焦原始任务,开始:
- 忘记自己为什么要读这些文件
- 重复读已经读过的文件
- 开始分析无关的代码
- 最终输出一个残缺的结果
核心矛盾:探索/调研类任务天然会产生大量中间数据——grep 输出、文件内容、ls 结果、pip list 等等。这些数据是完成任务必需的,但它们堆在 messages 里,又会淹没原始目标和已获得的关键发现。
这是上下文污染(context pollution)问题,不是 todo 能解决的——todo 帮你追踪"做到哪了",但没法帮你把已经产生的大量中间数据清理掉。
二、解决方案:子代理(Subagent)——全新的上下文,只返回摘要
s04 引入了一个新工具 task——不是自己去探索,而是派一个"子代理"去探索,子代理做完回来只汇报结果。
父代理 (Parent Agent) 子代理 (Subagent)
+--------------------------+ +--------------------------+
| messages=[...已有上下文] | | messages=[] ← 空白开始 |
| | 派发 | |
| tool: task | ----------> | while tool_use: |
| prompt="调研所有..." | | 执行 bash/read/edit |
| description="调研依赖" | | 结果追加到自己上下文 |
| | 返回 | |
| result = "总结报告..." | <---------- | 返回最后一段文本 |
+--------------------------+ +--------------------------+
| |
父代理上下文保持干净 子代理上下文用完就扔
核心洞见(原文注释):“Process isolation gives context isolation for free.” —— 进程隔离天然带来上下文隔离。
子代理的本质是什么?它是一个独立的 API 对话会话,从 messages=[] 开始,有自己的上下文空间。 子代理在探索过程中产生的所有中间数据都在它自己的 messages 里,不会污染父代理的上下文。子代理结束运行后,**只把最后一段文本(摘要)**返回给父代理。
结果:父代理的 messages 里只多了两样东西——一次 task 工具调用(很短的 prompt 文本),和一次工具结果(子代理的摘要报告,经过压缩)。几万行中间数据全部被丢弃,父代理的上下文保持干净。
三、和 s03 相比,多了什么?
| 组件 | s03 | s04 |
|---|---|---|
| 工具数量 | 5 (bash/read/write/edit/todo) | 5 (bash/read/write/edit/task),todo 被 task 替换 |
| 执行模型 | 单 Agent,所有步骤在一个会话里 | 父代理 + 可派发子代理,各跑各的会话 |
| 上下文管理 | todo 列表反复刷新到最新位置 | 子代理隔离中间数据,父代理上下文不膨胀 |
| 安全机制 | TodoManager 校验(最多 20 条等) | safe_path 路径沙箱 + 危险命令拦截 + 30 轮安全上限 |
| agent_loop | 多了 rounds_since_todo 和 nag | 多了 task 工具分发 → 触发 run_subagent |
s04 去掉 todo 不是因为 todo 不重要——恰恰相反,s04 的核心不是"代理如何规划任务",而是"代理如何隔离执行子任务"。把 todo 拿掉是为了让代码聚焦于展示 subagent 机制本身。真实产品中,todo 和 subagent 是共存的(如实际的 Claude Code)。
四、代码详解
4.1 全局基础设施
工作目录和客户端
WORKDIR = Path.cwd()
client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))
MODEL = os.environ["MODEL_ID"]
父代理和子代理共享同一个工作目录和同一个 Anthropic 客户端实例。这意味着它们操作同一个文件系统——子代理写文件,父代理能看到;父代理写文件,子代理也能看到。
System Prompt
SYSTEM = f"You are a coding agent at {WORKDIR}. Use the task tool to delegate exploration or subtasks."
SUBAGENT_SYSTEM = f"You are a coding subagent at {WORKDIR}. Complete the given task, then summarize your findings."
注意区别:
- 父代理的 prompt:告诉它用
task派发任务 - 子代理的 prompt:告诉它完成任务,然后总结发现
子代理的 prompt 里没有提到 task 工具——这是刻意的,因为子代理的 TOOLS 列表里本来就没有 task 工具(见下文)。
4.2 safe_path():工作区边界守卫
def safe_path(p: str) -> Path:
path = (WORKDIR / p).resolve()
if not path.is_relative_to(WORKDIR):
raise ValueError(f"Path escapes workspace: {p}")
return path
为什么需要这个函数? 子代理拥有 bash、read、write、edit 四个工具,可以操作文件系统。如果不限制路径范围,子代理可能(由于模型的幻觉或用户恶意 prompt)读取 /etc/passwd 或写入敏感目录。
safe_path 的逻辑:
- 把用户传入的路径拼到 WORKDIR 后面
- 调用
.resolve()解析所有..和符号链接,得到真实的绝对路径 - 检查真实路径是否在 WORKDIR 范围内
- 如果不是,抛异常
关键设计:resolve() 必须在 is_relative_to() 之前调用。如果只检查 (WORKDIR / p) 而不 resolve,攻击者可以传 ../../etc/passwd——拼接后是 /workspace/../../etc/passwd,但 is_relative_to 在某些 Python 版本中可能被绕过。先 resolve 后,/etc/passwd 显然不在 /workspace 下,一定会被拦截。
这建立了一个路径沙箱:无论父代理还是子代理,所有文件操作都被限制在工作目录内。
4.3 四个核心工具的实现
四个工具 run_bash、run_read、run_write、run_edit 是父代理和子代理共用的——同一份代码,同一套安全策略。
run_bash():带危险命令拦截的 shell 执行
def run_bash(command: str) -> str:
dangerous = ["rm -rf /", "sudo", "shutdown", "reboot", "> /dev/"]
if any(d in command for d in dangerous):
return "Error: Dangerous command blocked"
try:
r = subprocess.run(command, shell=True, cwd=WORKDIR,
capture_output=True, text=True, timeout=120)
out = (r.stdout + r.stderr).strip()
return out[:50000] if out else "(no output)"
except subprocess.TimeoutExpired:
return "Error: Timeout (120s)"
except (FileNotFoundError, OSError) as e:
return f"Error: {e}"
设计要点:
- 危险命令黑名单:
dangerous列表包含"rm -rf /"、"sudo"、"shutdown"、"reboot"、"> /dev/"。注意"> /dev/"——不是裸的> /dev,而是包含前缀" > ",防止误杀包含/dev/的正常路径字符串(如cat /dev/null) - 120 秒超时:防止子代理陷入死循环或启动长时间运行的命令
- 50000 字符截断:防止工具输出过大占用上下文。
[:50000]截断输出,让模型看到前面最重要的部分 - 异常捕获:
TimeoutExpired和FileNotFoundError/OSError都被 catch,返回错误信息而不是崩溃——整个系统的稳定性依赖于此
run_read():限制行数 + 输出截断
def run_read(path: str, limit: int = None) -> str:
try:
lines = safe_path(path).read_text().splitlines()
if limit and limit < len(lines):
lines = lines[:limit] + [f"... ({len(lines) - limit} more)"]
return "\n".join(lines)[:50000]
except Exception as e:
return f"Error: {e}"
safe_path(path)先做路径沙箱校验limit参数让模型可以选择先看前 N 行(对于大文件很实用)- 超过 limit 时加上提示
"... (N more)"——这很重要,告诉模型"文件不止这些,还有更多"。如果不加这个提示,模型可能误以为文件只有前 N 行,得出错误结论 - 最终输出仍做
[:50000]截断,二次保险
run_write():写文件 + 自动创建目录
def run_write(path: str, content: str) -> str:
try:
fp = safe_path(path)
fp.parent.mkdir(parents=True, exist_ok=True)
fp.write_text(content)
return f"Wrote {len(content)} bytes"
except Exception as e:
return f"Error: {e}"
fp.parent.mkdir(parents=True, exist_ok=True)——自动创建父目录,模型不用先bash: mkdir -p,减少了工具调用轮次- 返回值只告诉写了多少字节,不返回写入的内容(避免无意义的上下文重复)
run_edit():精确文本替换
def run_edit(path: str, old_text: str, new_text: str) -> str:
try:
fp = safe_path(path)
content = fp.read_text()
if old_text not in content:
return f"Error: Text not found in {path}"
fp.write_text(content.replace(old_text, new_text, 1))
return f"Edited {path}"
except Exception as e:
return f"Error: {e}"
- 先检查
old_text是否存在于文件中——找不到就报错,这是为了防止模型幻觉出不存在的代码片段然后试图替换 replace(old_text, new_text, 1)中的1表示只替换第一次出现,避免误伤多处相同文本- 这个简单的约束减少了"意外修改了不该改的代码"的风险
4.4 TOOL_HANDLERS:工具名到函数的映射
TOOL_HANDLERS = {
"bash": lambda **kw: run_bash(kw["command"]),
"read_file": lambda **kw: run_read(kw["path"], kw.get("limit")),
"write_file": lambda **kw: run_write(kw["path"], kw["content"]),
"edit_file": lambda **kw: run_edit(kw["path"], kw["old_text"], kw["new_text"]),
}
这里用 lambda 做了一层参数解包。Anthropic API 返回的 block.input 是一个包含所有参数的 dict(例如 {"command": "ls", ...}),但每个工具函数需要的参数不同。lambda 从 dict 中拆出对应字段传给真正的函数。
为什么不用普通函数?因为 lambda 可以直接嵌在 dict 值里,节省代码量。但本质上等价于:
def handle_bash(**kw):
return run_bash(kw["command"])
这个映射表被父代理和子代理共用——代码复用,安全策略统一。
4.5 CHILD_TOOLS:子代理的工具箱——核心洞察:没有 task
CHILD_TOOLS = [
{"name": "bash", ...},
{"name": "read_file", ...},
{"name": "write_file", ...},
{"name": "edit_file", ...},
]
子代理只有 bash、read、write、edit 四个基础工具。
关键设计:没有 task 工具。 为什么?
- 如果子代理也有
task,它就可以再派一个子代理,那个子代理又可以再派子代理……形成递归派发 - 递归派发会消耗大量的 API 调用次数和 token,而且模型很容易在这种递归结构中迷失
- 禁止递归是安全边界:子代理是最小的执行单元,只能做直接的文件操作和命令执行,不能再委托
这是 s04 架构中最重要的一条规则:父可以生子,子不能再生子。 层级深度被严格限制为 1 层。
4.6 run_subagent():子代理的核心引擎
这是 s04 的核心新增代码。
def run_subagent(prompt: str) -> str:
sub_messages = [{"role": "user", "content": prompt}] # 全新上下文
for _ in range(30): # 安全上限
response = client.messages.create(
model=MODEL, system=SUBAGENT_SYSTEM, messages=sub_messages,
tools=CHILD_TOOLS, max_tokens=8000,
)
sub_messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
break
results = []
for block in response.content:
if block.type == "tool_use":
handler = TOOL_HANDLERS.get(block.name)
output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)[:50000]})
sub_messages.append({"role": "user", "content": results})
# 只返回最后一段文本
return "".join(b.text for b in response.content if hasattr(b, "text")) or "(no summary)"
逐行分析:
第一行:sub_messages = [{"role": "user", "content": prompt}]
这就是上下文隔离的核心。 sub_messages 是一个全新创建的列表,从零开始,只有一条消息:父代理传过来的任务 prompt。父代理的完整对话历史、所有文件内容、所有 grep 输出——这些统统不在这里。
子代理是在一个信息真空中开始工作的,它知道的唯一信息就是父代理给它的任务描述。
第二行:for _ in range(30):
安全上限。子代理最多运行 30 轮(30 次 API 调用 + 工具执行)。如果 30 轮还没结束(模型还在调用工具),循环强制退出,返回最后的内容。
这个限制很重要:
- 防止子代理陷入死循环(模型反复调用相同工具,产生相同结果,永不停机)
- 限制 token 消耗(每次 API 调用都是钱)
- 30 轮够完成绝大多数调研/编辑任务,超过 30 轮说明任务本身有问题(太大,或模型卡住了)
注意:这不是 break 后的兜底——因为 for...else 语法不存在于这个循环中,30 轮结束后自然退出,然后执行最后的 return 语句。代码依赖 Python 的 for 循环正常结束来兜底。
stop_reason != "tool_use" 时的 break
API 返回的 stop_reason 有多种可能:
"tool_use":模型想调用工具,继续循环"end_turn":模型完成了思考,输出结束文本,退出循环"max_tokens":输出达到 max_tokens 上限,中间截断
当 stop_reason 不是 "tool_use" 时,说明子代理认为任务完成了(或被迫停止),此时退出循环。
工具执行块
results = []
for block in response.content:
if block.type == "tool_use":
handler = TOOL_HANDLERS.get(block.name)
output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
results.append({...})
sub_messages.append({"role": "user", "content": results})
这与父代理的工具分发逻辑一模一样。关键是 CHILD_TOOLS 里没有 task,所以 run_subagent 不会触发递归——子代理调不到 task 工具。
最后一行:return "...".join(b.text for ...) or "(no summary)"
子代理只返回文本摘要,不返回完整对话历史。 所有中间数据(grep 输出、文件内容、命令结果)都在 sub_messages 里,函数返回后 sub_messages 作为局部变量被垃圾回收。
如果有文本 block,拼接成字符串返回;如果是纯工具调用结束后没有文本(极端情况),返回 "(no summary)" 作为兜底。
4.7 PARENT_TOOLS:父代理的工具箱
PARENT_TOOLS = CHILD_TOOLS + [
{"name": "task", "description": "Spawn a subagent with fresh context. It shares the filesystem but not conversation history.",
"input_schema": {"type": "object", "properties": {
"prompt": {"type": "string"},
"description": {"type": "string", "description": "Short description of the task"}
}, "required": ["prompt"]}},
]
父代理 = 所有子代理的工具 + 一个额外的 task 工具。
task 工具有两个参数:
prompt(必填):给子代理的任务描述。这是子代理看到的唯一初始消息。description(选填):任务的简短描述,用于日志打印和调试。
注意 description 的 schema 里有 "description" 属性——这是给模型看的说明,告诉模型这个字段的用途。而 prompt 必填,description 选填,给模型留了自由度。
4.8 agent_loop():父代理的主循环
def agent_loop(messages: list):
while True:
response = client.messages.create(
model=MODEL, system=SYSTEM, messages=messages,
tools=PARENT_TOOLS, max_tokens=8000,
)
messages.append({"role": "assistant", "content": response.content})
if response.stop_reason != "tool_use":
return
results = []
for block in response.content:
if block.type == "tool_use":
if block.name == "task":
desc = block.input.get("description", "subtask")
prompt = block.input.get("prompt", "")
print(f"> task ({desc}): {prompt[:80]}")
output = run_subagent(prompt)
else:
handler = TOOL_HANDLERS.get(block.name)
output = handler(**block.input) if handler else f"Unknown tool: {block.name}"
print(f" {str(output)[:200]}")
results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)})
messages.append({"role": "user", "content": results})
与 s02/s03 的 agent_loop 相比,唯一的变化在工具分发:
if block.name == "task":
output = run_subagent(prompt) # 同步调用子代理
else:
# 其余工具走 TOOL_HANDLERS
run_subagent() 是同步阻塞调用。 父代理派发子代理后,父代理会等待子代理完全执行完毕(最多 30 轮),拿到返回的摘要,然后把摘要当成 task 工具的结果。
这意味着:
- 父代理在等待期间不做任何事(不能同时派发多个子代理)
- 一次
task调用 = 一个子代理的完整生命周期 - 子代理的结果像普通工具结果一样嵌入父代理的
messages
同步调用的设计选择:s04 选择了最简单的实现——同步、单任务、无并发。这是有意为之:先让学习者理解核心机制,并发管理是后面的优化。
4.9 主 REPL 循环
if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms04 >> \033[0m")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):
break
history.append({"role": "user", "content": query})
agent_loop(history)
response_content = history[-1]["content"]
if isinstance(response_content, list):
for block in response_content:
if hasattr(block, "text"):
print(block.text)
print()
- 青色提示符
\033[36m(和 s02/s03 保持一致风格的彩色终端提示) - 空输入退出(用户体验优化:按回车不发送空消息)
agent_loop(history)修改history后就返回,然后从history[-1]取出最后一条响应打印isinstance(response_content, list)检查很重要——Anthropic API 的content是一个 list of blocks,既有 text 也有 tool_use。只有 text block 才打印给用户看。
五、核心机制深度解析
5.1 上下文隔离:为什么 sub_messages = [] 是魔法
想象没有 subagent,父代理自己做探索:
messages = [
{"role": "user", "content": "调研所有 Python 文件的第三方库依赖"},
{"role": "assistant", "content": [ToolUse: bash grep import]},
{"role": "user", "content": [ToolResult: 2000 行 grep 输出]},
{"role": "assistant", "content": [ToolUse: bash grep from]},
{"role": "user", "content": [ToolResult: 3000 行 grep 输出]},
{"role": "assistant", "content": [ToolUse: read requirements.txt]},
{"role": "user", "content": [ToolResult: 500 行内容]},
... 又经过 10 轮,messages 有 20+ 条消息,几万行数据 ...
]
父代理的原始目标"调研第三方库"在第 1 条消息里。模型在做第 15 轮推理时,注意力机制主要聚焦在第 14、15 轮附近的内容。第 1 条消息虽然还在上下文中,但注意力的有效覆盖已经非常有限。
现在用 subagent:
父代理的 messages:
[
{"role": "user", "content": "调研所有 Python 文件的第三方库依赖"},
{"role": "assistant", "content": [ToolUse: task prompt="调研所有..." description="调研依赖"]},
{"role": "user", "content": [ToolResult: "总结报告: 共 8 个库: requests(3个文件), click(2个文件)..."]},
]
父代理只多了两轮——一次工具调用和一次结果。中间所有数据都发生在子代理的 sub_messages 里,而这个列表在 run_subagent() 返回后就被 Python GC 回收了。
子代理自己呢? 它的上下文也是干净的:
sub_messages = [
{"role": "user", "content": "调研所有 Python 文件的第三方库依赖"},
{"role": "assistant", "content": [ToolUse: bash grep import]},
{"role": "user", "content": [ToolResult: 2000 行 grep 输出]},
{"role": "assistant", "content": [ToolUse: bash grep from]},
{"role": "user", "content": [ToolResult: 3000 行 grep 输出]},
... 最多 30 轮 ...
]
子代理也是从零开始的,它的上下文里只有自己的探索过程,没有父代理的历史。这样它也能聚焦于当前任务。
这就是"Process isolation gives context isolation for free"的工程含义:创建一个新的函数调用栈(新的局部变量 sub_messages),天然地获得了上下文隔离。不需要手动清空任何东西,不需要复杂的上下文压缩——Python 的变量作用域替你做了所有事情。
5.2 工具过滤:为什么子代理不能有 task
比较两个工具列表:
CHILD_TOOLS = [bash, read_file, write_file, edit_file] # 4 个
PARENT_TOOLS = [bash, read_file, write_file, edit_file, task] # 5 个
run_subagent() 里使用 tools=CHILD_TOOLS,所以子代理根本不知道 task 工具的存在。Anthropic API 的 tool use 机制是:模型只能调用你传给它的 tools 列表中的工具。不给 task,模型就调不了 task。
禁止递归不是靠模型的自律,而是靠工具定义的物理限制。 这和 safe_path 的思路一致:不信任模型不会犯错,而是从系统层面直接隔绝犯错的可能性。
5.3 安全上限 range(30):为什么需要硬限制
子代理是一个"跑在沙箱里的自主进程"——bash、读、写、编辑,什么都行。问题是,LLM 有时候会陷入循环:
轮 1: bash grep "import requests" . → 0 结果
轮 2: bash find . -name "*.py" → 找到 50 个文件
轮 3: read file1.py → 没有 requests
轮 4: read file2.py → 没有 requests
轮 5: bash grep "from requests" . → 0 结果
轮 6: read file3.py → 没有 requests
... 无限循环 ...
如果没有 range(30),这个子代理会一直跑下去。30 是一个经验值——大多数调研任务在 10-15 轮内完成,30 轮给足了余量。超过 30 轮,子代理返回当前已有的内容,父代理可以决定重新派发或换一个策略。
5.4 摘要返回 vs 完整上下文返回
子代理返回的是 "...".join(b.text ...) ——只有最后一条响应中的文本部分。不是 sub_messages,不是完整的对话历史。
为什么只返回文本?
- 父代理需要的是结论,不是过程。调研依赖库,要的是最终的库列表,不是中间每一步的 grep 结果。
- 返回完整历史违背了子代理的设计初衷(隔离上下文)。
- 文本是自然压缩的——模型在最后一轮做总结时,已经自动把关键信息提炼出来了。
这就是"摘要模式":子代理看到一个 prompt,做一堆探索,最后写一段总结。父代理只读总结,不读日志。
六、完整流程走读(以具体任务为例)
假设用户在 s04 >> 提示符下输入:
“帮我找出这个仓库里所有 Python 文件中使用到的第三方库,列出每个库的名称、用途、以及哪些文件用到了它”
第 1 轮:父代理决定派发子代理
-
父代理收到用户消息,system prompt 告诉它 “Use the task tool to delegate exploration”
-
父代理意识到这是一个调研任务,会涉及大量文件读取和 grep,应该派子代理
-
父代理调用
task工具:{ "prompt": "探索当前目录下所有 Python 文件,找出所有第三方库的 import/from 语句..." } -
agent_loop中检测到block.name == "task",调用run_subagent(prompt)
第 2 步:子代理启动(run_subagent 内部)
-
sub_messages = [{"role": "user", "content": "探索当前目录下所有 Python 文件..."}] -
子代理第 1 次 API 调用,模型决定先 grep:
bash: grep -rn "^import\|^from" --include="*.py" .
-
工具执行,返回 grep 结果(比如 500 行),追加到
sub_messages
第 3 步:子代理分析结果
- 子代理第 2 次 API 调用,从 grep 结果中识别出第三方库:
requests、rich、pydantic、typer - 对于不确定的库,子代理继续探索:
bash: pip show requests→ 确认是第三方库bash: pip show os→ 确认是标准库,排除read_file: setup.py→ 查看项目声明的依赖
…子代理的 10-15 轮探索…
-
经过多轮探索,子代理完成调研,最后一次 API 调用不调工具,直接输出文本:
已完成调研,以下是发现的第三方库: 1. requests (3 个文件): main.py(client), utils.py(下载), api.py(HTTP请求) 2. rich (2 个文件): cli.py(表格输出), main.py(进度条) 3. pydantic (4 个文件): models/*.py (数据模型定义) 4. typer (1 个文件): cli.py(命令行入口) 共计 4 个第三方库。 -
子代理的
stop_reason == "end_turn",循环 break
run_subagent 返回
- 函数返回摘要文本(上面的 4 个库的总结)
sub_messages作为局部变量被 Python GC 回收
回到父代理
- 父代理收到
task工具的结果(子代理的摘要文本) - 父代理把结果展示给用户:
已完成调研,以下是发现的第三方库: ... - 父代理的上下文没有膨胀——只有一次 task 调用和一次结果
完整上下文对比
不用 subagent 的情况下,父代理的 messages 可能会有 20-30 条条目,包含几千行的 grep 输出和文件内容。用了 subagent,父代理的 messages 只有 3 条:
- 用户问题
- assistant: task 工具调用
- user: 摘要文本
七、设计洞察(为什么这样设计?)
7.1 上下文隔离不需要"压缩",只需要"不共享"
很多系统试图通过"上下文压缩"(总结旧消息、保留关键信息)来应对长对话。但压缩本身是个难题——什么该保留?什么该丢弃?
s04 的方案更彻底:不共享上下文,就是上下文隔离。 子代理有自己的 sub_messages,父代理有自己的 messages,互不相通。子代理完成后,只把结论(最后一轮文本)传回父代理。中间过程连压缩都不需要——直接丢弃。
7.2 安全性来自层级,而非列表
safe_path、危险命令黑名单、30 轮上限、禁止递归——这些安全机制不是零散地撒在代码里,而是层级分明:
| 层级 | 安全机制 | 防御什么 |
|---|---|---|
| 文件系统 | safe_path() + .resolve() |
路径穿越攻击 |
| 命令执行 | dangerous 黑名单 + 120s 超时 + 50000 截断 |
危险命令、资源耗尽 |
| 上下文 | sub_messages = [] + 局部变量 |
上下文污染、信息泄露 |
| 递归 | CHILD_TOOLS 不含 task |
递归派发、资源膨胀 |
| 运行时长 | range(30) 硬上限 |
死循环、API 费用 |
每一层独立运行,互不依赖。即使模型绕过了危险命令黑名单(比如用了未拦截的危险命令),safe_path 仍会限制它的文件操作范围。这就是纵深防御。
7.3 同步 vs 异步:简单的力量
run_subagent() 是同步阻塞的——父代理派发子代理后,必须等待它完成。这意味着:
- 不能同时派发多个子代理
- 父代理在子代理运行期间闲置
- 代码极其简单,没有并发、回调、协程
这是刻意的简化:先让学习者理解子代理的概念,再考虑并发优化。 就像学开车先学手动挡理解离合器原理。真实产品(如 Claude Code)中确实支持并发执行(通过异步或线程),但那是在理解基础模型之后的工程优化。
7.4 工具定义的两次分类
CHILD_TOOLS 和 PARENT_TOOLS 的关系不是"子集/超集"那么简单——它反映了一个权限模型:
PARENT: bash, read, write, edit, task → 可以派发
CHILD: bash, read, write, edit → 不能派发
这不是随意分配,而是遵循"最小权限原则":子代理只获得完成任务必需的工具。由于子代理的任务是"直接操作文件或执行命令",它不需要 task。反过来,如果子代理真的需要再次派发,那说明任务划分有问题——应该由父代理拆分后并行派发。
7.5 消息结构的统一性
注意这段代码在父代理和子代理里是完全一样的:
# 在 agent_loop 中
results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)})
messages.append({"role": "user", "content": results})
# 在 run_subagent 中
results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(output)[:50000]})
sub_messages.append({"role": "user", "content": results})
唯一的区别是子代理多了一个 [:50000] 截断(因为子代理的输出可能更不可控)。消息结构完全一致——都遵循 Anthropic API 的标准格式。这意味着父代理和子代理是同一套通讯协议的甲方和乙方,唯一的区别是对话起点不同(有历史 vs 空白)。
7.6 为何 s04 去掉了 todo 但保留了核心循环
s04 的 agent_loop 没有 todo 计数器、没有 nag 提醒、没有 TodoManager。这是有意的——s04 的概念焦点是"如何隔离执行",不是"如何自我规划"。加上 todo 不会让代码跑不了,但会在教学上分散注意力。
s04 保留了 s02 的核心循环骨架(while loop + 工具分发 + 结果回填),只在工具分发上加了 if block.name == "task" 的分支。这展示了 Harness 系列的教学哲学:每一课只改一个核心机制,其余部分保持简单。
八、总结
| 维度 | 核心要点 |
|---|---|
| 解决的问题 | 复杂探索任务产生大量中间数据,污染父代理上下文 |
| 核心方案 | 子代理 = 独立的 API 会话,从空白 context 开始,只返回摘要 |
| 关键代码行 | sub_messages = [{"role": "user", "content": prompt}] |
| 安全机制 | 路径沙箱、危险命令拦截、30 轮上限、禁止递归、输出截断 |
| 设计哲学 | 进程隔离天然带来上下文隔离;纵深防御;最小权限;教学简化 |
子代理的本质不是什么复杂的架构模式——它是一个拥有自己上下文的独立函数调用。Python 的变量作用域天然提供了隔离,你不需要构建复杂的"上下文管理器"或"对话树"。
下一课(s05)将引入 compact 机制——当单个对话自身太长(即使没有子代理)时,如何压缩历史而保留关键信息。子代理解决的是"把脏活交给别人",compact 解决的是"自己的记忆太多怎么办"。
更多推荐



所有评论(0)