v3: 子代理机制

约 450 行代码。新增 1 个工具。分而治之。

v2 添加了规划功能。但对于"探索代码库然后重构认证模块"这样的大型任务，单个代理会遇到上下文限制。探索会将 20 个文件转储到历史记录中，导致重构时失去焦点。

v3 添加了 Task 工具：用于生成具有独立上下文的子代理。

问题所在

单代理的上下文污染问题：

主代理历史记录：
  [探索中...] cat file1.py -> 500 行
  [探索中...] cat file2.py -> 300 行
  ... 还有 15 个文件 ...
  [现在开始重构...] "等等，file1 里有什么来着？"

解决方案：将探索任务委托给子代理：

主代理历史记录：
  [任务：探索代码库]
    -> 子代理探索 20 个文件
    -> 返回："认证模块在 src/auth/，数据库在 src/models/"
  [现在可以在干净的上下文中进行重构了]

代理类型注册表

每种代理类型定义了其能力：

AGENT_TYPES = {
    "explore": {
        "description": "用于搜索和分析的只读代理",
        "tools": ["bash", "read_file"],  # 无写入权限
        "prompt": "搜索和分析，永远不要修改文件。返回简洁的摘要。",
    },
    "code": {
        "description": "用于实现功能的完整代理",
        "tools": "*",  # 所有工具
        "prompt": "高效地实现所请求的更改。",
    },
    "plan": {
        "description": "用于设计工作的规划代理",
        "tools": ["bash", "read_file"],  # 只读
        "prompt": "分析并输出编号的实施计划。不要修改文件。",
    },
}

Task 工具

{
    "name": "Task",
    "description": "生成子代理处理聚焦的子任务",
    "input_schema": {
        "description": "简短任务名称（3-5 个词）",
        "prompt": "详细指令",
        "agent_type": "explore | code | plan"
    }
}

主代理调用 Task → 子代理运行 → 返回摘要。

子代理执行

Task 工具的核心：

def run_task(description, prompt, agent_type):
    config = AGENT_TYPES[agent_type]

    # 1. 代理特定的系统提示
    sub_system = f"You are a {agent_type} subagent.\n{config['prompt']}"

    # 2. 过滤后的工具
    sub_tools = get_tools_for_agent(agent_type)

    # 3. 独立的历史记录（关键：没有父上下文）
    sub_messages = [{"role": "user", "content": prompt}]

    # 4. 相同的查询循环
    while True:
        response = client.messages.create(
            model=MODEL, system=sub_system,
            messages=sub_messages, tools=sub_tools
        )
        if response.stop_reason != "tool_use":
            break
        # 执行工具，追加结果...

    # 5. 只返回最终文本
    return extract_final_text(response)

核心概念：

工具过滤

def get_tools_for_agent(agent_type):
    allowed = AGENT_TYPES[agent_type]["tools"]
    if allowed == "*":
        return BASE_TOOLS  # 没有 Task 工具（演示中防止无限递归）
    return [t for t in BASE_TOOLS if t["name"] in allowed]

• explore：仅限 bash + read_file
• code：所有工具
• plan：仅限 bash + read_file

子代理不获得 Task 工具（在演示中防止无限递归）。

进度显示

子代理的输出不会污染主聊天：

你：探索代码库
> Task: 探索代码库
  [explore] 探索代码库 ... 5 个工具, 3.2 秒
  [explore] 探索代码库 - 完成（8 个工具，5.1 秒）

这是我找到的：...

实时进度显示，最终输出干净。

典型流程

用户："将认证模块重构为使用 JWT"

主代理：
  1. Task(explore)："查找所有与认证相关的文件"
     -> 子代理读取 10 个文件
     -> 返回："认证在 src/auth/login.py，会话在..."

  2. Task(plan)："设计 JWT 迁移策略"
     -> 子代理分析结构
     -> 返回："1. 添加 jwt 库 2. 创建令牌工具..."

  3. Task(code)："实现 JWT 令牌"
     -> 子代理编写代码
     -> 返回："已创建 jwt_utils.py，已更新 login.py"

  4. 总结所做的更改

每个子代理都有干净的上下文。主代理保持专注。

对比

模式

