Phase 4 / Scale & Orchestration / s10

Sub-agents

隔离的 messages[]——在独立上下文中执行子任务

动机:为什么需要 Sub-agent

当任务变得复杂(“调研这个 API 的文档并写一个集成模块”),单个 agent 的 context window 会被中间步骤的产出塞满——API 文档内容、搜索结果、草稿代码。这些中间内容对最终结果很重要,但对主会话来说是噪声。

Sub-agent 的核心思想很简单:启动一个新的 agent loop,给它独立的 messages[],完成后只把结果报告回来

OpenClaw 的源码直接展示了这个架构。在 src/agents/system-prompt.ts 中,PromptMode 类型区分了 "full"(主 agent)和 "minimal"(sub-agent)模式。Sub-agent 使用精简的 system prompt,有独立的 session,工具集也不同。src/agents/tools/subagents-tool.tssrc/agents/tools/sessions-spawn-tool.ts 实现了子任务创建。

下面这张图展示了父子 agent 的隔离关系:

Sub-agent 上下文隔离独立 messages[]
Parent Agentuser: 重构这个模块assistant: 我来分析...assistant: spawn(调研 API)user: [sub-agent结果]messages.length: 4tools: bash, read, write, spawn, todoprompt: full (689 行)spawnSub-agentuser: 调研 FastAPI 的 APIassistant: tool_use: web_searchtool: FastAPI 是一个...assistant: 调研完成...messages.length: 4 (独立)tools: bash, read, write (无 spawn)prompt: minimal子任务的中间过程不会污染父 agent

左边是父 agent,有完整的 messages 历史和全量工具。右边是 sub-agent,有独立的 messages(从零开始)、精简的工具集(没有 spawn,防止无限嵌套)、精简的 system prompt(minimal 模式)。子任务完成后,只有结果摘要回到父 agent 的 messages 中。

Ground Truth:真实的 Sub-agent 实现

Sub-agent——跨代码库对比

子任务隔离在不同代码库中的实现

pythonagent/subagent.py
1class SubagentManager:
2 async def spawn(self, task: str, label=None,
3 origin_channel="cli",
4 origin_chat_id="direct"):
5 task_id = str(uuid.uuid4())[:8]
6 bg_task = asyncio.create_task(
7 self._run_subagent(task_id, task, label, origin)
8 )
9 self._running_tasks[task_id] = bg_task
10 return f"Subagent [{label}] started (id: {task_id})"
11
12 async def _run_subagent(self, task_id, task, label, origin):
13 # 独立的 ToolRegistry (没有 message/spawn 工具)
14 tools = ToolRegistry()
15 tools.register(ReadFileTool(...))
16 tools.register(WriteFileTool(...))
17 tools.register(ExecTool(...))
18 # 注意: 不注册 MessageTool 和 SpawnTool
19
20 # 独立的 system prompt
21 system = self._build_subagent_prompt(task)
22
23 # 独立的 messages (核心隔离)
24 messages = [
25 {"role": "system", "content": system},
26 {"role": "user", "content": task},
27 ]
28
29 # 运行独立的 agent loop
30 while iteration < 15:
31 response = await self.provider.chat(
32 messages=messages, tools=tools.get_definitions(),
33 )
34 if response.has_tool_calls:
35 # execute tools, append results
36 ...
37 else:
38 final_result = response.content
39 break
40
41 # 结果通过 message bus 报告给主 agent
42 await self._announce_result(
43 task_id, label, task, final_result, origin
44 )
typescriptsrc/agents/system-prompt.ts + tools/
1// system-prompt.ts: Sub-agent 用 minimal 模式
2type PromptMode = "full" | "minimal" | "none";
3// full = 主 agent (所有模块)
4// minimal = sub-agent (只有 Tooling, Workspace, Runtime)
5// none = 最简身份行
6
7// tools/sessions-spawn-tool.ts: 创建子 session
8// 子 session 有独立的 sessionKey
9// 通过 routing/session-key.ts 识别:
10function isSubagentSessionKey(key: string): boolean {
11 // sub-agent session 有特殊前缀
12}
13
14// tools/subagents-tool.ts: Sub-agent 管理
15// 列出运行中的 sub-agents
16// 每个 sub-agent:
17// 1. 独立的 session (独立 messages[])
18// 2. minimal system prompt
19// 3. 通过 sessions-send-tool 通信
20// 4. 无法访问父 agent 的对话历史
21
22// 关键: sub-agent 的隔离是架构级别的
23// sessionKey 不同 = messages[] 不同

关键观察:

构建:SubagentManager

