OpenClaw 多智能体协作：Agent 间通信与子任务编排实战
引言
在 AI Agent 的实际生产中，单一 Agent 往往难以应对复杂任务——它需要并行处理、独立审查、任务委派和结果汇聚。OpenClaw 为此提供了一套完整的多智能体（Multi-Agent）协作框架，通过子代理（Subagent）机制，让多个 Agent 协同完成原本需要人工拆分的复杂工作流。
本文将深入剖析 OpenClaw 的多智能体架构，从核心原理到生产级实战，带你掌握 Agent 间通信与子任务编排的全部技巧。
1. 多智能体架构概览
OpenClaw 的多智能体系统基于三个核心概念：
| 概念 | 说明 | 典型用途 |
|---|---|---|
| Subagent（子代理） | 独立的 Agent 会话，由父会话派生 | 并行任务执行、专项处理 |
| sessions_spawn | 创建子会话的 API | 动态启动子任务 |
| agents_wait | 收集子代理结果的机制 | 汇聚并行任务输出 |
架构图
┌─────────────────────────────────────────┐
│ 主会话 (Main) │
│ ┌─────────┐ ┌─────────┐ ┌─────────┐ │
│ │ 数据处理 │ │ 代码审查 │ │ 测试生成 │ │
│ │ Subagent │ │ Subagent │ │ Subagent │ │
│ └────┬────┘ └────┬────┘ └────┬────┘ │
│ │ │ │ │
│ └────────────┼────────────┘ │
│ ▼ │
│ agents_wait 汇聚 │
│ │ │
│ ▼ │
│ 综合结果 → 用户 │
└─────────────────────────────────────────┘
2. sessions_spawn：子代理的创建与调度
2.1 基本语法
# sessions_spawn 核心参数
sessions_spawn(
agentId="my-specialist-agent", # 目标 Agent 配置
model="volc-coding/deepseek-v4", # 可选：覆盖模型
context="isolated", # isolated | fork
mode="run", # run（一次性）| persistent
cleanup="delete", # delete | keep
collect=True # 需要收集结果时设为 True
)
2.2 上下文模式选择
OpenClaw 提供两种子代理上下文模式，选择正确的模式至关重要：
isolated 模式（推荐）：
– 子代理拥有全新的纯净上下文
– 大上下文时节约 Token
– 适合独立任务（代码审查、测试生成、数据处理）
fork 模式：
– 子代理继承父会话的完整历史
– 需要上下文延续的场景（续写、深入分析）
– Token 消耗较高
2.3 并行子代理实战
以下是一个典型的并行任务编排示例——同时进行代码审查、安全扫描和文档生成：
# 同时启动三个独立子任务
sessions_spawn:
- agentId: "code-reviewer"
context: "isolated"
collect: true
runId: "review-001"
- agentId: "security-scanner"
context: "isolated"
collect: true
runId: "scan-001"
- agentId: "doc-generator"
context: "isolated"
collect: true
runId: "doc-001"
# 等待任意一个完成（非阻塞等待其他）
results = agents_wait(
runIds=["review-001", "scan-001", "doc-001"]
)
# 返回最先完成的结果，包含 pending 列表
3. agents_wait：子任务结果收集
3.1 工作模式
agents_wait 是子代理结果收集的核心工具，支持两种模式：
any 模式（默认）：
– 等待任意一个子代理完成即返回
– 返回值包含已完成结果 + 未完成 runId 列表
– 适合需要逐步处理结果的场景
all 模式：
– 等待所有指定子代理完成
– 全部完成后统一返回
– 适合结果间有依赖关系的场景
3.2 实战：渐进式结果处理
# 第一阶段：启动所有子代理
spawn_results = sessions_spawn([
{"agentId": "data-processor", "runId": "dp-1", "collect": True},
{"agentId": "data-processor", "runId": "dp-2", "collect": True},
{"agentId": "data-processor", "runId": "dp-3", "collect": True},
])
# 第二阶段：逐个收集并处理
pending = ["dp-1", "dp-2", "dp-3"]
processed = []
while pending:
result = agents_wait(runIds=pending)
# 处理已完成的结果
processed.append(process_result(result.completed))
# 更新待处理列表
pending = result.pending
# 最终汇总
final_report = aggregate(processed)
4. Subagent 通信模式
4.1 父→子：参数传递
父代理通过 sessions_spawn 的 prompt/instruction 参数向子代理传递任务：
sessions_spawn(
agentId="code-reviewer",
context="isolated",
instruction="""
请审查以下代码文件的安全性：
- 文件路径：/project/src/auth/login.js
- 重点检查：SQL注入、XSS、认证绕过
- 输出格式：Markdown 报告，按严重程度排序
""",
collect=True
)
4.2 子→父：结果回传
子代理完成后的输出通过 agents_wait 收集：
result = agents_wait(runIds=["review-001"])
# result.completed.structuredOutput 包含子代理的结构化输出
# result.completed.transcript 包含完整对话记录
4.3 子→子：间接通信
子代理之间不直接通信，但可以通过两种方式间接协作：
- 父代理中转：父代理收集子A的输出，处理后传递给子B
- 共享文件系统：子代理写入文件，其他子代理读取
# 管道模式：子A输出 → 父中转 → 子B输入
spawn("data-extractor", runId="extract-1") # 从网站提取数据
result_a = agents_wait(runIds=["extract-1"]) # 收集提取结果
spawn("data-analyzer", # 分析提取的数据
instruction=f"分析以下数据：n{result_a.output}",
runId="analyze-1"
)
result_b = agents_wait(runIds=["analyze-1"]) # 收集分析结果
5. 高级编排模式
5.1 扇出-扇入（Fan-out/Fan-in）
最经典的多智能体模式：一个任务拆分成N个子任务，并行执行后汇聚结果。
┌──→ Subagent A ──┐
Main ────┼──→ Subagent B ──┼──→ Main (汇总)
└──→ Subagent C ──┘
适用场景：
– 多文件并行代码审查
– 多渠道数据采集
– A/B 测试并行分析
5.2 管道模式（Pipeline）
子代理按序执行，每个子代理的输出成为下一个的输入。
Main → Subagent A → Subagent B → Subagent C → Main
适用场景：
– ETL 数据管道
– 多阶段文档处理
– 渐进式代码优化
5.3 竞争模式（Competing）
多个子代理同时解决同一个问题，取最优结果。
┌──→ Subagent A (方案1) ──┐
Main ────┼──→ Subagent B (方案2) ──┼──→ 比较 & 选择最优
└──→ Subagent C (方案3) ──┘
适用场景：
– 代码方案比较（不同架构思路）
– 多模型效果对比
– 创意生成（取最好创意）
5.4 监督模式（Supervisor）
一个监督 Agent 协调多个执行 Agent，根据执行结果动态调整策略。
# 监督 Agent 伪代码
def supervisor_loop(task):
plan = decompose(task) # 拆解任务
for subtask in plan:
result = spawn_and_wait(subtask) # 委托执行
if result.quality < threshold: # 质量检查
plan = adjust_plan(result) # 动态调整
continue
collect(result) # 收集结果
return synthesize() # 最终综合
6. 实战案例：自动化代码发布流水线
以下是一个完整的自动化代码发布流水线，展示了多智能体协作的完整流程：
# === 阶段1：扇出 - 并行检查 ===
spawn_results = sessions_spawn([
{
"agentId": "code-reviewer",
"runId": "lint-check",
"instruction": "对 /project/src/ 进行 lint 检查",
"collect": True
},
{
"agentId": "security-scanner",
"runId": "sec-scan",
"instruction": "对 /project/src/ 进行安全扫描",
"collect": True
},
{
"agentId": "test-runner",
"runId": "unit-test",
"instruction": "运行 /project/ 单元测试套件",
"collect": True
}
])
# === 阶段2：扇入 - 收集结果 ===
pending = ["lint-check", "sec-scan", "unit-test"]
checks = {}
while pending:
result = agents_wait(runIds=pending)
checks[result.runId] = result.output
pending = result.pending
# === 阶段3：决策 - 是否通过 ===
if all(passed(checks.values())):
# === 阶段4：管道 - 构建 → 部署 → 通知 ===
build_result = spawn_and_wait("build-agent",
instruction="构建并打包项目")
if build_result.success:
deploy_result = spawn_and_wait("deploy-agent",
instruction=f"部署构建产物 {build_result.artifact}")
# === 阶段5：通知 ===
notify(f"✅ 发布成功：v{deploy_result.version}")
else:
# 生成失败报告
failure_report = spawn_and_wait("report-agent",
instruction=f"生成检查失败报告：n{format_checks(checks)}")
notify(f"❌ 发布受阻：{failure_report.summary}")
7. 最佳实践与避坑指南
7.1 Token 预算管理
每个子代理消耗独立 Token。并行3个子代理意味着3倍的Token消耗：
# 设置合理的 Token 预算
sessions_spawn(
agentId="reviewer",
token_budget=8000, # 限制单个子代理的Token消耗
context="isolated" # 使用isolated减少上下文Token
)
7.2 错误处理策略
def safe_spawn_with_retry(agentId, instruction, max_retries=2):
"""带重试的子代理启动"""
for attempt in range(max_retries):
try:
result = sessions_spawn(agentId=agentId, ...)
return agents_wait(runIds=[result.runId])
except Exception as e:
if attempt == max_retries - 1:
raise # 最后一次重试仍失败，向上抛出
log_warning(f"子代理失败，重试 {attempt + 1}/{max_retries}")
7.3 避免无限等待
# 设置超时，避免永久阻塞
result = agents_wait(
runIds=pending,
timeout="5m" # 5分钟后超时
)
if result.timed_out:
# 取消未完成的子代理
for rid in result.pending:
subagents(action="cancel", runId=rid)
7.4 何时使用多智能体 vs 单智能体
| 场景 | 推荐方案 | 原因 |
|---|---|---|
| 简单线性任务 | 单Agent | 多Agent带来额外开销 |
| 需要并行处理 | 多Agent | 显著提升速度 |
| 需要独立审查 | 多Agent | 避免自我审查偏差 |
| 任务间有数据依赖 | 单Agent + 函数调用 | 数据传递更高效 |
| CPU/IO密集型混合 | 多Agent | 独立资源，不阻塞 |
8. 与 Skills 系统的协同
多智能体与 Skills 系统可以完美结合：
# 子代理加载不同 Skill 执行不同任务
spawn("research-agent",
instruction="请使用 web-crawler 技能抓取竞品价格数据",
enabled_skills=["web-crawler", "data-analysis"]
)
spawn("report-agent",
instruction="请使用报告模板生成竞品分析报告",
enabled_skills=["report-generator"]
)
总结
OpenClaw 的多智能体协作框架为复杂自动化任务提供了优雅的解决方案。通过 sessions_spawn + agents_wait 的组合，配合扇出-扇入、管道、竞争、监督等编排模式，你可以构建出企业级的 AI Agent 流水线。
关键要点回顾：
1. 选对上下文模式：独立任务用 isolated，需要历史用 fork
2. 合理使用并行：扇出-扇入是最实用的模式
3. 做好错误处理：超时、重试、取消机制一个都不能少
4. 控制 Token 成本：设置预算、使用 isolated 模式
5. 与 Skills 协同：让不同子代理加载不同专项技能