AI 智能体 - 路由模式

在我们的上一篇文章中，我们探讨了“提示链式模式”（Prompt Chaining），一种通过将复杂任务分解为线性、顺序步骤来提高大型语言模型（LLM）可靠性的强大技术。这种“分而治之”的策略非常适合执行确定性的工作流，就像一条高效的工厂流水线。

但现实世界是混乱的，它很少遵循一条笔直的线性路径。

想象一下，你的“流水线”上突然来了一个完全不相关的请求。一个用户不会总是按照你设计的$A\to B\to C$的路径提问。他们可能会从 A 跳到 D，或者提出一个你的流水线根本没有准备好处理的问题。这时，严格的提示链就会“卡壳”。

这就是我们需要引入一个更适用概念的原因：路由模式（Routing Pattern）。如果说提示链是“路径”，那么路由就是“十字路口”和“交通枢纽”。它为 AI 智能体（Agent）引入了至关重要的条件逻辑，使其能够从一个僵化的执行者转变为一个能够动态决策、适应环境的智能系统。

一、当线性流水线“堵塞”：凸显路由的必要性

简单的顺序处理（提示链）有一个核心限制：它不具备决策能力。它只能盲目地将前一步的输出作为下一步的输入。

让我们来看一个更贴合实际的例子：一家电商公司构建了一个基于提示链的 AI 客服助手。

失败的线性工作流：

1. 提示 1（意图识别）： “识别用户是否在询问产品信息。”

2. 提示 2（信息提取）： “从[产品数据库]中提取[产品 A]的规格。”

3. 提示 3（回答生成）： “根据[规格]生成一段介绍。”

这个工作流在用户问“请介绍一下你们的新款相机”时表现良好。

但如果用户问：“我的订单到哪了？”

• 步骤 1 失败： 意图不是“产品信息”。工作流要么崩溃，要么给出一个牛头不对马嘴的回答（“我们的新款相机非常棒……”）。

如果用户问：“我的相机屏幕冻结了，我该怎么办？”

• 步骤 1 失败： 意图也不是“产品信息”。用户迫切需要技术支持，但这个线性的“产品介绍”流水线完全无法处理。

现实世界的 AI 系统必须能够处理这种多样性和不确定性。它们必须能够根据输入、环境状态或之前操作的结果，在多个潜在的操作路径之间进行仲裁。路由，就是实现这种动态决策能力的核心机制。

二、路由模式

路由模式将条件逻辑（if-then-else）引入了 AI 智能体的操作框架。它使智能体能够从固定的执行路径转移到一个动态模型，在该模型中，智能体首先评估输入，然后根据特定标准选择一个最合适的后续操作。

这使得系统行为更灵活，更具上下文感知能力。

让我们重新设计那个电商客服助手，这次使用路由模式：

1. 步骤 1：路由（分诊台）

2. AI 智能体（路由器）： “分析以下用户查询，并将其分类到最合适的处理路径。”

3. 用户查询： “我的相机屏幕冻结了，我该怎么办？”

4. 路由器决策： “意图为‘技术支持’。将此查询路由到‘技术支持子代理’。”

5. 步骤 2：执行（选择的路径）

6. 技术支持子代理（一个专门的提示链）：

7. 提示 1（诊断）： “根据查询[查询内容]，从[故障排除指南]中找到最相关的解决方案。”

8. 提示 2（行动建议）： “如果找不到解决方案，建议用户[升级到人工客服]。”

在这个新模型下，系统可以智能地处理各种查询：

通过这种方式，路由模式充当了系统的“中央调度员”或“分诊台”，确保每个请求都能被最合适的功能模块处理。

三、实现路由的四种方式

路由模式的核心是一个“评估和指导”机制。这个机制告诉工作流“接下来该往哪里走”。在实践中，我们可以通过以下四种主要方式来实现它：

1. 基于 LLM 的路由（智能路由）

这是最灵活但也最昂贵的方法。我们直接利用一个大型语言模型（LLM）的智能来进行决策。

• 工作原理： 我们向 LLM 提供一个专门的“路由提示”，要求它分析输入，并输出一个明确的指令或类别标识符，指示下一步或目标。

• 实际示例：

系统提示： "你是一个AI客服分诊台。你的唯一任务是分析以下用户查询，并从以下类别中选择最合适的一个：['订单状态', '产品信息', '技术支持', '财务问题', '其他']。你必须只输出这五个类别中的一个词。"
用户输入： "你们的发票怎么开？"
LLM输出： 财务问题 - 引导到财务子智能体

• 
  

• 优缺点：

• 优点： 极其灵活，能理解自然语言的细微差别，无需提前定义严格规则。

• 缺点： 速度较慢（需要一次额外的 LLM 调用），成本较高，且（像所有 LLM 一样）本质上是非确定性的。

