🤝 OpenClaw Agent协作协议

世界上有一种默契叫Agent协作，它们不是各干各的，而是像一支球队，传球、配合、共同完成目标...

"帮我做竞品分析"，用户一个请求，背后却是三个Agent在干活：搜索Agent找资料、分析Agent做对比、报告Agent写报告。它们之间怎么沟通？谁来协调？这就是Agent协作协议要解决的问题。

📋 功能介绍

🎯 A2A（Agent-to-Agent）协议核心

💡 协作模式

• 主从模式 - 主Agent指挥子Agent
• 对等模式 - Agent之间平等协作
• 链式模式 - Agent A → B → C，流水线
• 网状模式 - Agent之间自由通信
• 竞争模式 - 多个Agent竞争最优结果

🚀 使用方法

1. 任务委托配置

# 主Agent配置
agents:
  coordinator:
    name: 协调员
    
    # 可委托的Agent
    subordinates:
      - searcher    # 搜索Agent
      - analyzer    # 分析Agent
      - reporter    # 报告Agent
      
    # 委托配置
    delegation:
      # 任务分配策略
      strategy: match_capability  # 按能力匹配
      
      # 委托超时
      timeout: 300s
      
      # 失败重试
      retry: 2
      
    prompt: |
      作为协调员，你负责分解任务并委托给合适的Agent。
      
      收到任务后：
      1. 分析任务需要哪些能力
      2. 分解为子任务
      3. 委托给对应的Agent
      4. 汇总结果
      
      可委托的Agent：
      - searcher: 搜索信息
      - analyzer: 分析数据
      - reporter: 生成报告

2. A2A消息格式

# 标准化的Agent间消息
messages:
  # 任务请求
  task_request:
    type: request
    from: coordinator
    to: searcher
    payload:
      task: "搜索竞品A和B的信息"
      context:
        products: ["产品A", "产品B"]
        fields: ["价格", "功能", "评价"]
      deadline: 120s
      
  # 任务响应
  task_response:
    type: response
    from: searcher
    to: coordinator
    payload:
      status: success
      result:
        产品A: {价格: ¥99, 功能: [...], 评价: 4.5}
        产品B: {价格: ¥149, 功能: [...], 评价: 4.2}
        
  # 错误通知
  error_notify:
    type: error
    from: searcher
    to: coordinator
    payload:
      error_type: timeout
      message: "搜索超时，未完成"
      suggestion: "请缩小搜索范围"

3. 状态同步配置

# Agent状态同步
agents:
  team-member:
    collaboration:
      # 共享状态
      shared_state:
        enabled: true
        storage: memory  # 或 redis
        
        # 共享内容
        share:
          - progress      # 进度百分比
          - partial_result  # 中间结果
          - errors        # 遇到的错误
          
      # 状态更新频率
      update_interval: 10s
      
      # 状态订阅
      subscribe:
        - team_progress  # 团队整体进度
        - leader_status  # 主Agent状态

4. 协作工作流

# 协作工作流定义
workflows:
  competitor-analysis:
    name: 竞品分析协作流
    
    # 主Agent
    coordinator: coordinator-agent
    
    steps:
      # Step 1: 搜索
      - delegate:
          agent: searcher
          task: "搜索竞品信息"
          timeout: 60s
          
      # Step 2: 分析（依赖搜索结果）
      - delegate:
          agent: analyzer
          task: "对比分析"
          depends_on: searcher.result
          
      # Step 3: 报告（依赖分析结果）
      - delegate:
          agent: reporter
          task: "生成报告"
          depends_on: analyzer.result
          
    # 汇总输出
    output:
      format: report
      template: competitor-analysis.md

✨ 最佳实践

# 竞品分析协作过程
Coordinator: 收到任务"分析竞品A和B"
Coordinator: 分解为3个子任务
Coordinator: 委托searcher搜索信息

Searcher: 收到委托，开始搜索
Searcher: 搜索产品A信息...
Searcher: 搜索产品B信息...
Searcher: 完成搜索，返回结果
→ [产品A: 价格¥99, 功能5项, 评分4.5]
→ [产品B: 价格¥149, 功能8项, 评分4.2]

Coordinator: 收到搜索结果，委托analyzer分析
Analyzer: 开始对比分析
Analyzer: 生成对比表格
Analyzer: 返回分析结果

Coordinator: 收到分析结果，委托reporter写报告
Reporter: 根据分析生成报告
Reporter: 返回完整报告

Coordinator: 汇总完成，输出给用户
→ 竞品分析报告（含对比表格、优劣势分析、建议）

💻 代码示例

Python协作调用

from openclaw import Agent, Coordinator

# 创建协作团队
team = Coordinator(
    name="analysis-team",
    
    # 团队成员
    agents={
        "searcher": Agent.from_yaml("searcher.yaml"),
        "analyzer": Agent.from_yaml("analyzer.yaml"),
        "reporter": Agent.from_yaml("reporter.yaml")
    }
)

# 执行协作任务
result = team.run("分析产品A和B的竞品关系")

# 查看协作过程
print(result.workflow_trace)
# [
#   {"step": "delegate", "agent": "searcher", "duration": "45s"},
#   {"step": "delegate", "agent": "analyzer", "duration": "30s"},
#   {"step": "delegate", "agent": "reporter", "duration": "20s"}
# ]

# 查看各Agent输出
print(result.sub_results)

直接委托

from openclaw import Agent

# 主Agent
coordinator = Agent(
    name="coordinator",
    can_delegate=True
)

# 直接委托任务给其他Agent
result = coordinator.delegate(
    to="searcher",
    task="搜索最新的AI新闻",
    timeout=60,
    context={
        "keywords": ["OpenAI", "Claude", "Gemini"],
        "sources": ["HN", "Reddit"]
    }
)

print(result)

并行委托

# 并行委托多个Agent
results = coordinator.delegate_parallel(
    tasks=[
        {"agent": "searcher-a", "task": "搜索产品A"},
        {"agent": "searcher-b", "task": "搜索产品B"},
        {"agent": "searcher-c", "task": "搜索产品C"}
    ],
    timeout=60
)

# 汇总结果
all_data = results.merge()
print(all_data)

链式协作

# 流水线协作
pipeline = AgentPipeline([
    searcher,
    analyzer,
    reporter
])

result = pipeline.run("分析竞品")
# 自动传递：searcher → analyzer → reporter

自定义协作协议

from openclaw.collaboration import Protocol

class MyCollaborationProtocol(Protocol):
    """自定义协作协议"""
    
    def prepare_request(self, task, context):
        """准备委托请求"""
        return {
            "type": "task_request",
            "task": task,
            "context": context,
            "priority": self._calculate_priority(task)
        }
    
    def handle_response(self, response):
        """处理响应"""
        if response["status"] == "success":
            return response["result"]
        elif response["status"] == "error":
            # 自动重试
            return self._retry(response)
    
    def aggregate(self, results):
        """汇总多个结果"""
        return self._smart_merge(results)

# 使用自定义协议
team = Coordinator(
    protocol=MyCollaborationProtocol()
)

🔗 相关链接

• OpenClaw A2A协议详解 - 协议规范
• OpenClaw多Agent协作 - 协作模式
• OpenClaw子Agent编排 - 子Agent管理
• OpenClaw Agent编排调度 - 调度策略
• ClawHub入门指南 - 发现更多Skills

📊 协作模式对比