多 Agent 协作的消息路由架构：从顺序调用到动态编排-拓冰建站

多 Agent 协作的消息路由架构：从顺序调用到动态编排

一、单 Agent 在多任务场景中的能力边界

一个综合 AI 助手需要同时处理三类请求：技术问题解答（需要代码搜索 Agent）、文档摘要生成（需要长文本处理 Agent）、日程管理（需要日历操作 Agent）。如果用一个通用 Agent 处理所有请求，问题表现为：技术问题时 Agent 调用了日历工具（误判意图），文档摘要时没有触发分段处理（缺少领域知识），日程冲突检测逻辑散布在 Prompt 中难以维护。

单 Agent 架构的本质局限在于：一个 Agent 的 System Prompt 需要涵盖所有领域的规则，导致指令互相干扰。为代码搜索写的"给出可运行示例"指令，在文档摘要场景中变成了"摘要要包含可运行示例"的错误理解。

多 Agent 的解耦思路是：每个 Agent 只负责一个明确的领域，通过路由器将用户请求分发给最合适的 Agent。这样每个 Agent 的 Prompt 可以高度专业化，工具的权限范围也得到天然隔离。

二、消息路由的三种分发模式

多 Agent 系统的消息分发有三种模式，按复杂度递增：

graph TB USER[用户消息] --> ROUTER{消息路由器} subgraph 模式1：意图分类路由 ROUTER -->|分类：技术问题| A1[代码搜索 Agent] ROUTER -->|分类：文档摘要| A2[摘要生成 Agent] ROUTER -->|分类：日程管理| A3[日历管理 Agent] ROUTER -->|未分类| DEFAULT[通用兜底 Agent] end subgraph 模式2：顺序编排 ROUTER2[编排器] --> B1[意图识别 Agent] B1 --> B2[任务分解 Agent] B2 --> B3{是否为复杂任务?} B3 -->|是| B4[多 Agent 协作执行] B3 -->|否| B5[单一 Agent 执行] end subgraph 模式3：动态协作 C1[对话管理 Agent<br/>维护全局上下文] C1 --> C2[按需调用专业 Agent] C2 --> C1 C1 --> C3[结果聚合与冲突解决] end

模式 1 适合大多数场景——根据意图分类后直接路由。模式 2 适合需要规划和分解的复杂任务。模式 3 适合需要跨领域协作的场景（如技术问题涉及代码 + 文档）。

三、模式 1 的路由器实现

""" 多 Agent 消息路由器——基于意图分类的消息分发。 设计意图：路由器自身是一个轻量 Agent，专门做意图分类。 分类后将消息路由到对应专业 Agent，返回结果。 """ from typing import List, Dict, Optional from enum import Enum from dataclasses import dataclass import json class AgentCapability(str, Enum): """Agent 能力枚举——路由器根据此进行分发""" CODE_SEARCH = "code_search" DOCUMENT_SUMMARY = "document_summary" CALENDAR_MANAGEMENT = "calendar_management" GENERAL_CHAT = "general_chat" @dataclass class AgentConfig: """Agent 配置""" name: str capability: AgentCapability system_prompt: str # 该 Agent 可调用的工具列表 tools: List[str] # 优先级——同一消息可能匹配多个 Agent，取优先级最高的 priority: int = 0 # 路由关键词——快速过滤，减少 LLM 调用 keywords: List[str] = None @dataclass class RouteDecision: """路由决策结果""" agent: AgentConfig confidence: float reason: str class MessageRouter: """消息路由器""" def __init__(self, agents: List[AgentConfig]): self.agents = agents self.fallback_agent = next( (a for a in agents if a.capability == AgentCapability.GENERAL_CHAT), agents[0], ) async def route(self, message: str) -> RouteDecision: """路由用户消息到最合适的 Agent""" # 第一层：关键词快速过滤——节省 LLM 调用 keyword_match = self._keyword_filter(message) if keyword_match: return keyword_match # 第二层：LLM 意图分类——处理关键词无法覆盖的场景 return await self._llm_classify(message) def _keyword_filter(self, message: str) -> Optional[RouteDecision]: """关键词快速过滤——O(n) 复杂度，毫秒级""" message_lower = message.lower() for agent in self.agents: if not agent.keywords: continue # 计算关键词命中比例 hits = sum( 1 for kw in agent.keywords if kw.lower() in message_lower ) if hits > 0: return RouteDecision( agent=agent, confidence=min(0.9, 0.5 + hits * 0.1), reason=f"关键词匹配: {hits} 个命中", ) return None async def _llm_classify(self, message: str) -> RouteDecision: """LLM 意图分类——准确性高但需要 API 调用""" # 构造分类 Prompt——列出所有可用的 Agent agent_descriptions = "\n".join( f"- {agent.name}: {agent.capability.value} (优先级 {agent.priority})" for agent in self.agents ) classification_prompt = f"""分析用户消息，选择最合适的 Agent 处理。 可用 Agent: {agent_descriptions} 输出 JSON 格式: {{ "agent": "Agent名称", "confidence": 0.0-1.0, "reason": "分类理由（简短）" }} 如果无法确定意图，使用优先级最高的通用 Agent。 如果消息与所有 Agent 都不匹配，confidence 设为 0 并使用 GENERAL_CHAT。 """ # 调用轻量模型做分类——减少延迟和成本 try: response = await self._call_llm( system_prompt=classification_prompt, user_message=message, model="gpt-3.5-turbo", # 分类不需要强模型 max_tokens=100, ) data = json.loads(response) target_agent = next( (a for a in self.agents if a.name == data.get("agent")), self.fallback_agent, ) return RouteDecision( agent=target_agent, confidence=float(data.get("confidence", 0.5)), reason=data.get("reason", "LLM 分类"), ) except (json.JSONDecodeError, KeyError) as e: # 分类失败——降级到通用 Agent return RouteDecision( agent=self.fallback_agent, confidence=0.3, reason=f"分类异常降级: {e}", ) async def _call_llm( self, system_prompt: str, user_message: str, model: str, max_tokens: int, ) -> str: """调用 LLM——简化版，实际应接入 API 客户端""" # 此处为占位实现 return '{"agent": "通用助手", "confidence": 0.5, "reason": "默认"}'

三个设计要点：