2. 基于规则的路由（确定性路由）

这是最传统、最快的方法，涉及使用预定义的规则（如 if-else 语句、正则表达式）来进行决策。

• 工作原理： 工程师预先编写逻辑，检查输入中是否存在特定的关键字、模式或结构化数据。

• 实际示例：

    def route_query(query: str) -> str:        query_lower = query.lower()        if "订单" in query_lower or "物流" in query_lower:            return "ORDER_STATUS"        elif "发票" in query_lower or "付款" in query_lower:            return "BILLING"        elif re.search(r'\b(冻结|死机|不工作)\b', query_lower):            return "TECH_SUPPORT"        else:            return "GENERAL_INFO"

• 优缺点：

• 优点： 速度极快，成本为零（只是本地代码执行），完全确定和可预测。

• 缺点： 非常僵化，只能覆盖已知规则，缺乏灵活性。如果用户说“我的包裹在哪？”，它会因为没有“订单”或“物流”关键字而匹配失败。它无法理解语义。

3. 基于嵌入的路由（语义路由）

这是一种在灵活性和成本之间取得巧妙平衡的现代方法。它利用了嵌入（Embeddings）的语义相似性能力。

• 工作原理：

• 预定义： 我们为每个可能的路由（如“订单状态”、“技术支持”）创建一个描述性短语，并预先计算它们的向量嵌入。

• 运行时： 我们获取用户查询，并将其转换为向量嵌入。

• 比较： 我们计算用户查询嵌入与所有预定义路由嵌入之间的“余弦相似度”。

• 决策： 查询被路由到语义上最相似的路径。

• 实际示例：

• 用户查询：“我的包裹卡住了”

• 它的嵌入向量在语义空间上会非常接近“订单状态”的嵌入，而不是“产品信息”。

• 优缺点：

• 优点： 速度快（嵌入计算比 LLM 调用快得多），能理解语义（“包裹卡住”和“订单物流”很接近），比规则灵活得多。

• 缺点： 不如 LLM 路由那样能处理极其复杂或多步骤的逻辑判断。

4. 基于机器学习模型的路由（专业路由）

这种方法涉及训练一个专门的、小型的分类模型（例如，一个微调过的 BERT 或 DistilBERT 模型）来执行路由任务。

• 工作原理： 你收集一个标记数据集（例如，1000 个用户查询及其对应的正确路由），然后使用这些数据来训练一个多类别分类器。

• 实际示例： 一个专门的 IntentClassifier 模型，它在推理时接收查询，并输出一个包含每个路由概率的 JSON，例如：{"ORDER_STATUS": 0.95, "TECH_SUPPORT": 0.05}。

• 优缺点：

• 优点： 在速度和准确性之间达到了最佳平衡。一旦训练完成，推理速度极快，并且是为你的特定任务高度优化的。

• 缺点： 需要数据科学专业知识，需要收集训练数据、进行模型训练和维护，是实现成本最高（指人力和流程）的方案。

四、路由模式的实际应用场景

路由模式是自适应 AI 系统设计的关键控制机制。它的效用跨越多个领域，为系统提供了必要的条件逻辑层。

1. 人机交互：智能虚拟助手

这是最直观的应用。无论是 Siri 还是企业内部的 AI 助手，路由都用于解读用户意图。

• 场景： 一个 AI 驱动的“智能辅导系统”。

• 路由逻辑：

• 输入： 学生提交了对某个数学问题的解答。

• 路由： AI 分析学生的答案。

• 路径 1（正确）： 如果答案正确，路由到 Congratulate_Chain（祝贺链），并解锁下一个模块。

• 路径 2（部分正确）： 如果答案的步骤正确但计算错误，路由到 Hint_Chain（提示链），指出计算错误。

• 路径 3（完全错误）： 如果概念错误，路由到 Reteach_Chain（重教链），返回到该概念的核心教学视频。

2. 自动化数据和文档处理管道

在企业后台，路由充当着数据分类和分发的功能。

• 场景： 一个公司的共享电子邮箱 xxx@company

• 路由逻辑：

• 输入： 收到一封新邮件。

• 路由： AI 分析邮件内容和附件。

• 路径 1（发票）： 如果附件是 PDF 且内容包含“Invoice”或“发票”，路由到 AccountsPayable_Workflow（应付账款工作流），自动提取金额和日期，并录入财务系统。

• 路径 2（简历）： 如果附件包含“Resume”或“CV”字样，路由到 HR_Workflow（人力资源工作流），提取候选人信息并存入 ATS（申请人跟踪系统）。

• 路径 3（销售线索）： 如果邮件正文包含“价格”、“演示”或“采购”等关键字，路由到 Sales_Workflow（销售工作流），在 CRM 中创建一条新的销售线索，并分配给销售人员。

