动机:为什么需要 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.ts 和 src/agents/tools/sessions-spawn-tool.ts 实现了子任务创建。
下面这张图展示了父子 agent 的隔离关系:
左边是父 agent,有完整的 messages 历史和全量工具。右边是 sub-agent,有独立的 messages(从零开始)、精简的工具集(没有 spawn,防止无限嵌套)、精简的 system prompt(minimal 模式)。子任务完成后,只有结果摘要回到父 agent 的 messages 中。
Ground Truth:真实的 Sub-agent 实现
Sub-agent——跨代码库对比
子任务隔离在不同代码库中的实现
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_task10 return f"Subagent [{label}] started (id: {task_id})"1112 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 和 SpawnTool1920 # 独立的 system prompt21 system = self._build_subagent_prompt(task)2223 # 独立的 messages (核心隔离)24 messages = [25 {"role": "system", "content": system},26 {"role": "user", "content": task},27 ]2829 # 运行独立的 agent loop30 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 results36 ...37 else:38 final_result = response.content39 break4041 # 结果通过 message bus 报告给主 agent42 await self._announce_result(43 task_id, label, task, final_result, origin44 )
1// system-prompt.ts: Sub-agent 用 minimal 模式2type PromptMode = "full" | "minimal" | "none";3// full = 主 agent (所有模块)4// minimal = sub-agent (只有 Tooling, Workspace, Runtime)5// none = 最简身份行67// tools/sessions-spawn-tool.ts: 创建子 session8// 子 session 有独立的 sessionKey9// 通过 routing/session-key.ts 识别:10function isSubagentSessionKey(key: string): boolean {11 // sub-agent session 有特殊前缀12}1314// tools/subagents-tool.ts: Sub-agent 管理15// 列出运行中的 sub-agents16// 每个 sub-agent:17// 1. 独立的 session (独立 messages[])18// 2. minimal system prompt19// 3. 通过 sessions-send-tool 通信20// 4. 无法访问父 agent 的对话历史2122// 关键: sub-agent 的隔离是架构级别的23// sessionKey 不同 = messages[] 不同
关键观察:
- 核心是隔离:sub-agent 有全新的
messages[],看不到父 agent 的历史 - 工具集受限:sub-agent 不能给用户发消息(没有 MessageTool),不能再创建 sub-agent(没有 SpawnTool)
- 异步执行:nanobot 用
asyncio.create_task做后台执行 - 结果通信:通过 message bus 把结果报告回主 agent,而不是直接修改父 agent 的 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 模式 | 单 agent | SubagentManager 多 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"
可以尝试的提示:
- “Research Python web frameworks, spawn a sub-agent to do it”
- “帮我调研 FastAPI vs Flask,用子任务分别调研”
- 观察父 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 系统中的自然表达。