第2章：Routing

Routing 模式概述

虽然通过 prompt chaining 进行顺序处理是执行确定性线性工作流的基础技术，但其适用性在需要自适应响应的场景中受到限制。真实的 Agent 系统通常必须根据偶然因素（如环境状态、用户输入或先前操作的结果）在多个潜在操作之间进行仲裁。这种动态决策能力（控制流向不同专门功能、工具或子流程的流向）是通过一种称为 routing 的机制实现的。

Routing 将条件逻辑引入 Agent 的操作框架中，使系统能够从固定的执行路径转变为一种模型，在该模型中 Agent 动态评估特定标准，从一组可能的后续操作中进行选择。这使得系统行为更加灵活和具有上下文感知能力。

例如，一个为客户咨询设计的 Agent，当配备 routing 功能时，可以首先对传入的查询进行分类以确定用户的意图。基于此分类，它可以将查询定向到一个专门直接回答问题的 Agent、一个用于账户信息的数据库检索工具，或者一个用于复杂问题的升级程序，而不是默认使用单一、预定的响应路径。因此，使用 routing 的更复杂的 Agent 可以：

分析用户的查询。
基于其意图路由查询：
- 如果意图是"检查订单状态"，则路由到一个与订单数据库交互的子 Agent 或工具链。
- 如果意图是"产品信息"，则路由到一个搜索产品目录的子 Agent 或链。
- 如果意图是"技术支持"，则路由到一个访问故障排除指南或升级到人工的不同链。
- 如果意图不明确，则路由到一个澄清子 Agent 或 prompt chain。

Routing 模式的核心组件是执行评估并指导流向的机制。该机制可以通过多种方式实现：

基于 LLM 的 Routing：语言模型本身可以被提示分析输入并输出一个特定的标识符或指令，指示下一步或目的地。例如，一个 prompt 可能要求 LLM "Analyze the following user query and output only the category: 'Order Status', 'Product Info', 'Technical Support', or 'Other'." 然后 Agent 系统读取此输出并相应地指导工作流。
基于 Embedding 的 Routing：输入查询可以被转换为向量 embedding（参见 RAG，第14章）。然后将此 embedding 与表示不同路由或能力的 embedding 进行比较。查询被路由到 embedding 最相似的路由。这对于语义 routing 很有用，其中决策基于输入的含义而不仅仅是关键词。
基于规则的 Routing：这涉及使用基于关键词、模式或从输入中提取的结构化数据的预定义规则或逻辑（例如 if-else 语句、switch 语句）。这比基于 LLM 的 routing 更快、更确定，但处理细微或新奇的输入时灵活性较差。
基于机器学习模型的 Routing：使用一个判别模型（如分类器），该模型已在一个小型标注数据语料库上专门训练以执行 routing 任务。虽然它与基于 embedding 的方法在概念上有相似之处，但其关键特征是监督微调过程，该过程调整模型参数以创建专门的 routing 功能。该技术不同于基于 LLM 的 routing，因为决策组件不是在推理时执行 prompt 的生成模型。相反，routing 逻辑被编码在微调模型学习到的权重中。虽然 LLM 可能用于预处理步骤中生成合成数据以扩充训练集，但它们不参与实时的 routing 决策。

Routing 机制可以在 Agent 操作周期的多个节点上实现。它们可以在开始处应用于分类主要任务，在处理链的中间点确定后续操作，或在子程序中选择来自给定集合的最合适工具。

诸如 LangChain、LangGraph 和 Google 的 Agent Developer Kit (ADK) 等计算框架提供了定义和管理此类条件逻辑的显式构造。凭借其基于状态的图架构，LangGraph 特别适合复杂的 routing 场景，其中决策取决于整个系统累积的状态。类似地，Google 的 ADK 提供了基础组件来构建 Agent 的能力和交互模型，这些构成了实现 routing 逻辑的基础。在这些框架提供的执行环境中，开发者定义可能的操作路径以及决定计算图中节点间转换的函数或基于模型的评估。

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

实际应用与用例