复杂任务
  └─ 主代理（协调者）
       ├─ 子代理 A（探索） -> 摘要
       ├─ 子代理 B（规划） -> 计划
       └─ 子代理 C（编码） -> 结果

相同的代理循环，不同的上下文。这就是全部诀窍。

学习笔记

按此顺序阅读源代码

打开 v3_subagent.py，先阅读以下部分：

1. AGENT_TYPES
2. get_agent_descriptions
3. TASK_TOOL
4. get_tools_for_agent
5. run_task
6. execute_tool
7. agent_loop

如果将 v3 视为 v2 加上一个新功能，最容易理解：

Task(description, prompt, agent_type)

其他所有内容都是围绕这个工具的支持代码。

代理注册表是控制面板

源代码在 AGENT_TYPES 中定义子代理行为：

AGENT_TYPES = {
    "explore": {
        "tools": ["bash", "read_file"],
        "prompt": "搜索和分析，但永远不要修改文件...",
    },
    "code": {
        "tools": "*",
        "prompt": "高效地实现所请求的更改。",
    },
    "plan": {
        "tools": ["bash", "read_file"],
        "prompt": "分析并输出编号的实施计划...",
    },
}

这个注册表控制三个方面：

这是核心思想：子代理不是特殊的魔法对象。它们是相同的循环，只是有不同的指令、不同的工具和不同的上下文。

Task 是启动另一个循环的工具

主代理将 Task 视为普通工具：

TASK_TOOL = {
    "name": "Task",
    "description": "生成子代理处理聚焦的子任务。",
    ...
}

当模型调用 Task 时，execute_tool 将其分派到：

return run_task(args["description"], args["prompt"], args["agent_type"])

这意味着 v3 仍然是相同的代理循环。唯一的变化是其中一个工具启动了一个嵌套循环。

主代理循环
  -> Task 工具
      -> 子代理循环
          -> 工具
          -> 最终摘要
  -> 工具结果返回给主代理

上下文隔离发生在一行代码中

run_task 中最重要的一行是：

sub_messages = [{"role": "user", "content": prompt}]

子代理不接收父代理的完整对话。它从全新的历史记录开始，只有任务提示。

这给你带来：

• 更干净的父上下文
• 更便宜的主对话
• 更少的意外残留
• 探索和实现之间的自然边界

父代理只接收返回的摘要，而不是每个中间文件读取。

工具过滤是安全和聚焦的保障

函数 get_tools_for_agent 决定每个子代理可以使用什么：

allowed = AGENT_TYPES[agent_type]["tools"]
if allowed == "*":
    return BASE_TOOLS
return [t for t in BASE_TOOLS if t["name"] in allowed]

这就是为什么 explore 和 plan 是只读的：

explore -> bash + read_file
plan    -> bash + read_file
code    -> 所有基础工具

工具过滤做两件事：

1. 降低风险。研究代理无法编辑文件。
2. 改善行为。规划代理被推向分析，而不是行动。

为什么子代理不获得 Task 工具

在这个演示中，子代理接收 BASE_TOOLS，而不是 ALL_TOOLS。

这意味着子代理不接收 Task 工具。这避免了学习版本中的递归子代 理生成。

生产系统可能允许更深的树，但初学者版本将树保持在一层深度：

主代理
 ├── explore 子代理
 ├── plan 子代理
 └── code 子代理

父代理实际看到什么

在 run_task 内部，子代理可以进行许多工具调用。但在最后：

for block in response.content:
    if hasattr(block, "text"):
        return block.text

父代理将子代理的最终文本作为单个工具结果接收。

这就是 v3 的全部价值：

许多子代理观察 -> 一个父代理摘要

良好的子代理边界

将子代理用于以下任务：

• 聚焦的
• 可独立检查的
• 上下文繁重的
• 易于总结的

好的例子：

查找所有与认证相关的文件。
设计迁移计划。
检查测试失败并总结可能的原因。

不好的例子：

修复所有问题。
理解整个仓库。
完成项目。

如果提示太宽泛，子代理将返回模糊的摘要。

学习检查

阅读代码后，确保你能回答：

• 子代理类型在哪里定义？
• 哪一行代码创建了隔离的子上下文？
• 为什么 explore 和 plan 有只读工具？
• 为什么演示避免给子代理 Task 工具？
• 究竟是什么成为返回给父代理的 tool_result？

