通过 Nanobot 源码学习架构 ---（4）SubAgent

OpenClaw 应该有40万行代码阅读理解起来难度过大因此本系列通过Nanobot来学习 OpenClaw 的特色。Nanobot是由香港大学数据科学实验室(HKUDS)开源的超轻量级个人 AI 助手框架定位为Ultra-Lightweight OpenClaw。非常适合学习Agent架构。nanobot 的 Subagent 实现是一个简洁但强大的后台任务执行机制。通过复用主 Agent 的 LLM provider 但限制工具集和迭代次数实现了任务隔离和资源控制。消息总线机制确保子代理结果能够顺利通知主 Agent最终传达给用户。这种设计使得主 Agent 可以保持专注的对话交互同时将复杂任务委派给后台子代理执行。Parent agent Subagent ------------------ ------------------ | messages[...] | | messages[] | -- fresh | | dispatch | | | tool: task | ---------- | while tool_use: | | prompt... | | call tools | | | summary | append results | | result ... | ---------- | return last text | ------------------ ------------------ Parent context stays clean. Subagent context is discarded.注本系列借鉴的文章过多可能在参考文献中有遗漏的文章如果有还请大家指出。0x01 基础背景SubAgent子智能体是从现有 Agent 运行中生成的后台独立运行实例。它们在独立的会话中执行任务完成后将结果自动通告回请求者的聊天渠道。1.1 原理为什么需要 SubAgent表面上看SubAgent 解决的是「并行执行」问题——多个任务同时推进提升效率。然而事实上拆分 Agent 不只是为了分工更是为了上下文压缩或者说上下文隔离。我们思考下当一个 Agent 试图全包所有角色时会发生什么随着对话轮次增加系统提示词和历史记录会越来越臃肿。这直接导致三个连锁反应模型更容易遗忘早期约束推理漂移——偏离原始任务目标成本上升——Token 消耗持续累积最终单一上下文窗口已经无法承载当前任务的复杂度需求。这引出了单 Agent 架构的本质权衡维度表现优势最原生的架构、开发链路最短、运行效率极高适合快速构建 Demo 或处理知识依赖较少的场景劣势极度依赖上下文窗口的质量与长度。一旦涉及大量领域知识的注入极易引发上下文爆炸导致模型注意力分散稳定性大幅下降因此关键问题浮现当单点突破遇到上下文瓶颈时我们该如何通过架构演进在保持灵活性的同时解决知识承载的问题这正是 SubAgent 被引入的核心动机。1.2 架构拓扑的演进从协作模式看SubAgent 的引入形成了两种典型架构架构类型特征适用场景主从协作模式存在中央 Orchestrator 作为主 Agent需要统一决策、结果整合的复杂任务纯 SubAgent 模式只有平行的 SubAgent无中央协调任务天然可完全并行无需统一收口后一种模式的核心逻辑在于路由分发与领域隔离主 AgentOrchestrator扮演大脑角色仅负责意图识别与任务路由判断这个问题该交给谁而无需背负所有领域的知识重担。子 AgentSub-Agent拥有独立的 Identity 空间内化特定领域的专业知识。每个子 Agent 只需专注于解决某一类垂直场景其 Prompt 指令更精简领域知识更聚焦。1.3 领域隔离从上下文工程Context Engineering的角度看SubAgent 实现了Isolate 机制——上下文隔离。这种隔离通常由三种触发条件驱动隔离噪声——避免失败路径或中间探索污染后续推理隔离关注点——让专业化的工具集各司其职减少干扰突破物理限制——通过并行扩展单 Agent 的 Token 上限如何识别需要拆分的信号 当以下现象出现时便是架构调整的时机上下文窗口接近极限表现为幻觉率上升、忽略早期指令工具集过大频繁选错工具且工具集内有明显的专业领域区分需要覆盖大信息空间搜索覆盖面不足单 Agent 无法遍历把任务拆给专业 Agent让它在独立上下文中完成子任务再返回结果相当于把上下文按职责切片。这不仅通常会更稳、更便宜更是一种主动的上下文噪声隔离。1.4 工作场景Skills vs SubAgent如何选择我们用操作系统类比来理解两者的定位差异Skills 是应用程序装在主系统里按需调用SubAgent 是虚拟机独立运行完再把结果交回来。一句话总结选择逻辑任务简单用应用任务复杂开虚拟机。维度SkillsSubAgent任务复杂度简单主 Agent 全程掌控复杂、耗时长、中间过程繁琐知识复用可以复用按需加载独立封装领域隔离上下文管理节省上下文动态加载完全隔离主 Agent 零负担并行需求串行执行支持多任务并行主 Agent 状态持续参与细节保持思维清晰只收结果1.5 工作流主从协作的典型模式SubAgent 是典型的主从协作管理模式。我们以一个具体场景为例用户要求帮我分析这个代码仓库同时整理几份竞品资料然后给我一份对比报告执行流程如下主 Agent 继续和用户保持对话确认细节同时spawn一个 SubAgent 去分析仓库结构再spawn一个 SubAgent 去整理竞品资料两个 SubAgent并行执行各自拥有精简的 System Prompt最后统一收口主 Agent 基于两份摘要完成对比分析我们拆解这个流程的 Context Engineering 价值通过 层层外包 只传结果 的机制将大任务分解后的中间过程隔离在子 Agent 内主 Agent 的 Context 始终保持精简。主Agent接收任务 ↓ [tool_use] Spawn(分析仓库结构), Spawn(整理竞品资料) ↓ 两个Sub-agent并行执行各自有精简System Prompt ↓ 返回仓库结构、竞品资料 ↓ 主Agent Context中只有库结构、竞品资料的摘要没有详尽的信息 ↓ 主Agent完成比较分析因此SubAgent 的本质定义是Agent 可以召唤的子实例以精简 System Prompt 专注单一任务主 Agent 只接收摘要结果Context Window 中不保留子任务的完整执行过程。0x02 Nanobot SubAgent 功能SubAgent 是 nanobot 的后台任务执行机制允许主 Agent 派生独立的子代理来执行耗时或独立的任务而不阻塞主对话流程。2.1 SubAgent 与主 Agent 的区别2.1.1 设计目的主 Agent专注于用户对话提供即时响应管理会话状态和记忆Subagent专注于执行耗时任务不阻塞主对话独立完成后通知主 Agent子代理有独立的工具集和执行限制2.1.2 Subagent 优势响应性主 Agent 不会被耗时任务阻塞保持与用户的实时交互并发性多个子代理可以同时运行执行不同任务隔离性子代理有独立的工具集和限制不会干扰主对话可取消性通过 session_key 实现会话级的任务取消结果聚合子代理结果通过主 Agent 统一格式化后发送给用户2.2 SubagentManager 类2.2.1 初始化参数class SubagentManager: Manages background subagent execution. def __init__( self, provider: LLMProvider, # LLM 提供商复用主 Agent 的 workspace: Path, # 工作空间路径 bus: MessageBus, # 消息总线用于通知主 Agent model: str | None None, # 模型名称 temperature: float 0.7, # 温度参数 max_tokens: int 4096, # 最大 token 数 brave_api_key: str | None None, # 网络搜索 API 密钥 exec_config: ExecToolConfig | None None, # Shell 执行配置 restrict_to_workspace: bool False, # 是否限制到工作空间 ):2.2.2 内部状态管理_running_tasks映射 task_id 到 asyncio.Task存储所有运行的子代理任务_session_tasks映射 session_key 到 task_id 集合追踪每个会话关联的子代理self._running_tasks: dict[str, asyncio.Task | None] {} self._session_tasks: dict[str, set[str]] {}2.3 创建子代理流程2.3.1 spawn() 方法详解spawn()方法是创建子代理的入口点async def spawn( self, task: str, # 子代理要执行的任务描述 label: str | None None, # 显示标签用于用户识别 origin_channel: str cli, # 原始渠道用于结果通知 origin_chat_id: str direct, # 原始聊天 ID session_key: str | None None, # 会话键用于会话级取消 ) - str:2.3.2 子代理创建步骤生成唯一标识符task_id str(uuid.uuid4())[:8] # 生成 8 字符的 UUID4如 a1b2c3d4 display_label label or task[:30] (... if len(task) 30 else )记录原始来源origin {channel: origin_channel, chat_id: origin_chat_id}用于后续将结果通知回正确的用户/渠道。创建并启动后台任务bg_task asyncio.create_task( self._run_subagent(task_id, task, display_label, origin) ) self._running_tasks[task_id] bg_task创建异步任务来运行子代理并将其注册到_running_tasks字典中。关联到会话if session_key: self._session_tasks.setdefault(session_key, set()).add(task_id)如果提供了 session_key将 task_id 加入该会话的子代理集合。这使得/stop命令可以取消整个会话的所有子代理。设置清理回调def _cleanup(_: asyncio.Task) - None: self._running_tasks.pop(task_id, None) if session_key and (ids : self._session_tasks.get(session_key)): ids.discard(task_id) if not ids: del self._session_tasks[session_key] bg_task.add_done_callback(_cleanup)当子代理任务完成无论成功或失败时回调函数执行从_running_tasks移除 task_id从会话的 task_id 集合中移除如果该会话没有剩余的子代理删除会话集条目返回用户反馈return fSubagent [{display_label}] started (id: {task_id}). Ill notify you when it completes.2.4 子代理执行逻辑_run_subagent() 是子代理的核心执行方法负责完整的 Agent 循环其具体逻辑如下2.4.1. 构建子代理专用工具集tools ToolRegistry() allowed_dir self.workspace if self.restrict_to_workspace else None tools.register(ReadFileTool(workspaceself.workspace, allowed_dirallowed_dir)) tools.register(WriteFileTool(workspaceself.workspace, allowed_dirallowed_dir)) tools.register(EditFileTool(workspaceself.workspace, allowed_dirallowed_dir)) tools.register(ListDirTool(workspaceself.workspace, allowed_dirallowed_dir)) tools.register(ExecTool( working_dirstr(self.workspace), timeoutself.exec_config.timeout, restrict_to_workspaceself.restrict_to_workspace, path_appendself.exec_config.path_append, )) tools.register(WebSearchTool(api_keyself.brave_api_key)) tools.register(WebFetchTool())重要设计子代理的工具集与主 Agent 不同包含文件读写、目录列表、Shell 执行、网络搜索和获取排除MessageTool不能直接发送消息给用户排除SpawnTool不能派生更多子代理排除CronTool不能创建定时任务这种设计确保子代理专注于执行任务不会干扰主对话流程或创建递归任务。2.4.2. 构建子代理专用提示system_prompt self._build_subagent_prompt(task) messages [ {role: system, content: system_prompt}, {role: user, content: task}, ]系统提示明确子代理的角色和限制# Subagent ## Current Time {now} ({tz}) You are a subagent spawned by main agent to complete a specific task. ## Rules 1. Stay focused - complete only the assigned task, nothing else 2. Your final response will be reported back to main agent 3. Do not initiate conversations or take on side tasks 4. Be concise but informative in your findings ## What You Can Do - Read and write files in workspace - Execute shell commands - Search web and fetch web pages - Complete task thoroughly ## What You Cannot Do - Send messages directly to users (no message tool available) - Spawn other subagents - Access main agents conversation history ## Workspace Your workspace is at: {workspace} Skills are available at: {workspace}/skills/ (read SKILL.md files as needed) When you have completed the task, provide a clear summary of your findings or actions.这个提示确保子代理专注于分配的任务不会发起新对话不会尝试与用户直接交互知道自己的能力边界2.4.3. 运行 Agent 循环限制迭代次数子代理使用与主 Agent 相同的 LLM provider但迭代次数限制为 15 次避免子代理运行过久。max_iterations 15 # 子代理的最大迭代次数主 Agent 是 40 iteration 0 final_result: str | None None while iteration max_iterations: iteration 1 response await self.provider.chat( messagesmessages, toolstools.get_definitions(), modelself.model, temperatureself.temperature, max_tokensself.max_tokens, )2.4.4. 处理工具调用工具调用处理逻辑与主 Agent 类似将工具调用添加到消息历史逐个执行工具将工具结果添加到消息历史继续循环等待 LLM 下一轮响应if response.has_tool_calls: # 构建工具调用消息 tool_call_dicts [ { id: tc.id, type: function, function: { name: tc.name, arguments: json.dumps(tc.arguments, ensure_asciiFalse), }, } for tc in response.tool_calls ] messages.append({ role: assistant, content: response.content or , tool_calls: tool_call_dicts, }) # 执行工具 for tool_call in response.tool_calls: result await tools.execute(tool_call.name, tool_call.arguments) messages.append({ role: tool, tool_call_id: tool_call.id, name: tool_call.name, content: result, })2.4.5. 处理完成条件当 LLM 返回文本响应而没有工具调用时视为任务完成退出循环。else: final_result response.content break2.4.6. 处理未完成情况如果达到最大迭代次数仍未产生最终响应使用默认消息。if final_result is None: final_result Task completed but no final response was generated.2.4.7. 通知结果成功完成时调用_announce_result()方法通知主 Agent。logger.info(Subagent [{}] completed successfully, task_id) await self._announce_result(task_id, label, task, final_result, origin, ok)2.4.8. 错误处理如果执行过程中发生异常捕获并通知主 Agent 错误信息。except Exception as e: error_msg fError: {str(e)} logger.error(Subagent [{}] failed: {}, task_id, e) await self._announce_result(task_id, label, task, error_msg, origin, error)2.5 结果通知机制子代理完成任务后需要将结果通知给主 Agent主 Agent 再转发给用户。_announce_result() 方法完成了此功能。async def _announce_result( self, task_id: str, label: str, task: str, result: str, origin: dict[str, str], status: str, ) - None:2.5.1 通知内容构建status_text completed successfully if status ok else failed announce_content f[Subagent {label} {status_text}] Task: {task} Result: {result} Summarize this naturally for the user. Keep it brief (1-2 sentences). Do not mention technical details like subagent or task IDs.通知内容包含子代理标签和状态原始任务描述执行结果指示主 Agent 如何处理简洁地总结给用户2.5.2 注入消息总线InboundMessage 通过bus.publish_inbound()将通知发布到入站队列。这会被 AgentLoop 接收并处理最终将总结转发给用户。msg InboundMessage( channelsystem, # 使用 system 渠道标识 sender_idsubagent, # 标识来自子代理 chat_idf{origin[channel]}:{origin[chat_id]}, # 原始渠道和聊天 ID contentannounce_content, # 通知内容 ) await self.bus.publish_inbound(msg)2.6 会话级取消机制cancel_by_session() 方法实现了会话级取消机制。这个方法被主 Agent 的/stop命令处理调用实现会话级的任务清理。async def cancel_by_session(self, session_key: str) - int: Cancel all subagents for a given session. Returns count cancelled. tasks [ self._running_tasks[tid] for tid in self._session_tasks.get(session_key, []) if tid in self._running_tasks and not self._running_tasks[tid].done() ] for t in tasks: t.cancel() if tasks: await asyncio.gather(*tasks, return_exceptionsTrue) return len(tasks)取消流程如下从_session_tasks获取该会话关联的所有 task_id筛选出未完成的任务对每个任务调用cancel()方法等待所有任务取消完成返回取消的任务数量2.7 状态查询方法get_running_count() 返回当前运行的子代理数量可用于监控和状态报告。def get_running_count(self) - int: Return number of currently running subagents. return len(self._running_tasks)0x03 Subagent 与 Main Agent 的关系分析Subagent 不是 Main Agent 的完全克隆而是一个共享部分组件但功能受限的独立执行单元。Subagent 的设计模式是共享基础组件 独立执行环境共享资源LLM Provider、MessageBus、Workspace、配置参数隔离执行独立的工具集、消息历史、系统提示受限能力不能发送消息、不能派生子代理、无对话历史结果聚合通过 MessageBus 通知 Main Agent由 Main Agent 统一输出这种设计避免了子代理干扰主对话流程同时确保资源高效利用。架构关系图如下主Agent和SubAgent的对比如下特性主 AgentSubagent消息来源用户通过聊天平台主 Agent 的 spawn 调用目标对话交互执行特定任务迭代次数4015消息发送可用 MessageTool不可用子代理派生可用 SpawnTool不可用定时任务可用 CronTool不可用会话历史完整访问无访问权限结果输出直接发送给用户通知主 Agent运行方式同步阻塞消息处理异步后台执行3.1 相同的组件3.1.1 LLM Provider 完全共享共享原因避免重复创建 API 连接节省资源和维护成本。LLM 调用是无状态的多个 Agent 可以安全地共享同一个 provider。# SubagentManager 初始化时 self.provider provider # 和 Main Agent 使用同一个实例 # Subagent 执行时 response await self.provider.chat( messagesmessages, toolstools.get_definitions(), modelself.model, temperatureself.temperature, max_tokensself.max_tokens, )3.1.2 MessageBus 共享共享原因Subagent 需要通过 MessageBus 将结果通知给 Main Agent不是用来处理用户消息。self.bus bus # 和 Main Agent 使用同一个实例 # Subagent 完成任务后 await self.bus.publish_inbound(msg) # 通知 Main Agent3.1.3 Workspace 路径相同共享原因Subagent 访问相同的文件系统能够读写主 Agent 工作空间中的文件。self.workspace workspace # 和 Main Agent 使用同一个路径 # Subagent 中的工具 tools.register(ReadFileTool(workspaceself.workspace, ...))3.1.4 配置参数主Agent和Subagent的配置参数相同虽然具体数值会有不同。参数Main AgentSubagent说明modelconfig.agents.defaults.modelprovider.get_default_model()模型名称temperatureconfig.agents.defaults.temperature0.7温度参数max_tokensconfig.agents.defaults.max_tokens4096最大 token 数workspaceconfig.workspace_pathworkspace工作空间exec_configconfig.tools.exec传入的 exec_configShell 执行配置restrict_to_workspaceconfig.tools.restrict_to_workspace传入的值工作空间限制3.2 不同的组件3.2.1 工具集ToolRegistry完全不同且受限设计意图Subagent 是工具型执行单元专注于完成任务不进行交互式对话或启动更多子任务。Subagent 使用的工具tools ToolRegistry() tools.register(ReadFileTool(...)) # 文件读取 tools.register(WriteFileTool(...)) # 文件写入 tools.register(EditFileTool(...)) # 文件编辑 tools.register(ListDirTool(...)) # 目录列表 tools.register(ExecTool(...)) # Shell 执行 tools.register(WebSearchTool(...)) # 网络搜索 tools.register(WebFetchTool(...)) # 网页获取Subagent 排除的工具❌MessageTool不能直接发送消息给用户❌SpawnTool不能派生更多子代理避免递归爆炸❌CronTool不能创建定时任务Main Agent 包含的工具# 除了上述 7 个工具外还包含 - MessageTool (发送消息给渠道) - SpawnTool (派生子代理) - CronTool (引用 cron_service) - MCPServersTool (连接 MCP 服务器) - SkillsTool (动态加载技能)3.2.2 System Prompt 完全不同区别Subagent 的提示是聚焦式的强调专注任务、不发起对话Main Agent 的提示是对话式的包含完整指南和记忆。Subagent 的系统提示def _build_subagent_prompt(self, task: str) - str: return f# Subagent ## Current Time {now} ({tz}) You are a subagent spawned by main agent to complete a specific task. ## Rules 1. Stay focused - complete only assigned task, nothing else 2. Your final response will be reported back to main agent 3. Do not initiate conversations or take on side tasks 4. Be concise but informative in your findings ## What You Can Do - Read and write files in workspace - Execute shell commands - Search web and fetch web pages - Complete task thoroughly ## What You Cannot Do - Send messages directly to users (no message tool available) - Spawn other subagents - Access main agents conversation history ...Main Agent 的系统提示# nanobot You are nanobot, a helpful AI assistant. ## Runtime {system} {machine}, Python {version} ## Workspace Your workspace is at: {workspace_path} ## nanobot Guidelines - State intent before tool calls... - Before modifying a file, read it first... ... (还会包含 Bootstrap Files、Long-term Memory、Skills 等)3.2.3 消息历史完全隔离区别Subagent 没有对话历史每次都是全新的开始Main Agent 有完整的会话记忆支持多轮对话。Subagent 的消息messages [ {role: system, content: subagent_system_prompt}, {role: user, content: task}, # 只有当前任务描述 ]Main Agent 的消息messages [ {role: system, content: main_system_prompt}, *history, # 完整的会话历史可能数百条 {role: user, content: runtime_context}, # 运行时元数据 {role: user, content: current_message}, ]3.2.4 最大迭代次数不同原因Subagent 执行的任务应该是相对独立的和快速的避免子代理运行过久占用资源。Agent最大迭代次数Main Agent40Subagent153.2.5 结果处理方式不同Subagent 的结果# Subagent 不直接发送结果给用户 final_result response.content await self._announce_result(task_id, label, task, final_result, origin, ok) # 通过 MessageBus 通知 Main AgentMain Agent 再转发给用户Main Agent 的结果# Main Agent 直接发送结果给用户 await self.bus.publish_outbound(OutboundMessage( channelmsg.channel, chat_idmsg.chat_id, contentfinal_content, ))3.2.6 会话管理方式不同Subagent没有 SessionManager 引用消息历史在内存中messages列表不持久化会话到磁盘不涉及记忆归档Main Agent有 SessionManager 引用会话持久化到 JSONL 文件支持 memory 归档到 MEMORY.md支持多轮对话记忆3.3 创建流程对比3.3.1 Main Agent 创建在 gateway()agent AgentLoop( busbus, # 共享 providerprovider, # 共享 workspaceworkspace, # 共享 modelconfig.agents.defaults.model, temperatureconfig.agents.defaults.temperature, max_tokensconfig.agents.defaults.max_tokens, max_iterations40, # 完整对话需要更多迭代 memory_windowconfig.agents.defaults.memory_window, session_managerSessionManager(...), # 独有会话持久化 cron_servicecron, # 独有定时任务服务 mcp_serversconfig.tools.mcp_servers, # 独有MCP 服务器 channels_configconfig.channels, # 独有渠道配置 )3.3.2 Subagent 创建在 SubagentManager.spawn()# 通过 _run_subagent() 内部直接创建不通过构造函数 async def _run_subagent(self, task_id, task, label, origin): # 工具集受限的 7 个工具 tools ToolRegistry() tools.register(ReadFileTool(...)) # ... (不含 MessageTool, SpawnTool, CronTool) # System Prompt聚焦式提示 system_prompt self._build_subagent_prompt(task) # 消息只有当前任务无历史 messages [ {role: system, content: system_prompt}, {role: user, content: task}, ] # 迭代次数15 次 max_iterations 15 # 循环执行 while iteration max_iterations: response await self.provider.chat(...) # 使用共享的 provider # ...0xFF 参考