agent设计模式：第二章节—路由

1.1 路由模式概述

       通过提示链进行顺序处理是使用语言模型执行确定性、线性工作流的基础技术，但在需要自适应响应的场景中，其适用性有限。现实中的代理系统必须根据偶然因素（如环境状态、用户输入或先前操作的结果）在多个潜在动作之间进行仲裁。这种动态决策能力，即控制流转向不同专业功能、工具或子过程的能力，是通过一种称为路由的机制实现的。
路由将条件逻辑引入代理的操作框架，使代理从固定的执行路径转变为动态评估特定标准以从一组可能的后续动作中进行选择的模型。这允许更灵活和上下文感知的系统行为。
例如，一个用于客户咨询的代理，当配备路由功能时，可以首先对收到的查询进行分类，以确定用户的意图。基于这种分类，它可以将查询直接路由到专门的问题解答代理、用于账户信息的数据库检索工具，或用于复杂问题的升级流程，而不是默认到单一、预设的响应路径。因此，使用路由的更复杂的代理可以：

1.1.1 分析用户的查询

1.1.2 根据其意图路由查询

• 如果意图是"查询订单状态"，则路由到与订单数据库交互的子代理或工具链
• 如果意图是"产品信息"，路由到搜索产品目录的子代理或链。
• 如果意图是"技术支持"，路由到访问故障排除指南的不同链，或升级到人工。
• 如果意图不明确，路由到澄清子代理或提示链。

      路由模式的核心组件是一个执行评估和指导流程的机制。这种机制可以以多种方式实现：

• 基于 LLM 的路由：语言模型本身可以被提示来分析输入，并输出一个特定的标识符或指令，该标识符或指令指示下一步或目的地。例如，一个提示可能会要求 LLM“分析以下用户查询，并仅输出类别：‘订单状态’、‘产品信息’、‘技术支持’或’其他’。”然后，智能体系统读取这个输出，并相应地指导工作流程。
• 基于嵌入的路由：输入查询可以被转换为向量嵌入（参见 RAG，第 14 章）。然后，将这个嵌入与代表不同路由或能力的嵌入进行比较。查询被路由到其嵌入最相似的路由。这对于语义路由很有用，其中决策基于输入的含义，而不仅仅是关键词。
• 基于规则的路由：这涉及使用预定义的规则或逻辑（例如，if-else 语句、switch 案例），基于关键词、模式或从输入中提取的结构化数据。这可以比基于 LLM 的路由更快、更确定，但对于处理细微差别或新颖的输入则不太灵活。
• 基于机器学习模型的路径选择：它采用一个判别模型，例如分类器，该模型在少量标记数据上经过专门训练以执行路径选择任务。虽然它与基于嵌入的方法在概念上相似，但其关键特征是监督微调过程，该过程调整模型的参数以创建专门的路径选择功能。这种技术与基于 LLM 的路径选择不同，因为决策组件不是在推理时执行提示的生成模型。相反，路径选择逻辑被编码在微调模型的 learned 权重中。虽然 LLM 可能用于预处理步骤以生成合成数据来增强训练集，但它们不参与实时路径选择决策。

      路径选择机制可以在智能体操作周期的多个节点实现。它们可以应用于开始时对主要任务进行分类，在处理链的中间点确定后续动作，或在子程序期间从给定集合中选择最合适的工具。

      计算框架如 LangChain、LangGraph 和谷歌的 Agent 开发者套件（ADK）提供了明确的构造来定义和管理此类条件逻辑。凭借其基于状态的图架构，LangGraph 特别适用于复杂的路由场景，其中决策取决于整个系统的累积状态。类似地，谷歌的 ADK 提供了构建代理能力和交互模型的基础组件，这些组件是实现路由逻辑的基础。在这些框架提供的执行环境中，开发者定义可能的操作路径以及决定计算图中节点之间转换的功能或基于模型的评估。

      路由的实现使系统能够超越确定性顺序处理。它促进了更具适应性的执行流程的开发，这些流程能够动态且适当地响应更广泛的输入和状态变化。