Routing 模式是自适应 Agent 系统设计中的关键控制机制，使它们能够根据可变输入和内部状态动态改变其执行路径。通过提供必要的条件逻辑层，其实用性跨多个领域。

在人机交互中，如虚拟助手或 AI 驱动的导师，routing 被用于解释用户意图。对自然语言查询的初始分析确定最合适的后续操作，无论是调用特定的信息检索工具、升级到人工操作员，还是根据用户表现选择课程中的下一个模块。这使得系统能够超越线性对话流并进行上下文响应。

在自动化数据和文档处理 pipeline 中，routing 充当分类和分发功能。传入数据（如电子邮件、支持工单或 API 负载）根据内容、元数据或格式进行分析。然后系统将每项定向到相应的工作流，如销售线索导入流程、针对 JSON 或 CSV 格式的特定数据转换函数，或紧急问题升级路径。

在涉及多个专门工具或 Agent 的复杂系统中，routing 充当高级调度器。一个由用于搜索、摘要和分析信息的不同 Agent 组成的研究系统将使用 router 根据当前目标将任务分配给最合适的 Agent。类似地，AI 编码助手使用 routing 在将代码片段传递给正确的专门工具之前，识别编程语言和用户意图——调试、解释或翻译。

最终，routing 提供了创建功能多样且具有上下文感知能力的系统所必需的逻辑仲裁能力。它将 Agent 从预定义序列的静态执行者转变为能够根据变化条件决定完成任务的最有效方法的动态系统。

实操代码示例（LangChain）

在代码中实现 routing 涉及定义可能的路径和决定走哪条路径的逻辑。诸如 LangChain 和 LangGraph 等框架为此提供了特定的组件和结构。LangGraph 的基于状态的图结构对于可视化和实现 routing 逻辑特别直观。

以下代码演示了一个使用 LangChain 和 Google 的 Generative AI 的简单 Agent 类系统。它设置了一个"协调器"（coordinator），根据请求的意图（预订、信息或不明确）将用户请求路由到不同的模拟"子 Agent"处理程序。系统使用语言模型对请求进行分类，然后将其委托给适当的处理程序函数，模拟了多 Agent 架构中常见的委派模式。

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

bash

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

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

python

# 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"]),  # 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.")
        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 库和 Google 的 Generative AI 模型（特别是 gemini-2.5-flash）构建了一个简单的 Agent 类系统。具体来说，它定义了三个模拟的子 Agent 处理程序：booking_handler、info_handler 和 unclear_handler，每个设计用于处理特定类型的请求。

核心组件是 coordinator_router_chain，它使用 ChatPromptTemplate 指示语言模型将传入的用户请求分类为三个类别之一：'booker'、'info' 或 'unclear'。此 router chain 的输出随后被 RunnableBranch 用于将原始请求委托给相应的处理程序函数。RunnableBranch 检查语言模型的决策，并将请求数据定向到 booking_handler、info_handler 或 unclear_handler。coordinator_agent 组合了这些组件，首先对请求进行 routing 以做出决策，然后将请求传递给所选的处理程序。最终输出从处理程序的响应中提取。

main 函数使用三个示例请求演示了系统的使用，展示了不同输入如何被路由并由模拟 Agent 处理。包含语言模型初始化的错误处理以确保健壮性。代码结构模拟了一个基本的多 Agent 框架，其中中央协调器根据意图将任务委派给专门 Agent。

实操代码示例（Google ADK）

Agent Development Kit (ADK) 是一个用于工程化 Agent 系统的框架，为定义 Agent 的能力和行为提供了结构化环境。与基于显式计算图的架构相比，ADK 范式中的 routing 通常通过定义一组离散的"工具"（tools）来实现，这些工具代表 Agent 的功能。为响应特定用户查询选择合适的工具由框架的内部逻辑管理，该逻辑利用底层模型匹配用户意图与正确的功能处理程序。

以下 Python 代码演示了一个使用 Google ADK 库的 ADK 应用示例。它设置了一个"协调器"（Coordinator）Agent，根据定义的指令将用户请求路由到专门的子 Agent（"Booker" 用于预订，"Info" 用于一般信息）。然后子 Agent 使用特定工具模拟处理请求，展示了 Agent 系统内的基本委派模式。

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

