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

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

想象一下，你的“流水线”上突然来了一个完全不相关的请求。一个用户不会总是按照你设计的A — B — 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. 步骤2：执行（选择的路径）

    • 技术支持子智能体（一个专门的提示链）：

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

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

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

用户查询 路由器决策（意图） 路由目标（子流程/工具）
“我的订单什么时候到？” 订单状态 OrderDB_Tool（订单数据库工具）
“介绍一下新款相机。” 产品信息 ProductInfo_Chain（产品信息链）
“相机屏幕冻结了！” 技术支持 TechSupport_Agent（技术支持子智能体）
“你们的发票怎么开？” 财务问题 Billing_Chain（财务处理链）
“呃，那个东西……” 意图不明 Clarification_Chain（澄清提示链）

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

三、实现路由的四种方式

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

1. 基于LLM的路由（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）的语义相似性能力。

  • 工作原理：

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

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

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

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

  • 实际示例：

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

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

  • 优缺点：

    • 优点： 速度快（嵌入计算比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新闻和学术论文” 路由到Search_Agent（搜索代理）。

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

    • 路由3： 任务“提取内部数据库中相关的项目数据” 路由到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代码示例，它演示了这种基于“委派”的路由模式：

# Copyright (c) 2025 Marco Fago
# This code is licensed under the MIT License.
# See the LICENSE file in the repository for the full license text.

import uuid
from typing import Dict, Any, Optional
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.adk.tools import FunctionTool
from google.genai import types
from google.adk.events import Event
import nest_asyncio
import 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》

# 人工智能 # AI智能体
Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

控制层和方向层的分野：OpenProse、Natural-Language Agent Harnesses 与 Attractor-Guided Engineering

本文探讨了OpenProse、NLAHs和AGE三种框架在解决AI agent产出不可靠问题上的不同思路。OpenProse通过确定性runtime维护目标不变性，NLAHs将控制策略外化为可审查的自然语言文档，而AGE则关注系统长期结构稳定性的工程方法。关键区别在于：OpenProse和NLAHs采用控制目标（fixed point）范式，依赖外部指令纠正偏差；AGE则借鉴数学吸引子概念，通过文

avatar AI Agent技术社区
cover

小白程序员转行AI Agent工程师的完整学习路线，高薪进阶必备

avatar AI Agent技术社区

AI 智能体（AI Agent）完整课程讲义

为解决 LLM 无法落地真实业务操作的问题，核心解决方案就是AI 智能体。内置海量现成模板：全自动生成 AI 短视频、智能分类整理邮件等；自由组合 LLM、第三方接口、本地工具，按需定制专属业务智能体；适合有个性化业务流程、需要高度自定义 Agent 的新手。

avatar AI Agent技术社区