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 的逻辑:

  1. 把用户传入的路径拼到 WORKDIR 后面
  2. 调用 .resolve() 解析所有 .. 和符号链接,得到真实的绝对路径
  3. 检查真实路径是否在 WORKDIR 范围内
  4. 如果不是,抛异常

关键设计resolve() 必须在 is_relative_to() 之前调用。如果只检查 (WORKDIR / p) 而不 resolve,攻击者可以传 ../../etc/passwd——拼接后是 /workspace/../../etc/passwd,但 is_relative_to 在某些 Python 版本中可能被绕过。先 resolve 后,/etc/passwd 显然不在 /workspace 下,一定会被拦截。

这建立了一个路径沙箱:无论父代理还是子代理,所有文件操作都被限制在工作目录内。

4.3 四个核心工具的实现

四个工具 run_bashrun_readrun_writerun_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] 截断输出,让模型看到前面最重要的部分
  • 异常捕获TimeoutExpiredFileNotFoundError/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 轮:父代理决定派发子代理

  1. 父代理收到用户消息,system prompt 告诉它 “Use the task tool to delegate exploration”

  2. 父代理意识到这是一个调研任务,会涉及大量文件读取和 grep,应该派子代理

  3. 父代理调用 task 工具:

    {
      "prompt": "探索当前目录下所有 Python 文件,找出所有第三方库的 import/from 语句..."
    }
    
  4. agent_loop 中检测到 block.name == "task",调用 run_subagent(prompt)

第 2 步:子代理启动(run_subagent 内部)

  1. sub_messages = [{"role": "user", "content": "探索当前目录下所有 Python 文件..."}]

  2. 子代理第 1 次 API 调用,模型决定先 grep:

    • bash: grep -rn "^import\|^from" --include="*.py" .
  3. 工具执行,返回 grep 结果(比如 500 行),追加到 sub_messages

第 3 步:子代理分析结果

  1. 子代理第 2 次 API 调用,从 grep 结果中识别出第三方库:requestsrichpydantictyper
  2. 对于不确定的库,子代理继续探索:
    • bash: pip show requests → 确认是第三方库
    • bash: pip show os → 确认是标准库,排除
    • read_file: setup.py → 查看项目声明的依赖

…子代理的 10-15 轮探索…

  1. 经过多轮探索,子代理完成调研,最后一次 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 个第三方库。
    
  2. 子代理的 stop_reason == "end_turn",循环 break

run_subagent 返回

  1. 函数返回摘要文本(上面的 4 个库的总结)
  2. sub_messages 作为局部变量被 Python GC 回收

回到父代理

  1. 父代理收到 task 工具的结果(子代理的摘要文本)
  2. 父代理把结果展示给用户:
    已完成调研,以下是发现的第三方库:
    ...
    
  3. 父代理的上下文没有膨胀——只有一次 task 调用和一次结果

完整上下文对比

不用 subagent 的情况下,父代理的 messages 可能会有 20-30 条条目,包含几千行的 grep 输出和文件内容。用了 subagent,父代理的 messages 只有 3 条:

  1. 用户问题
  2. assistant: task 工具调用
  3. 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_TOOLSPARENT_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 解决的是"自己的记忆太多怎么办"。

Logo

汇聚全球AI编程工具,助力开发者即刻编程。

更多推荐