多 Agent 协作模式技术调研报告

┌────────────────────────────────────────────────────────────────┐
│                     多 Agent 协作模式分类                        │
├────────────────────────────────────────────────────────────────┤
│                                                                │
│  1. 星型 (Star)           2. 流水线 (Pipeline)                 │
│     ┌───┐                    A → B → C → D                     │
│     │ A │                                                        │
│    ╱│╲│╲                  3. 广播 (Broadcast)                  │
│   B C D E                   A → B                               │
│                             A → C                               │
│  4. 网状 (Mesh)             A → D                               │
│     A ─── B                                                       │
│     │ ╲ ╱ │                  5. 混合 (Hybrid)                   │
│     C ─── D                 星型 + 流水线（生产环境主流）          │
│                                                                │
└────────────────────────────────────────────────────────────────┘

开始
  │
  ├─ 需要集中控制任务路由吗？
  │   ├─ 是 → 星型架构
  │   └─ 否 → 继续
  │
  ├─ 工作流程是确定性的吗？
  │   ├─ 是 → 流水线架构
  │   └─ 否 → 继续
  │
  ├─ 需要一对多信息分发吗？
  │   ├─ 是 → 广播架构
  │   └─ 否 → 继续
  │
  ├─ Agent 需要自主协作吗？
  │   ├─ 是 → 网状架构
  │   └─ 否 → 继续
  │
  └─ 默认推荐 → 混合架构（星型 + 流水线）

                    ┌─────────────┐
                    │  Orchestrator│
                    │  (主 Agent)   │
                    └──────┬──────┘
                           │
          ┌────────────────┼────────────────┐
          │                │                │
          ▼                ▼                ▼
    ┌──────────┐    ┌──────────┐    ┌──────────┐
    │ Agent A  │    │ Agent B  │    │ Agent C  │
    │ (编程)    │    │ (写作)    │    │ (测试)    │
    └──────────┘    └──────────┘    └──────────┘

{
  "agents": {
    "list": [
      {
        "id": "main",
        "default": true,
        "workspace": "/root/.openclaw/workspace",
        "model": { "primary": "aliyun/qwen3.5-plus" },
        "identity": { "name": "戴蒙", "emoji": "🤖" },
        "groupChat": {
          "mentionPatterns": ["@damon:conduit.local"]
        }
      },
      {
        "id": "coding",
        "workspace": "/root/.openclaw/workspace-coding",
        "model": { "primary": "aliyun/qwen3-coder-plus" },
        "identity": { "name": "编程专家", "emoji": "💻" },
        "groupChat": {
          "mentionPatterns": [
            "@coding:conduit.local",
            "@coder:conduit.local",
            "@tester:conduit.local"
          ]
        }
      },
      {
        "id": "tech-blog-writer",
        "workspace": "/root/.openclaw/workspace-tech-blog-writer",
        "model": { "primary": "aliyun/qwen3.5-plus" },
        "identity": { "name": "技术博客作家", "emoji": "📝" }
      }
    ]
  },
  "tools": {
    "agentToAgent": {
      "enabled": true,
      "allow": ["main", "coding", "tech-blog-writer"]
    }
  },
  "bindings": [
    {
      "agentId": "main",
      "match": { "channel": "matrix", "accountId": "damon" }
    },
    {
      "agentId": "coding",
      "match": { "channel": "matrix", "accountId": "coding" }
    }
  ]
}

# 伪代码：星型架构的任务路由
class StarOrchestrator:
    def __init__(self):
        self.agents = {
            "coding": CodingAgent(),
            "writing": WritingAgent(),
            "testing": TestingAgent()
        }

    def route_task(self, user_message: str) -> str:
        # 1. 意图识别
        intent = self.classify_intent(user_message)

        # 2. 选择 Agent
        if intent == "coding":
            target_agent = self.agents["coding"]
        elif intent == "writing":
            target_agent = self.agents["writing"]
        elif intent == "testing":
            target_agent = self.agents["testing"]
        else:
            target_agent = self  # 主 Agent 自己处理

        # 3. 委托任务
        result = target_agent.execute(user_message)

        # 4. 返回结果
        return result

    def classify_intent(self, message: str) -> str:
        # 基于关键词或 LLM 分类
        if any(kw in message for kw in ["代码", "开发", "实现", "修复"]):
            return "coding"
        elif any(kw in message for kw in ["文章", "博客", "写作"]):
            return "writing"
        elif any(kw in message for kw in ["测试", "bug", "调试"]):
            return "testing"
        else:
            return "general"

输入 → [Agent A] → [Agent B] → [Agent C] → 输出
       (收集)      (分析)      (生成)

from langgraph.graph import StateGraph, MessagesState, START, END
from typing import TypedDict, Annotated, List
from langgraph.graph.message import add_messages

# 1. 定义状态
class PipelineState(TypedDict):
    messages: Annotated[List, add_messages]
    collected_data: dict
    analysis_result: str
    final_output: str
    current_stage: str

