《Agentic Design Patterns：构建智能系统的实战指南》- 第二章路由

书籍名称：Agentic Design Patterns: A Hands-On Guide to Building Intelligent Systems
本书作者：Antonio Gulli
链接地址：https://docs.google.com/document/d/1rsaK53T3Lg5KoGwvf8ukOUvbELRtH-V0LnOIFDxBryE
内容摘要：本文是对《智能体设计模式》一书第二章节的翻译，此章节主要介绍了路由（Routing）模式。

摘要

智能体系统通常需要应对种类繁多的输入与情境，这些情况无法通过单一的线性流程来处理。简单的顺序工作流缺乏根据上下文进行决策的能力。如果系统没有一种机制来选择针对特定任务的合适工具或子流程，它就会显得僵化且缺乏适应性。这一局限性使得构建能够处理真实世界中复杂多变用户请求的高级应用变得困难。路由（Routing）模式通过在智能体的运行框架中引入条件逻辑，为上述问题提供了标准化的解决方案。它使系统能够先对输入请求进行分析，以判断其意图或性质，然后根据分析结果，动态地将控制流程引导至最合适的专用工具、函数或子智能体。该决策过程可以通过多种方式实现，包括使用大语言模型（LLM）提示、应用预定义规则，或基于向量嵌入的语义相似度判断。最终，路由机制将原本静态、预设的执行路径转化为灵活、具备上下文感知能力的工作流，使系统能够自动选择最优行动方案。

经验法则：当一个智能体需要根据用户输入或当前状态，在多个不同的工作流、工具或子智能体之间进行选择时，应采用“路由”模式。这对于需要将输入请求进行分类、分流的应用场景尤为关键，例如让客服机器人区分处理销售咨询、技术支持和账户管理等不同类型的问题。

《Agentic Design Patterns：构建智能系统的实战指南》- 第二章路由 — 图 1：使用大语言模型作为路由器的“路由”模式示意图

路由模式概述

虽然提示链为语言模型提供了一种实现线性、确定性流程的基础手段，然而，在需要灵活应对变化或进行动态决策的场景中，它的作用往往显得不足。现实中的智能体系统通常需要根据各种条件因素（例如环境状态、用户输入或前一步操作的结果）在多个潜在的行动之间进行决策。这种动态决策能力控制了流程的走向，将任务分配给不同的功能模块、工具或子流程，这一机制被称为路由。路由在智能体的运行框架中引入了条件逻辑（conditional logic），使系统不再沿着固定路径执行，而是能够根据特定条件进行动态评估与决策，从多个后续操作中选择最合适的一项。这种机制让系统表现出更高的灵活性与情境感知能力。

举例来说，一个用于回答客户咨询的智能体，如果配备了路由功能，就能够先对用户输入的请求进行意图识别。根据识别结果，智能体可以将问题转交给专门的问答子智能体、调用数据库检索工具获取账户信息、或在遇到复杂问题时接入人工服务，而不是像传统系统那样，始终按照单一路径生成预设回答。

一个具备路由功能的高级智能体可以执行如下操作：

1. 分析用户的查询内容

2. 根据意图进行路由：

若意图为“查询订单状态”，则路由到与订单数据库交互的子智能体或工具链；
若意图为“产品信息”，则路由到查询产品目录的子智能体或链；
若意图为“技术支持”，则路由到故障排查指南或人工支持的链；
若意图不明确，则路由到澄清意图的子智能体或提示链。

路由模式的实现方式

路由模式的核心在于评估与引导机制，该机制负责分析输入并决定信息流的走向。其实现方式包括以下几种：

1. 基于 LLM 的路由（LLM-based Routing）

大语言模型本身可以通过提示来执行输入分析，并输出一个标识符或指令，以指示下一步操作。例如，提示内容可能是：

“请分析以下用户查询，只输出其所属类别：‘订单状态（Order Status）’、‘产品信息（Product Info）’、‘技术支持（Technical Support）’或‘其他（Other）’。”

系统随后读取模型输出，根据结果将工作流导向对应的分支。

2. 基于规则的路由（Rule-based Routing）

通过预定义规则或逻辑结构（如 if-else 语句、switch 分支）来执行路由。这类方法可依据关键字、模式或输入中提取的结构化数据进行判断。其优点是速度快、确定性强，但对于语义复杂或新颖的输入，灵活性不足。

3. 基于嵌入的路由（Embedding-based Routing）