完整源代码

#!/usr/bin/env python3
"""
v3_subagent.py - Mini Claude Code：子代理机制（约 450 行）

核心理念："通过上下文隔离实现分而治之"
==========================================
v2 添加了规划功能。但对于"探索代码库然后重构认证"这样的大型任务，
单个代理会遇到问题：

问题 - 上下文污染：
-------------------
    单代理历史记录：
      [探索中...] cat file1.py -> 500 行
      [探索中...] cat file2.py -> 300 行
      ... 还有 15 个文件 ...
      [现在开始重构...] "等等，file1 里有什么来着？"

模型的上下文被探索细节填满，留给实际任务的空间很小。
这就是"上下文污染"。

解决方案 - 具有独立上下文的子代理：
------------------------------------
    主代理历史记录：
      [任务：探索代码库]
        -> 子代理探索 20 个文件（在自己的上下文中）
        -> 只返回："认证在 src/auth/，数据库在 src/models/"
      [现在可以在干净的上下文中进行重构了]

每个子代理有：
  1. 自己的全新消息历史
  2. 过滤后的工具（explore 不能写入）
  3. 专门的系统提示
  4. 只向父代理返回最终摘要

关键洞察：
---------
    流程隔离 = 上下文隔离

通过生成子任务，我们获得：
  - 主代理的干净上下文
  - 可能的并行探索
  - 自然的任务分解
  - 相同的代理循环，不同的上下文

代理类型注册表：
---------------
    | 类型    | 工具                | 用途                        |
    |---------|---------------------|---------------------------- |
    | explore | bash, read_file     | 只读探索                    |
    | code    | all tools           | 完全实施权限                |
    | plan    | bash, read_file     | 设计但不修改                |

典型流程：
---------
    用户："将认证重构为使用 JWT"

    主代理：
      1. Task(explore)："查找所有与认证相关的文件"
         -> 子代理读取 10 个文件
         -> 返回："认证在 src/auth/login.py..."

      2. Task(plan)："设计 JWT 迁移"
         -> 子代理分析结构
         -> 返回："1. 添加 jwt 库 2. 创建工具..."

      3. Task(code)："实现 JWT 令牌"
         -> 子代理编写代码
         -> 返回："已创建 jwt_utils.py，已更新 login.py"

      4. 向用户总结更改

使用方法：
    python v3_subagent.py
"""

import os
import subprocess
import sys
import time
from pathlib import Path

from dotenv import load_dotenv

load_dotenv()

try:
    from anthropic import Anthropic
except ImportError:
    sys.exit("请安装：pip install anthropic python-dotenv")


# =============================================================================
# 配置
# =============================================================================

API_KEY = os.getenv("ANTHROPIC_API_KEY")
BASE_URL = os.getenv("ANTHROPIC_BASE_URL")
MODEL = os.getenv("MODEL_NAME", "claude-sonnet-4-20250514")
WORKDIR = Path.cwd()

client = Anthropic(api_key=API_KEY, base_url=BASE_URL) if BASE_URL else Anthropic(api_key=API_KEY)


# =============================================================================
# 代理类型注册表 - 子代理机制的核心
# =============================================================================

AGENT_TYPES = {
    # Explore：用于搜索和分析的只读代理
    # 不能修改文件 - 适用于广泛探索
    "explore": {
        "description": "用于探索代码、查找文件、搜索的只读代理",
        "tools": ["bash", "read_file"],  # 无写入权限
        "prompt": "你是一个探索代理。搜索和分析，但永远不要修改文件。返回简洁的摘要。",
    },

    # Code：用于实现的全功能代理
    # 拥有所有工具 - 用于实际编码工作
    "code": {
        "description": "用于实现功能和修复错误的完整代理",
        "tools": "*",  # 所有工具
        "prompt": "你是一个编码代理。高效地实现所请求的更改。",
    },

    # Plan：用于设计工作的规划代理
    # 只读，专注于生成计划和策略
    "plan": {
        "description": "用于设计实施策略的规划代理",
        "tools": ["bash", "read_file"],  # 只读
        "prompt": "你是一个规划代理。分析代码库并输出编号的实施计划。不要做任何更改。",
    },
}