# 2. 定义节点（每个节点是一个 Agent）
def collect_agent(state: PipelineState) -> dict:
    """信息收集阶段"""
    data = search_web(state['messages'][-1].content)
    return {
        "collected_data": data,
        "current_stage": "collect"
    }

def analyze_agent(state: PipelineState) -> dict:
    """分析阶段"""
    analysis = analyze_data(state['collected_data'])
    return {
        "analysis_result": analysis,
        "current_stage": "analyze"
    }

def generate_agent(state: PipelineState) -> dict:
    """内容生成阶段"""
    output = generate_content(state['analysis_result'])
    return {
        "final_output": output,
        "current_stage": "generate"
    }

# 3. 组装流水线
graph = StateGraph(PipelineState)
graph.add_node("collect", collect_agent)
graph.add_node("analyze", analyze_agent)
graph.add_node("generate", generate_agent)

# 4. 定义执行顺序
graph.add_edge(START, "collect")
graph.add_edge("collect", "analyze")
graph.add_edge("analyze", "generate")
graph.add_edge("generate", END)

# 5. 编译并运行
app = graph.compile()
result = app.invoke({"messages": [{"role": "user", "content": "调研 AI Agent 市场"}]})

                    ┌─────────────┐
                    │   Sender    │
                    │  (广播源)    │
                    └──────┬──────┘
                           │
          ┌────────────────┼────────────────┐
          │                │                │
          ▼                ▼                ▼
    ┌──────────┐    ┌──────────┐    ┌──────────┐
    │Receiver A│    │Receiver B│    │Receiver C│
    └──────────┘    └──────────┘    └──────────┘

import asyncio
from typing import List, Dict

class BroadcastCoordinator:
    def __init__(self):
        self.receivers = {
            "market_research": ResearchAgent(),
            "competitor_analysis": CompetitorAgent(),
            "trend_detection": TrendAgent()
        }

    async def broadcast_task(self, topic: str) -> Dict:
        """广播任务给所有接收者"""
        tasks = {
            name: agent.analyze(topic)
            for name, agent in self.receivers.items()
        }

        # 并行执行
        results = await asyncio.gather(*tasks.values())

        # 汇总结果
        return {
            name: result
            for name, result in zip(tasks.keys(), results)
        }

    A ───── B
    │ ╲   ╱ │
    │   X   │
    │ ╱   ╲ │
    C ───── D

from crewai import Agent, Task, Crew

# 定义多个专业 Agent
researcher = Agent(
    role='Senior Research Analyst',
    goal='Discover cutting-edge developments in AI',
    backstory='Expert in analyzing industry trends',
    verbose=True,
    allow_delegation=True
)

writer = Agent(
    role='Tech Content Strategist',
    goal='Craft compelling content on tech advancements',
    backstory='Expert in translating technical concepts',
    verbose=True,
    allow_delegation=True
)

editor = Agent(
    role='Chief Editor',
    goal='Ensure content quality and consistency',
    backstory='Expert in editing and quality control',
    verbose=True
)

# 定义任务（任务间有依赖关系）
research_task = Task(
    description='Analyze 2026 AI trends',
    expected_output='List of top 5 trends',
    agent=researcher
)

writing_task = Task(
    description='Write blog post about AI trends',
    expected_output='Full blog post',
    agent=writer,
    context=[research_task]
)

editing_task = Task(
    description='Review and edit the blog post',
    expected_output='Final polished article',
    agent=editor,
    context=[writing_task]
)

# 创建 Crew（自动协调 Agent 协作）
crew = Crew(
    agents=[researcher, writer, editor],
    tasks=[research_task, writing_task, editing_task],
    verbose=2
)

result = crew.kickoff()

┌─────────────────────────────────────────────────────────────┐
│                      OpenClaw 架构                           │
├─────────────────────────────────────────────────────────────┤
│  用户请求 (Matrix/Feishu)                                   │
│         │                                                   │
│         ▼                                                   │
│  ┌─────────────┐                                           │
│  │ Main Agent  │ ← 星型架构中心                             │
│  │   (戴蒙)    │                                           │
│  └──────┬──────┘                                           │
│         │                                                   │
│    ┌────┴────┬────────────┬────────────┐                   │
│    │         │            │            │                   │
│    ▼         ▼            ▼            ▼                   │
│ ┌──────┐ ┌──────┐    ┌──────┐    ┌──────┐                 │
│ │Coding│ │Novel │    │Tech  │    │ HF   │                 │
│ │Agent │ │Writer│    │Blog  │    │Chat  │                 │
│ └──────┘ └──────┘    └──────┘    └──────┘                 │
│                                                             │
│  每个 Agent 内部采用流水线架构                               │
│  (意图识别 → 任务执行 → 质量检查 → 输出)                     │
└─────────────────────────────────────────────────────────────┘

class AgentMessage:
    sender: str
    receiver: str
    task_type: str
    payload: dict
    priority: str  # low/medium/high/urgent
    timeout: int   # 超时时间（秒）