多Agent系统与协作机制

深入理解多Agent系统的架构设计、协作模式和通信机制，构建高效的分布式智能系统

📖 概述
- 什么时候才该用多 Agent
🏗️ 2026 多 Agent 协作模式
📨 消息 Schema、冲突解决与共享记忆
🧪 可运行示例
- 示例 A：OpenAI Agents SDK 的 Handoff（Python）
- 示例 B：LangGraph 的 Supervisor（Python）
📜 历史回顾：FIPA ACL（了解即可）
🎯 学习检验
- 理论理解检验
- 实践能力检验
🚀 实践项目建议
- 基础实战项目
- 高级综合项目
📚 延伸阅读

📖 概述

多 Agent 系统（Multi-Agent System）在 2026 年的主流含义已经从"学术多智能体"转向"多个 LLM Agent 协作完成一个复杂任务"。本文聚焦工程实践：什么时候该上多 Agent、有哪些协作模式（Supervisor / Handoff / Crew）、消息怎么传、冲突怎么解、记忆边界在哪、成本如何随 Agent 数上升，并给出 OpenAI Agents SDK 与 LangGraph 的可运行示例。

面试一句话：多 Agent ≠ 更聪明。它解决的是"单 Agent 上下文/工具过载、需要角色隔离"的问题，代价是上下文传递、协调复杂度和成本都上升。先把单 Agent 做稳，再考虑拆。

什么时候才该用多 Agent

该上多 Agent	不该上（单 Agent 足够）
工具/职责太多，单 Agent 选错工具频繁	工具少、流程线性
需要不同"人设/权限"隔离（如执行 vs 审核）	一个角色能搞定
子任务可并行、能显著降延迟	串行依赖强、并行收益小
不同子任务适配不同模型（降本）	同一模型通吃

🏗️ 2026 多 Agent 协作模式

模式一：Supervisor / Orchestrator-Worker（主管-工人）

一个 Supervisor（编排者） 负责把任务拆分、分发给若干 Worker（专家 Agent），再汇总结果。控制权始终回到 Supervisor。这是目前最常用、最可控的多 Agent 模式。

模式二：Handoff（移交）

Agent 之间直接移交控制权：当前 Agent 判断任务超出自己职责，就把对话整体"交给"更专业的 Agent。OpenAI Agents SDK 把它做成一等公民（handoffs=[...]）。与 Supervisor 的区别：控制权转移后不一定回来，更像客服转接。

模式三：Crew（角色分工流水线）

像组建一个团队：研究员 → 写作 → 审校，按固定流程（Process）串/并行流转。CrewAI 是代表。适合内容生产这类角色清晰、阶段固定的任务。

三种模式对比

模式	控制流	代表框架	适用	风险
Supervisor	中心化，控制权回主管	LangGraph、OpenAI Agents SDK	通用复杂任务、需汇总	主管成瓶颈
Handoff	去中心，控制权转移	OpenAI Agents SDK	意图分流、客服转接	转错无法回收
Crew	流水线，阶段流转	CrewAI	内容生产、固定流程	流程僵化

LangGraph 的 Send / Command（编排原语）

LangGraph 用图来编排多 Agent，两个关键原语：

Send：动态向某个节点扇出多个并行子任务（map-reduce 式分发，数量运行时才知道）。
Command：节点返回时同时决定"更新哪些状态 + 跳到哪个节点"，用来实现 Agent 间的移交与路由。

📨 消息 Schema、冲突解决与共享记忆

统一消息 Schema

多 Agent 之间务必用结构化消息，而不是自由文本，便于路由、追踪与重放：

{
  "msg_id": "m-20260609-001",
  "conversation_id": "task-42",
  "from": "supervisor",
  "to": "researcher",
  "type": "task_request",
  "payload": { "goal": "查2026 Q1新能源车销量", "constraints": ["至少3个来源"] },
  "reply_to": null,
  "ts": "2026-06-09T05:00:00Z"
}

type 建议固定枚举：task_request / task_result / clarify / handoff / error。

冲突解决

多个 Agent 给出矛盾结论时的常见裁决策略：