输入查询会被转换为向量嵌入（vector embedding）（参见第14章 RAG）。该嵌入向量会与表示不同功能路径或能力的嵌入进行比较，并将查询路由到最相似的路径。这种方法适用于语义路由（semantic routing），即根据输入的语义而非关键字进行决策。

4. 基于机器学习模型的路由（Machine Learning Model-Based Routing）

此方法使用判别模型，例如在少量标注数据上训练的分类器，用于执行特定的路由任务。与基于嵌入的方法类似，但其核心在于监督微调，即通过调整模型参数来获得专用的路由能力。这种方法与基于 LLM 的路由存在一些不同：决策逻辑不是通过提示驱动的生成模型在推理时完成，而是直接编码在模型的权重中。LLM 可能在前期用于生成合成数据以扩充训练集，但在实时路由决策中不参与推理。

路由的应用位置

路由机制可以在智能体的运行周期中多个阶段进行实现。它既可以在任务开始时，用于识别与分类主要任务类型；也可以在处理链的中间环节中，用于决定下一步执行的动作；还可以在子流程中，用于从多个可选工具中挑选最合适的一个。目前，一些计算框架（如 LangChain、LangGraph 以及 Google 的 Agent Developer Kit（ADK））都提供了明确的结构，用于定义与管理此类条件逻辑。其中，LangGraph 采用基于状态的图结构，特别适合处理那些依赖于整个系统累积状态进行决策的复杂路由场景。而 Google ADK 则提供了用于构建智能体能力与交互模型的基础组件，为实现路由逻辑奠定了架构基础。在这些框架所提供的执行环境中，开发者可以定义多种可能的操作路径，并指定函数或基于模型的评估逻辑，以决定计算图中不同节点之间的转移关系。

通过实现路由机制，系统得以突破传统的确定性顺序处理模式，从而构建出更加灵活、自适应的执行流程，能够根据不同的输入与系统状态变化进行动态且恰当的响应。

路由的实际运用

路由模式是构建自适应智能体系统的重要控制机制。它使系统能够根据不同的输入内容和变化的内部状态，动态调整任务执行的路径，从而在不同领域中发挥关键作用，为模型提供必要的条件判断逻辑。

在人机交互场景中，例如虚拟助手或智能教学系统，路由机制用于识别并解析用户意图。系统会首先对自然语言输入进行初步分析，然后选择最合适的后续操作：可能是调用特定的信息检索工具，也可能是将请求转交给人工处理，或根据用户的表现选择教学流程中的下一个模块。这种方式使系统能够超越单一、线性的对话流程，实现更加灵活和符合语境的响应。

在自动化数据或文档处理流程中，路由机制承担着分类与分发的职能。系统会对输入数据的内容、元数据或格式（如电子邮件、工单或 API 负载）进行分析，然后将不同类型的数据指派给相应的处理流程，比如录入销售线索、针对 JSON 或 CSV 格式数据进行转换、对紧急问题快速上报。

在包含多个专用工具或智能体的复杂系统中，路由机制则起到高层调度器的作用。例如，一个科研系统可能由负责检索、摘要和分析的不同智能体组成，路由器会根据当前任务目标，将具体操作分配给最合适的智能体。同样，AI 编程助手也依赖路由机制来识别编程语言和用户意图，判断其需求是调试、解释还是代码转换，然后将代码片段传递给对应的专用工具进行处理。

总体来看，路由机制赋予了智能体逻辑决策与任务裁定的能力。它是实现多功能、具备情境感知能力系统的基础。通过路由，智能体不再只是被动执行预设指令的工具，而能够在动态环境中自主决策，选择最合适的方式完成任务，从而成为真正具备适应性和智能性的系统。

实战代码示例（LangChain）

在代码中实现路由时，核心在于定义可选的执行路径以及决定路径选择的逻辑。像 LangChain 和 LangGraph 这样的框架，为此提供了专门的组件与结构。其中，LangGraph 的基于状态的图结构尤其直观，便于可视化和实现路由逻辑。

下面这段代码演示了一个基于 LangChain 和 Google 生成式 AI 的简单代理系统。它设置了一个“协调器”（coordinator），根据用户请求的意图（预订、信息查询或不明确操作）将请求分派给不同的“子代理”处理程序（sub-agent handler）。系统使用语言模型来判断请求类别，然后将任务委派给对应的处理函数，模拟多代理架构中常见的“任务分派”模式。

