Agentic 设计模式详解 - 路由 (Routing)
什么是路由？
上一篇我们聊了提示词链（Prompt Chaining）——把复杂任务拆成一串小步骤，按顺序执行。这在流程固定的场景下很好用，但现实世界的问题往往不是线性的：用户问的问题五花八门，你不可能让每个请求都走同一条路。
路由（Routing）就是给 Agent 加上"岔路口"的能力。 它让 Agent 先判断"这个请求属于什么类型"，然后把请求分发到对应的处理流程。
打个比方：你去医院挂号，分诊台的护士会先问你哪里不舒服，然后告诉你去内科、外科还是急诊。护士不会亲自给你看病，但她决定了你接下来走哪条路。路由在 Agent 系统里扮演的就是这个"分诊台"的角色。
拿客服场景举例，一个带路由的 Agent 会这样工作：
- 分析用户的查询。
- 根据意图路由到不同的处理流程：
- "查订单状态" → 路由到订单数据库查询的子 Agent
- "想了解产品" → 路由到产品目录搜索的子 Agent
- "东西坏了要修" → 路由到技术支持或转人工
- 意图不明确 → 路由到一个专门澄清问题的子 Agent
没有路由，你只能写一个"万能 Agent"试图处理所有情况，结果就是什么都做不好。有了路由，每个子 Agent 只需要专注于自己擅长的事。
四种常见的路由实现方式
路由的核心问题就一个：怎么判断这个请求该走哪条路？ 常见的做法有四种：
1. 让 LLM 来判断
最直接的方式。给 LLM 一个 Prompt："分析用户的问题，告诉我它属于哪个类别——'订单查询'、'产品咨询'、'技术支持'还是'其他'，只输出类别名。" 系统拿到 LLM 的输出后，用简单的 if-else 分发到对应流程。灵活，能处理模糊的自然语言输入，但每次路由都要调一次 LLM，有延迟和成本。
2. 用 Embedding 做语义匹配
把用户的问题转成向量，然后跟预先准备好的各条"路线"的向量做相似度比较，选最像的那条。好处是基于语义而非关键词——用户说"我的包裹到哪了"和"帮我查物流"会被匹配到同一条路由，即使措辞完全不同。
3. 写规则硬编码
最传统的方式：if-else、正则匹配、关键词检测。速度快、结果确定、不花钱，但只能处理你预想到的情况，遇到新的表达方式就傻了。适合输入格式比较固定的场景。
4. 训练一个专用分类模型
用标注数据训练一个小型分类器来做路由决策。跟让 LLM 判断不同，这里的路由逻辑是"学"出来的，不是靠 Prompt 临场发挥。推理速度快、成本低，但需要前期准备训练数据。适合请求量大、路由类别稳定的生产环境。
实际项目中，这四种方式经常混合使用。比如先用规则过滤掉明显的类别，剩下模糊的再交给 LLM 判断。
路由能用在哪些场景？
- 客服 / 虚拟助手： 根据用户意图分流到不同的处理 Agent——查订单、退换货、技术支持、转人工，各走各的路。
- 数据处理流水线： 传入的数据根据格式（JSON / CSV / XML）或内容（紧急工单 / 普通反馈 / 销售线索）分发到不同的处理函数。
- 多 Agent 协作系统： 一个研究系统里有搜索 Agent、摘要 Agent、分析 Agent，路由器根据当前任务把活儿派给最合适的那个。
- AI 编码助手： 先识别用户是要调试、解释还是翻译代码，再把请求交给对应的专用工具。
代码示例（LangChain）
下面用 LangChain 实现一个最基本的路由：一个"协调器"接收用户请求，用 LLM 判断意图，然后分发给对应的处理函数。
pip install langchain langgraph google-cloud-aiplatform langchain-google-genai google-adk deprecated pydantic
## 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"]
)
## Combine the router chain and the delegation branch into a single runnable
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()
这段代码的核心逻辑分三步：
- LLM 做分类：
coordinator_router_chain让 LLM 分析用户请求，输出一个类别词（'booker'、'info' 或 'unclear'）。 - 按类别分发：
RunnableBranch根据 LLM 的输出，把请求转发给对应的处理函数。 - 组装流水线：
coordinator_agent把分类和分发串成一条链——先决策，再执行。
这就是"基于 LLM 的路由"最典型的实现：LLM 不直接回答用户问题，而是充当一个分诊员，决定谁来回答。
代码示例（Google ADK）
Google ADK 提供了另一种实现路由的思路。与 LangChain 需要你显式编写路由逻辑不同，ADK 的做法更"声明式"：你定义好一个 Coordinator Agent 和若干子 Agent，在 Coordinator 的 instruction 里写清楚分发规则，框架的 Auto-Flow 机制会自动完成路由。
## 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 ---
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."
## --- 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.",
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:
if hasattr(event.content, 'text') and event.content.text:
final_result = event.content.text
elif event.content.parts:
text_parts = [part.text for part in event.content.parts if part.text]
final_result = "".join(text_parts)
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) ---")
runner = InMemoryRunner(coordinator)
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}")
if __name__ == "__main__":
import nest_asyncio
nest_asyncio.apply()
await main()
对比两种实现，核心区别在于：
- LangChain 需要你自己写路由判断逻辑（
RunnableBranch+ 条件函数），控制力更强，适合需要精细定制路由规则的场景。 - Google ADK 你只需要声明子 Agent 和它们的职责描述，框架自动根据
sub_agents的description来做路由决策。写起来更简洁，但路由逻辑对你来说是个黑盒。
什么时候该用路由？
一句话：当你的 Agent 需要根据输入的不同，走完全不同的处理流程时，就该用路由。
典型信号：
- 你发现自己在一个 Prompt 里写了一堆 "如果用户问的是 X 就……如果是 Y 就……"
- 你的 Agent 需要调用多个工具，但每次只需要用其中一个
- 不同类型的请求需要完全不同的处理逻辑
路由不是什么高深的概念，本质上就是程序设计里最基本的条件分支——只不过在 Agentic 系统里，这个"条件判断"可以由 LLM 来完成，从而处理自然语言这种非结构化的输入。