# --- 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=(
        "You are the main coordinator. Your only task is to analyze incoming 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 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()

此脚本由一个主要的 Coordinator Agent 和两个专门的 sub_agents 组成：Booker 和 Info。每个专门的 Agent 配备了一个 FunctionTool，该工具包装了一个模拟操作的 Python 函数。booking_handler 函数模拟处理航班和酒店预订，而 info_handler 函数模拟检索一般信息。unclear_handler 被包含作为协调器无法委派的请求的后备处理，尽管当前的协调器逻辑在主要的 run_coordinator 函数中没有显式将其用于委派失败的情况。

Coordinator Agent 的主要角色（如其指令中定义）是分析传入的用户消息并将其委派给 Booker 或 Info Agent。这种委派由 ADK 的 Auto-Flow 机制自动处理，因为 Coordinator 定义了 sub_agents。run_coordinator 函数设置了一个 InMemoryRunner，创建用户和会话 ID，然后使用 runner 通过协调器 Agent 处理用户的请求。runner.run 方法处理请求并产生事件，代码从 event.content 中提取最终响应文本。

main 函数通过使用不同请求运行协调器来演示系统的使用，展示它如何将预订请求委派给 Booker，将信息请求委派给 Info Agent。

概览

是什么：Agent 系统通常必须响应无法由单一线性流程处理的各种输入和情况。简单的顺序工作流缺乏基于上下文做出决策的能力。如果没有选择正确工具或子流程来执行特定任务的机制，系统会保持僵化和非自适应。这种局限性使得构建能够管理真实世界用户请求的复杂性和变异性的复杂应用变得困难。

为什么：Routing 模式通过将条件逻辑引入 Agent 的操作框架，提供了一个标准化的解决方案。它使系统能够首先分析传入的查询以确定其意图或性质。基于此分析，Agent 动态地将控制流定向到最合适的专门工具、函数或子 Agent。此决策可以通过多种方法驱动，包括提示 LLM、应用预定义规则或使用基于 embedding 的语义相似性。最终，routing 将静态的、预定的执行路径转变为能够选择最佳可能操作的灵活且具有上下文感知能力的工作流。

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

可视化总结

图1：Router 模式，使用 LLM 作为 Router

关键要点

Routing 使 Agent 能够根据条件对工作流中的下一步做出动态决策。
它允许 Agent 处理多样化的输入并调整其行为，超越线性执行。
Routing 逻辑可以使用 LLM、基于规则的系统或 embedding 相似性来实现。
诸如 LangGraph 和 Google ADK 等框架提供了在 Agent 工作流中定义和管理 routing 的结构化方法，尽管采用不同的架构方法。

结论

Routing 模式是构建真正动态且响应式 Agent 系统的关键一步。通过实现 routing，我们超越了简单的线性执行流，赋予我们的 Agent 智能决策的能力，使其能够决定如何处理信息、响应用户输入以及利用可用的工具或子 Agent。

我们已经看到 routing 如何应用于各种领域，从客户服务聊天机器人到复杂的数据处理 pipeline。分析输入并有条件地指导工作流的能力是创建能够处理真实世界任务内在变异性的 Agent 的基础。

使用 LangChain 和 Google ADK 的代码示例演示了两种不同但有效的实现 routing 的方法。LangGraph 的基于图的结构提供了一种可视化且显式的方式来定义状态和转换，使其适用于具有复杂 routing 逻辑的复杂多步工作流。另一方面，Google ADK 通常侧重于定义不同的能力（Tools），并依赖框架将用户请求路由到适当工具处理程序的能力，这对于具有明确定义的离散操作集的 Agent 来说可能更简单。

掌握 Routing 模式对于构建能够智能地在不同场景中导航并根据上下文提供定制响应或操作的 Agent 至关重要。它是创建多功能且健壮的 Agent 应用的关键组成部分。

参考文献

LangGraph Documentation: https://www.langchain.com/
Google Agent Developer Kit Documentation: https://google.github.io/adk-docs/