在开始之前，请确保安装以下依赖库：

pip install langchain langgraph google-cloud-aiplatform langchain-google-genai google-adk deprecated pydantic

你还需要设置语言模型的 API 密钥环境变量（例如 OpenAI、Google Gemini 或 Anthropic）。

# 版权所有 (c) 2025 Marco Fago
# https://www.linkedin.com/in/marco-fago/
#
# 本代码基于 MIT 许可证授权
# 有关完整许可条款，请参阅仓库中的 LICENSE 文件

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableBranch

# --- 配置部分 ---
# 确保你的 API 密钥环境变量已正确设置（例如 GOOGLE_API_KEY）
try:
   llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0)
   print(f"语言模型已初始化: {llm.model}")
except Exception as e:
   print(f"语言模型初始化失败: {e}")
   llm = None

# --- 定义模拟的子代理处理程序（相当于 ADK 的 sub_agents） ---

def booking_handler(request: str) -> str:
   """模拟预订代理处理请求"""
   print("\n--- 委派至预订处理程序 ---")
   returnf"预订处理程序已处理请求: '{request}'。结果：模拟的预订操作。"

def info_handler(request: str) -> str:
   """模拟信息查询代理处理请求"""
   print("\n--- 委派至信息处理程序 ---")
   returnf"信息处理程序已处理请求: '{request}'。结果：模拟的信息检索操作。"

def unclear_handler(request: str) -> str:
   """处理无法识别或不明确的请求"""
   print("\n--- 处理不明确的请求 ---")
   returnf"协调器无法委派该请求: '{request}'。请进一步说明。"

# --- 定义协调器路由链（相当于 ADK 协调器的指令） ---
# 该链用于决定将请求委派给哪个处理程序。
coordinator_router_prompt = ChatPromptTemplate.from_messages([
   ("system", """分析用户请求并确定应由哪个专业处理程序负责。
    - 若请求涉及预订航班或酒店，请输出 'booker'；
    - 若请求为一般信息查询，请输出 'info'；
    - 若请求不清晰或不属于上述类别，请输出 'unclear'。
    仅输出一个单词：'booker'、'info' 或 'unclear'。"""),
   ("user", "{request}")
])

if llm:
   coordinator_router_chain = coordinator_router_prompt | llm | StrOutputParser()

# --- 定义委派逻辑（相当于 ADK 的自动流程，根据 sub_agents 分派） ---
# 使用 RunnableBranch 根据路由结果将任务分派至对应处理程序。

# 定义各个分支逻辑
branches = {
   "booker": RunnablePassthrough.assign(output=lambda x: booking_handler(x['request']['request'])),
   "info": RunnablePassthrough.assign(output=lambda x: info_handler(x['request']['request'])),
   "unclear": RunnablePassthrough.assign(output=lambda x: unclear_handler(x['request']['request'])),
}

# 创建 RunnableBranch。
# 它接收路由链的输出，并根据分类结果将原始请求转发给对应的处理函数。
delegation_branch = RunnableBranch(
   (lambda x: x['decision'].strip() == 'booker', branches["booker"]), # 增加 .strip() 防止多余空格
   (lambda x: x['decision'].strip() == 'info', branches["info"]),     # 同上
   branches["unclear"] # 默认分支，处理“不明确”或其他输出
)

# 将路由链与委派分支组合为一个完整的可运行单元
# 路由链的输出（decision）与原始输入（request）一并传递给委派逻辑
coordinator_agent = {
   "decision": coordinator_router_chain,
   "request": RunnablePassthrough()
} | delegation_branch | (lambda x: x['output'])  # 提取最终输出

# --- 示例运行 ---
def main():
   ifnot llm:
       print("\n由于语言模型初始化失败，跳过执行。")
       return

   print("--- 运行预订请求示例 ---")
   request_a = "帮我订一张去伦敦的机票。"
   result_a = coordinator_agent.invoke({"request": request_a})
   print(f"最终结果 A: {result_a}")

   print("\n--- 运行信息查询请求示例 ---")
   request_b = "意大利的首都是哪里？"
   result_b = coordinator_agent.invoke({"request": request_b})
   print(f"最终结果 B: {result_b}")

   print("\n--- 运行不明确请求示例 ---")
   request_c = "告诉我关于量子物理的事情。"
   result_c = coordinator_agent.invoke({"request": request_c})
   print(f"最终结果 C: {result_c}")

