第 13 篇 · 多 Agent：协作原理与任务委派实现

当Agent需要处理的任务变复杂后，单个 Agent 既要理解目标、拆分任务，又要检索资料、调用工具、检查结果和组织答案。

所有步骤共用同一份 messages 消息数组、同一组工具和同一个执行状态，就很容易出现上下文膨胀、注意力分散和局部失败相互影响等问题。

多 Agent 系统提供了另一种组织方式，把复杂目标交给多个相对独立的 Agent 执行单元，通过明确的分工、通信协议和协调机制共同完成任务。

我们假设用户提出一个任务：

找出最近一周修改过 services/ 这个目录下面代码 的所有 PR，并逐个总结改动与风险，最后生成一份统一报告。

多 Agent系统可以让一个协调者Agent负责拆解与汇总，让多个执行者Agent分别分析 PR，每个执行者只处理自己的局部上下文，最后把标准化结论交回协调者。

本文分成两部分：

1. 先介绍什么是多 Agent、为什么需要多 Agent、常见实现方式和关键工程问题
2. 再沿着源码说明 mini-openclaw 如何用 主 Agent 委派子 Agent 实现一种具体的多 Agent 架构。

第一部分：多 Agent 的原理

什么是多 Agent

多 Agent 系统包含多个能够独立运行的 Agent 执行单元，它们围绕同一个目标分工协作。

一个 Agent 执行单元通常包含以下几个部分：

• 自己的任务或角色
• 独立的上下文或状态
• 可调用的模型、工具与 Skill
• 接收任务和返回结果的通信接口
• 明确的启动、运行和终止条件

Agent 可以使用不同模型，也可以复用同一个模型配置。判断一个系统是否属于多 Agent， 关键在于它是否包含多个边界清晰、可自主循环运行的 Agent，以及这些 Agent 是否具备彼此通信、协同推进任务的能力。

为什么需要多 Agent

多 Agent 关注的是复杂任务的组织方式，即使单个Agent有能力完成任务，拆分执行仍可能让流程更可靠、更容易控制。

1. 分解复杂任务

复杂目标往往包含检索、分析、实现、测试和总结等不同阶段。把它们拆成输入和输出明确的子任务，可以降低每个 Agent 单次决策的难度。

2. 隔离上下文

假设分析 5 个 PR，每个 PR 的 diff 和工具结果占 4,000 token。最终报告只需要每个 PR 的结论，不需要把所有原始 diff 长期保留在协调者的 messages 中。

独立 Agent 可以在自己的上下文中处理大量中间材料，只把摘要、证据和风险返回给协调者。这种做法控制了各执行单元的上下文规模，但整个系统的 token 总消耗未必会减少。

3. 专业化能力

不同 Agent 可以使用不同的 system prompt、模型、工具或 Skill。例如：

• 检索 Agent 只负责查找资料并给出来源
• 代码 Agent 只负责修改指定目录
• 审查 Agent 不写代码，只检查风险
• 汇总 Agent 不访问外部系统，只整合已有结果

专业化可以减少无关工具和指令对模型的干扰。

4. 并发执行

相互独立的子任务可以同时运行，处理 10 个互不依赖的文件时，fan-out 并行通常比一个 Agent 依次处理更快。

多 Agent 也可能串行执行，异步调度、并发执行和结果汇聚机制共同决定了这类任务能否获得时延收益。

5. 故障与权限隔离

一个局部任务失败时，协调者可以单独重试、降级或跳过，主流程的其他状态仍然保留。每个 Agent 也可以只获得完成任务所需的最小权限，降低误操作范围。

多 Agent 同样有代价，它会增加模型调用、通信和调度开销，还会引入结果冲突、重复劳动、共享状态一致性和错误传播等新问题。使用多 Agent 前，需要确认分工收益足以覆盖这些成本。

多 Agent 系统由什么组成

