1. 引言为什么需要 Multi-Agent 协作单 Agent 的能力边界受限于上下文窗口、工具调用链长度和单一角色视角。Multi-Agent 协作通过拆分角色、并行执行、互相校验把复杂任务拆解为可管理的子任务从而显著提升任务完成质量和稳定性。在真实业务场景中一个 Agent 往往难以同时兼顾「规划、执行、审查、总结」等多重职责。例如一个负责写代码的 Agent 如果同时要检查代码质量很容易陷入「自己写、自己审」的盲区。Multi-Agent 协作的核心思想是让每个 Agent 只做自己最擅长的事再通过协作机制把结果整合起来。本文将从协作模式分类、核心工程组件、代码实战三个层面系统讲解 Multi-Agent 协作的落地方法。全文包含可直接运行的 Python 代码示例帮助你从零搭建一个多 Agent 协作系统。2. Multi-Agent 协作的三种核心模式根据任务拆分方式和信息流向Multi-Agent 协作主要分为以下三种模式模式核心思想适用场景典型代表编排模式Orchestration一个中心调度器Orchestrator负责任务拆分、分配和结果汇总任务层次清晰、子任务相对独立AutoGen GroupChat、LangGraph流水线模式Pipeline任务按固定顺序依次传递前一个 Agent 的输出是后一个 Agent 的输入流程固定、阶段明确的场景LangChain SequentialChain辩论/评审模式Debate/Review多个 Agent 从不同视角对同一问题提出观点互相质疑、迭代收敛需要高准确率、多角度验证的场景CAMEL、Multi-Agent Debate实际工程中这三种模式往往组合使用。例如先用编排模式拆分任务再用流水线模式执行子任务最后用评审模式校验结果。3. 核心工程组件无论采用哪种协作模式一个健壮的 Multi-Agent 系统都离不开以下核心组件3.1 Agent 角色定义每个 Agent 需要明确的角色描述、能力边界和输出格式。角色定义越清晰协作效率越高。from dataclasses import dataclass, field from typing import List, Optional dataclass class AgentRole: Agent 角色定义 name: str system_prompt: str tools: List[str] field(default_factorylist) max_iterations: int 5 temperature: float 0.7 示例定义三个协作角色 planner AgentRole( nameplanner, system_prompt你是一个资深项目经理负责把复杂任务拆解为可执行的子任务并明确每个子任务的输入输出。, tools[task_split, dependency_check], temperature0.3, ) coder AgentRole( namecoder, system_prompt你是一个高级软件工程师擅长编写高质量、可运行的代码。你只负责编码不负责需求分析。, tools[write_code, run_code], temperature0.5, ) reviewer AgentRole( namereviewer, system_prompt你是一个严格的代码审查专家负责检查代码的正确性、安全性和可维护性并给出修改建议。, tools[code_review, security_check], temperature0.2, )3.2 消息传递机制Agent 之间通过结构化消息进行通信。消息需要包含发送者、接收者、内容类型和元数据以便追踪协作过程。from enum import Enum from datetime import datetime from typing import Dict, Any class MessageType(Enum): TASK task # 任务分配 RESULT result # 任务结果 REVIEW review # 评审意见 REVISION revision # 修改请求 DONE done # 完成信号 dataclass class Message: Agent 间传递的结构化消息 sender: str receiver: str msg_type: MessageType content: str metadata: Dict[str, Any] field(default_factorydict) timestamp: str field(default_factorylambda: datetime.now().isoformat()) def to_dict(self) - Dict[str, Any]: return { sender: self.sender, receiver: self.receiver, msg_type: self.msg_type.value, content: self.content, metadata: self.metadata, timestamp: self.timestamp, }/code/pre 3.3 共享记忆与状态管理 多 Agent 协作需要共享上下文避免每个 Agent 各自为政。共享记忆通常分为短期记忆当前任务上下文和长期记忆历史经验库。 class SharedMemory: 多 Agent 共享记忆 def __init__(self): self.short_term: List[Message] [] # 短期记忆当前任务的消息流 self.long_term: Dict[str, Any] {} # 长期记忆历史经验与结论 def add_message(self, msg: Message): self.short_term.append(msg) # 自动归档关键结论到长期记忆 if msg.msg_type MessageType.DONE: self.long_term[msg.metadata.get(task_id, unknown)] msg.content def get_context(self, agent_name: str, window: int 10) - str: 获取指定 Agent 的最近上下文 relevant [m for m in self.short_term if m.receiver agent_name or m.sender agent_name] return \n.join(f[{m.sender}→{m.receiver}] {m.content} for m in relevant[-window:])/code/pre 4. 代码实战基于 LangGraph 搭建编排模式 LangGraph 是构建有状态、可编排多 Agent 系统的优秀框架。下面我们实现一个「规划-编码-审查」三 Agent 协作系统。 4.1 环境准备 pip install langgraph langchain-openai 4.2 定义 Agent 节点 from langgraph.graph import StateGraph, END from typing import TypedDict, Annotated, List import operator class WorkflowState(TypedDict): 工作流状态 task: str plan: str code: str review: str revision_count: int messages: Annotated[List[dict], operator.add] 规划 Agent def planner_node(state: WorkflowState) - dict: 将任务拆解为实施计划 task state[task] 实际项目中这里调用 LLM plan f针对任务「{task}」的实施计划\n1. 分析需求\n2. 设计接口\n3. 编写实现\n4. 编写测试 return {plan: plan, messages: [{role: planner, content: plan}]} 编码 Agent def coder_node(state: WorkflowState) - dict: 根据计划编写代码 plan state[plan] 实际项目中这里调用 LLM code def add(a: int, b: int) - int: 两数相加 return a b def multiply(a: int, b: int) - int: 两数相乘 return a * b return {code: code, messages: [{role: coder, content: 已完成编码}]} 审查 Agent def reviewer_node(state: WorkflowState) - dict: 审查代码质量 code state[code] 实际项目中这里调用 LLM if def not in code: review 代码缺少函数定义需要补充。 else: review 代码结构清晰函数定义完整通过审查。 return {review: review, messages: [{role: reviewer, content: review}]} 4.3 构建协作图 def should_revise(state: WorkflowState) - str: 根据审查结果决定是否返工 if 通过 in state[review] or state[revision_count] 3: return end return revise 构建状态图 workflow StateGraph(WorkflowState) 添加节点 workflow.add_node(planner, planner_node) workflow.add_node(coder, coder_node) workflow.add_node(reviewer, reviewer_node) 添加边 workflow.set_entry_point(planner) workflow.add_edge(planner, coder) workflow.add_edge(coder, reviewer) workflow.add_conditional_edges( reviewer, should_revise, { end: END, revise: coder, # 审查不通过则返回编码节点 }, ) 编译图 app workflow.compile() 执行 result app.invoke({ task: 实现一个简单的计算器模块, revision_count: 0, messages: [], }) print( 最终代码 ) print(result[code]) print( 审查意见 ) print(result[review]) 运行上述代码你会看到三个 Agent 按「规划 → 编码 → 审查」的顺序协作审查不通过时自动返回编码节点进行修订最多重试 3 次。 5. 代码实战基于 AutoGen 实现辩论评审模式 AutoGen 的 GroupChat 机制天然支持多 Agent 辩论。下面实现一个「正反方辩论 裁判裁决」的评审系统。 5.1 环境准备 pip install pyautogen 5.2 定义辩论 Agent from autogen import ConversableAgent, GroupChat, GroupChatManager 正方 Agent主张使用方案 A proponent ConversableAgent( nameproponent, system_message你是方案 A 的支持者。你坚信方案 A 在性能上更优。 你的论证要点 方案 A 的响应延迟更低 方案 A 的资源占用更少 方案 A 的社区生态更成熟 请用数据和逻辑说服对方。, llm_config{config_list: [{model: gpt-4, api_key: YOUR_API_KEY}]}, human_input_modeNEVER, ) 反方 Agent主张使用方案 B opponent ConversableAgent( nameopponent, system_message你是方案 B 的支持者。你坚信方案 B 在可维护性上更优。 你的论证要点 方案 B 的代码结构更清晰 方案 B 的扩展性更好 方案 B 的团队学习成本更低 请用数据和逻辑说服对方。, llm_config{config_list: [{model: gpt-4, api_key: YOUR_API_KEY}]}, human_input_modeNEVER, ) 裁判 Agent中立裁决 judge ConversableAgent( namejudge, system_message你是技术决策委员会主席负责听取双方辩论后做出最终裁决。 请从性能、可维护性、成本、风险四个维度综合评估给出最终技术选型建议。, llm_config{config_list: [{model: gpt-4, api_key: YOUR_API_KEY}]}, human_input_modeNEVER, ) 5.3 启动辩论 创建群聊 group_chat GroupChat( agents[proponent, opponent, judge], messages[], max_round6, # 最多辩论 6 轮 ) 创建群聊管理器 manager GroupChatManager( groupchatgroup_chat, llm_config{config_list: [{model: gpt-4, api_key: YOUR_API_KEY}]}, ) 发起辩论 result judge.initiate_chat( manager, message我们正在评估两个技术方案方案 A高性能和方案 B高可维护性。请双方展开辩论最终由我做出裁决。, ) 打印辩论记录 for msg in group_chat.messages: print(f[{msg[name]}] {msg[content]}) print(---) 辩论模式特别适合技术选型、方案评审、代码审查等需要多角度验证的场景。通过多轮交锋可以暴露单 Agent 视角下的盲区。 6. 工程实践要点与避坑指南 6.1 通信协议设计 Agent 之间的消息必须结构化。建议使用 JSON Schema 定义消息格式并包含版本号以便后续演进。 { schema_version: 1.0, message_id: msg_20260809_001, sender: planner, receiver: coder, type: task, payload: { task_id: task_001, description: 实现用户登录接口, acceptance_criteria: [输入校验, 密码加密, 返回 token] }, trace_id: trace_20260809_001 } 6.2 超时与重试机制 任何 Agent 都可能卡死或返回异常结果。必须为每个 Agent 调用设置超时和重试策略。 import time from functools import wraps def with_retry(max_retries: int 3, timeout: int 30): 为 Agent 调用添加重试和超时机制 def decorator(func): wraps(func) def wrapper(*args, **kwargs): for attempt in range(max_retries): try: 实际项目中用 asyncio.wait_for 实现超时 return func(*args, **kwargs) except Exception as e: if attempt max_retries - 1: raise time.sleep(2 ** attempt) # 指数退避 return None return wrapper return decorator with_retry(max_retries3, timeout30) def call_agent(agent_name: str, prompt: str) - str: 调用指定 Agent示例实现 实际项目中这里调用 LLM API return f{agent_name} 的响应 6.3 成本控制 多 Agent 协作会显著增加 Token 消耗。建议采取以下措施 上下文裁剪只传递当前任务相关的上下文避免全量历史。 结果缓存对相同输入的 Agent 调用结果做缓存。 模型分级简单任务用轻量模型复杂任务用强模型。 提前终止达到验收标准后立即停止不追求多余轮次。 6.4 可观测性 必须记录完整的协作轨迹便于问题排查和效果优化。 import json from datetime import datetime class CollaborationLogger: 协作过程日志记录器 def init(self, log_file: str collab_log.jsonl): self.log_file log_file def log(self, event: dict): 记录协作事件 event[timestamp] datetime.now().isoformat() with open(self.log_file, a, encodingutf-8) as f: f.write(json.dumps(event, ensure_asciiFalse) \n) def get_trace(self, trace_id: str) - list: 按 trace_id 查询完整协作链路 traces [] with open(self.log_file, r, encodingutf-8) as f: for line in f: event json.loads(line) if event.get(trace_id) trace_id: traces.append(event) return traces 使用示例 logger CollaborationLogger() logger.log({ trace_id: trace_001, event: agent_start, agent: planner, input: 实现登录功能, }) 7. 总结与展望 Multi-Agent 协作是提升 AI 系统能力上限的关键路径。本文从三种核心协作模式出发通过 LangGraph 和 AutoGen 两个框架的实战代码演示了编排模式和辩论评审模式的落地方法。 在实际工程中建议遵循以下原则 角色单一职责每个 Agent 只做一件事职责越单一协作越稳定。 消息结构化用统一的 Schema 定义消息避免自由文本带来的解析问题。 失败可恢复为每个环节设计重试、降级和人工介入机制。 成本可度量建立 Token 消耗和延迟的监控指标持续优化。 未来Multi-Agent 协作将向「自适应编排」和「群体智能」方向发展——系统能够根据任务复杂度动态调整 Agent 数量和协作方式甚至让 Agent 自主发现新的协作模式。掌握本文的工程方法你就已经站在了这一趋势的前沿。