策略	做法	适用
主管裁决	Supervisor/Reviewer 最终拍板	中心化架构默认
证据优先	看谁的来源更可验证	检索/事实类
投票/多数	多个 Agent 表决	主观判断、可并行
升级人工	低置信度/高风险转人工	关键决策

原则：一定要有唯一最终裁决者，否则会陷入互相"反驳"的死循环，烧 token 还不收敛。

共享记忆边界

私有记忆：每个 Agent 的中间推理不互相污染（关键，否则上下文爆炸）。
共享记忆：只把"结论/产物"写入共享区（如黑板、共享 state、向量库），不共享原始思考过程。
隔离子 Agent：子 Agent 用独立上下文窗口处理脏活，只回传精炼结果——这是控制上下文与成本的核心手段。

成本随 Agent 数上升的权衡

上下文每移交一次就要重新喂背景，token 成本近似随 Agent 数与轮数相乘增长。
并行 Agent 降延迟，但并发 token 峰值更高。
经验法则：能用 1 个 Agent + 多工具解决，就别拆 3 个 Agent；拆分要换来"质量或延迟"的实质收益。

🧪 可运行示例

示例 A：OpenAI Agents SDK 的 Handoff（Python）

# pip install openai-agents
from agents import Agent, Runner

# 两个专家 Agent
billing_agent = Agent(
    name="Billing Agent",
    instructions="你只负责账单、退款、发票问题，用中文回答。",
)
tech_agent = Agent(
    name="Tech Support Agent",
    instructions="你只负责技术故障排查，用中文回答。",
)

# 分诊 Agent：根据意图移交给专家
triage_agent = Agent(
    name="Triage Agent",
    instructions="判断用户意图，账单问题移交 Billing Agent，技术问题移交 Tech Support Agent。",
    handoffs=[billing_agent, tech_agent],   # Handoff：一等公民
)

result = Runner.run_sync(triage_agent, "我上个月被重复扣费了，怎么退款？")
print(result.final_output)   # 由 Billing Agent 接手回答

示例 B：LangGraph 的 Supervisor（Python）

# pip install langgraph langchain-openai langgraph-supervisor
from langchain_openai import ChatOpenAI
from langgraph.prebuilt import create_react_agent
from langgraph_supervisor import create_supervisor

model = ChatOpenAI(model="gpt-5")

def web_search(query: str) -> str:
    """检索网络信息（示例占位，实际接真实搜索 API）。"""
    return f"关于「{query}」的检索结果……"

# 两个专家 worker
researcher = create_react_agent(
    model, tools=[web_search], name="researcher",
    prompt="你是检索专家，只负责查资料并给出带来源的结论。",
)
writer = create_react_agent(
    model, tools=[], name="writer",
    prompt="你是写作专家，把检索结论整理成结构化中文报告。",
)

# Supervisor 编排：决定何时调用哪个 worker，并汇总
supervisor = create_supervisor(
    agents=[researcher, writer],
    model=model,
    prompt="你是主管：先让 researcher 查资料，再让 writer 写报告，最后汇总给用户。",
).compile()

for chunk in supervisor.stream(
    {"messages": [{"role": "user", "content": "写一份2026年AI Agent框架现状简报"}]}
):
    print(chunk)

两个示例分别对应"去中心移交"和"中心化编排"，是面试里要求手写多 Agent 时的两种标准答案。

展开：通用多 Agent 系统骨架（教学示意，非生产代码）

// 教学示意：通用多 Agent 系统骨架，便于理解注册/通信/生命周期
class MultiAgentSystem {
  constructor(name, architecture = 'hierarchical') {
    this.name = name;
    this.architecture = architecture;
    this.agents = new Map();
    this.communicationLayer = new CommunicationLayer();
    this.coordinationLayer = new CoordinationLayer();
    this.environmentLayer = new EnvironmentLayer();
    this.systemState = 'inactive';
    this.metrics = new SystemMetrics();
  }