一个可工作的多 Agent 系统通常包含五个层面：

只创建多个模型调用还不够。没有任务边界、通信协议和终止条件，多个 Agent 很容易重复工作或互相覆盖结果。

常见的多 Agent 实现方式

多 Agent 没有唯一架构。下面几种方式可以单独使用，也可以组合。

1. Supervisor–Worker：监督者与执行者

一个 Supervisor 理解总目标、拆分任务、选择 Worker 并汇总结果。Worker 只处理被分配的局部任务。

它的优点是控制关系和用户入口清晰，便于统一权限、预算和最终输出。缺点是 Supervisor 可能成为上下文、性能和决策瓶颈。

2. Pipeline：流水线

任务按照固定阶段传递，前一个 Agent 的输出成为后一个 Agent 的输入：

检索 Agent → 分析 Agent → 写作 Agent → 审查 Agent

流水线适合步骤稳定、依赖关系明确的任务，容易测试和重放。缺点是上游错误会沿链路传播，流程也不擅长处理动态变化。

3. Router–Experts：路由器与专家

Router 先判断任务类型，再把请求交给一个或多个专业 Agent。例如把问题路由给代码、数据、法务或客服 Agent。

这种方式的重点是选择合适的执行者，有些请求会被完整交给一个专家处理。它适合请求类型多、专家边界清晰的系统。主要风险是路由错误，以及跨领域任务被分给单一专家。

4. Fan-out / Fan-in：并行分发与汇聚

调度器把同类或互相独立的任务同时发给多个 Agent，等待结果后统一聚合。例如并行分析多个 PR，或让多个 Agent 独立提出方案，再由评审 Agent 比较。

它可以降低批量任务的总时延，也能用多份独立答案提高覆盖率。代价是调用成本更高，并且需要处理超时、部分失败、重复结果和冲突合并。

5. Peer-to-Peer / Shared Workspace：对等协作或共享工作区

这种架构不设唯一的 Supervisor。各 Agent 通过消息、事件队列、黑板或共享工作区协作：一个 Agent 发布任务，其他 Agent 认领任务或继续处理。

这种方式适合开放式、动态协作，但工程复杂度最高。系统必须处理并发写入、状态一致性、任务抢占、死锁、重复消费和全局终止等问题。

6. Hierarchical Teams：分层团队

顶层 Agent 管理多个组长，组长继续管理各自的执行者。它适合规模较大的任务树，但链路越深，信息损失、预算失控和故障定位就越困难。

Agent 之间如何通信

Agent 协作常见有三种通信方式：

1. 消息传递：一个 Agent 把任务或结果直接发送给另一个 Agent；
2. 结构化任务与结果：通过固定 schema 传递状态、证据、风险和下一步建议；
3. 共享状态：多个 Agent 读写同一个数据库、文件工作区、任务队列或记忆系统。

直接把原始对话全文交给另一个 Agent，通常会带入无关信息、扩大上下文，也让接收方难以区分事实、指令和中间推理。

一般可以按照下面这个格式来处理：

返回结果也应尽量结构化，至少包含状态、结论、证据或发现、风险、缺失信息和下一步建议。结构化协议让协调者可以判断成功与失败，也便于持久化、观测和自动评估。

多 Agent 的关键工程问题

上下文边界

上下文太少，执行者无法完成任务，上下文太多，又会失去隔离意义。通常只传递任务必需事实、输入引用、约束和验收标准，不直接复制协调者的完整 messages。

权限边界

委派不能成为权限升级通道，层级委派中通常要求：

Worker 权限 ⊆ Supervisor 权限

并进一步按照最小权限原则，只分配本次子任务需要的能力。

调度与状态

系统需要知道任务处于等待、运行、成功、失败、取消还是超时状态。并行任务还要处理结果顺序、部分成功和重试幂等性。

终止与预算

