多 Agent 协作
单个 Agent 能做的事有限,真正的力量来自多个 Agent 的协作。OpenClaw 原生支持多 Agent 架构,让你可以像组建团队一样编排多个专业 Agent,实现 1+1>10 的效果。
前置知识
阅读本页前,建议先了解 Agent 循环 中单个 Agent 的工作机制。本页聚焦于多个 Agent 之间的协作模式和配置方法。
为什么需要多 Agent
单个 Agent 存在四个核心局限:
| 局限 | 说明 |
|---|---|
| 上下文窗口有限 | 超长任务会遗忘早期信息 |
| 专业深度不足 | 一个通才 Agent 不如多个专才 Agent 组合 |
| 并行效率低 | 只能串行处理,速度慢 |
| 角色冲突 | 同一个 Agent 又要创作又要批评,容易自我妥协 |
多 Agent 架构带来的优势:
- 分工明确:每个 Agent 专注自己最擅长的部分
- 并行处理:多个 Agent 同时工作,效率倍增
- 相互验证:一个 Agent 的输出被另一个检查,提高质量
- 无限扩展:需求增加时添加新 Agent,而非让一个 Agent 更复杂
三种协作模式
Pipeline(流水线模式)
任务像流水线一样按顺序流经多个 Agent,每个 Agent 负责一个处理阶段。
输入 → Agent A → Agent B → Agent C → 输出
↓ ↓ ↓
(处理1) (处理2) (处理3)适用场景:内容生产流水线(撰写 → 审核 → SEO 优化 → 发布)
配置示例:
{
"agents": {
"list": [
{
"id": "writer",
"name": "写作师",
"workspace": "~/.openclaw/workspace-writer",
"model": "anthropic/claude-sonnet-4-5",
"tools": {
"allow": ["read", "write", "edit", "apply_patch"],
"deny": ["exec"]
}
},
{
"id": "reviewer",
"name": "审核员",
"workspace": "~/.openclaw/workspace-reviewer",
"model": "anthropic/claude-sonnet-4-5",
"tools": {
"allow": ["read"],
"deny": ["write", "edit", "exec"]
}
},
{
"id": "seo-optimizer",
"name": "SEO优化师",
"workspace": "~/.openclaw/workspace-seo",
"model": "anthropic/claude-sonnet-4-5"
}
]
}
}Parallel(并行模式)
多个 Agent 同时处理不同的子任务,最终由汇总节点合并结果。
┌→ Agent A(任务1)→┐
输入 → 分发 → 汇总 → 输出
├→ Agent B(任务2)→┤
└→ Agent C(任务3)→┘适用场景:市场调研(同时调研多个维度)、信息采集(多源并行搜索)
配置示例:
# openclaw-agents/coordinator.py
import asyncio
from openclaw import AgentPool, SharedMemory
async def run_parallel_research(topic: str):
# 初始化共享内存
memory = SharedMemory("research_project")
# 创建 Agent 池
pool = AgentPool()
# 并行执行搜索任务
search_tasks = [
pool.run("researcher", f"搜索{topic}的市场规模数据"),
pool.run("researcher", f"搜索{topic}的主要竞争玩家"),
pool.run("researcher", f"搜索{topic}的最新政策动向"),
pool.run("researcher", f"搜索{topic}的技术发展趋势"),
]
# 并行等待所有搜索完成
search_results = await asyncio.gather(*search_tasks)
# 存入共享内存
memory.set("search_results", search_results)Hierarchical(层级模式)
主 Agent 作为协调者,将任务分配给多个子 Agent,子 Agent 各自拥有独立的工具集。
主 Agent(协调者)
/ | \
Agent A Agent B Agent C
/ \ / \
工具1 工具2 工具3 工具4适用场景:复杂项目管理、研究报告生成、自动化内容运营
配置示例:
{
"agents": {
"list": [
{
"id": "coordinator",
"name": "协调员",
"default": true,
"workspace": "~/.openclaw/workspace-coordinator",
"model": "anthropic/claude-sonnet-4-5"
},
{
"id": "researcher",
"name": "研究员",
"workspace": "~/.openclaw/workspace-researcher",
"model": "anthropic/claude-sonnet-4-5"
},
{
"id": "analyst",
"name": "分析师",
"workspace": "~/.openclaw/workspace-analyst",
"model": "anthropic/claude-sonnet-4-5"
},
{
"id": "writer",
"name": "写作师",
"workspace": "~/.openclaw/workspace-writer",
"model": "anthropic/claude-sonnet-4-5"
}
]
}
}配置详解
openclaw.json 多 Agent 配置
多 Agent 配置写在 ~/.openclaw/openclaw.json 中。核心概念:
- 每个 Agent 拥有独立的工作区(workspace)、状态目录(agentDir)和会话存储
- 绑定规则(bindings)决定消息如何路由到对应 Agent
- 路由优先级:peer 匹配 > guildId > teamId > accountId > 渠道匹配 > 默认 Agent
完整配置示例:
{
"agents": {
"list": [
{
"id": "researcher",
"name": "研究员",
"default": true,
"workspace": "~/.openclaw/workspace-researcher",
"model": "anthropic/claude-sonnet-4-5"
},
{
"id": "analyst",
"name": "分析师",
"workspace": "~/.openclaw/workspace-analyst",
"model": "anthropic/claude-sonnet-4-5"
},
{
"id": "writer",
"name": "写作师",
"workspace": "~/.openclaw/workspace-writer",
"model": "anthropic/claude-sonnet-4-5",
"tools": {
"allow": ["read", "write", "edit", "apply_patch"],
"deny": ["exec"]
}
},
{
"id": "reviewer",
"name": "审核员",
"workspace": "~/.openclaw/workspace-reviewer",
"model": "anthropic/claude-sonnet-4-5",
"tools": {
"allow": ["read"],
"deny": ["write", "edit", "exec"]
}
}
],
"bindings": [
{ "agentId": "researcher", "match": { "channel": "whatsapp" } },
{ "agentId": "analyst", "match": { "channel": "telegram" } }
]
}
}Bindings 与路由
Bindings 决定来自不同渠道或用户的消息应该由哪个 Agent 处理:
{
"bindings": [
// 来自 WhatsApp 的消息交给研究员处理
{ "agentId": "researcher", "match": { "channel": "whatsapp" } },
// 来自 Telegram 的消息交给分析师处理
{ "agentId": "analyst", "match": { "channel": "telegram" } }
]
}路由匹配按优先级从高到低:
peer-- 精确匹配特定用户guildId-- 匹配群组 IDteamId-- 匹配团队 IDaccountId-- 匹配账号 IDchannel-- 匹配消息渠道- 默认 Agent(
"default": true)
工作空间隔离
每个 Agent 拥有独立的工作空间,互不干扰:
{
"agents": {
"list": [
{
"id": "researcher",
"workspace": "~/.openclaw/workspace-researcher",
"sandbox": { "mode": "off" }
},
{
"id": "writer",
"workspace": "~/.openclaw/workspace-writer",
"sandbox": {
"mode": "all",
"scope": "agent"
}
}
]
}
}sandbox 配置控制每个 Agent 的隔离级别:
"mode": "off"-- 不隔离,Agent 可以访问系统资源"mode": "all"-- 完全隔离,限制在 workspace 范围内"scope": "agent"-- 隔离范围为单个 Agent
工具权限控制
通过 tools.allow 和 tools.deny 精细控制每个 Agent 可以使用的工具:
{
"tools": {
"allow": ["read", "write", "edit"],
"deny": ["exec", "bash", "process"]
}
}OpenClaw 提供了便捷的工具组简写:
| 工具组 | 包含工具 | 说明 |
|---|---|---|
group:runtime | exec / bash / process | 命令执行类 |
group:fs | read / write / edit / apply_patch | 文件操作类 |
group:sessions | 会话管理相关 | 会话管理类 |
group:memory | 记忆搜索相关 | 记忆操作类 |
group:ui | 浏览器 / 画布 | 界面交互类 |
使用工具组简写的示例:
{
"id": "tracker",
"name": "数据追踪",
"tools": { "allow": ["read", "group:runtime"] }
},
{
"id": "analyst",
"name": "复盘分析",
"tools": { "allow": ["read", "group:runtime", "group:fs"] }
}deny 优先
当 allow 和 deny 冲突时,deny 优先。建议对敏感 Agent(如审核员)只开放 read 权限,确保其只能查看不能修改。
实战案例
案例一:研究报告系统
一个可以在 2 小时内生成专业研究报告的多 Agent 系统,原本需要 3-5 天。
架构设计:
用户输入"生成关于[X]的市场研究报告"
↓
协调 Agent(任务拆解)
╱ ↓ ↓ ↓ ╲
搜索 竞品 数据 案例 整合
Agent Agent Agent Agent Agent
╲ ╱ ↓ ╲ ╱
审核 Agent(质量把关)
↓
写作 Agent(最终成文)
↓
用户获得完整报告协调 Agent 的 SOUL.md:
你是一个研究项目协调员,负责将用户的研究需求拆解成可并行执行的子任务。
工作流程:
1. 接收用户的研究主题
2. 拆解成 5-8 个子任务(可以并行执行的部分)
3. 分配给对应专业 Agent
4. 追踪进度,汇总结果
5. 协调写作 Agent 生成最终报告
任务分配规则:
- 信息搜集类 → 搜索 Agent(可多个并行)
- 数据分析类 → 数据 Agent
- 竞品分析类 → 竞品 Agent
- 案例研究类 → 案例 Agent
- 最终撰写 → 写作 AgentAgent 间通信实现:
# openclaw-agents/coordinator.py
import asyncio
from openclaw import AgentPool, SharedMemory
async def run_research(topic: str):
memory = SharedMemory("research_project")
pool = AgentPool()
# 并行执行搜索任务
search_tasks = [
pool.run("researcher", f"搜索{topic}的市场规模数据"),
pool.run("researcher", f"搜索{topic}的主要竞争玩家"),
pool.run("researcher", f"搜索{topic}的最新政策动向"),
pool.run("researcher", f"搜索{topic}的技术发展趋势"),
]
search_results = await asyncio.gather(*search_tasks)
memory.set("search_results", search_results)
# 数据分析(依赖搜索结果,串行)
analysis = await pool.run(
"analyst",
"分析以下搜索结果,提取关键数据和趋势",
context={"search_results": search_results}
)
memory.set("analysis", analysis)
# 并行执行专项分析
specialized_tasks = [
pool.run("competitor_analyst", f"深度分析{topic}的竞争格局"),
pool.run("case_researcher", f"找出{topic}领域3个成功案例"),
]
specialized_results = await asyncio.gather(*specialized_tasks)
# 写作(依赖所有前序结果)
report_draft = await pool.run(
"writer",
"根据以下材料生成完整研究报告",
context={
"topic": topic,
"research": search_results,
"analysis": analysis,
"specialized": specialized_results,
}
)
# 审核
reviewed_report = await pool.run(
"reviewer",
"审核以下报告的质量和准确性",
context={"draft": report_draft}
)
return reviewed_report案例二:内容运营系统
一个真实可运行的内容运营多 Agent 系统,每天自动产出内容并发布。
系统流程:
每日 06:00
↓
热点监控 Agent(扫描微博/Twitter热搜)
↓
选题 Agent(评分热点,选出3个最适合的)
↓
├→ 小红书写作 Agent → 自动发布
├→ 公众号写作 Agent → 存草稿(人工确认后发)
└→ 抖音脚本 Agent → 生成脚本文件
↓
数据追踪 Agent(每晚统计阅读/互动数据)
↓
复盘 Agent(每周生成内容效果分析报告)多 Agent 配置(openclaw.json):
{
"agents": {
"list": [
{
"id": "coordinator",
"name": "内容协调员",
"default": true,
"workspace": "~/.openclaw/workspace-content-ops",
"model": "anthropic/claude-sonnet-4-5"
},
{
"id": "tracker",
"name": "数据追踪",
"workspace": "~/.openclaw/workspace-tracker",
"model": "anthropic/claude-sonnet-4-5",
"tools": { "allow": ["read", "group:runtime"] }
},
{
"id": "analyst",
"name": "复盘分析",
"workspace": "~/.openclaw/workspace-analyst",
"model": "anthropic/claude-sonnet-4-5",
"tools": { "allow": ["read", "group:runtime", "group:fs"] }
}
],
"bindings": [
{ "agentId": "coordinator", "match": { "channel": "whatsapp" } },
{ "agentId": "tracker", "match": { "channel": "telegram" } }
]
}
}选题评分逻辑:
# tools/topic_scorer.py
def score_topic(topic: dict, account_profile: dict) -> dict:
"""
对热点话题进行多维度评分
返回综合分数和是否推荐创作
"""
scores = {}
# 1. 热度分(0-30分)
heat_score = min(topic['heat'] / 1000000 * 30, 30)
scores['heat'] = heat_score
# 2. 相关性分(0-30分):与账号人设的匹配度
relevance = calculate_relevance(
topic['title'],
account_profile['keywords']
)
scores['relevance'] = relevance * 30
# 3. 竞争度分(0-20分):已有多少同类内容
competition = check_competition(topic['title'])
scores['competition'] = (1 - competition) * 20 # 竞争少得分高
# 4. 创作难度分(0-20分):账号有能力写吗
feasibility = assess_feasibility(topic, account_profile['expertise'])
scores['feasibility'] = feasibility * 20
total = sum(scores.values())
return {
'topic': topic['title'],
'totalScore': round(total, 1),
'scores': scores,
'recommendation': 'recommended' if total > 60 else 'skip',
'reason': generate_recommendation_reason(scores, total),
'suggestedAngle': suggest_content_angle(topic, account_profile),
}最佳实践
信息冗余策略
问题:每个 Agent 都把所有信息传给下一个 Agent,导致上下文爆炸。
解决方案:使用共享内存 + 摘要传递。
# 错误做法:传递完整内容
next_agent.run(context={"full_research": 50000_char_content})
# 正确做法:传递摘要 + 共享内存 ID
memory.store("research_full", full_content)
next_agent.run(context={
"research_summary": "200字摘要",
"full_research_id": "memory://research_full" # 需要时再取
})失败处理
Agent 可能因超时、API 错误等原因失败,需要 fallback 机制:
async def run_with_fallback(agent_name: str, task: str, fallback_agent: str):
try:
result = await pool.run(agent_name, task, timeout=60)
return result
except TimeoutError:
print(f"Agent {agent_name} 超时,尝试 {fallback_agent}")
return await pool.run(fallback_agent, task, timeout=120)
except AgentError as e:
print(f"Agent失败: {e}")
# 记录错误,让协调者决定如何处理
return {"error": str(e), "status": "failed"}成本控制
多 Agent 系统的成本是单 Agent 的 N 倍,必须精细控制:
# 每个 Agent 的成本上限
agents:
researcher:
costLimit:
perTask: 0.02 # 每个搜索任务最多 $0.02
daily: 1.0 # 每日最多 $1
writer:
costLimit:
perTask: 0.10 # 写作任务最多 $0.1
daily: 5.0 # 每日最多 $5
# 整体项目成本上限
project:
costLimit:
perProject: 0.50 # 每个研究项目最多 $0.5
daily: 20.0 # 每日最多 $20Token 成本
Anthropic 的实践数据:普通聊天消耗 1x Token,单 Agent 约 4x,多 Agent 系统达 15x。拆分前务必权衡成本和收益。
命令行管理 Agent:
# 添加新 Agent
openclaw agents add work
# 查看所有 Agent 及绑定关系
openclaw agents list --bindings
# 查看 Agent 状态
openclaw agent status my-agent
# 启动 / 停止 Agent
openclaw agent start my-agent
openclaw agent stop my-agent