第4课：子Agent —— 把大任务拆给"小弟"去干

你将学到什么

• 什么是子Agent（Subagent），为什么需要"上下文隔离"
• 父Agent和子Agent之间的关系：共享文件系统，不共享对话历史
• 如何实现 run_subagent() 函数来生成子Agent
• 子Agent的工具集为什么要比父Agent少（禁止递归生成）
• task 工具的设计：对父Agent来说，子Agent就是一个普通工具调用

核心概念：项目经理和实习生

想象一下你是一个项目经理（父Agent），手头有个大活儿要干。你不需要所有事情都亲力亲为，你可以派一个实习生（子Agent）去做一些独立的子任务。

比如你在做一个Web项目，需要调研"用哪个数据库比较好"。你会怎么做？

1. 派实习生去调研 —— "你去查查 SQLite、PostgreSQL、MongoDB 各自的优缺点"
2. 实习生自己去折腾 —— 他自己去查资料、写测试代码、试错、对比
3. 实习生交报告 —— "经理，我调研完了，推荐用 PostgreSQL，原因是..."
4. 你只看报告 —— 你不需要知道他中间试了多少次、走了多少弯路

这就是子Agent的核心模式：

• 全新的上下文（messages=[]）—— 实习生不需要知道你之前和客户聊了什么
• 共享文件系统 —— 实习生和你在同一个办公室，能访问同样的文件
• 只返回总结 —— 实习生的中间过程全部丢弃，只有最终报告传回来
• 父Agent上下文保持干净 —— 你的脑子里不会被实习生的调研细节塞满

关键洞察："进程隔离天然带来了上下文隔离。" 子Agent有自己独立的 messages 列表，做完就扔，父Agent的对话历史不会膨胀。

ASCII 流程图

用户输入："帮我分析这个项目的代码质量"
   |
   v
+----------------------------+
| 父Agent（项目经理）         |
| messages = [用户的提问...]  |
| 工具：bash/read/write/     |
|       edit/task            |
+----------------------------+
   |
   | 调用 task 工具：
   | "去检查所有 Python 文件的代码风格"
   |
   v
+----------------------------+          +----------------------------+
| 父Agent 等待结果...        |          | 子Agent（实习生）           |
|                            |          | messages = []   <-- 全新！  |
|                            |  派活儿   | 工具：bash/read/write/edit |
|                            | -------> | （注意：没有 task 工具！）   |
|                            |          |                            |
|                            |          | 第1步：ls *.py             |
|                            |          | 第2步：读取每个文件         |
|                            |          | 第3步：运行 pylint         |
|                            |          | 第4步：整理结果             |
|                            |          | 第5步：返回总结报告         |
|                            |  报告    |                            |
|  tool_result = "报告..."   | <------- | "检查了5个文件，发现..."    |
+----------------------------+          +----------------------------+
   |                                              |
   |                                      子Agent上下文丢弃
   v                                     （中间过程全没了）
+----------------------------+
| 父Agent 继续工作            |
| 上下文里只多了一条           |
| tool_result（总结报告）      |
| 上下文依然干净！             |
+----------------------------+
   |
   v
 返回给用户：最终分析报告

和上一课的对比

上一课（第3课）我们学了 TodoWrite，让 Agent 自己管理任务进度。这一课的核心变化是：

简单说：上一课的 Agent 是个"事必躬亲的劳模"，这一课变成了"会分配任务的项目经理"。

完整代码

python复制
#!/usr/bin/env python3
"""s04_subagent.py - 子Agent（Subagent）

核心思想：生成一个"子Agent"来做子任务。
子Agent有全新的上下文（messages=[]），
跟父Agent共享文件系统但不共享对话历史。
子Agent做完了只返回一个总结给父Agent。

关键洞察："进程隔离天然带来上下文隔离。"
"""

import os          # 读取环境变量
import subprocess  # 执行 shell 命令
from pathlib import Path  # 路径操作