  // 注册Agent
  registerAgent(agent) {
    if (this.agents.has(agent.id)) {
      throw new Error(`Agent ${agent.id} already registered`);
    }
    
    agent.setSystem(this);
    this.agents.set(agent.id, agent);
    
    // 根据架构模式设置Agent关系
    this.setupAgentRelationships(agent);
    
    console.log(`🤖 注册Agent: ${agent.id} (${agent.type})`);
    return this;
  }

  // 移除Agent
  unregisterAgent(agentId) {
    const agent = this.agents.get(agentId);
    if (agent) {
      agent.setSystem(null);
      this.agents.delete(agentId);
      this.cleanupAgentRelationships(agentId);
      console.log(`🗑️ 移除Agent: ${agentId}`);
    }
    return this;
  }

  // 设置Agent关系
  setupAgentRelationships(agent) {
    switch (this.architecture) {
      case 'hierarchical':
        this.setupHierarchicalRelationships(agent);
        break;
      case 'peer_to_peer':
        this.setupPeerToPeerRelationships(agent);
        break;
      case 'blackboard':
        this.setupBlackboardRelationships(agent);
        break;
      case 'market':
        this.setupMarketRelationships(agent);
        break;
    }
  }

  // 层次化架构关系设置
  setupHierarchicalRelationships(agent) {
    if (agent.role === 'coordinator') {
      // 协调者可以与所有Agent通信
      this.agents.forEach(existingAgent => {
        if (existingAgent.id !== agent.id) {
          this.communicationLayer.establishConnection(agent.id, existingAgent.id);
        }
      });
    } else {
      // 普通Agent只与协调者通信
      this.agents.forEach(existingAgent => {
        if (existingAgent.role === 'coordinator') {
          this.communicationLayer.establishConnection(agent.id, existingAgent.id);
        }
      });
    }
  }

  // 点对点架构关系设置
  setupPeerToPeerRelationships(agent) {
    // 每个Agent都可以与其他Agent直接通信
    this.agents.forEach(existingAgent => {
      if (existingAgent.id !== agent.id) {
        this.communicationLayer.establishConnection(agent.id, existingAgent.id);
      }
    });
  }

  // 黑板架构关系设置
  setupBlackboardRelationships(agent) {
    // 所有Agent都通过黑板进行通信
    this.communicationLayer.connectToBlackboard(agent.id);
  }

  // 市场架构关系设置
  setupMarketRelationships(agent) {
    // Agent根据角色建立不同的连接
    if (agent.role === 'auctioneer') {
      this.agents.forEach(existingAgent => {
        if (existingAgent.id !== agent.id) {
          this.communicationLayer.establishConnection(agent.id, existingAgent.id);
        }
      });
    } else {
      // 参与者与拍卖师连接
      this.agents.forEach(existingAgent => {
        if (existingAgent.role === 'auctioneer') {
          this.communicationLayer.establishConnection(agent.id, existingAgent.id);
        }
      });
    }
  }

  // 启动系统
  async start() {
    if (this.systemState === 'active') {
      console.log('⚠️ 系统已经启动');
      return;
    }
    
    console.log(`🚀 启动多Agent系统: ${this.name}`);
    this.systemState = 'starting';
    
    try {
      // 启动各层
      await this.environmentLayer.start();
      await this.communicationLayer.start();
      await this.coordinationLayer.start();
      
      // 启动所有Agent
      const startPromises = Array.from(this.agents.values())
        .map(agent => agent.start());
      
      await Promise.all(startPromises);
      
      this.systemState = 'active';
      this.metrics.recordSystemStart();
      
      console.log('✅ 多Agent系统启动完成');
      
    } catch (error) {
      this.systemState = 'error';
      console.error('❌ 系统启动失败:', error.message);
      throw error;
    }
  }

  // 停止系统
  async stop() {
    if (this.systemState === 'inactive') {
      console.log('⚠️ 系统已经停止');
      return;
    }
    
    console.log('🛑 停止多Agent系统');
    this.systemState = 'stopping';
    
    try {
      // 停止所有Agent
      const stopPromises = Array.from(this.agents.values())
        .map(agent => agent.stop());
      
      await Promise.all(stopPromises);
      
      // 停止各层
      await this.coordinationLayer.stop();
      await this.communicationLayer.stop();
      await this.environmentLayer.stop();
      
      this.systemState = 'inactive';
      this.metrics.recordSystemStop();
      
      console.log('✅ 多Agent系统停止完成');
      
    } catch (error) {
      this.systemState = 'error';
      console.error('❌ 系统停止失败:', error.message);
      throw error;
    }
  }