每个 Agent 都可能重复搜索或调用工具。系统应限制单 Agent 轮数、总委派数、递归深度、token、费用和执行时间，并定义全局完成条件。

结果可信度

多个 Agent 不会自动带来正确答案，系统仍要校验证据、处理相互冲突的结论，并防止下游把上游的猜测当成事实。常见做法包括 schema 校验、评审 Agent、规则校验和外部测试。

可观测性

一次用户请求可能产生多条 Agent 运行链路。事件与 Trace 至少要能回答：

• 谁创建了哪个任务
• 每个 Agent 收到了什么上下文和权限
• 调用了哪些模型与工具
• 结果如何传递
• 哪个步骤失败、超时或消耗过高

什么时候不需要多 Agent

以下任务通常留在单 Agent 中更合适：

• 问题简单，一次模型循环即可完成
• 子任务高度耦合，需要频繁共享不断变化的状态
• 无法写清子任务的输入、边界和验收标准
• 委派出去的 context 几乎等于完整父对话
• 多次调用和协调成本高于分工收益

可以先问两个问题：

任务能否拆成 输入明确、输出明确 的独立单元？

拆分带来的隔离、专业化或并发收益，是否大于通信与调度成本？

如果答案是否定的，不必为了使用多 Agent 而增加 Agent。

第二部分：mini-openclaw 如何实现多 Agent

前面介绍的是通用原理。下面回到 mini-openclaw，沿着一次 delegate_task 调用实际经过的顺序阅读源码。

源码结构与完整调用链

相关代码分散在多个文件中，每个文件负责委派流程的一部分：

一次委派的主调用链如下：

chat_service.stream_chat()
    → 主模型返回 tool__delegate_task
    → tool_executor.dispatch()
    → delegate_task_tools.execute()
    → delegation_service.delegate_task()
    → delegation_service._run_delegated_task()
    → delegation_service._execute_child_agent()

子 Agent 如果需要调用工具，会在 _execute_child_agent() 内运行自己的工具循环：

子模型返回 tool call
    → tool_executor.dispatch()
    → 具体工具或 Skill
    → 工具结果追加到子 messages
    → 再次调用子模型

子模型不再返回工具调用时，结果沿原路径返回：

子模型最终文本
    → _normalize_child_result()
    → DelegateTaskResult
    → 序列化为 JSON 字符串
    → 成为主 Agent 的 tool result
    → 主模型继续生成最终答复

后面的代码讲解按这三段展开：主 Agent 发起委派、服务层创建并运行子 Agent、结果回到主 Agent。

把上面三段调用链合在一起，可以得到一次委派的完整流程：

delegate_task_tools.execute() 是工具层与委派服务层的边界，_execute_child_agent() 是委派准备阶段与子 Agent 独立循环的边界。后面逐步阅读源码时，可以用这两个位置判断当前代码仍在主循环中，还是已经进入子循环。

第一步：让主 Agent 知道何时可以委派

backend/app/services/delegation_service.py 定义了 ORCHESTRATION_SYSTEM_PROMPT，告诉主 Agent：

• 哪些情况适合或不适合使用子 Agent
• 创建子 Agent 时应该提供哪些字段
• 子权限不能超过父权限
• 子 Agent 不直接回复最终用户
• 主 Agent 必须整合结果后再回答

backend/app/services/chat_service.py 的 get_agent_system_prompt() 会把这段规则加入主 Agent 的 system prompt：

sections.append(ORCHESTRATION_SYSTEM_PROMPT)

提示词只负责告诉模型“什么时候适合委派”，模型还需要在本轮请求的 tools 中看到委派工具，才能发起调用。

delegate_task_tools.py 把该工具声明为默认可用的系统工具：

"id": "delegate_task",
"is_system": True,
"configurable": False,
"default_available": True,

chat_service.stream_chat() 调用 build_tool_definitions() 时，系统工具会进入 openai_tools，工具构建器统一添加 tool__ 前缀，所以模型看到的函数名是：