from anthropic import Anthropic          # Anthropic 官方 SDK
from dotenv import load_dotenv           # 加载 .env 环境变量

# ============================================================
# 第一步：初始化配置（和前几课一样）
# ============================================================

# 加载 .env 文件，override=True 让 .env 的值覆盖系统环境变量
load_dotenv(override=True)

# 如果用了代理地址，移除可能冲突的 AUTH_TOKEN
if os.getenv("ANTHROPIC_BASE_URL"):
    os.environ.pop("ANTHROPIC_AUTH_TOKEN", None)

# 当前工作目录 —— Agent 只能在这个目录下操作文件
WORKDIR = Path.cwd()

# 创建 Anthropic 客户端
client = Anthropic(base_url=os.getenv("ANTHROPIC_BASE_URL"))

# 从环境变量读取模型名称
MODEL = os.environ["MODEL_ID"]


# ============================================================
# 第二步：系统提示词 —— 父Agent和子Agent各一套
# ============================================================

# 父Agent的提示词：告诉它可以用 task 工具来"分派任务"
# 就像告诉项目经理："你有实习生可以用"
SYSTEM = f"You are a coding agent at {WORKDIR}. Use the task tool to delegate exploration or subtasks."

# 子Agent的提示词：告诉它完成任务后要总结发现
# 就像告诉实习生："做完了写个报告给我"
SUBAGENT_SYSTEM = f"You are a coding subagent at {WORKDIR}. Complete the given task, then summarize your findings."


# ============================================================
# 第三步：工具实现 —— 父Agent和子Agent共用同一套
# ============================================================
# 父子Agent共享文件系统，所以路径安全检查和工具实现完全相同。
# 这就像项目经理和实习生在同一个办公室工作，操作同样的文件。

def safe_path(p: str) -> Path:
    """
    路径安全检查：确保不会操作工作目录之外的文件。
    父Agent和子Agent都要遵守同样的安全规则。
    """
    path = (WORKDIR / p).resolve()
    if not path.is_relative_to(WORKDIR):
        raise ValueError(f"Path escapes workspace: {p}")
    return path

def run_bash(command: str) -> str:
    """
    执行 shell 命令。
    不管是父Agent还是子Agent调用，安全限制都一样。
    """
    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)"

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}"

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}"

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}"


# ============================================================
# 第四步：工具处理器映射
# ============================================================

# 工具名 -> 执行函数 的映射
# 注意：这里只有基础工具，不包含 task 工具
# task 工具在父Agent的 agent_loop 里特殊处理（因为它要启动子Agent）
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"]),
}


# ============================================================
# 第五步：定义工具集 —— 子Agent和父Agent的工具不同！
# ============================================================

# ★ 子Agent的工具集：只有基础工具，没有 task ★
# 这是一个关键设计决策：子Agent不能再派活儿给"孙Agent"
# 如果允许递归派活，会产生无限嵌套的Agent树，失去控制
CHILD_TOOLS = [
    {"name": "bash", "description": "Run a shell command.",
     "input_schema": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}},
    {"name": "read_file", "description": "Read file contents.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "limit": {"type": "integer"}}, "required": ["path"]}},
    {"name": "write_file", "description": "Write content to file.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "content": {"type": "string"}}, "required": ["path", "content"]}},
    {"name": "edit_file", "description": "Replace exact text in file.",
     "input_schema": {"type": "object", "properties": {"path": {"type": "string"}, "old_text": {"type": "string"}, "new_text": {"type": "string"}}, "required": ["path", "old_text", "new_text"]}},
]

# ★ 父Agent的工具集：基础工具 + task 工具 ★
# task 工具就是父Agent的"派活儿"能力
# 对父Agent来说，调用 task 和调用 bash 没有区别 —— 都是工具调用
# 但在底层，task 会启动一个全新的子Agent
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"},  # 派给子Agent的任务描述
             "description": {               # 简短描述（方便日志显示）
                 "type": "string",
                 "description": "Short description of the task"
             }
         },
         "required": ["prompt"]
     }},
]