def get_agent_descriptions() -> str:
    """为 Task 工具生成代理类型描述。"""
    return "\n".join(
        f"- {name}: {cfg['description']}"
        for name, cfg in AGENT_TYPES.items()
    )


# =============================================================================
# TodoManager（来自 v2，未修改）
# =============================================================================

class TodoManager:
    """带约束的任务列表管理器。详见 v2。"""

    def __init__(self):
        self.items = []

    def update(self, items: list) -> str:
        validated = []
        in_progress = 0

        for i, item in enumerate(items):
            content = str(item.get("content", "")).strip()
            status = str(item.get("status", "pending")).lower()
            active = str(item.get("activeForm", "")).strip()

            if not content or not active:
                raise ValueError(f"项目 {i}：需要 content 和 activeForm")
            if status not in ("pending", "in_progress", "completed"):
                raise ValueError(f"项目 {i}：无效状态")
            if status == "in_progress":
                in_progress += 1

            validated.append({
                "content": content,
                "status": status,
                "activeForm": active
            })

        if in_progress > 1:
            raise ValueError("只能有一个任务处于 in_progress 状态")

        self.items = validated[:20]
        return self.render()

    def render(self) -> str:
        if not self.items:
            return "无待办事项。"
        lines = []
        for t in self.items:
            mark = "[x]" if t["status"] == "completed" else \
                   "[>]" if t["status"] == "in_progress" else "[ ]"
            lines.append(f"{mark} {t['content']}")
        done = sum(1 for t in self.items if t["status"] == "completed")
        return "\n".join(lines) + f"\n（{done}/{len(self.items)} 已完成）"


TODO = TodoManager()


# =============================================================================
# 系统提示
# =============================================================================

SYSTEM = f"""你是位于 {WORKDIR} 的编码代理。

循环：规划 -> 使用工具执行 -> 报告。

你可以为复杂的子任务生成子代理：
{get_agent_descriptions()}

规则：
- 对需要聚焦探索或实施的子任务使用 Task 工具
- 使用 TodoWrite 跟踪多步骤工作
- 优先使用工具而非文字说明。行动，不要只是解释。
- 完成后，总结所做的更改。"""


# =============================================================================
# 基础工具定义
# =============================================================================

BASE_TOOLS = [
    {
        "name": "bash",
        "description": "运行 shell 命令。",
        "input_schema": {
            "type": "object",
            "properties": {"command": {"type": "string"}},
            "required": ["command"],
        },
    },
    {
        "name": "read_file",
        "description": "读取文件内容。",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "limit": {"type": "integer"}
            },
            "required": ["path"],
        },
    },
    {
        "name": "write_file",
        "description": "写入文件。",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "content": {"type": "string"}
            },
            "required": ["path", "content"],
        },
    },
    {
        "name": "edit_file",
        "description": "替换文件中的文本。",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string"},
                "old_text": {"type": "string"},
                "new_text": {"type": "string"},
            },
            "required": ["path", "old_text", "new_text"],
        },
    },
    {
        "name": "TodoWrite",
        "description": "更新任务列表。",
        "input_schema": {
            "type": "object",
            "properties": {
                "items": {
                    "type": "array",
                    "items": {
                        "type": "object",
                        "properties": {
                            "content": {"type": "string"},
                            "status": {
                                "type": "string",
                                "enum": ["pending", "in_progress", "completed"]
                            },
                            "activeForm": {"type": "string"},
                        },
                        "required": ["content", "status", "activeForm"],
                    },
                }
            },
            "required": ["items"],
        },
    },
]


# =============================================================================
# Task 工具 - v3 中的核心新增
# =============================================================================

TASK_TOOL = {
    "name": "Task",
    "description": f"""生成子代理处理聚焦的子任务。

子代理在独立上下文中运行 - 它们看不到父代理的历史记录。
使用此功能保持主对话干净。

代理类型：
{get_agent_descriptions()}

示例用途：
- Task(explore)："查找所有使用认证模块的文件"
- Task(plan)："设计数据库迁移策略"
- Task(code)："实现用户注册表单"
""",
    "input_schema": {
        "type": "object",
        "properties": {
            "description": {
                "type": "string",
                "description": "简短任务名称（3-5 个词），用于进度显示"
            },
            "prompt": {
                "type": "string",
                "description": "给子代理的详细指令"
            },
            "agent_type": {
                "type": "string",
                "enum": list(AGENT_TYPES.keys()),
                "description": "要生成的代理类型"
            },
        },
        "required": ["description", "prompt", "agent_type"],
    },
}