tool__delegate_task

主 Agent 同时拿到了编排规则和工具定义，前者说明使用条件，后者说明工具调用参数。

第二步：主模型发起 delegate_task 工具调用

backend/app/tools/delegate_task_tools.py 定义了五个参数：

{
    "task": "子 Agent 要完成的具体任务",
    "context": "必要上下文",
    "tools": ["允许的工具"],
    "skills": ["允许的技能"],
    "expected_output": "期望输出格式"
}

schema 只强制要求 task，其余字段都有默认值。缺少 context 或 expected_output 时，请求仍能进入服务层，子 Agent 获得的信息也会相应减少。

主模型返回这个函数调用后，chat_service.stream_chat() 先交给统一分发器：

result = dispatch(
    func_name,
    args,
    registry,
    session_id=session_id,
    execution_context=execution_context,
    tool_call_id=tc["id"],
)

registry 把 tool__delegate_task 映射回工具 ID delegate_task，随后调用链进入 delegate_task_tools.execute()。

工具的 execute() 不直接运行模型，只把参数和当前执行上下文交给 delegation_service.delegate_task()：

return delegate_task(
    args,
    session_id=kwargs.get("session_id"),
    execution_context=kwargs.get("execution_context"),
    tool_call_id=kwargs.get("tool_call_id"),
)

这里的三项附加参数用途如下：

• session_id：记录事件、创建子任务和确定文件输出目录；
• execution_context：保存父 Agent 的 ID 及权限配置；
• tool_call_id：关联主 Agent 的这次工具调用与整条委派链路。

第三步：校验请求并登记委派

backend/app/schemas/delegation.py 定义：

class DelegateTaskRequest(BaseModel):
    task: str
    context: str = ""
    tools: list[str] = Field(default_factory=list)
    skills: list[str] = Field(default_factory=list)
    expected_output: str = ""

delegate_task() 先执行：

request = DelegateTaskRequest.model_validate(args)

参数类型不合法时，服务直接返回 Error: delegate_task 参数不合法，不会启动子模型。当前模型没有为 task 设置 min_length，所以空字符串仍能通过 Pydantic 校验。随后还会检查当前执行上下文是否包含 agent_id，因为子 Agent 需要继承父 Agent 的模型配置。

校验通过后，delegate_task() 生成 delegation_id。通常直接复用主模型生成的 tool_call_id；没有该 ID 时才创建 UUID。

如果存在 session_id，系统接着做两项登记：

1. 写入 delegation_start 事件，保存任务、上下文、工具、Skill 和父 Agent ID；
2. 在根任务下创建一个状态为 running 的子任务。

子任务保存 delegation_id、tool_call_id，并在 attributes 中记录 expected output、工具和 Skill。后面的Trace 和任务看板据此关联主 Agent 的工具调用与子 Agent 的执行记录。

创建任务记录的代码包在 try/except 中。任务树写入失败时，child_task_id 会变成 None，子 Agent 仍会继续运行。任务记录属于观测能力，不参与子任务的业务执行。

登记完成后，delegate_task() 调用 _run_delegated_task()。后续的权限校验、运行环境构造和子模型循环都从这里开始。

第四步：从父权限中解析子权限

4.1 计算父 Agent 的有效权限

_collect_parent_permissions() 使用父执行上下文重新构建工具 registry：

_, registry = build_tool_definitions(
    execution_context.tool_ids,
    execution_context.skill_ids,
)

随后从 registry 中提取 tool ID 和 skill ID。build_tool_definitions() 会把系统工具、强制工具和 Agent 配置中的工具一起加入 registry，权限校验以这份结果为准。

这里需要区分“配置权限”和“有效权限”。父 Agent 配置文件中的 tool_ids 只是一部分，系统工具和强制工具也会进入主 Agent 的 registry。因此，子权限的上限以父 Agent 实际能够调用的工具为准。