Skip to content
今天更新

多 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 负责一个处理阶段。

text
输入 → Agent A → Agent B → Agent C → 输出
         ↓           ↓           ↓
      (处理1)    (处理2)    (处理3)

适用场景:内容生产流水线(撰写 → 审核 → SEO 优化 → 发布)

配置示例

json
{
  "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 同时处理不同的子任务,最终由汇总节点合并结果。

text
        ┌→ Agent A(任务1)→┐
输入 → 分发                   → 汇总 → 输出
        ├→ Agent B(任务2)→┤
        └→ Agent C(任务3)→┘

适用场景:市场调研(同时调研多个维度)、信息采集(多源并行搜索)

配置示例

python
# 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 各自拥有独立的工具集。

text
          主 Agent(协调者)
           /     |     \
      Agent A  Agent B  Agent C
       / \              / \
    工具1 工具2       工具3 工具4

适用场景:复杂项目管理、研究报告生成、自动化内容运营

配置示例

json
{
  "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

完整配置示例:

json
{
  "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 处理:

json
{
  "bindings": [
    // 来自 WhatsApp 的消息交给研究员处理
    { "agentId": "researcher", "match": { "channel": "whatsapp" } },
    // 来自 Telegram 的消息交给分析师处理
    { "agentId": "analyst", "match": { "channel": "telegram" } }
  ]
}

路由匹配按优先级从高到低:

  1. peer -- 精确匹配特定用户
  2. guildId -- 匹配群组 ID
  3. teamId -- 匹配团队 ID
  4. accountId -- 匹配账号 ID
  5. channel -- 匹配消息渠道
  6. 默认 Agent("default": true

工作空间隔离

每个 Agent 拥有独立的工作空间,互不干扰:

json
{
  "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.allowtools.deny 精细控制每个 Agent 可以使用的工具:

json
{
  "tools": {
    "allow": ["read", "write", "edit"],
    "deny": ["exec", "bash", "process"]
  }
}

OpenClaw 提供了便捷的工具组简写

工具组包含工具说明
group:runtimeexec / bash / process命令执行类
group:fsread / write / edit / apply_patch文件操作类
group:sessions会话管理相关会话管理类
group:memory记忆搜索相关记忆操作类
group:ui浏览器 / 画布界面交互类

使用工具组简写的示例:

json
{
  "id": "tracker",
  "name": "数据追踪",
  "tools": { "allow": ["read", "group:runtime"] }
},
{
  "id": "analyst",
  "name": "复盘分析",
  "tools": { "allow": ["read", "group:runtime", "group:fs"] }
}

deny 优先

allowdeny 冲突时,deny 优先。建议对敏感 Agent(如审核员)只开放 read 权限,确保其只能查看不能修改。

实战案例

案例一:研究报告系统

一个可以在 2 小时内生成专业研究报告的多 Agent 系统,原本需要 3-5 天。

架构设计

text
用户输入"生成关于[X]的市场研究报告"

  协调 Agent(任务拆解)
  ╱    ↓     ↓     ↓    ╲
搜索  竞品  数据  案例  整合
Agent Agent Agent Agent Agent
  ╲    ╱     ↓     ╲    ╱
   审核 Agent(质量把关)

   写作 Agent(最终成文)

   用户获得完整报告

协调 Agent 的 SOUL.md

markdown
你是一个研究项目协调员,负责将用户的研究需求拆解成可并行执行的子任务。

工作流程:
1. 接收用户的研究主题
2. 拆解成 5-8 个子任务(可以并行执行的部分)
3. 分配给对应专业 Agent
4. 追踪进度,汇总结果
5. 协调写作 Agent 生成最终报告

任务分配规则:
- 信息搜集类 → 搜索 Agent(可多个并行)
- 数据分析类 → 数据 Agent
- 竞品分析类 → 竞品 Agent
- 案例研究类 → 案例 Agent
- 最终撰写 → 写作 Agent

Agent 间通信实现

python
# 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 系统,每天自动产出内容并发布。

系统流程

text
每日 06:00

热点监控 Agent(扫描微博/Twitter热搜)

选题 Agent(评分热点,选出3个最适合的)

   ├→ 小红书写作 Agent → 自动发布
   ├→ 公众号写作 Agent → 存草稿(人工确认后发)
   └→ 抖音脚本 Agent → 生成脚本文件

数据追踪 Agent(每晚统计阅读/互动数据)

复盘 Agent(每周生成内容效果分析报告)

多 Agent 配置openclaw.json):

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" } }
    ]
  }
}

选题评分逻辑

python
# 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,导致上下文爆炸。

解决方案:使用共享内存 + 摘要传递。

python
# 错误做法:传递完整内容
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 机制:

python
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 倍,必须精细控制:

yaml
# 每个 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           # 每日最多 $20

Token 成本

Anthropic 的实践数据:普通聊天消耗 1x Token,单 Agent 约 4x,多 Agent 系统达 15x。拆分前务必权衡成本和收益。

命令行管理 Agent

bash
# 添加新 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
觉得有帮助?