# 主代理获得所有工具，包括 Task
ALL_TOOLS = BASE_TOOLS + [TASK_TOOL]


def get_tools_for_agent(agent_type: str) -> list:
    """
    根据代理类型过滤工具。

    每种代理类型都有允许的工具白名单。
    '*' 表示所有工具（但子代理不获得 Task，以防止演示中的无限递归）。
    """
    allowed = AGENT_TYPES.get(agent_type, {}).get("tools", "*")

    if allowed == "*":
        return BASE_TOOLS  # 所有基础工具，但不包括 Task（演示中无递归）

    return [t for t in BASE_TOOLS if t["name"] in allowed]


# =============================================================================
# 工具实现
# =============================================================================

def safe_path(p: str) -> Path:
    """确保路径保持在工作区内。"""
    path = (WORKDIR / p).resolve()
    if not path.is_relative_to(WORKDIR):
        raise ValueError(f"路径超出工作区：{p}")
    return path


def run_bash(cmd: str) -> str:
    """执行带安全检查的 shell 命令。"""
    if any(d in cmd for d in ["rm -rf /", "sudo", "shutdown"]):
        return "错误：危险命令"
    try:
        r = subprocess.run(
            cmd, shell=True, cwd=WORKDIR,
            capture_output=True, text=True, timeout=60
        )
        return ((r.stdout + r.stderr).strip() or "(无输出)")[:50000]
    except Exception as e:
        return f"错误：{e}"


def run_read(path: str, limit: int = None) -> str:
    """读取文件内容。"""
    try:
        lines = safe_path(path).read_text().splitlines()
        if limit:
            lines = lines[:limit]
        return "\n".join(lines)[:50000]
    except Exception as e:
        return f"错误：{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"已向 {path} 写入 {len(content)} 字节"
    except Exception as e:
        return f"错误：{e}"


def run_edit(path: str, old_text: str, new_text: str) -> str:
    """替换文件中的精确文本。"""
    try:
        fp = safe_path(path)
        text = fp.read_text()
        if old_text not in text:
            return f"错误：在 {path} 中未找到文本"
        fp.write_text(text.replace(old_text, new_text, 1))
        return f"已编辑 {path}"
    except Exception as e:
        return f"错误：{e}"


def run_todo(items: list) -> str:
    """更新待办事项列表。"""
    try:
        return TODO.update(items)
    except Exception as e:
        return f"错误：{e}"


# =============================================================================
# 子代理执行 - v3 的核心
# =============================================================================