1.2 实际应用与用例

      路由模式是自适应智能体系统设计中的关键控制机制，使其能够根据可变输入和内部状态动态改变执行路径。它在多个领域具有实用性，通过提供必要的条件逻辑层来实现这一功能。
在人机交互领域，例如虚拟助手或 AI 驱动的导师，路由被用于解释用户意图。对自然语言查询的初步分析确定了最合适的后续操作，无论是调用特定的信息检索工具、升级到人工操作员，还是根据用户表现选择课程中的下一个模块。这使系统能够超越线性的对话流程，并做出情境化的响应。
在自动化数据与文档处理流程中，路由充当分类和分发功能。传入的数据，如电子邮件、支持工单或 API 负载，会根据内容、元数据或格式进行分析。系统随后将每个项目导向相应的流程，例如销售线索摄入流程、针对 JSON 或 CSV 格式的特定数据转换功能，或紧急问题升级路径。
在涉及多种专用工具或代理的复杂系统中，路由充当高级调度器。一个由搜索、摘要和分析信息等不同代理组成的科研系统会使用路由器根据当前目标将任务分配给最合适的代理。类似地，AI 编程助手使用路由来识别编程语言和用户意图——是调试、解释还是翻译——然后将代码片段传递给正确的专用工具。
最终，路由提供了逻辑仲裁的能力，这对于创建功能多样且具有上下文感知的系统至关重要。它将一个代理从一个静态执行预定义序列的实体转变为一个能够在不断变化的环境中做出关于最有效完成任务方法的动态系统。

1.3 动手代码示例（LangChain）

      在代码中实现路由涉及定义可能的路径以及决定选择哪条路径的逻辑。LangChain 和 LangGraph 等框架提供了特定的组件和结构来实现这一点。LangGraph 基于状态图的架构结构特别直观，便于可视化和实现路由逻辑。
这段代码展示了使用 LangChain 和 Google 的生成式 AI 的一个简单的代理系统。它设置了一个"协调器"，根据请求的意图（预订、信息或模糊不清）将用户请求路由到不同的模拟"子代理"处理程序。系统使用语言模型对请求进行分类，然后将其委托给适当的处理程序函数，模拟了多代理架构中常见的基本委托模式。

首先，确保已安装必要的库：

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

你还需要设置你的环境，使用你选择的语言模型的 API 密钥（例如，OpenAI、Google Gemini、Anthropic）

# 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 ChatGoogleGenerativeAIfrom langchain_core.prompts import ChatPromptTemplatefrom langchain_core.output_parsers import StrOutputParserfrom 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 RunnableBranchbranches = {"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"]), # Added .strip()(lambda x: x['decision'].strip() == 'info', branches["info"]),     # Added .strip()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']) # Extract the final output# --- Example Usage ---def main():if not llm:print("\nSkipping execution due to LLM initialization failure.")returnprint("--- 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 库和 Google 的生成式 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 结合了这些组件，首先路由请求以获取决策，然后将请求传递给选定的处理函数。最终输出是从处理函数的响应中提取的。
主函数通过三个示例请求演示了系统的使用，展示了不同的输入如何被模拟代理路由和处理。包含了语言模型初始化的错误处理以确保系统的健壮性。代码结构模拟了一个基本的多代理框架，其中中央协调器根据意图将任务分配给专门的代理。

1.4 实际代码示例（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 toolsbooking_agent = Agent(name="Booker",model="gemini-2.0-flash",description="A specialized agent that handles all flightand 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 informationand answers user questions by calling the info tool.",tools=[info_tool])# Define the parent agent with explicit delegation instructionscoordinator = Agent(name="Coordinator",model="gemini-2.0-flash",instruction=("You are the main coordinator. Your only task is to analyzeincoming user requests ""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 thecorrect specialist agent.",# The presence of sub_agents enables LLM-driven delegation (Auto-Flow) by default.sub_agents=[booking_agent, info_agent])# --- Execution Logic ---asyncdef 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())awaitrunner.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 partsif hasattr(event.content, 'text') and event.content.text:final_result = event.content.textelif 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 responsebreakprint(f"Coordinator Final Response: {final_result}")return final_resultexcept Exception as e:print(f"An error occurred while processing your request: {e}")return f"An error occurred while processing your request: {e}"asyncdef 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 Usageresult_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 Infoprint(f"Final Output C: {result_c}")result_d = await run_coordinator(runner, "Find flights to Tokyo next month.") # Should go to Bookerprint(f"Final Output D: {result_d}")if __name__ == "__main__":import nest_asyncionest_asyncio.apply()await main()

      该脚本包含一个主协调器代理和两个专门子代理：预订者和信息代理。每个专门代理都配备了一个 FunctionTool，该工具封装了一个模拟操作的 Python 函数。booking_handler 函数模拟处理航班和酒店预订，而 info_handler 函数模拟获取一般信息。unclear_handler 作为备用选项，用于协调器无法委托的请求，尽管当前协调器逻辑在 main run_coordinator 函数中未明确使用它来处理委托失败的情况。
