引言单一AI Agent已经不够用了。在2026年复杂的企业AI应用越来越多地采用多智能体系统Multi-Agent System将复杂任务分解给多个专业化Agent并行处理由协调者AgentOrchestrator负责调度和结果整合。LangGraph、AutoGen、CrewAI的成熟让多智能体系统的构建成本大幅降低。本文从架构设计角度深度讲解如何构建生产级多智能体系统。—## 一、多智能体系统的核心架构模式### 1.1 主从模式Orchestrator-Worker最常见的模式一个Orchestrator负责分解任务和协调多个Worker Agent负责执行用户请求 → Orchestrator ├── Research Agent负责信息搜集 ├── Analysis Agent负责数据分析 ├── Writing Agent负责内容生成 └── Review Agent负责质量审核 → 整合结果 → 用户适用场景工作流相对固定任务可以明确分解### 1.2 市场竞标模式Market-BasedAgent通过竞标获取任务pythonclass TaskMarket: 任务市场Agent竞标执行任务 async def publish_task(self, task: Task): # 向所有Agent广播任务 bids await asyncio.gather( *[agent.bid(task) for agent in self.agents] ) # 选择报价最优的Agent winner min(bids, keylambda b: b.estimated_cost) return await winner.agent.execute(task)### 1.3 辩论模式Debate多个Agent对同一问题提出不同观点通过讨论收敛到更好的答案初始回答 → Agent A正方 → Agent B反方 → Agent C仲裁 → 最终结论综合三方观点适用场景需要高准确性、多视角分析的场景投资分析、风险评估—## 二、使用LangGraph构建多智能体系统### 2.1 研究报告生成系统pythonfrom langgraph.graph import StateGraph, ENDfrom langgraph.prebuilt import create_react_agentfrom typing import TypedDict, Annotatedimport operator# 定义共享状态class ResearchState(TypedDict): topic: str search_results: list analysis: str draft: str final_report: str review_notes: str iteration: int# 创建专业化Agent# 搜索Agent负责信息采集search_agent create_react_agent( llm, tools[web_search, arxiv_search, news_search], state_modifier你是一个专业的信息搜集Agent负责搜集最新、最权威的信息)# 分析Agent负责深度分析analysis_agent create_react_agent( llm, tools[data_analysis, chart_generator], state_modifier你是一个数据分析专家负责对收集到的信息进行深度分析)# 写作Agent负责内容生成writing_agent create_react_agent( llm, tools[format_markdown, citation_manager], state_modifier你是一个技术写作专家负责将分析结果写成高质量报告)# 审核Agent负责质量把关review_agent create_react_agent( llm, tools[fact_check, grammar_check], state_modifier你是一个严格的编辑负责发现报告中的错误和不足)# 构建工作流图def build_research_graph(): graph StateGraph(ResearchState) # 添加节点 graph.add_node(search, search_node) graph.add_node(analyze, analyze_node) graph.add_node(write, write_node) graph.add_node(review, review_node) graph.add_node(revise, revise_node) # 定义流程 graph.set_entry_point(search) graph.add_edge(search, analyze) graph.add_edge(analyze, write) graph.add_edge(write, review) # 条件分支审核通过则结束否则修改 graph.add_conditional_edges( review, lambda state: revise if state[review_notes] else END, { revise: revise, END: END } ) # 限制修改次数防止无限循环 graph.add_conditional_edges( revise, lambda state: review if state[iteration] 3 else END, { review: review, END: END } ) return graph.compile()async def search_node(state: ResearchState) - dict: 搜索节点执行信息搜集 result await search_agent.ainvoke({ messages: [(user, f搜集关于「{state[topic]}」的最新信息重点关注最新进展和数据)] }) return { search_results: result[messages][-1].content }async def review_node(state: ResearchState) - dict: 审核节点质量把关 result await review_agent.ainvoke({ messages: [(user, f请审核以下报告指出需要改进的地方报告内容{state[draft]}请从以下角度评估1. 事实准确性有无错误数据或过时信息2. 逻辑连贯性3. 覆盖完整性有无重要遗漏4. 可读性如果报告已达到发布标准请回复APPROVED否则列出具体改进建议。)] }) review_content result[messages][-1].content if APPROVED in review_content: return {review_notes: , final_report: state[draft]} else: return {review_notes: review_content}### 2.2 智能客服多Agent系统pythonclass CustomerServiceOrchestrator: 智能客服多智能体系统 def __init__(self): self.intent_classifier IntentClassifier() self.agents { order: OrderAgent(), product: ProductAgent(), refund: RefundAgent(), complaint: ComplaintAgent(), technical: TechnicalSupportAgent() } self.escalation_agent HumanEscalationAgent() async def handle_request(self, user_message: str, session: Session) - str: # 步骤1意图分类 intent await self.intent_classifier.classify( user_message, contextsession.history ) # 步骤2路由到专业Agent if intent.type in self.agents: agent self.agents[intent.type] # 步骤3执行处理 result await agent.handle( messageuser_message, intentintent, sessionsession ) # 步骤4检查是否需要升级人工 if result.needs_human_escalation: return await self.escalation_agent.handle(session, result) return result.response else: # 未知意图转通用Agent return await self.agents.get(general, self.escalation_agent).handle( messageuser_message, sessionsession ) async def parallel_process(self, user_message: str, session: Session) - str: 并行处理同时查询多个系统然后整合结果 # 并行获取用户信息、订单信息、库存信息 user_info, orders, inventory await asyncio.gather( self.agents[profile].get_user_info(session.user_id), self.agents[order].get_recent_orders(session.user_id), self.agents[product].check_inventory(session.context.get(product_id)) ) # 使用Orchestrator LLM整合所有信息回答用户 context f用户信息{user_info}最近订单{orders}库存状态{inventory} return await self.orchestrator_llm.answer( questionuser_message, contextcontext )—## 三、Agent间通信协议### 3.1 消息传递模式pythonfrom dataclasses import dataclassfrom enum import Enumfrom typing import Anyclass MessageType(Enum): TASK task RESULT result ERROR error QUERY query UPDATE updatedataclassclass AgentMessage: Agent间通信的标准消息格式 sender: str recipient: str message_type: MessageType content: Any trace_id: str # 用于全链路追踪 timestamp: float correlation_id: str None # 关联到哪个请求class AgentMessageBus: Agent消息总线 def __init__(self): self.subscribers {} self.message_history [] async def publish(self, message: AgentMessage): 发布消息 self.message_history.append(message) # 发送给指定接收者 if message.recipient in self.subscribers: for handler in self.subscribers[message.recipient]: await handler(message) def subscribe(self, agent_id: str, handler: callable): 订阅消息 if agent_id not in self.subscribers: self.subscribers[agent_id] [] self.subscribers[agent_id].append(handler)### 3.2 共享状态管理pythonimport asynciofrom typing import Optionalclass SharedAgentMemory: 多Agent共享的工作记忆 def __init__(self): self._store {} self._lock asyncio.Lock() self._version 0 async def update(self, key: str, value: Any, agent_id: str): 线程安全地更新共享状态 async with self._lock: self._store[key] { value: value, updated_by: agent_id, timestamp: asyncio.get_event_loop().time(), version: self._version } self._version 1 async def get(self, key: str) - Optional[Any]: async with self._lock: entry self._store.get(key) return entry[value] if entry else None async def get_all_context(self) - dict: 获取所有共享上下文用于注入Agent提示词 async with self._lock: return {k: v[value] for k, v in self._store.items()}—## 四、错误处理与容错pythonclass FaultTolerantOrchestrator: 容错的多Agent协调器 async def execute_with_fallback( self, primary_agent: BaseAgent, fallback_agent: BaseAgent, task: Task, timeout: float 30.0 ) - AgentResult: 主Agent失败时自动切换到备用Agent try: # 尝试主Agent设置超时 result await asyncio.wait_for( primary_agent.execute(task), timeouttimeout ) return result except asyncio.TimeoutError: logger.warning(f主Agent {primary_agent.name} 超时切换到备用) return await fallback_agent.execute(task) except AgentError as e: logger.error(f主Agent {primary_agent.name} 失败: {e}) if e.is_recoverable: return await fallback_agent.execute(task) raise async def retry_with_reflection( self, agent: BaseAgent, task: Task, max_retries: int 3 ) - AgentResult: 失败时让Agent反思并重试 last_error None for attempt in range(max_retries): try: return await agent.execute(task) except Exception as e: last_error e # 让Agent反思失败原因 reflection await self.reflection_llm.analyze( tasktask, errorstr(e), attemptattempt 1 ) # 根据反思修改任务策略 task task.with_hint(reflection) raise RuntimeError(fAgent在{max_retries}次尝试后仍然失败: {last_error})—## 五、性能优化pythonclass OptimizedMultiAgentSystem: 高性能多Agent系统 async def parallel_sub_tasks(self, task: ComplexTask) - list: 并行执行独立的子任务 # 构建任务依赖图 dag task.build_dependency_graph() results {} # 按拓扑顺序执行 for level in dag.topological_levels(): # 同一层的任务可以并行 level_tasks [task.subtasks[t_id] for t_id in level] level_results await asyncio.gather( *[self.agents[t.agent_type].execute(t) for t in level_tasks], return_exceptionsTrue ) for t, result in zip(level_tasks, level_results): if isinstance(result, Exception): logger.error(f子任务 {t.id} 失败: {result}) else: results[t.id] result return results—## 总结多智能体系统的核心价值在于1.并行化多个Agent同时处理不同子任务显著缩短总时间2.专业化每个Agent专注于自己最擅长的领域质量更高3.容错性一个Agent失败不影响整体支持降级和重试4.可扩展性轻松添加新的专业Agent扩展系统能力构建多智能体系统的关键是清晰的任务分解 标准化的通信协议 健壮的错误处理。在这三点做好之前不要过早地追求架构的复杂性。