# ============================================================
# 第六步：run_subagent() —— 本课的核心！生成子Agent
# ============================================================

def run_subagent(prompt: str) -> str:
    """
    生成并运行一个子Agent。

    这是整个文件最核心的函数。理解了它，就理解了子Agent的全部。

    工作流程：
    1. 创建全新的 messages 列表（只包含派给它的任务）
    2. 进入 Agent Loop（和父Agent的循环结构一模一样）
    3. 子Agent自己调工具、自己处理结果、自己迭代
    4. 循环结束后，只把最后一条文本消息返回给父Agent

    参数：
        prompt: 派给子Agent的任务描述

    返回：
        子Agent的总结文本（中间过程全部丢弃）
    """
    # ★ 关键！全新的 messages 列表 ★
    # 子Agent看不到父Agent之前的任何对话
    # 它只知道"有人给了我一个任务"
    sub_messages = [{"role": "user", "content": prompt}]  # 全新上下文

    # 子Agent自己的 Agent Loop
    # 最多循环 30 次（安全上限，防止无限循环）
    for _ in range(30):
        # 调用 Claude API —— 注意这里用的是：
        # - SUBAGENT_SYSTEM：子Agent专用的系统提示词
        # - sub_messages：子Agent自己的对话历史（和父Agent无关）
        # - CHILD_TOOLS：子Agent的工具集（没有 task 工具！）
        response = client.messages.create(
            model=MODEL, system=SUBAGENT_SYSTEM, messages=sub_messages,
            tools=CHILD_TOOLS, max_tokens=8000,
        )

        # 把 AI 回复加入子Agent自己的对话历史
        sub_messages.append({"role": "assistant", "content": response.content})

        # 如果不再需要调用工具 → 子Agent完成了任务，退出循环
        if response.stop_reason != "tool_use":
            break

        # 处理子Agent的工具调用（和父Agent的处理方式一模一样）
        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}"
                # 子Agent的工具结果只进入子Agent的 messages
                # 父Agent完全看不到这些中间过程
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(output)[:50000]
                })

        # 把工具结果拼回子Agent的对话历史
        sub_messages.append({"role": "user", "content": results})

    # ★ 关键！只返回最终的文本总结 ★
    # 子Agent可能折腾了 10 轮、调了 20 次工具，但父Agent只拿到这一句话
    # sub_messages 里的所有中间过程？全部丢弃，永远不会进入父Agent的上下文
    return "".join(b.text for b in response.content if hasattr(b, "text")) or "(no summary)"


# ============================================================
# 第七步：父Agent的 Agent Loop
# ============================================================

def agent_loop(messages: list):
    """
    父Agent的主循环。

    和前几课的 Agent Loop 结构几乎一样，唯一的区别是：
    遇到 task 工具时，不是直接执行，而是调用 run_subagent() 启动子Agent。

    对父Agent来说，task 就是一次普通的工具调用：
    - 发出工具调用请求（tool_use）
    - 等待工具返回结果（tool_result）
    - 根据结果继续决策

    它完全不知道（也不需要知道）子Agent在底下折腾了多少轮。
    """
    while True:
        # 调用 Claude API（父Agent用 PARENT_TOOLS，包含 task 工具）
        response = client.messages.create(
            model=MODEL, system=SYSTEM, messages=messages,
            tools=PARENT_TOOLS, max_tokens=8000,
        )

        # 把 AI 回复加入父Agent的对话历史
        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":
                    # ★ task 工具：启动子Agent ★
                    # 从父Agent的视角，这就是一次普通的工具调用
                    # 但在底层，它启动了一个全新的Agent来执行任务
                    desc = block.input.get("description", "subtask")
                    print(f"> task ({desc}): {block.input['prompt'][:80]}")
                    # 调用 run_subagent()，等它做完，拿到总结
                    output = run_subagent(block.input["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]}")

                # 把工具结果拼成 tool_result 格式
                results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": str(output)
                })

        # ★ 注意：父Agent的 messages 里只会添加子Agent的总结 ★
        # 子Agent内部的 10 轮对话、20 次工具调用？
        # 全都不在这里。父Agent的上下文保持干净。
        messages.append({"role": "user", "content": results})