  // 广播消息
  async broadcastMessage(message, excludeAgents = []) {
    const recipients = Array.from(this.agents.keys())
      .filter(agentId => !excludeAgents.includes(agentId));
    
    return await this.communicationLayer.broadcast(message, recipients);
  }

  // 发送消息给特定Agent
  async sendMessage(senderId, receiverId, message) {
    return await this.communicationLayer.sendMessage(senderId, receiverId, message);
  }

  // 获取系统状态
  getSystemStatus() {
    const agentStatuses = {};
    this.agents.forEach((agent, id) => {
      agentStatuses[id] = {
        state: agent.state,
        role: agent.role,
        type: agent.type,
        capabilities: Array.from(agent.capabilities)
      };
    });
    
    return {
      systemState: this.systemState,
      architecture: this.architecture,
      totalAgents: this.agents.size,
      agentStatuses,
      metrics: this.metrics.getMetrics()
    };
  }

  // 获取Agent
  getAgent(agentId) {
    return this.agents.get(agentId);
  }

  // 获取所有Agent
  getAllAgents() {
    return Array.from(this.agents.values());
  }

  // 按类型获取Agent
  getAgentsByType(type) {
    return Array.from(this.agents.values())
      .filter(agent => agent.type === type);
  }

  // 按角色获取Agent
  getAgentsByRole(role) {
    return Array.from(this.agents.values())
      .filter(agent => agent.role === role);
  }

  // 清理Agent关系
  cleanupAgentRelationships(agentId) {
    this.communicationLayer.removeAgentConnections(agentId);
  }
}

📜 历史回顾：FIPA ACL（了解即可）

在 LLM 之前，多智能体研究有一套通信标准 FIPA ACL（Agent Communication Language）：用 inform/request/cfp/propose/accept-proposal 等"言语行为（performative）"定义消息语义，配合**合同网协议（Contract Net）**做任务招投标。

面试一句话：FIPA ACL / 合同网是符号主义时代的产物，思想（结构化消息、招投标分配）至今有借鉴价值，但 2026 的 LLM 多 Agent 实践已不直接用它，而是用上文的 Supervisor/Handoff/Crew + JSON 消息 schema。

展开：FIPA ACL 与协作 Agent 的示意代码（史料性质）

// 史料示意：FIPA ACL 消息与协作 Agent 骨架，非生产代码
class CollaborativeAgent {
  constructor(id, type, role = 'participant') {
    this.id = id;
    this.type = type;
    this.role = role;
    this.state = 'inactive';
    this.capabilities = new Set();
    this.currentTasks = new Map();
    this.collaborationHistory = [];
    this.trustNetwork = new Map();
    this.reputation = 0.5;
    this.system = null;
  }

  // 处理任务
  async handleTask(task) {
    if (!this.canHandleTask(task)) {
      throw new Error(`Agent ${this.id} cannot handle task ${task.name}`);
    }
    
    this.currentTasks.set(task.id, {
      task,
      startTime: Date.now(),
      status: 'in_progress'
    });
    
    console.log(`📋 ${this.id} 开始处理任务: ${task.name}`);
    
    try {
      const result = await this.executeTask(task);
      
      this.currentTasks.set(task.id, {
        ...this.currentTasks.get(task.id),
        status: 'completed',
        result,
        endTime: Date.now()
      });
      
      console.log(`✅ ${this.id} 完成任务: ${task.name}`);
      return result;
      
    } catch (error) {
      this.currentTasks.set(task.id, {
        ...this.currentTasks.get(task.id),
        status: 'failed',
        error: error.message,
        endTime: Date.now()
      });
      
      console.error(`❌ ${this.id} 任务失败: ${task.name}`, error.message);
      throw error;
    }
  }
}

