第2章：路由 Routing
路由模式概述
虽然通过提示链进行顺序处理是执行确定性线性工作流的一种基础技术，但在需要自适应响应的场景中，其适用性有限。现实世界中的智能智能体系统通常需要根据环境状态、用户输入或前一个操作的结果，在多个潜在操作之间进行仲裁。这种动态决策能力通过一种称为路由的机制实现，控制流程被引导到不同的专用功能、工具或子流程。
路由在智能体的操作框架中引入了条件逻辑，使其从固定的执行路径转变为动态评估特定标准以从一组可能的后续操作中进行选择的模型。这种机制使系统行为更加灵活且能够感知上下文。
例如，一个为客户查询设计的智能体在配备路由功能后，可以首先对收到的查询进行分类以确定用户的意图。根据这一分类，它可以将查询引导至专门的直接问答智能体、用于账户信息的数据库检索工具，或复杂问题的升级处理程序，而不是默认采用单一的预设响应路径。因此，一个使用路由的更高级智能体可以：
- 分析用户的查询。
- 根据查询的意图进行路由：
- 如果意图是“检查订单状态”，则路由到与订单数据库交互的子智能体或工具链。
- 如果意图是“产品信息”，则路由到搜索产品目录的子智能体或工具链。
- 如果意图是“技术支持”，则路由到访问故障排除指南或升级至人工支持的不同工具链。
- 如果意图不明确，则路由到澄清子智能体或提示链。
路由模式的核心组件是执行评估并引导流程的机制。该机制可以通过多种方式实现：
- 基于LLM的路由： 语言模型本身可以通过提示分析输入，并输出特定标识符或指令，以指示下一步或目的地。例如，可以提示LLM执行以下任务：“分析以下用户查询，并仅输出类别：‘订单状态’、‘产品信息’、‘技术支持’或‘其他’。”智能智能体系统随后读取此输出并相应地引导工作流程。
- 基于嵌入的路由： 输入查询可以转换为向量嵌入（详见RAG，第14章）。然后将该嵌入与代表不同路由或功能的嵌入进行比较。查询将被路由到与其嵌入最相似的路由。这种方法适用于语义路由，其决策基于输入的意义而不仅仅是关键词。
- 基于规则的路由： 使用预定义的规则或逻辑（例如，if-else语句、switch语句），基于从输入中提取的关键词、模式或结构化数据进行路由。这种方法比基于LLM的路由更快、更具确定性，但在处理复杂或新颖输入时灵活性较差。
- 基于机器学习模型的路由： 使用一种判别模型，例如分类器，该模型专门在一小部分标注数据上进行训练以执行路由任务。虽然与基于嵌入的方法在概念上有相似之处，但其关键特征是监督微调过程，通过调整模型参数来创建专门的路由功能。这种技术与基于LLM的路由不同，因为决策组件不是在推理时执行提示的生成模型。相反，路由逻辑编码在经过微调的模型的学习权重中。虽然LLM可能在预处理步骤中用于生成合成数据以扩充训练集，但它们不会参与实时路由决策。
路由机制可以在智能智能体的操作周期内的多个节点上实现。它可以在开始时应用于分类主要任务，在处理链的中间点确定后续动作，或在子程序中选择给定工具集中最合适的工具。
LangChain、LangGraph和Google的Agent Developer Kit（ADK）等计算框架提供了明确的结构，用于定义和管理此类条件逻辑。LangGraph凭借其基于状态的图架构，特别适合于复杂的路由场景，在这些场景中，决策取决于整个系统的累积状态。同样，Google的ADK提供了用于构建智能体功能和交互模型的基础组件，这些组件是实现路由逻辑的基础。在这些框架提供的执行环境中，开发人员定义可能的操作路径，以及决定节点之间转换的函数或基于模型的评估。
路由的实现使系统能够超越确定性的顺序处理。它促进了更具适应性的执行流程的开发，可以动态且适当地响应更广泛的输入和状态变化。
实际应用及使用场景
路由模式是设计自适应智能智能体系统中的关键控制机制，使其能够根据可变的输入和内部状态动态调整执行路径。其用途涵盖多个领域，为系统提供必要的条件逻辑层。
在人机交互中，例如虚拟助手或基于人工智能的导师，路由用于解释用户意图。对自然语言查询的初步分析决定了最适合的后续操作，无论是调用特定的信息检索工具、升级到人工操作员，还是根据用户表现选择课程中的下一个模块。这使系统能够超越线性对话流程，并根据上下文进行响应。
在自动化数据和文档处理管道中，路由充当分类和分发功能。传入的数据（如电子邮件、支持票据或 API 负载）根据内容、元数据或格式进行分析。系统随后将每项内容引导至相应的工作流程，例如销售线索摄取过程、针对 JSON 或 CSV 格式的特定数据转换功能，或紧急问题升级路径。
在涉及多个专业工具或智能体的复杂系统中，路由充当高级调度器。一个由搜索、摘要和信息分析等不同智能体组成的研究系统会使用路由器根据当前目标将任务分配给最合适的智能体。同样，AI 编程助手使用路由来识别编程语言和用户意图（调试、解释或翻译），然后将代码片段传递给正确的专业工具。
最终，路由提供了逻辑仲裁的能力，这是创建功能多样且具有上下文感知系统的关键。它将智能体从预定义序列的静态执行者转变为动态系统，能够根据变化的条件决定完成任务的最有效方法。
实践代码示例（LangChain）
在代码中实现路由涉及定义可能的路径以及决定选择哪条路径的逻辑。像 LangChain 和 LangGraph 这样的框架提供了具体的组件和结构来实现这一点。LangGraph 的基于状态的图结构特别直观，适合可视化和实现路由逻辑。
以下代码展示了一个使用 LangChain 和 Google 的生成式 AI 构建的简单智能体系统。它设置了一个“协调器”，根据请求的意图（预订、信息或不明确）将用户请求路由到不同的模拟“子智能体”处理程序。系统使用语言模型对请求进行分类，然后将其委派给适当的处理函数，模拟了多智能体架构中常见的基本委派模式。
首先，确保安装了必要的库：
pip install langchain langgraph google-cloud-aiplatform langchain-google-genai google-adk deprecated pydantic
您还需要使用您选择的语言模型（例如 OpenAI、Google Gemini、Anthropic）的 API 密钥设置环境。
## 版权所有 (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 子智能体） ---
def booking_handler(request: str) -> str:
"""模拟预订智能体处理请求。"""
print("\n--- 委派至预订处理程序 ---")
return f"预订处理程序已处理请求: '{request}'。结果: 模拟预订操作。"
def info_handler(request: str) -> str:
"""模拟信息智能体处理请求。"""
print("\n--- 转交至信息处理程序 ---")
return f"信息处理程序已处理请求: '{request}'。结果: 模拟信息检索。"
def unclear_handler(request: str) -> str:
"""处理无法转交的请求。"""
print("\n--- 处理不明确的请求 ---")
return f"协调器无法转交请求: '{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 自动流） ---
## 使用 RunnableBranch 根据路由链的输出进行路由。
## 定义 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。它接收路由链的输出，
## 并将原始输入（'request'）路由到相应的处理程序。
delegation_branch = RunnableBranch(
(lambda x: x['decision'].strip() == 'booker', branches["booker"]), # 添加 .strip()
(lambda x: x['decision'].strip() == 'info', branches["info"]), # 添加 .strip()
branches["unclear"] # 默认分支用于 'unclear' 或任何其他输出
)
## 将路由链和转交分支组合成一个可运行的实体
## 路由链的输出（'decision'）与原始输入（'request'）一起传递到 delegation_branch。
coordinator_agent = {
"decision": coordinator_router_chain,
"request": RunnablePassthrough()
} | delegation_branch | (lambda x: x['output']) # 提取最终输出
## --- 示例使用 ---
def main():
if not llm:
print("\n由于 LLM 初始化失败，跳过执行。")
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()
核心组件是 coordinator_router_chain，它利用 ChatPromptTemplate 指导语言模型将用户请求分类为以下三类之一：booker、info 或 unclear。该路由链的输出随后由 RunnableBranch 使用，用于将原始请求委派给相应的处理函数。RunnableBranch 检查语言模型的决策，并将请求数据分别引导至 booking_handler、info_handler 或 unclear_handler。coordinator_agent 将这些组件结合起来，首先对请求进行路由决策，然后将请求传递给选定的处理器。最终的输出则从处理器的响应中提取。
主函数通过三个示例请求展示了系统的使用方式，演示了不同输入如何被模拟智能体路由和处理。代码还包含语言模型初始化的错误处理，以确保系统的鲁棒性。代码结构模仿了一个基本的多智能体框架，其中一个中央协调器根据意图将任务委派给专业智能体。
实操代码示例（Google ADK）
Agent Development Kit (ADK) 是一个用于设计智能体系统的框架，为定义智能体的能力和行为提供了结构化环境。与基于显式计算图的架构不同，ADK 范式中的路由通常通过定义一组离散的“工具”来实现，这些工具代表智能体的功能。框架的内部逻辑利用底层模型来匹配用户意图与正确的功能处理器，从而管理对适当工具的选择。
以下 Python 代码展示了一个使用 Google ADK 库的 Agent Development Kit (ADK) 应用示例。代码设置了一个“协调器”智能体，该智能体根据定义的指令将用户请求路由到专业子智能体（“Booker”用于预订，“Info”用于一般信息）。子智能体随后使用特定工具模拟处理请求，展示了智能体系统中的基本任务委派模式。
## Copyright (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("-------------------------- 预订处理器被调用 ----------------------------")
return f"已模拟处理预订操作：'{request}'。"
def info_handler(request: str) -> str:
"""
处理一般信息请求。
参数:
request: 用户的问题。
返回:
表示信息请求已处理的消息。
"""
print("-------------------------- 信息处理器被调用 ----------------------------")
return f"信息请求：'{request}'。结果：已模拟信息检索。"
def unclear_handler(request: str) -> str:
"""处理无法委派的请求。"""
return f"协调器无法委派请求：'{request}'。请明确说明。"
## --- 从函数创建工具 ---
booking_tool = FunctionTool(booking_handler)
info_tool = FunctionTool(info_handler)
## 定义配备专属工具的专业子智能体
booking_agent = Agent(
name="Booker",
model="gemini-2.0-flash",
description="一个专门处理所有航班和酒店预订请求的智能体，通过调用预订工具完成任务。",
tools=[booking_tool]
)
info_agent = Agent(
name="Info",
model="gemini-2.0-flash",
description="一个专门提供一般信息并通过调用信息工具回答用户问题的智能体。",
tools=[info_tool]
)
## 定义具有明确委派指令的主智能体
coordinator = Agent(
name="Coordinator",
model="gemini-2.0-flash",
instruction=(
"你是主要的协调员。你的唯一任务是分析用户的请求并将其委派给合适的专业智能体。"
"不要尝试直接回答用户的问题。\n"
"- 对于任何与预订航班或酒店相关的请求，请委派给 'Booker' 智能体。\n"
"- 对于所有其他的一般信息问题，请委派给 'Info' 智能体。"
),
description="一个协调员，用于将用户请求路由到正确的专业智能体。",
# 子智能体的存在默认启用了基于 LLM 的委派（自动流程）。
sub_agents=[booking_agent, info_agent]
)
## --- 执行逻辑 ---
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
)
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 获取文本以避免迭代 parts
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}")
return f"处理请求时发生错误: {e}"
async def main():
"""运行 ADK 示例的主函数。"""
print("--- Google ADK 路由示例（ADK 自动流程风格） ---")
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 agent）和两个专门的子智能体（Booker 和 Info）组成。每个专门智能体都配备了一个 FunctionTool，用于封装一个模拟操作的 Python 函数。booking_handler 函数模拟处理航班和酒店预订，而 info_handler 函数模拟检索一般信息。unclear_handler 作为协调智能体无法委派请求时的回退处理函数，尽管当前的协调智能体逻辑并未在主 run_coordinator 函数中明确使用它来处理委派失败的情况。
协调智能体的主要职责（根据其指令定义）是分析传入的用户消息，并将其委派给 Booker 或 Info 智能体。由于协调智能体定义了子智能体，这种委派由 ADK 的自动流程机制（Auto-Flow）自动处理。run_coordinator 函数设置了一个 InMemoryRunner，创建了用户和会话 ID，然后使用运行器通过协调智能体处理用户请求。runner.run 方法处理请求并生成事件，代码从 event.content 中提取最终的响应文本。
主函数通过运行协调智能体处理不同的请求，展示了系统如何将预订请求委派给 Booker，以及将信息请求委派给 Info 智能体。
概览
定义（What）
智能智能体系统通常需要响应各种输入和情况，而这些情况无法通过单一的线性流程来处理。简单的顺序工作流缺乏基于上下文进行决策的能力。如果没有选择正确工具或子流程的机制，系统将保持僵化且缺乏适应性。这种局限性使得构建能够管理现实世界用户请求复杂性和多样性的高级应用变得困难。
设计意图（Why）
路由模式通过在智能体的操作框架中引入条件逻辑提供了标准化解决方案。它使系统能够首先分析传入查询以确定其意图或性质。基于此分析，智能体动态地将控制流导向最适合的专用工具、函数或子智能体。此决策可以通过多种方法驱动，包括提示 LLM（大语言模型）、应用预定义规则或使用基于嵌入的语义相似性。最终，路由将静态的预定执行路径转变为灵活且具有上下文感知的工作流，能够选择最佳的操作。
使用原则（Rule of Thumb）
当智能体需要根据用户输入或当前状态在多个不同的工作流、工具或子智能体之间进行决策时，使用路由模式。对于需要分流或分类传入请求以处理不同类型任务的应用程序（例如客服机器人区分销售咨询、技术支持和账户管理问题），路由模式至关重要。
图解 （Visual Summary）
图1：路由模式，使用 LLM 作为路由器
关键要点
- 路由使智能体能够根据条件对工作流的下一步进行动态决策。
- 它使智能体能够处理多样化的输入并适应其行为，超越线性执行的限制。
- 路由逻辑可以通过 LLM、基于规则的系统或嵌入相似性来实现。
- 像 LangGraph 和 Google ADK 这样的框架提供了定义和管理智能体工作流中的路由的结构化方法，尽管其架构方法有所不同。
结论
路由模式是构建真正动态且响应迅速的智能智能体系统的关键步骤。通过实现路由，我们可以超越简单的线性执行流程，使智能体能够对如何处理信息、响应用户输入以及利用可用工具或子智能体做出智能决策。
我们已经看到路由如何应用于多个领域，从客户服务聊天机器人到复杂的数据处理管道。分析输入并有条件地引导工作流程的能力是创建能够处理现实世界任务固有多样性的智能智能体的基础。
使用 LangChain 和 Google ADK 的代码示例展示了两种不同但有效的实现路由的方法。LangGraph 的基于图的结构提供了一种可视化且显式的方法来定义状态和转换，使其非常适合具有复杂、多步骤工作流程和复杂路由逻辑的场景。而 Google ADK 通常专注于定义明确的功能（工具），并依赖框架将用户请求路由到适当的工具处理程序，这对于具有一组明确离散操作的智能体来说可能更简单。
掌握路由模式对于构建能够智能导航不同场景并根据上下文提供定制响应或操作的智能体至关重要。这是创建多功能且强大的智能应用程序的关键组成部分。
参考文献
- LangGraph 文档：https://www.langchain.com/
- Google Agent Developer Kit 文档：https://google.github.io/adk-docs/