3. 复杂的多代理系统（高级调度器）

在涉及多个专业工具或 AI 代理的复杂系统中，路由起着“高级调度器”或“项目经理”的作用。

• 场景： 一个 AI 研究助手，任务是“撰写一份关于 AIGC 行业最新趋势的综合报告”。

• 路由逻辑（由一个“协调器”Agent 执行）：

• 协调器： “要完成这份报告，我需要三个专业代理的帮助。”

• 路由 1： 任务“搜索最新的 AIGC 新闻和学术论文” $\to$ 路由到 Search_Agent（搜索代理）。

• 路由 2： 任务“分析竞争对手（如 OpenAI, Google）的最新动态” $\to$ 路由到 Competitor_Agent（竞争对手分析代理）。

• 路由 3： 任务“提取内部数据库中相关的项目数据” $\to$ 路由到 InternalDB_Tool（内部数据库工具）。

• 最终步骤（串行）： 协调器收集所有三个路径的输出，然后路由到 ReportWriter_Agent（报告撰写代理），生成最终报告。

五、落地实现：框架与代码示例

在代码中实现路由涉及定义可能的路径和决定选择哪条路径的逻辑。像 LangChain、LangGraph 和 Google ADK 这样的框架为此提供了结构化的组件。

1. LangChain / LangGraph

LangChain 提供了 RunnableBranch，这是一种在“LangChain 表达式语言（LCEL）”中实现条件逻辑的直接方式。而其兄弟项目 LangGraph 则更进一步，它提供了一个基于状态的图形架构，特别适合于复杂的、多步骤的、有循环的路由场景，因为它将工作流显式地建模为节点和边的图。

• LangChain 示例（概念）：

• 在 LangChain 中，你可以设置一个 RunnableBranch。

• 它会首先运行一个“路由链”（例如，一个基于 LLM 的分类器）。

• 然后，RunnableBranch 会根据该分类器的输出，将原始输入引导到两个或多个不同的处理链中的一个（例如，booking_handler 或 info_handler）。

• 这对于简单的 if-else 分支非常有效。

2. Google ADK (Agent Development Kit) 示例

Google 的 ADK 框架在处理路由时采用了不同的范式。它通常不依赖于显式定义的计算图（如 LangGraph），而是更专注于定义一组离散的“工具”和“子代理”。

路由（或称为“委派”）是通过框架的内部逻辑（称为“Auto-Flow”）来实现的。你通过自然语言指令告诉一个“协调器”代理它的工作职责和它手下的“团队”（子代理），ADK 的底层模型会利用这些信息将用户的意图与正确的功能处理程序（工具或子代理）相匹配。

以下是一个使用 Google ADK 的 Python 代码示例，它演示了这种基于“委派”的路由模式：