if __name__ == "__main__":
   main()

代码说明

如上所示，这段 Python 代码利用 LangChain 库与 Google 的生成式 AI 模型（gemini-2.5-flash）构建了一个简易的代理系统。代码中定义了三个模拟的子代理处理程序：booking_handler、info_handler 和 unclear_handler，分别用于处理不同类型的请求。核心部分是 coordinator_router_chain，它通过一个 ChatPromptTemplate 指令，要求语言模型将用户请求分类为三种类型之一：booker、info 或 unclear。模型的输出结果会传递给 RunnableBranch，由它将请求分派至相应的处理函数。RunnableBranch 根据分类结果，将数据定向传输给对应的处理程序。coordinator_agent 将这些组件整合起来，先通过语言模型进行决策，然后再执行委派，最后提取处理结果作为系统输出。

主函数 main() 展示了三个示例请求，分别对应不同的路由与处理逻辑。代码中还包含了语言模型初始化失败时的错误处理，以保证系统的稳健性。整体结构模拟了一个基础的多代理框架：由中央协调器根据意图将任务分派给专业化的子代理进行处理。

实操代码示例（Google ADK）

Agent Development Kit（简称 ADK）是一个用于构建智能体系统的框架，它为定义智能体的能力和行为提供了结构化的开发环境。与基于显式计算图的架构不同，在 ADK 范式中，路由通常是通过定义一组离散的“工具”来实现的，这些工具代表智能体可执行的具体功能。针对用户查询选择合适的工具这一过程由框架的内部逻辑自动管理，它利用底层模型将用户意图匹配到正确的功能处理器上。

下面这段 Python 代码演示了一个使用 Google ADK 库实现的应用示例。代码设置了一个名为Coordinator（协调者）的主智能体，用于根据定义的指令，将用户请求路由到专门的子智能体（如负责预订任务的 Booker 和提供信息的 Info）。子智能体通过各自的工具函数模拟请求的处理过程，展示了智能体系统中的基本任务委派模式。

# 版权所有 (c) 2025 Marco Fago
#
# 本代码基于 MIT 许可证发布。
# 详细许可信息请参见仓库中的 LICENSE 文件。

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

# --- 定义工具函数 ---
# 以下函数用于模拟各个专业子智能体的行为。

def booking_handler(request: str) -> str:
   """
   处理航班和酒店预订请求。
   参数：
       request：用户的预订请求。
   返回：
       一条确认消息，表示该预订请求已被模拟处理。
   """
   print("-------------------------- 调用 Booking Handler ----------------------------")
   returnf"已模拟执行预订请求：'{request}'。"

def info_handler(request: str) -> str:
   """
   处理一般信息类请求。
   参数：
       request：用户的问题。
   返回：
       一条消息，表示信息请求已被模拟处理。
   """
   print("-------------------------- 调用 Info Handler ----------------------------")
   returnf"信息请求：'{request}'。结果：已模拟信息检索。"

def unclear_handler(request: str) -> str:
   """处理无法被分配的请求。"""
   returnf"协调者无法分配请求：'{request}'。请进一步说明。"

# --- 从函数创建工具 ---
booking_tool = FunctionTool(booking_handler)
info_tool = FunctionTool(info_handler)

# 定义具备相应工具的专业子智能体
booking_agent = Agent(
   name="Booker",
   model="gemini-2.0-flash",
   description="专门处理航班和酒店预订请求的智能体，调用 booking 工具执行任务。",
   tools=[booking_tool]
)

info_agent = Agent(
   name="Info",
   model="gemini-2.0-flash",
   description="专门提供一般信息和回答用户问题的智能体，调用 info 工具执行任务。",
   tools=[info_tool]
)

# 定义主协调智能体，并明确指定其委派规则
coordinator = Agent(
   name="Coordinator",
   model="gemini-2.0-flash",
   instruction=(
       "你是主协调者。你的唯一任务是分析用户的请求，"
       "并将其委派给合适的专业智能体。请不要直接回答用户。\n"
       "- 如果请求与航班或酒店预订相关，请委派给 'Booker' 智能体。\n"
       "- 其他一般信息类问题，请委派给 'Info' 智能体。"
   ),
   description="一个负责将用户请求路由到正确专业智能体的协调者。",
   # 定义子智能体后，系统将默认启用基于 LLM 的自动委派（Auto-Flow）。
   sub_agents=[booking_agent, info_agent]
)

