第2章：路由
路由模式概述
虽然通过提示链进行顺序处理是执行具有确定性、线性工作流程的语言模型的基础技术，但其适用性在需要自适应响应的场景中受到限制。现实世界的智能体系统通常必须根据环境状态、用户输入或先前操作的结果等偶然因素在多个潜在行动之间进行权衡。这种动态决策能力，它控制着控制流流向不同的专业功能、工具或子过程，是通过一种称为路由的机制实现的。
路由将条件逻辑引入智能体的操作框架中，使得智能体从固定的执行路径转向一种动态评估特定标准以从一系列可能的后继动作中选择模型。这允许系统行为更加灵活和具有情境感知。
例如，一个针对客户咨询设计的智能体，当配备了路由功能后，可以首先对收到的查询进行分类，以确定用户的意图。基于这种分类，它可以将查询直接导向专门的智能体进行直接问答，导向用于账户信息的数据库检索工具，或者导向用于复杂问题的升级流程，而不是默认使用单一的、预先设定的响应路径。因此，使用路由功能的更高级智能体可以：
分析用户的查询。
路由查询，基于其意图： 如果意图是“检查订单状态”，则路由到与订单数据库交互的子智能体或工具链。 * 如果意图是“产品信息”，则路由到子智能体或链，以搜索产品目录。 * 如果意图是“技术支持”，则将其路由到另一个链，该链可以访问故障排除指南或升级至人工服务。 * 如果意图不明确，则路由至澄清子智能体或提示链。
路由模式的核心组件是一种执行评估并引导流量的机制。这种机制可以通过多种方式实现：
- 基于LLM的路由： 语言模型本身可以被提示来分析输入并输出一个特定的标识符或指令，指示下一步或目的地。例如，一个提示可能会要求LLM“分析以下用户查询，并只输出类别：'订单状态'、'产品信息'、'技术支持'或'其他'。”然后，智能体系统读取这个输出并根据此指导工作流程。
- 基于嵌入的路由： 输入查询可以被转换为向量嵌入（参见RAG，第14章）。然后，将该嵌入与代表不同路由或能力的嵌入进行比较。查询将被路由到与该嵌入最相似的路径。这对于语义路由非常有用，因为决策是基于输入的意义，而不仅仅是关键词。
- 基于规则的路由： 这涉及使用预定义的规则或逻辑（例如，if-else语句、switch cases），基于从输入中提取的关键词、模式或结构化数据。这比基于LLM的路由更快、更确定，但在处理细微或新颖的输入时灵活性较低。
- 基于机器学习模型的路由：它采用了一种判别模型，例如分类器，该模型专门在少量标记数据语料库上训练，以执行路由任务。虽然它与基于嵌入的方法在概念上相似，但其关键特征是监督微调过程，该过程调整模型的参数以创建一个专门的路由功能。这种技术与基于LLM的路由不同，因为决策组件不是一个在推理时执行提示的生成模型。相反，路由逻辑被编码在微调模型的学习权重中。虽然LLM可能在预处理步骤中使用来生成合成数据以增强训练集，但它们并不参与实际的实时路由决策。
路由机制可以在智能体的操作周期中的多个环节实现。它们可以在一开始用于分类主要任务，在处理链中的中间点用于确定后续动作，或者在子程序中从给定集合中选择最合适的工具。
计算框架如LangChain、LangGraph以及谷歌的智能体开发者工具包（ADK）为定义和管理此类条件逻辑提供了明确的结构。LangGraph凭借其基于状态的图架构，特别适合于复杂的路由场景，在这些场景中，决策取决于整个系统的累积状态。同样，谷歌的ADK提供了构建智能体能力和交互模型的基础组件，这些组件是实施路由逻辑的基础。在这些框架提供的执行环境中，开发者定义了可能的操作路径以及决定计算图中节点之间转换的函数或基于模型的评估。
路由的实现使系统能够超越确定性顺序处理。它促进了更适应的执行流程的发展，能够动态且恰当地响应更广泛的输入和状态变化。
实际应用与用例
路由模式是自适应智能体系统设计中的一种关键控制机制，它使系统能够根据可变输入和内部状态动态改变其执行路径。其效用跨越多个领域，通过提供必要的条件逻辑层。
在人与计算机交互中，例如与虚拟助手或AI驱动的导师交互时，路由被用于解释用户意图。对自然语言查询的初步分析确定最合适的后续操作，无论是调用特定的信息检索工具、升级到人工操作员，还是根据用户表现选择课程中的下一个模块。这使系统能够超越线性的对话流程，并做出情境响应。
在自动化数据与文档处理管道中，路由充当分类和分发功能。 incoming data，例如电子邮件、支持票据或API有效载荷，将根据内容、元数据或格式进行分析。然后，系统将每个项目引导到相应的流程，例如销售线索摄取流程、针对JSON或CSV格式的特定数据转换功能，或紧急问题升级路径。
在涉及多个专用工具或智能体的复杂系统中，路由充当高级调度器。由专门用于搜索、摘要和分析信息的智能体组成的研发系统会使用路由器根据当前目标将任务分配给最合适的智能体。同样，人工智能编码助手使用路由来识别编程语言和用户的意图——是调试、解释还是翻译——在将代码片段传递给正确的专用工具之前。
最终，路由提供了进行逻辑仲裁的能力，这对于创建功能多样化和情境感知的系统至关重要。它将智能体从静态执行预定义序列的执行者转变为能够在变化条件下做出关于完成任务的最高效方法的决策的动态系统。
动手代码示例（LangChain）
在代码中实现路由涉及定义可能的路径以及决定采取哪条路径的逻辑。LangChain和LangGraph等框架为此提供了特定的组件和结构。LangGraph基于状态的图结构特别直观，适用于可视化实现路由逻辑。
此代码演示了一个使用LangChain和谷歌的生成式AI的简单智能体系统。它设置了一个“协调器”，根据请求的意图（预订、信息或模糊不清）将用户请求路由到不同的模拟“子智能体”处理器。该系统使用语言模型对请求进行分类，然后将其委托给适当的处理器函数，模拟了在多智能体架构中常见的基本委托模式。
首先，请确保您已安装必要的库：
pip install \
langchain \
langgraph \
google-cloud-aiplatform \
langchain-google-genai \
google-adk \
deprecated \
pydantic
您还需要设置您的环境，并使用您选择的语言模型（例如，OpenAI、Google Gemini、Anthropic）的API密钥。
# Copyright (c) 2025 Marco Fago # https://www.linkedin.com/in/marco-fago/ # # This code is licensed under the MIT License. # See the LICENSE file in the repository for the full license text.
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
# --- Configuration ---
# Ensure your API key environment variable is set (e.g., GOOGLE_API_KEY)
try:
llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0)
print(f"Language model initialized: {llm.model}")
except Exception as e:
print(f"Error initializing language model: {e}")
llm = None
# --- Define Simulated Sub-Agent Handlers (equivalent to ADK sub_agents) ---
def booking_handler(request: str) -> str:
"""Simulates the Booking Agent handling a request."""
print("\n--- DELEGATING TO BOOKING HANDLER ---")
return f"Booking Handler processed request: '{request}'. Result: Simulated booking action."
def info_handler(request: str) -> str:
"""Simulates the Info Agent handling a request."""
print("\n--- DELEGATING TO INFO HANDLER ---")
return f"Info Handler processed request: '{request}'. Result: Simulated information retrieval."
def unclear_handler(request: str) -> str:
"""Handles requests that couldn't be delegated."""
print("\n--- HANDLING UNCLEAR REQUEST ---")
return f"Coordinator could not delegate request: '{request}'. Please clarify."
# --- Define Coordinator Router Chain (equivalent to ADK coordinator's instruction) ---
# This chain decides which handler to delegate to.
coordinator_router_prompt = ChatPromptTemplate.from_messages([
("system", """Analyze the user's request and determine which specialist handler should process it.
- If the request is related to booking flights or hotels,
output 'booker'.
- For all other general information questions, output 'info'.
- If the request is unclear or doesn't fit either category,
output 'unclear'.
ONLY output one word: 'booker', 'info', or 'unclear'."""),
("user", "{request}") ])
if llm:
coordinator_router_chain = coordinator_router_prompt | llm | StrOutputParser()
# --- Define the Delegation Logic (equivalent to ADK's Auto-Flow based on sub_agents) ---
# Use RunnableBranch to route based on the router chain's output.
# Define the branches for the 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'])),
}
# Create the RunnableBranch. It takes the output of the router chain
# and routes the original input ('request') to the corresponding handler.
delegation_branch = RunnableBranch(
(lambda x: x['decision'].strip() == 'booker', branches["booker"]),
(lambda x: x['decision'].strip() == 'info', branches["info"]),
branches["unclear"] # Default branch for 'unclear' or any other output
)
# Combine the router chain and the delegation branch into a single runnable
# The router chain's output ('decision') is passed along with the original input ('request')
# to the delegation_branch.
coordinator_agent = {
"decision": coordinator_router_chain,
"request": RunnablePassthrough()
} | delegation_branch | (lambda x: x['output'])
# --- Example Usage ---
def main():
if not llm:
print("\nSkipping execution due to LLM initialization failure.")
return
print("--- Running with a booking request ---")
request_a = "Book me a flight to London."
result_a = coordinator_agent.invoke({"request": request_a})
print(f"Final Result A: {result_a}")
print("\n--- Running with an info request ---")
request_b = "What is the capital of Italy?"
result_b = coordinator_agent.invoke({"request": request_b})
print(f"Final Result B: {result_b}")
print("\n--- Running with an unclear request ---")
request_c = "Tell me about quantum physics."
result_c = coordinator_agent.invoke({"request": request_c})
print(f"Final Result C: {result_c}")
if __name__ == "__main__":
main()
如前所述，这段Python代码使用LangChain库和谷歌的生成式AI模型gemini-2.5-flash构建了一个类似智能体的简单系统。具体来说，它定义了三个模拟的子智能体处理器：booking_handler（预订处理器）、info_handler（信息处理器）和unclear_handler（不明确处理器），每个处理器都设计用于处理特定类型的请求。
核心组件是 coordinator_router_chain，它使用 ChatPromptTemplate 指示语言模型将用户请求分类到三个类别之一：'booker'（预订者）、'info'（信息）或 'unclear'（不明确）。该路由链的输出随后被 RunnableBranch 用于将原始请求委派给相应的处理函数。RunnableBranch 会检查语言模型的决定，并将请求数据导向 booking_handler、info_handler 或 unclear_handler。coordinator_agent 结合这些组件，首先路由请求以做出决定，然后将请求传递给选定的处理器。最终输出从处理器的响应中提取。
主要功能通过三个示例请求展示了系统的使用方法，展示了不同输入是如何被模拟智能体路由和处理的。包括了语言模型初始化的错误处理，以确保系统的鲁棒性。代码结构模仿了一个基本的智能体框架，其中中央协调器根据意图将任务委托给专门的智能体。
动手代码示例（Google ADK）
智能体开发工具包（ADK）是一个用于构建智能体系统的框架，它提供了一个结构化的环境来定义智能体的能力和行为。与基于显式计算图的架构相比，ADK范式中的路由通常是通过定义一组离散的“工具”来实现的，这些工具代表了智能体的功能。在用户查询响应中，适当工具的选择由框架的内部逻辑管理，该逻辑利用底层模型将用户意图与正确的功能处理程序相匹配。
这段Python代码演示了使用Google的ADK库开发的智能体开发工具包（ADK）应用示例。它设置了一个“协调器”智能体，根据定义的指令将用户请求路由到专门的子智能体（“预订者”用于预订和“信息”用于一般信息）。然后，子智能体使用特定工具来模拟处理请求，展示了智能体系统中的基本委派模式。
# 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
# --- Define Tool Functions ---
# These functions simulate the actions of the specialist agents.
def booking_handler(request: str) -> str:
"""Handles booking requests for flights and hotels.
Args:
request: The user's request for a booking.
Returns:
A confirmation message that the booking was handled.
"""
print("-------------------------- Booking Handler Called ----------------------------")
return f"Booking action for '{request}' has been simulated."
def info_handler(request: str) -> str:
"""Handles general information requests.
Args:
request: The user's question.
Returns:
A message indicating the information request was handled.
"""
print("-------------------------- Info Handler Called ----------------------------")
return f"Information request for '{request}'. Result: Simulated information retrieval."
def unclear_handler(request: str) -> str:
"""Handles requests that couldn't be delegated."""
return f"Coordinator could not delegate request: '{request}'. Please clarify."
# --- Create Tools from Functions ---
booking_tool = FunctionTool(booking_handler)
info_tool = FunctionTool(info_handler)
# Define specialized sub-agents equipped with their respective tools
booking_agent = Agent(
name="Booker",
model="gemini-2.0-flash",
description="A specialized agent that handles all flight and hotel booking requests by calling the booking tool.",
tools=[booking_tool]
)
info_agent = Agent(
name="Info",
model="gemini-2.0-flash",
description="A specialized agent that provides general information and answers user questions by calling the info tool.",
tools=[info_tool]
)
# Define the parent agent with explicit delegation instructions
coordinator = Agent(
name="Coordinator",
model="gemini-2.0-flash",
instruction=(f"You are the main coordinator. Your only task is to analyze incoming user requests "
f"and delegate them to the appropriate specialist agent. Do not try to answer the user directly.\n"
"- For any requests related to booking flights or hotels, delegate to the 'Booker' agent.\n"
"- For all other general information questions, delegate to the 'Info' agent."),
description="A coordinator that routes user requests to the correct specialist agent.",
# The presence of sub_agents enables LLM-driven delegation (Auto-Flow) by default.
sub_agents=[booking_agent, info_agent]
)
# --- Execution Logic ---
async def run_coordinator(runner: InMemoryRunner, request: str):
"""Runs the coordinator agent with a given request and delegates."""
print(f"\n--- Running Coordinator with request: '{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:
# Try to get text directly from event.content
# to avoid iterating parts
if hasattr(event.content, 'text') and event.content.text:
final_result = event.content.text
elif event.content.parts:
# Fallback: Iterate through parts and extract text (might trigger warning)
text_parts = [part.text for part in event.content.parts if part.text]
final_result = "".join(text_parts)
# Assuming the loop should break after the final response
break
print(f"Coordinator Final Response: {final_result}")
return final_result
except Exception as e:
print(f"An error occurred while processing your request: {e}")
return f"An error occurred while processing your request: {e}"
async def main():
"""Main function to run the ADK example."""
print("--- Google ADK Routing Example (ADK Auto-Flow Style) ---")
print("Note: This requires Google ADK installed and authenticated.")
runner = InMemoryRunner(coordinator)
# Example Usage
result_a = await run_coordinator(runner, "Book me a hotel in Paris.")
print(f"Final Output A: {result_a}")
result_b = await run_coordinator(runner, "What is the highest mountain in the world?")
print(f"Final Output B: {result_b}")
result_c = await run_coordinator(runner, "Tell me a random fact.") # Should go to Info
print(f"Final Output C: {result_c}")
result_d = await run_coordinator(runner, "Find flights to Tokyo next month.") # Should go to Booker
print(f"Final Output D: {result_d}")
if __name__ == "__main__":
import nest_asyncio
nest_asyncio.apply()
await main()
此脚本包含一个主协调器智能体和两个专用子智能体：预订员和信息服务。每个专用智能体都配备了一个FunctionTool，它封装了一个模拟动作的Python函数。booking_handler函数模拟处理航班和酒店预订，而info_handler函数模拟检索一般信息。unclear_handler被包含作为协调器无法委派的请求的备用方案，尽管当前协调器逻辑在main_run_coordinator函数中并未明确使用它来处理委派失败。
协调器智能体的主要角色，如其在指令中定义的那样，是分析传入的用户消息并将它们委派给预订智能体或信息智能体。由于协调器已定义子智能体，这种委派由ADK的自动流机制自动处理。run_coordinator函数设置一个内存运行器，创建用户和会话ID，然后使用运行器通过协调器智能体处理用户的请求。runner.run方法处理请求并产生事件，代码从event.content中提取最终响应文本。
主要功能通过运行协调器并使用不同的请求来演示系统的使用方法，展示了它如何将预订请求委派给预订者，并将信息请求委派给信息智能体。
概览
内容：智能体系统通常需要响应各种输入和情况，而这些输入和情况无法通过单一、线性的过程来处理。简单的顺序工作流程缺乏根据上下文做出决策的能力。如果没有选择正确工具或子过程来执行特定任务的机制，系统将保持僵化和不可适应。这种限制使得构建能够管理现实世界用户请求的复杂性和多变性的复杂应用程序变得困难。
原因： 路由模式通过在智能体的操作框架中引入条件逻辑，提供了一种标准化的解决方案。它使系统能够首先分析传入的查询以确定其意图或性质。基于这种分析，智能体动态地将控制流导向最合适的专用工具、功能或子智能体。这一决策可以由多种方法驱动，包括提示LLM（大型语言模型）、应用预定义规则或使用基于嵌入的语义相似度。最终，路由将静态、预定的执行路径转变为灵活且具有上下文感知的工作流程，能够选择最佳可能的操作。
经验法则： 当智能体必须根据用户的输入或当前状态在多个不同的工作流程、工具或子智能体之间做出选择时，请使用路由模式。这对于需要对传入请求进行分类或分诊以处理不同类型任务的应用程序至关重要，例如，客户支持智能体在区分销售咨询、技术支持和账户管理问题时的应用。
视觉摘要：
图1：路由模式，使用LLM作为路由器
关键要点
路由使智能体能够根据条件动态决定工作流程中的下一步操作。 它允许智能体处理多样化的输入并适应其行为，超越了线性执行。 路由逻辑可以使用大型语言模型（LLM）、基于规则的系统或嵌入相似度来实现。 * 类似于LangGraph和Google ADK的框架，为在智能体工作流中定义和管理路由提供了结构化的方法，尽管它们采用了不同的架构方法。
结论
路由模式是构建真正动态和响应式智能体系统的关键步骤。通过实现路由，我们超越了简单的线性执行流程，赋予我们的智能体在如何处理信息、响应用户输入以及利用可用工具或子智能体方面做出智能决策的能力。
我们已经看到了路由如何在各个领域得到应用，从客户服务聊天机器人到复杂的数据处理管道。分析输入并条件性地引导工作流程的能力是创建能够处理现实世界任务固有可变性的智能体的基础。
使用LangChain和Google ADK的代码示例展示了两种不同但有效的实现路由的方法。LangGraph的基于图的结构提供了一种直观和明确的方式来定义状态和转换，使其非常适合具有复杂路由逻辑的多步骤工作流程。另一方面，Google ADK通常专注于定义不同的功能（工具），并依赖于框架将用户请求路由到适当的工具处理器，这对于具有明确离散动作集的智能体来说可能更简单。
掌握路由模式对于构建能够智能导航不同场景并提供基于上下文的定制化响应或行动的智能体至关重要。它是创建灵活且稳健的智能体应用的关键组件。