// FIPA ACL 消息（言语行为 + 会话上下文）
class FIPAACLMessage {
  constructor(performative, sender, receiver, content) {
    this.performative = performative; // 言语行为类型
    this.sender = sender;
    this.receiver = receiver;
    this.content = content;
    this.language = 'javascript'; // 内容语言
    this.ontology = 'default'; // 本体
    this.protocol = 'fipa-acl';
    this.conversationId = null;
    this.replyWith = null;
    this.inReplyTo = null;
    this.timestamp = Date.now();
  }

  // 验证消息格式
  validate() {
    const validPerformatives = [
      'inform', 'request', 'query-if', 'query-ref',
      'cfp', 'propose', 'accept-proposal', 'reject-proposal',
      'agree', 'refuse', 'confirm', 'disconfirm'
    ];

    if (!validPerformatives.includes(this.performative)) {
      throw new Error(`Invalid performative: ${this.performative}`);
    }

    if (!this.sender || !this.receiver) {
      throw new Error('Sender and receiver are required');
    }

    return true;
  }
}

🎯 学习检验

理论理解检验

多Agent系统架构：能否理解不同架构模式的特点和适用场景？
协作机制：能否掌握各种Agent协作模式和协议？
通信协议：能否理解FIPA ACL和合同网协议的工作原理？
冲突解决：能否设计有效的冲突解决机制？

实践能力检验

系统设计：能否设计复杂的多Agent系统架构？
协作实现：能否实现Agent间的有效协作？
协议应用：能否正确应用各种通信协议？
性能优化：能否优化多Agent系统的性能？

🚀 实践项目建议

基础实战项目

智能客服系统：实现多Agent协作的客服平台
任务分配系统：使用合同网协议的任务分配
协作问答系统：多个专业Agent协作回答问题
资源管理系统：多Agent协作管理共享资源

高级综合项目

智能制造系统：工厂环境下的多Agent协作
智慧城市平台：城市管理的大规模Agent系统
金融交易系统：高频交易的多Agent协作
游戏AI系统：游戏中的多Agent智能体系统

📚 延伸阅读

理论基础

"Multiagent Systems" - Gerhard Weiss
"An Introduction to MultiAgent Systems" - Michael Wooldridge
"Distributed Artificial Intelligence" - 各种学术论文
"Agent Communication Languages" - FIPA标准文档

实现技术（2026）

OpenAI Agents SDK 文档 — Handoff、Guardrails、tracing
LangGraph 文档 — Supervisor、Send / Command、多 Agent 状态图
CrewAI 文档 — 角色、任务与 Process 编排
Anthropic：Building Effective Agents — orchestrator-worker 等模式

本站延伸

💡 学习提示：多Agent系统是复杂的分布式智能系统，需要深入理解Agent间的协作机制和通信协议。建议从简单的双Agent协作开始，逐步扩展到复杂的多Agent系统。重视系统架构设计，选择合适的协作模式和通信协议。在实际开发中，要考虑系统的可扩展性、容错性和性能优化。通过实际项目练习，深入理解多Agent系统的设计原理和实现方法。

目录​

📖 概述​

什么时候才该用多 Agent​

🏗️ 2026 多 Agent 协作模式​

模式一：Supervisor / Orchestrator-Worker（主管-工人）​

模式二：Handoff（移交）​

模式三：Crew（角色分工流水线）​

三种模式对比​

LangGraph 的 Send / Command（编排原语）​

📨 消息 Schema、冲突解决与共享记忆​

统一消息 Schema​

冲突解决​

共享记忆边界​

成本随 Agent 数上升的权衡​

🧪 可运行示例​

示例 A：OpenAI Agents SDK 的 Handoff（Python）​

示例 B：LangGraph 的 Supervisor（Python）​

📜 历史回顾：FIPA ACL（了解即可）​

🎯 学习检验​

理论理解检验​

实践能力检验​

🚀 实践项目建议​

基础实战项目​

高级综合项目​

📚 延伸阅读​

理论基础​

实现技术（2026）​

本站延伸​

目录