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

泠不丁

463人浏览 · 2026-07-13 10:35:58

泠不丁 · 2026-07-13 10:35:58 发布

多 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": "默认"}'

三个设计要点：