# ============================================================
# 第八步：交互式主循环
# ============================================================

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循环
        agent_loop(history)
        # 打印 AI 的最终回答
        # Agent Loop 结束后，history 最后一条是 assistant 的回复
        response_content = history[-1]["content"]
        if isinstance(response_content, list):
            for block in response_content:
                if hasattr(block, "text"):
                    print(block.text)
        print()

代码逐行拆解

模块一：两套系统提示词

python复制
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."

父Agent和子Agent有不同的"人设"：

• 父Agent：被告知"你可以用 task 工具分派任务"。它是项目经理，知道自己有小弟可用。
• 子Agent：被告知"完成任务后总结你的发现"。它是实习生，只管埋头干活，做完交报告。

为什么要分开？因为角色不同，行为模式也不同。你不会告诉实习生"你也可以派实习生"（那就乱套了）。

模块二：两套工具集

python复制
# 子Agent工具：bash / read_file / write_file / edit_file
CHILD_TOOLS = [...]

# 父Agent工具：子Agent的所有工具 + task
PARENT_TOOLS = CHILD_TOOLS + [
    {"name": "task", "description": "Spawn a subagent with fresh context...", ...}
]

这里有一个精心的设计：子Agent没有 task 工具。

为什么？想象一下如果子Agent也能派活儿：

text复制
父Agent → 派子Agent → 子Agent又派孙Agent → 孙Agent又派曾孙Agent → ...

这就是无限递归，会失控。所以我们在工具层面就切断了这条链路：子Agent只能用基础工具干活，不能再分派。

模块三：run_subagent() —— 整个文件的核心

这个函数做了三件关键的事：

第一件：创建全新上下文

python复制
sub_messages = [{"role": "user", "content": prompt}]  # 全新上下文

就这一行，是子Agent和父Agent最本质的区别。sub_messages 是一个全新的列表，里面只有派给子Agent的任务描述。父Agent之前的所有对话？子Agent完全看不到。

这就像你跟实习生说"你去查一下 PostgreSQL 的性能数据"，实习生不需要知道你之前和客户聊了两个小时的项目背景。

第二件：子Agent自己的 Agent Loop

python复制
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
    # ... 处理工具调用 ...
    sub_messages.append({"role": "user", "content": results})

子Agent的循环结构和父Agent一模一样：调 API → 处理工具 → 拼回结果 → 再调 API...

区别只有三点：

1. 用 SUBAGENT_SYSTEM（子Agent的系统提示词）
2. 用 sub_messages（子Agent自己的对话历史）
3. 用 CHILD_TOOLS（没有 task 工具）

第三件：只返回最终总结

python复制
return "".join(b.text for b in response.content if hasattr(b, "text")) or "(no summary)"

子Agent可能折腾了 10 轮，调了 20 次 bash，读了 15 个文件。但这些统统不重要。父Agent只拿到最后一句总结。

sub_messages 这个列表？函数返回后就被垃圾回收了。子Agent的所有中间过程，永远不会进入父Agent的上下文。

这就是"上下文隔离"的精髓。

模块四：父Agent的 agent_loop() 中对 task 的处理

python复制
if block.name == "task":
    desc = block.input.get("description", "subtask")
    print(f"> task ({desc}): {block.input['prompt'][:80]}")
    output = run_subagent(block.input["prompt"])
else:
    handler = TOOL_HANDLERS.get(block.name)
    output = handler(**block.input) if handler else f"Unknown tool: {block.name}"

对父Agent来说，task 和 bash 没什么两样 —— 都是"调用一个函数，拿到一个字符串结果"。

• 调 bash("ls -la") → 得到文件列表字符串
• 调 task("分析代码质量") → 得到分析报告字符串

