openclaw平替之nanobot源码解析（六）：子智能体（Subagents）如何把任务拆给独立上下文

1. 从 openclaw 平替说起为什么需要子智能体如果你正在找 openclaw 的平替方案nanobot 的子智能体Subagents机制值得单独拎出来讲。它解决的问题很具体一个主 Agent 面对搜集 10 个网站信息并对比这类任务时如果全部串行处理用户要等很久而且中间任何一步出错都会污染主对话的上下文。子智能体的思路是把大任务拆成若干独立小任务每个小任务跑在自己的上下文里跑完只把结果回传给主 Agent。这套机制的核心价值在于上下文隔离。主 Agent 的对话历史里不会塞进子智能体读过的文件内容、执行过的 Shell 输出、搜索到的原始网页。子智能体在自己的上下文里消化这些噪音只把结论交回来。对 openclaw 平替场景来说这意味着你不需要引入 LangGraph、AutoGen 这类编排框架nanobot 用 asyncio 消息总线就实现了类似的多智能体协作边界。适合谁看已经在本地跑通 nanobot、想理解 spawn 工具背后发生了什么的人或者正在评估 openclaw 平替方案、关心多智能体协作实现成本的人。这篇会从源码链路切入给出可复制的配置片段和一次任务分发的验证步骤让你能亲眼看到上下文隔离的效果。先明确三个关键文件的位置后面所有分析都围绕它们展开nanobot/agent/tools/spawn.py # spawn 工具定义主 Agent 召唤分身的入口 nanobot/agent/subagent.py # SubagentManager子智能体生命周期管理 nanobot/agent/loop.py # 主循环异步任务调度的观察点子智能体不是主 Agent 的克隆。它是一个精简版 Agent 实例有读写文件、执行 Shell、网页搜索的能力但默认没有 spawn 工具禁止递归分身和 message 工具不能直接给用户发消息。它的 System Prompt 明确告诉它你是被派去完成特定任务的子智能体保持专注。这个设计边界很重要后面会展开。2. TaoToken 前置给子智能体接上模型能力子智能体要跑起来底层得有模型调用能力。nanobot 本身不绑定特定模型服务你可以通过 TaoToken 这类兼容 OpenAI 接口的服务来提供模型。官网在 https://taotoken.net/?utm_sourcetaotoken_aicg_blog_endutm_mediumcsdnutm_campaignrewriteutm_content API 端点是 https://taotoken.net/api 。为什么在源码解析里要提这个因为子智能体的独立上下文意味着它会发起独立的模型请求。主 Agent 一次对话可能触发 3 到 5 个子智能体每个子智能体都有自己的 System Prompt 和任务描述这些都会消耗 token。如果你用本地小模型子智能体的推理质量会明显下降如果用统一的大模型服务需要确保并发请求不会互相阻塞。TaoToken 的接入方式对 nanobot 比较友好因为它兼容 OpenAI 的 chat completions 格式。你需要在 nanobot 的配置里指定 Base URL、API Key 和 Model ID 三件套。先拿到 API Key访问 https://taotoken.net/api-keys 创建然后确认你要用的模型 ID。模型对话页面在 https://taotoken.net/models 可以查看可用模型。这里有个实际经验子智能体的任务通常比主对话更重因为它要读文件、跑命令、搜网页。建议给子智能体配一个上下文窗口足够大的模型否则任务描述加上工具返回结果很容易撑爆窗口。主 Agent 可以用轻量模型做调度子智能体用能力更强的模型做执行这样成本和质量能平衡。配置写在哪里nanobot 的模型配置通常在项目根目录的配置文件里或者通过环境变量注入。下面给出一个可复制的配置片段路径按你的实际项目结构调整。如果你用的是 Claude Code 风格的配置settings 文件里需要包含完整的 Base URL、Key 和 Model ID。{ model: { provider: openai-compatible, base_url: https://taotoken.net/api, api_key: sk-your-key-here, model_id: claude-sonnet-4-20250514, max_tokens: 8192, temperature: 0.3 }, subagent: { enabled: true, max_concurrent: 3, timeout_seconds: 300 } }注意max_concurrent这个参数。子智能体是异步并发的如果你同时 spawn 太多模型服务的并发限制可能触发 429。实测下来 3 到 5 个并发比较稳。timeout_seconds也要设否则某个子智能体卡在网页请求上会一直占着资源。如果你用的是 TOML 格式的配置等价写法如下[model] provider openai-compatible base_url https://taotoken.net/api api_key sk-your-key-here model_id claude-sonnet-4-20250514 max_tokens 8192 temperature 0.3 [subagent] enabled true max_concurrent 3 timeout_seconds 300配置完成后先别急着跑子智能体。用一次普通对话验证模型连通性确认主 Agent 能正常回复。如果主 Agent 都调不通子智能体肯定也跑不起来。验证方法在第四节展开。3. 可复制配置spawn 工具与 SubagentManager 参数这一节直接给可复制的配置和源码关键片段。先看 spawn 工具的定义它接收两个参数task和label。task是具体任务描述label是任务标签用于日志和结果回传时标识是哪个子智能体。# nanobot/agent/tools/spawn.py 关键结构 class SpawnTool: name spawn description Spawn a subagent to handle a complex or time-consuming task. parameters { task: { type: string, description: The specific task for the subagent to complete. }, label: { type: string, description: A short label to identify this subagent in logs and results. } } async def execute(self, task: str, label: str) - str: task_id self.manager.create_task(task, label) return fSubagent {label} spawned with task_id{task_id}SubagentManager 负责生命周期管理。它为每个子智能体生成唯一task_id用asyncio.create_task在后台启动异步任务。这意味着子智能体在后台跑的时候主 Agent 可以继续和用户聊天或者继续 spawn 更多子智能体。# nanobot/agent/subagent.py 关键结构 class SubagentManager: def __init__(self, bus, model_config): self.bus bus self.model_config model_config self.tasks {} def create_task(self, task: str, label: str) - str: task_id fsub-{uuid4().hex[:8]} self.tasks[task_id] { label: label, task: task, status: running } asyncio.create_task(self._run_subagent(task_id, task, label)) return task_id async def _run_subagent(self, task_id, task, label): # 构建独立上下文 messages [ {role: system, content: SUBAGENT_SYSTEM_PROMPT}, {role: user, content: task} ] # 独立模型调用 result await self._call_model(messages) # 结果回传 await self._announce_result(task_id, label, task, result)子智能体的 System Prompt 是隔离的关键。它明确告诉模型你是子智能体专注完成任务最终回复会报告给主 Agent。这个 Prompt 里还包含了工作区路径和可用技能列表。注意技能列表里有availabletrue和availablefalse的区分后者表示依赖的 CLI 工具没装。# Subagent System Prompt 核心片段 You are a subagent spawned by the main agent to complete a specific task. Stay focused on the assigned task. Your final response will be reported back to the main agent. ## Workspace .nanobot ## Skills Read SKILL.md with read_file to use a skill.子智能体的工具箱和主 Agent 有区别。它有读写文件、执行 Shell、网页搜索但没有 spawn禁止递归和 message不能直接给用户发消息。这个边界设计防止了无限套娃也保证了用户只和主 Agent 交互。配置片段方面如果你想让子智能体使用特定的模型可以在配置里单独指定{ subagent: { enabled: true, max_concurrent: 3, timeout_seconds: 300, model_override: { model_id: claude-sonnet-4-20250514, max_tokens: 8192 }, allowed_tools: [read_file, write_file, shell, web_search], denied_tools: [spawn, message] } }allowed_tools和denied_tools是白名单和黑名单机制。默认情况下 spawn 和 message 在黑名单里。如果你有特殊需求比如允许子智能体再 spawn 一层可以调整但强烈不建议递归失控的调试成本很高。4. 验证请求跑通一次任务分发并观察上下文隔离配置写好后怎么验证子智能体真的在独立上下文里跑最直接的方法是给主 Agent 一个需要拆分的任务然后观察日志和结果。启动 nanobot 后在主对话里输入类似这样的任务帮我搜集三个不同来源的天气信息分别整理成简短摘要最后汇总对比。主 Agent 应该会调用 spawn 工具创建多个子智能体。你会在日志里看到类似输出[INFO] SpawnTool: spawning subagent labelweather-source-1 [INFO] SubagentManager: created task_idsub-a1b2c3d4 [INFO] Subagent [weather-source-1] starting task [INFO] SpawnTool: spawning subagent labelweather-source-2 [INFO] SubagentManager: created task_idsub-e5f6g7h8 [INFO] Subagent [weather-source-2] starting task [INFO] Subagent [weather-source-1] completed successfully [INFO] Subagent [weather-source-2] completed successfully关键观察点主 Agent 的思考和子智能体的执行在日志里是交织的。因为 asyncio 的异步特性主 Agent 在等模型返回的时候事件循环会切到子智能体执行子智能体等模型返回的时候又切回主 Agent。你可以在 debug 控制台的 Threads Variables 里看当前执行的是loop.py还是subagent.py。想更清楚地看到上下文隔离在subagent.py的 while 循环里打个断点。当子智能体执行到断点时检查它的messages变量你会发现里面只有子智能体的 System Prompt 和任务描述没有主 Agent 的对话历史。这就是上下文隔离的直接证据。结果回传走的是 MessageBus。子智能体完成后调用_announce_result把结果封装成 System 消息发到 inbound 队列# nanobot/agent/subagent.py 结果回传 async def _announce_result(self, task_id, label, task, result, origin, status): 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. msg InboundMessage( channelsystem, sender_idsubagent, chat_idf{origin[channel]}:{origin[chat_id]}, contentannounce_content, ) await self.bus.publish_inbound(msg)为什么发给 inbound因为 inbound 是 Agent 的输入端。主 Agent 像接收普通用户消息一样接收到这条系统消息然后触发新一轮思考总结结果并回复用户。整个链路里用户只看到主 Agent 的最终回复子智能体的中间过程完全隔离。验证成功的标志主 Agent 回复里包含了三个来源的天气摘要和对比但你在主对话历史里找不到子智能体读过的原始网页内容。这说明上下文隔离生效了。5. 常见错排查401、local proxy failed 与 reading choices子智能体跑不起来报错通常集中在几个地方。这一节对照真实报错给排查路径。401 Unauthorized模型服务认证失败。检查配置里的api_key是否正确Base URL 是否指向 https://taotoken.net/api 。如果你用的是环境变量注入确认变量名和代码里读取的一致。子智能体用的是独立的模型调用如果主 Agent 能跑但子智能体报 401检查subagent.model_override里是否覆盖了错误的 key。local proxy failed这个报错通常出现在网络请求层。检查你的运行环境是否能正常访问模型服务端点。如果你在容器里跑 nanobot确认容器的网络配置允许出站请求。另外检查timeout_seconds是否设得太短子智能体的任务可能涉及多次模型调用和工具执行超时太短会误报。reading choices 相关报错这通常意味着模型返回的响应格式不符合预期。子智能体的 System Prompt 要求它返回最终结果但如果模型返回了工具调用格式而子智能体没有对应工具就会解析失败。检查allowed_tools配置确保子智能体声明的工具和实际注册的工具一致。如果你在model_override里换了模型确认新模型支持你用的工具调用格式。OAuth 相关报错如果你用的是需要 OAuth 的模型服务子智能体的独立请求可能没有携带正确的 token。检查 token 刷新逻辑是否对子智能体生效。这种情况下建议改用 API Key 认证避免 OAuth 的复杂性。子智能体卡住不返回检查max_concurrent是否设得太高导致模型服务限流。另外在subagent.py的 while 循环里加日志看子智能体执行到哪一步卡住。常见原因是网页搜索请求没有超时设置一直挂着。结果没有回传给主 Agent检查 MessageBus 的 inbound 队列是否正常。如果子智能体完成了但主 Agent 没反应可能是_announce_result里的origin信息不对导致消息发到了错误的 chat_id。在_announce_result里加日志确认chat_id和当前会话匹配。排查时有个通用技巧先把max_concurrent设为 1只跑一个子智能体排除并发干扰。单个能跑通后再逐步增加并发。另外子智能体的日志级别调到 DEBUG能看到完整的模型请求和响应内容。6. 多智能体协作的边界与后续nanobot 的子智能体系统展示了如何用异步编程加消息总线构建协作模型。它不依赖复杂编排框架通过工具调用和系统消息注入就让 AI 有了并行处理复杂任务的能力。对 openclaw 平替场景来说这套机制的实现成本很低核心代码集中在 spawn.py 和 subagent.py 两个文件。但边界也很明确。子智能体默认不能递归不能直接和用户交互这限制了它的协作深度。如果你需要更复杂的多智能体拓扑比如子智能体之间互相通信nanobot 当前的设计不支持需要自己扩展 MessageBus 的路由逻辑。另一个边界是上下文隔离的代价。每个子智能体都是独立的模型调用token 消耗会随子智能体数量线性增长。任务拆分粒度太细调度开销会超过并行收益。实测下来单个子智能体的任务执行时间在 10 秒以上时并行才有明显收益。如果你想把子智能体接到长期编码或 Agent 工作流里可以了解 Coding Plan 方案https://taotoken.net/coding-plan?utm_sourcetaotoken_aicg_blog_endutm_contentcoding-planutm_campaignrewrite 。接入文档在 https://taotoken.net/doc?utm_sourcetaotoken_aicg_blog_endutm_contentdocutm_campaignrewrite 里面有完整的 Base URL、Key 和 Model ID 配置说明。调试子智能体时记住在subagent.py的 while 循环里打断点这是观察独立上下文最直接的方式。日志里Subagent [...] starting task和completed successfully之间的内容就是子智能体在自己的上下文里做的事主 Agent 看不到这些细节只收到最终结果。