协调器代理的主要作用，在其指令中定义，是分析传入的用户消息，并将它们分配给预订者或信息代理。这种分配由 ADK 的 Auto-Flow 机制自动处理，因为协调器定义了子代理。run_coordinator 函数设置了一个 InMemoryRunner，创建用户和会话 ID，然后使用该运行器通过协调器代理处理用户的请求。runner.run 方法处理请求并生成事件，代码从事件.content 中提取最终响应文本。
主函数通过使用不同的请求运行协调器，展示了系统如何将预订请求委托给预订者，将信息请求委托给信息代理。

1.5 一目了然

       是什么：智能体系统必须经常应对各种输入和情况，而这些无法通过单一、线性的流程处理。简单的顺序工作流缺乏根据上下文做决策的能力。如果没有机制来为特定任务选择正确的工具或子流程，系统将保持僵化和非适应性。这种局限性使得构建能够管理现实世界用户请求的复杂性和可变性的高级应用变得困难。

       原因：路由模式通过将条件逻辑引入代理的操作框架，提供了一种标准化的解决方案。它使系统能够首先分析传入的查询以确定其意图或性质。根据这种分析，代理动态地将控制流引导至最合适的专用工具、功能或子代理。这一决策可以通过多种方法驱动，包括提示 LLMs、应用预定义规则或使用基于嵌入的语义相似度。最终，路由将静态、预定的执行路径转变为灵活且具有上下文感知能力的工作流程，能够选择最佳可能的操作。

       经验法则：当代理必须根据用户输入或当前状态在多个不同的工作流、工具或子代理之间做出决策时，应使用路由模式。这对于需要分诊或分类处理不同类型任务的应用程序至关重要，例如客服机器人区分销售咨询、技术支持和账户管理问题。

视觉摘要：

图 1：路由模式，使用 LLM 作为路由器

1.6 关键要点

• 路由使代理能够根据条件动态地做出关于工作流程下一步的决策
• 它允许代理处理多样化的输入并调整其行为，超越线性执行。
• 路由逻辑可以使用 LLM、基于规则的系统或嵌入相似性来实现。
• 像 LangGraph 和 Google ADK 这样的框架提供了结构化的方式来定义和管理代理工作流中的路由，尽管它们采用了不同的架构方法。

1.7 结论

      路由模式是构建真正动态和响应式代理系统的关键步骤。通过实现路由，我们超越了简单的线性执行流程，并赋予我们的代理智能决策如何处理信息、响应用户输入以及利用可用工具或子代理的能力。
我们已经看到路由如何在各种领域中应用，从客户服务聊天机器人到复杂的数据处理管道。分析输入并条件性地引导工作流的能力是创建能够处理现实世界任务固有变化性的代理的基础。
使用 LangChain 和 Google ADK 的代码示例展示了两种不同但同样有效的实现路由方法。LangGraph 基于图的结构提供了一种可视化和明确的方式来定义状态和转换，非常适合具有复杂、多步骤工作流和复杂路由逻辑的场景。另一方面，Google ADK 通常专注于定义不同的能力（工具），并依赖于框架将用户请求路由到适当的工具处理程序，这对于具有明确定义的一系列离散动作的代理来说可能更简单。
掌握路由模式对于构建能够智能导航不同场景并提供基于上下文的定制化响应或操作的智能体至关重要。它是创建多功能且健壮的智能体应用的关键组成部分。