class SubagentManager:
    """管理子任务的创建和执行"""

    def __init__(self, provider, workspace: Path, registry: ToolRegistry):
        self.provider = provider
        self.workspace = workspace
        self.parent_registry = registry
        self._results: dict[str, str] = {}

    def spawn(self, task: str) -> str:
        """启动一个 sub-agent 执行任务"""
        task_id = str(uuid.uuid4())[:8]

        # 独立的工具集——比父 agent 受限
        tools = ToolRegistry()
        tools.register(BashTool())
        tools.register(ReadFileTool())
        tools.register(WriteFileTool())

        # 独立的 system prompt
        system = f"""你是一个 sub-agent,专门执行以下任务。
完成后给出简洁的结果摘要。
工作区: {self.workspace}"""

        # 独立的 messages——核心隔离
        messages = [
            {"role": "system", "content": system},
            {"role": "user", "content": task},
        ]

        # 运行独立的 agent loop
        result = self._run_loop(messages, tools, max_iterations=15)
        self._results[task_id] = result
        return f"[Sub-agent {task_id}] 完成: {result[:200]}"

    def _run_loop(self, messages, tools, max_iterations=15) -> str:
        """独立的 agent loop"""
        for _ in range(max_iterations):
            response = self.provider.chat(
                messages=messages, tools=tools.get_definitions()
            )
            messages.append({"role": "assistant", "content": response.content})
            if response.stop_reason != "tool_use":
                return response.content[-1].text
            # execute tools, append results...
        return "达到最大迭代次数"

测试:验证上下文隔离

# 父 agent 的 messages
parent_messages = [
    {"role": "user", "content": "我正在处理一个大型重构"},
    {"role": "assistant", "content": "好的,让我分析代码结构"},
]

# 启动 sub-agent
mgr = SubagentManager(provider, Path("."), registry)
result = mgr.spawn("列出 src/ 目录下的所有 Python 文件")

# sub-agent 完成后:
# 1. 父 agent 的 messages 没有被修改
assert len(parent_messages) == 2  # 不变

# 2. sub-agent 的执行过程不会出现在父 agent 的上下文中
# 3. 只有最终结果被传回

变更内容

组件之前 (s09)之后 (s10)
Agent 模式单 agentSubagentManager 多 agent
子任务SpawnTool 创建子 agent
上下文共享 messages隔离的 messages[]

本课代码: agents/s10_sub_agents.py — 344 行 (新增 41 行)

试一试

cd public/code
python agents/s10_sub_agents.py "Research Python web frameworks, spawn a sub-agent to do it"

可以尝试的提示:

  1. “Research Python web frameworks, spawn a sub-agent to do it”
  2. “帮我调研 FastAPI vs Flask,用子任务分别调研”
  3. 观察父 agent 的 messages 长度——子任务的中间过程不会出现

距离生产

Sub-agent 看起来很简单——就是启动一个新的 agent loop。但生产实现的复杂性远超预期。

异步与生命周期管理。我们的 spawn() 是同步的——父 agent 停下来等子任务完成。nanobot 用 asyncio.create_task() 做真正的后台执行,并维护一个 _running_tasks 字典追踪所有活跃子任务。这不只是性能优化——它允许父 agent 在等待子任务的同时继续响应用户。nanobot 还实现了 cancel_by_session():当用户发送 /stop 时,可以取消某个 session 下的所有子任务。这在我们的同步实现中不可能做到。

通信模式。我们让子任务直接返回字符串。nanobot 用 message bus(self.bus.publish_inbound)把结果注入回主 agent 的消息队列——子任务完成后,主 agent 收到一条 “system” 消息,包含结果摘要。OpenClaw 更进一步:sessions-spawn-tool.ts 创建独立的 session,通过 sessions-send-tool.ts 进行 session 间通信。这种松耦合设计让子 agent 可以运行在不同的进程甚至不同的机器上。

System prompt 差异化。OpenClaw 在 system-prompt.ts 中定义了 PromptMode:主 agent 用 "full"(完整的身份、工具说明、记忆、技能),子 agent 用 "minimal"(只有工具和工作区信息)。这不只是节省 token——更精简的 prompt 让子 agent 更聚焦,减少”走神”的概率。

第一性原理思考:Sub-agent 的本质是用空间换时间——用额外的 context window 容量(独立的 messages[])换取不污染主对话的干净执行。但这引出一个更深的问题:为什么不给主 agent 一个更大的 context window?理论上,200K token 的 context 足够处理大多数任务。答案是:context window 的大小不是唯一的限制因素。更长的 context 意味着更高的 API 成本、更慢的响应速度、以及”注意力稀释”——模型在超长 context 中找到关键信息的能力会下降。Sub-agent 通过隔离 context 同时解决了这三个问题。这是”分而治之”策略在 LLM 系统中的自然表达。