父Agent完全不知道（也不关心）task 背后启动了一个完整的Agent循环。它只知道"我派了个活儿出去，结果回来了"。

运行效果

示例：让 Agent 分析一个项目的结构和代码质量

text复制
s04 >> 帮我分析当前目录下的代码，告诉我项目结构和代码质量

> task (分析项目结构): 列出当前目录下的所有文件和目录结构，分析项目的组织方式
  [子Agent在后台工作：执行 ls、tree 等命令，读取关键文件]
  该项目是一个 Python 教程项目，包含 agents/ 和 tutorials/ 两个主目录...

> task (代码质量检查): 检查所有 Python 文件的代码风格、潜在bug和可改进的地方
  [子Agent在后台工作：逐个读取.py文件，运行 pylint 等工具]
  共检查了 5 个 Python 文件。整体代码质量良好，有几处建议：...

项目分析完成！以下是总结：

1. **项目结构**：这是一个循序渐进的 AI Agent 教程项目...
2. **代码质量**：整体良好，代码风格统一...

注意关键行为：

1. 父Agent收到任务后，决定派两个子Agent（一个分析结构，一个检查质量）
2. 每个子Agent独立工作 —— 有自己的对话历史，自己调工具
3. 子Agent只返回总结 —— 父Agent看到的是精炼的报告，不是原始的 ls 输出
4. 父Agent汇总后给用户 —— 把两份子报告整合成最终答案

上下文隔离的效果

假设子Agent在分析过程中执行了以下操作：

text复制
子Agent内部（父Agent看不到这些）：
  1. bash("find . -name '*.py'")           → 找到 5 个文件
  2. read_file("agents/s01_basic.py")       → 读取 200 行代码
  3. read_file("agents/s02_tools.py")       → 读取 300 行代码
  4. read_file("agents/s03_todo_write.py")  → 读取 350 行代码
  5. bash("python -m pylint agents/")       → 运行代码检查
  6. 生成总结报告

如果没有子Agent，这 6 步的所有输出（可能有上千行）都会塞进父Agent的上下文里。有了子Agent，父Agent的 messages 里只多了一条：

text复制
tool_result: "共检查了 5 个 Python 文件。整体代码质量良好..."

这就是上下文隔离的价值 —— 父Agent的上下文保持干净，不会被子任务的细节淹没。

关键收获

1. 子Agent就是一个普通工具调用 —— 对父Agent来说，task("做个调研") 和 bash("ls") 没有本质区别。都是调用工具、拿到结果。子Agent的复杂性被完全封装在 run_subagent() 函数里。

1. 上下文隔离是核心价值 —— 子Agent的 messages=[] 是全新的，做完就丢。父Agent永远不会被子任务的中间过程污染。这在处理复杂任务时极其重要，否则上下文窗口很快就会被塞满。

1. 共享文件系统是协作桥梁 —— 父子Agent虽然不共享对话历史，但共享文件系统。子Agent可以创建文件，父Agent可以读取。这是它们之间除了"总结文本"之外的第二条通信通道。

1. 禁止递归生成是必要的约束 —— 子Agent没有 task 工具，不能再派活儿。这防止了无限嵌套的Agent树。好的架构不是给出无限自由，而是在关键处设置合理的边界。

1. 子Agent循环和父Agent循环结构完全一样 —— 都是"调API → 处理工具 → 拼回结果 → 再调API"的 while 循环。唯一的区别是上下文（messages）、系统提示词（system）和工具集（tools）不同。这体现了优雅的设计：同一套模式，通过不同的配置实现不同的行为。

下一课预告

现在我们的Agent已经会分派子任务了，但还有一个问题：子Agent一次只能处理一个任务，如果有多个独立子任务，效率太低了。

第5课：并行子Agent —— 让多个子Agent同时工作。就像项目经理同时派三个实习生分别去做三件不相关的事，而不是等第一个做完了再派第二个。这将大幅提升复杂任务的处理速度。