0x00 概要

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，无中央协调	任务天然可完全并行，无需统一收口

后一种模式的核心逻辑在于"路由分发"与"领域隔离"：

• 主 Agent（Orchestrator）：扮演"大脑"角色，仅负责意图识别与任务路由，判断"这个问题该交给谁"，而无需背负所有领域的知识重担。
• 子 Agent（Sub-Agent）：拥有独立的 Identity 空间，内化特定领域的专业知识。每个子 Agent 只需专注于解决某一类垂直场景，其 Prompt 指令更精简，领域知识更聚焦。

---

1.3 领域隔离

从上下文工程（Context Engineering）的角度看，SubAgent 实现了 Isolate 机制——上下文隔离。这种隔离通常由三种触发条件驱动：

1. 隔离噪声——避免失败路径或中间探索污染后续推理
2. 隔离关注点——让专业化的工具集各司其职，减少干扰
3. 突破物理限制——通过并行扩展单 Agent 的 Token 上限

如何识别需要拆分的信号？ 当以下现象出现时，便是架构调整的时机：

• 上下文窗口接近极限（表现为幻觉率上升、忽略早期指令）
• 工具集过大（频繁选错工具，且工具集内有明显的专业领域区分）
• 需要覆盖大信息空间（搜索覆盖面不足，单 Agent 无法遍历）

把任务拆给专业 Agent，让它在独立上下文中完成子任务再返回结果，相当于把上下文按职责切片。这不仅通常会更稳、更便宜，更是一种主动的上下文噪声隔离。

1.4 工作场景：Skills vs SubAgent，如何选择？

我们用操作系统类比来理解两者的定位差异：Skills 是应用程序，装在主系统里按需调用；SubAgent 是虚拟机，独立运行完再把结果交回来。

一句话总结选择逻辑：任务简单用应用，任务复杂开虚拟机。

维度	Skills	SubAgent
任务复杂度	简单，主 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 优势

1. 响应性：主 Agent 不会被耗时任务阻塞，保持与用户的实时交互
2. 并发性：多个子代理可以同时运行，执行不同任务
3. 隔离性：子代理有独立的工具集和限制，不会干扰主对话
4. 可取消性：通过 session_key 实现会话级的任务取消
5. 结果聚合：子代理结果通过主 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 子代理创建步骤

1. 生成唯一标识符

task_id = str(uuid.uuid4())[:8]  # 生成 8 字符的 UUID4，如 "a1b2c3d4"
display_label = label or task[:30] + ("..." if len(task) > 30 else "")

1. 记录原始来源

origin = {"channel": origin_channel, "chat_id": origin_chat_id}

用于后续将结果通知回正确的用户/渠道。

1. 创建并启动后台任务

bg_task = asyncio.create_task(
    self._run_subagent(task_id, task, display_label, origin)
)
self._running_tasks[task_id] = bg_task

创建异步任务来运行子代理，并将其注册到 _running_tasks 字典中。

1. 关联到会话

if session_key:
    self._session_tasks.setdefault(session_key, set()).add(task_id)

如果提供了 session_key，将 task_id 加入该会话的子代理集合。这使得 /stop 命令可以取消整个会话的所有子代理。

1. 设置清理回调

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 集合中移除
• 如果该会话没有剩余的子代理，删除会话集条目

1. 返回用户反馈

return f"Subagent [{display_label}] started (id: {task_id}). I'll 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(workspace=self.workspace, allowed_dir=allowed_dir))
tools.register(WriteFileTool(workspace=self.workspace, allowed_dir=allowed_dir))
tools.register(EditFileTool(workspace=self.workspace, allowed_dir=allowed_dir))
tools.register(ListDirTool(workspace=self.workspace, allowed_dir=allowed_dir))
tools.register(ExecTool(
    working_dir=str(self.workspace),
    timeout=self.exec_config.timeout,
    restrict_to_workspace=self.restrict_to_workspace,
    path_append=self.exec_config.path_append,
))
tools.register(WebSearchTool(api_key=self.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 agent's 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(
        messages=messages,
        tools=tools.get_definitions(),
        model=self.model,
        temperature=self.temperature,
        max_tokens=self.max_tokens,
    )

2.4.4. 处理工具调用

工具调用处理逻辑与主 Agent 类似：