def run_task(description: str, prompt: str, agent_type: str) -> str:
    """
    在独立上下文中执行子代理任务。

    这是子代理机制的核心：

    1. 创建独立的消息历史（关键：没有父上下文！）
    2. 使用代理特定的系统提示
    3. 根据代理类型过滤可用工具
    4. 运行与主代理相同的查询循环
    5. 只返回最终文本（不是中间细节）

    父代理只看到摘要，保持其上下文干净。

    进度显示：
    ----------------
    运行时，我们显示：
      [explore] 查找认证文件 ... 5 个工具, 3.2 秒

    这提供了可见性，而不会污染主对话。
    """
    if agent_type not in AGENT_TYPES:
        return f"错误：未知代理类型 '{agent_type}'"

    config = AGENT_TYPES[agent_type]

    # 代理特定的系统提示
    sub_system = f"""你是位于 {WORKDIR} 的 {agent_type} 子代理。

{config["prompt"]}

完成任务并返回清晰、简洁的摘要。"""

    # 为此代理类型过滤后的工具
    sub_tools = get_tools_for_agent(agent_type)

    # 独立的消息历史 - 这是关键！
    # 子代理从新开始，看不到父代理的对话
    sub_messages = [{"role": "user", "content": prompt}]

    # 进度跟踪
    print(f"  [{agent_type}] {description}")
    start = time.time()
    tool_count = 0

    # 运行相同的代理循环（静默 - 不打印到主聊天）
    while True:
        response = client.messages.create(
            model=MODEL,
            system=sub_system,
            messages=sub_messages,
            tools=sub_tools,
            max_tokens=8000,
        )

        if response.stop_reason != "tool_use":
            break

        tool_calls = [b for b in response.content if b.type == "tool_use"]
        results = []

        for tc in tool_calls:
            tool_count += 1
            output = execute_tool(tc.name, tc.input)
            results.append({
                "type": "tool_result",
                "tool_use_id": tc.id,
                "content": output
            })

            # 更新进度行（原地）
            elapsed = time.time() - start
            sys.stdout.write(
                f"\r  [{agent_type}] {description} ... {tool_count} 个工具, {elapsed:.1f}秒"
            )
            sys.stdout.flush()

        sub_messages.append({"role": "assistant", "content": response.content})
        sub_messages.append({"role": "user", "content": results})

    # 最终进度更新
    elapsed = time.time() - start
    sys.stdout.write(
        f"\r  [{agent_type}] {description} - 完成（{tool_count} 个工具，{elapsed:.1f}秒）\n"
    )

    # 提取并只返回最终文本
    # 这是父代理看到的 - 干净的摘要
    for block in response.content:
        if hasattr(block, "text"):
            return block.text

    return "(子代理未返回文本)"


def execute_tool(name: str, args: dict) -> str:
    """将工具调用分派到实现。"""
    if name == "bash":
        return run_bash(args["command"])
    if name == "read_file":
        return run_read(args["path"], args.get("limit"))
    if name == "write_file":
        return run_write(args["path"], args["content"])
    if name == "edit_file":
        return run_edit(args["path"], args["old_text"], args["new_text"])
    if name == "TodoWrite":
        return run_todo(args["items"])
    if name == "Task":
        return run_task(args["description"], args["prompt"], args["agent_type"])
    return f"未知工具：{name}"


# =============================================================================
# 主代理循环
# =============================================================================

def agent_loop(messages: list) -> list:
    """
    支持子代理的主代理循环。

    与 v1/v2 模式相同，但现在包含 Task 工具。
    当模型调用 Task 时，它会生成一个具有独立上下文的子代理。
    """
    while True:
        response = client.messages.create(
            model=MODEL,
            system=SYSTEM,
            messages=messages,
            tools=ALL_TOOLS,
            max_tokens=8000,
        )

        tool_calls = []
        for block in response.content:
            if hasattr(block, "text"):
                print(block.text)
            if block.type == "tool_use":
                tool_calls.append(block)

        if response.stop_reason != "tool_use":
            messages.append({"role": "assistant", "content": response.content})
            return messages

        results = []
        for tc in tool_calls:
            # Task 工具有特殊的显示处理
            if tc.name == "Task":
                print(f"\n> Task: {tc.input.get('description', '子任务')}")
            else:
                print(f"\n> {tc.name}")

            output = execute_tool(tc.name, tc.input)

            # 不打印完整的 Task 输出（它管理自己的显示）
            if tc.name != "Task":
                preview = output[:200] + "..." if len(output) > 200 else output
                print(f"  {preview}")

            results.append({
                "type": "tool_result",
                "tool_use_id": tc.id,
                "content": output
            })

        messages.append({"role": "assistant", "content": response.content})
        messages.append({"role": "user", "content": results})


# =============================================================================
# 主 REPL
# =============================================================================

def main():
    print(f"Mini Claude Code v3（带子代理）- {WORKDIR}")
    print(f"代理类型：{', '.join(AGENT_TYPES.keys())}")
    print("输入 'exit' 退出。\n")

    history = []

    while True:
        try:
            user_input = input("你：").strip()
        except (EOFError, KeyboardInterrupt):
            break

        if not user_input or user_input.lower() in ("exit", "quit", "q"):
            break

        history.append({"role": "user", "content": user_input})

        try:
            agent_loop(history)
        except Exception as e:
            print(f"错误：{e}")

        print()


if __name__ == "__main__":
    main()

分而治之。上下文隔离。

← v2 | 返回 README | v0 →