多 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[对话管理 Agentbr/维护全局上下文] 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( agentagent, confidencemin(0.9, 0.5 hits * 0.1), reasonf关键词匹配: {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_promptclassification_prompt, user_messagemessage, modelgpt-3.5-turbo, # 分类不需要强模型 max_tokens100, ) data json.loads(response) target_agent next( (a for a in self.agents if a.name data.get(agent)), self.fallback_agent, ) return RouteDecision( agenttarget_agent, confidencefloat(data.get(confidence, 0.5)), reasondata.get(reason, LLM 分类), ) except (json.JSONDecodeError, KeyError) as e: # 分类失败——降级到通用 Agent return RouteDecision( agentself.fallback_agent, confidence0.3, reasonf分类异常降级: {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: 默认}三个设计要点两层路由——关键词快速过滤处理高频场景毫秒级LLM 分类处理语义模糊场景秒级减少不必要的 API 调用分类失败降级——JSON 解析失败时回退到通用 Agent不影响用户获取基本回复分类使用轻量模型——意图分类不需要强模型的推理能力。四、多 Agent 的协作成本考量每次路由的额外开销。引入路由器后每次用户消息增加了一次意图分类的 LLM 调用约 0.5-1 秒延迟约 500 Token 消耗。对于日活 1000 的系统日增 50 万 Token 的额外成本。Agent 间上下文传递的断裂。用户先问代码问题路由到代码 Agent再追问文档相关的问题路由到文档 Agent两个 Agent 没有共享上下文。需要在上层维护一个对话管理 Agent 来聚合所有子 Agent 的上下文。Router 自身的健壮性。如果 Router 将 10% 的消息错误分类用户将收到不相关的回复。Router 的准确率应持续监控当误分类率超过 5% 时需要调整分类 Prompt 或更新关键词列表。五、总结多 Agent 消息路由的架构设计根据业务领域划分 Agent 职责每个 Agent 有独立的 System Prompt 和工具权限路由器使用两层过滤关键词 LLM 分类分发消息分类失败时降级到通用 Agent不中断用户体验。落地步骤列出所有需要专业处理的消息类型按类型定义 Agent为每个 Agent 添加关键词列表覆盖 80% 的高频场景实现 LLM 分类作为关键词的补充处理剩余 20% 的语义模糊场景监控误分类率——定期抽样审查路由决策的准确性。