import uuidfrom typing import Dict, Any, Optionalfrom google.adk.agents import Agentfrom google.adk.runners import InMemoryRunnerfrom google.adk.tools import FunctionToolfrom google.genai import typesfrom google.adk.events import Eventimport nest_asyncioimport asyncio
# --- 1. 定义工具函数 (模拟子代理的实际工作) ---# 这些函数模拟了专业代理的最终操作。
def booking_handler(request: str) -> str:    """处理航班和酒店的预订请求。"""    print("-------------------------- 预订处理程序被调用 ----------------------------")    return f"预订操作已模拟：'{request}'。"
def info_handler(request: str) -> str:    """处理一般信息请求。"""    print("-------------------------- 信息处理程序被调用 ----------------------------")    return f"信息请求已模拟：'{request}'。结果：模拟信息检索。"
# --- 2. 从函数创建工具 ---booking_tool = FunctionTool(booking_handler)info_tool = FunctionTool(info_handler)
# --- 3. 定义专业的子代理 ---# 每个子代理都配备了自己专有的工具。
booking_agent = Agent(    name="Booker",    model="gemini-2.0-flash", # 假设使用 gemini-2.0-flash    description="一个专业代理，通过调用预订工具来处理所有航班和酒店的预订请求。",    tools=[booking_tool])
info_agent = Agent(    name="Info",    model="gemini-2.0-flash", # 假设使用 gemini-2.0-flash    description="一个专业代理，通过调用信息工具来提供一般信息并回答用户问题。",    tools=[info_tool])
# --- 4. 定义协调器代理 (路由器) ---# 这是路由逻辑的核心。注意看 'instruction' 参数。
coordinator = Agent(    name="Coordinator",    model="gemini-2.0-flash", # 假设使用 gemini-2.0-flash    instruction=(        "你是一个主协调器。你唯一的任务是分析传入的用户请求 "        "并将它们委派给相应的专业代理。"        "绝对不要尝试自己直接回答用户的问题。\n"        "- 对于任何与预订航班或酒店相关的请求，委派给 'Booker' 代理。\n"        "- 对于所有其他一般信息问题，委派给 'Info' 代理。"    ),    description="一个将用户请求路由到正确专业代理的协调器。",    # 提供了sub_agents列表，ADK将自动启用LLM驱动的委派（Auto-Flow）    sub_agents=[booking_agent, info_agent])
# --- 5. 执行逻辑 ---async def run_coordinator(runner: InMemoryRunner, request: str):    """使用给定的请求运行协调器并处理委派。"""    print(f"\n--- 运行协调器，请求: '{request}' ---")    final_result = ""    try:        user_id = "user_123"        session_id = str(uuid.uuid4())        await runner.session_service.create_session(            app_name=runner.app_name, user_id=user_id, session_id=session_id        )
        # 运行ADK并处理事件流        for event in runner.run(            user_id=user_id,            session_id=session_id,            new_message=types.Content(                role='user',                parts=[types.Part(text=request)]            ),        ):            # 检查这是否是包含内容的最终响应事件            if event.is_final_response() and event.content:                if hasattr(event.content, 'text') and event.content.text:                    final_result = event.content.text                elif event.content.parts:                    text_parts = [part.text for part in event.content.parts if part.text]                    final_result = "".join(text_parts)                                print(f"协调器最终响应: {final_result}")                break # 获得最终响应后退出循环                return final_result    except Exception as e:        print(f"处理请求时发生错误: {e}")        return f"处理请求时发生错误: {e}"
async def main():    """运行ADK示例的主函数。"""    print("--- Google ADK 路由示例 (ADK Auto-Flow 风格) ---")    print("注意: 这需要已安装并认证 Google ADK。")        runner = InMemoryRunner(coordinator)        # --- 示例用法 ---        result_a = await run_coordinator(runner, "帮我预订一个巴黎的酒店。")    print(f"最终输出 A: {result_a}") # 应该路由到 Booker
    result_b = await run_coordinator(runner, "世界上最高的山是哪座？")    print(f"最终输出 B: {result_b}") # 应该路由到 Info
    result_c = await run_coordinator(runner, "告诉我一个随机的冷知识。")    print(f"最终输出 C: {result_c}") # 应该路由到 Info
    result_d = await run_coordinator(runner, "查找下个月去东京的航班。")    print(f"最终输出 D: {result_d}") # 应该路由到 Booker
if __name__ == "__main__":    # 允许在Jupyter/IPython等环境中运行asyncio    nest_asyncio.apply()    asyncio.run(main())

代码解释：

• 1-3 步（设置）： 我们定义了两个模拟的“最终操作”（booking_handler, info_handler），将它们包装成 FunctionTool，然后创建了两个专业的 Agent（Booker, Info），每个代理只拥有自己的工具。

• 第 4 步（路由核心）： 我们创建了 Coordinator（协调器）代理。它的魔力在于 instruction 参数。我们用自然语言明确指示它“不要自己回答问题”，而是根据请求内容将任务委派给 Booker 或 Info。

• Auto-Flow： 因为 Coordinator 在其 sub_agents 列表中定义了 Booker 和 Info，ADK 的“Auto-Flow”机制会自动接管。它会理解 Coordinator 的指令，并在运行时自动执行“LLM 路由”，将用户的请求正确地委派给相应的子代理，子代理再调用其工具返回结果。

• 第 5 步（执行）： main 函数展示了这一点。当我们发送“帮我预订一个巴黎的酒店”时，Coordinator（路由）会将其交给 Booker（子代理），后者调用 booking_tool（工具）并返回模拟的确认信息。

结论：从“执行者”到“决策者”

路由模式是 AI 智能体开发中一个关键的飞跃。它将系统从“线性执行者”转变为“动态决策者”。

通过实现路由，我们超越了简单的顺序工作流，使 AI 智能体能够就如何处理信息、响应用户输入以及利用可用工具集做出智能决策。我们已经看到，无论是通过显式的 RunnableBranch（LangChain）、复杂的状态图（LangGraph），还是通过基于自然语言指令的自动委派（Google ADK），其核心思想都是一致的：为 AI 引入条件逻辑和决策能力。

掌握路由模式对于构建能够智能地导航不同场景、处理现实世界任务固有变异性的 AI 应用来说，是必不可少的一步。它是创建多功能、健壮且真正有用的 AI 系统的关键组件。

参考资料

1. LangChain 文档: https://python.langchain.com/v0.2/docs/core_modules/expression_language/

2. LangGraph 文档: https://langchain-ai.github.io/langgraph

3. Google ADK 文档: https://google.github.io/adk-docs

4. Antonio Gulli 《Agentic Design Patterns》