# --- 执行逻辑 ---

asyncdef 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
       )

       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:
               # 优先从 event.content 直接获取文本，避免不必要的遍历
               if hasattr(event.content, 'text') and event.content.text:
                    final_result = event.content.text
               elif event.content.parts:
                   # 备用方案：遍历 parts 提取文本（可能会产生警告）
                   text_parts = [part.text for part in event.content.parts if part.text]
                   final_result = "".join(text_parts)
               # 获取最终响应后退出循环
               break

       print(f"协调者最终响应：{final_result}")
       return final_result
   except Exception as e:
       print(f"处理请求时发生错误：{e}")
       returnf"处理请求时发生错误：{e}"

asyncdef main():
   """主函数：运行 ADK 路由示例。"""
   print("--- Google ADK 路由示例（Auto-Flow 风格） ---")
   print("注意：运行前需安装并认证 Google ADK。")

   runner = InMemoryRunner(coordinator)
   # 示例使用
   result_a = await run_coordinator(runner, "帮我预订巴黎的酒店。")
   print(f"最终输出 A: {result_a}")
   result_b = await run_coordinator(runner, "世界上海拔最高的山是什么？")
   print(f"最终输出 B: {result_b}")
   result_c = await run_coordinator(runner, "告诉我一个随机的趣闻。")  # 应由 Info 处理
   print(f"最终输出 C: {result_c}")
   result_d = await run_coordinator(runner, "查找下个月去东京的航班。")  # 应由 Booker 处理
   print(f"最终输出 D: {result_d}")

if __name__ == "__main__":
   import nest_asyncio
   nest_asyncio.apply()
   await main()

该脚本由一个主协调智能体 Coordinator 和两个专业子智能体 Booker 与 Info 组成。每个子智能体都配备了一个 FunctionTool，用于包装对应的 Python 函数来模拟实际操作。其中，booking_handler 模拟航班和酒店预订的处理，info_handler 模拟一般信息的检索，而 unclear_handler 则作为协调者无法分配请求时的备用处理器（虽然当前逻辑中未被显式使用）。

Coordinator 智能体的主要职责是分析输入的用户请求，并将其委派给 Booker 或 Info 智能体。由于 Coordinator 定义了子智能体，因此 ADK 会自动启用 Auto-Flow 自动委派机制来执行分派。run_coordinator 函数通过创建用户与会话 ID，利用 InMemoryRunner 运行协调者智能体，并从事件流中提取最终响应文本。

main 函数演示了系统的实际运行方式，通过不同的用户请求展示了路由机制如何自动将预订类请求委派给 Booker，而信息类请求则交由 Info 智能体处理，从而实现智能化、上下文感知的任务分派。

本章重点

路由使智能体能够根据条件动态决定下一步执行的操作。
它让智能体具备处理多样化输入并自适应行为的能力，从而突破线性执行模式。
路由逻辑可以通过大语言模型 、基于规则的系统或嵌入相似度等方式实现。
框架如 LangGraph 与 Google ADK 均提供了在智能体工作流中定义和管理路由的结构化方案，尽管二者的架构设计各有不同。

总结

路由模式是在构建真正动态、具备响应能力的智能体系统中至关重要的一步。通过实现路由机制，我们能够突破单一、线性的执行流程，使智能体具备智能决策的能力——能够自主判断如何处理信息、如何响应用户输入，以及如何调用可用的工具或子智能体。

路由的应用范围十分广泛，从客服聊天机器人到复杂的数据处理流水线，都能体现其价值。分析输入并根据条件动态引导工作流的能力，是打造能够应对真实世界任务多样性和不确定性的智能体的基础。

文中展示的 LangChain 与 Google ADK 示例分别代表了两种不同但同样有效的路由实现方式。LangGraph 采用基于图的结构，能够以可视化、显式的方式定义状态与转移，非常适合具有复杂多步逻辑的工作流。而 Google ADK 则更注重定义明确的功能模块，并依托框架本身将用户请求自动路由至合适的工具处理器，对于功能边界清晰、操作集较固定的智能体而言，这种方式更加简洁高效。

掌握路由模式是构建智能体系统的关键能力之一。它让智能体能够在不同场景中进行智能决策，并根据上下文生成合适的响应或执行相应的操作。可以说，路由机制是构建灵活、稳健且具备真实智能特征的智能体应用的核心组成部分。