[智能体设计模式]第2章-路由（Route）
本系列内容知识来源于《智能体设计模式》一书，仅做知识分享。
目录
路由模式概述
路由模式的核心组件
路由机制的应用阶段
实践与应用场景
1. 人机交互场景（虚拟助手、AI教师等）
2. 自动化数据与文档处理流程
3. 多工具/多智能体协同系统
实战代码示例（LangChain）
前置准备
完整代码
代码说明
详细说明
先明确核心角色
1. 第一步：创建“接线员”——coordinator_router_chain
2. 第二步：创建“专业部门”——branches字典
3. 第三步：创建“调度员”——delegation_branch
4. 第四步：组装整个“服务中心”——coordinator_agent
第一步：并行准备数据（左边的字典）
第二步：调度员处理（| delegation_branch）
第三步：提取最终结果（| (lambda x: x['output'])）
路由模式概述
虽然通过提示链实现的顺序处理是利用语言模型执行确定性、线性工作流的基础技术，但在需要自适应响应的场景下，其适用性有限。现实中的智能体系统通常需要根据环境状态、用户输入或前序操作结果等因素，在多个潜在动作之间进行仲裁。
这种动态决策能力——即根据特定条件将控制流导向不同的专用函数、工具或子流程——就是通过 “路由”机制 实现的。
示例：一个用于客户咨询的智能体在具备路由功能后，可以首先对用户查询进行分类以判断意图。根据分类结果，查询可以被导向专门的问答智能体、用于账户信息检索的数据库工具，或用于复杂问题升级的流程，而不是始终采用单一、预设的响应路径。
路由模式的核心组件
路由模式的核心是执行评估并引导流程的机制，其实现方式包括以下4类：
| 实现方式 | 核心逻辑与特点 |
| 基于LLM的路由 | 通过提示语言模型分析输入，输出指示下一步的标识符/指令（如分类标签）。 |
| 基于嵌入的路由 | 将输入查询转为向量嵌入，与不同路由/能力的嵌入比对，路由到最相似路径。 |
| 基于规则的路由 | 使用预定义规则（if-else、switch-case），根据关键词、模式或结构化数据路由。 |
| 基于机器学习模型的路由 | 用分类器等判别模型（小规模标注数据训练）实现路由。 |
路由机制的应用阶段
路由可在智能体操作周期的多个阶段实现：
- 初始任务分类（如用户查询意图识别）
- 处理链中间决策（如选择下一步工具）
- 子流程内路径选择（如适配不同格式的数据处理）
实践与应用场景
路由模式的核心价值是实现“动态分流”，适用于需要自适应决策的复杂系统，常见场景如下：
1. 人机交互场景（虚拟助手、AI教师等）
- 核心作用：解析用户意图，动态调整响应策略
- 应用示例：
-
- 虚拟助手：根据查询意图调用信息检索工具、升级人工坐席，或跳转服务流程
- AI教师：根据学生答题表现选择下一个课程模块，实现个性化学习路径
2. 自动化数据与文档处理流程
- 核心作用：分类与分发，适配不同类型的处理需求
- 应用示例：
-
- 邮件/工单处理：根据内容优先级、主题分类，导向销售线索导入、紧急问题升级等工作流
- 数据格式适配：根据API返回数据格式（JSON/CSV），路由到对应的数据转换流程
3. 多工具/多智能体协同系统
- 核心作用：充当高级调度器，分配任务给专用组件
- 应用示例：
-
- 研究系统：由检索、摘要、分析等智能体组成，路由器根据当前目标分配任务
- AI编程助手：先识别编程语言和用户意图（调试、解释、翻译），再将代码片段交给对应工具处理
实战代码示例（LangChain）
本示例通过 LangChain 构建一个简单的多智能体系统：设置“协调者”组件，根据用户请求意图（预订、信息查询、不明确），将请求路由到对应的“子智能体”处理器，模拟多智能体架构中的委托模式。
前置准备
- 安装所需库：
pip install langchain langgraph google-cloud-aiplatform langchain-google-genai google-adk deprecated pydantic- 配置API密钥：
- 确保环境变量中已设置
GOOGLE_API_KEY（用于调用 Gemini 模型）
完整代码
# 导入LangChain相关库：
# 1. 谷歌Gemini模型的LangChain封装（用于调用大模型）
from langchain_google_genai import ChatGoogleGenerativeAI
# 2. 聊天型提示模板（支持system/user/assistant多角色对话格式）
from langchain_core.prompts import ChatPromptTemplate
# 3. 字符串输出解析器（将大模型输出的原始响应转为字符串）
from langchain_core.output_parsers import StrOutputParser
# 4. 核心流程组件：
# - RunnablePassthrough：透传数据（不修改输入，直接传递给下一个组件）
# - RunnableBranch：分支路由（根据条件将数据导向不同处理流程）
from langchain_core.runnables import RunnablePassthrough, RunnableBranch# --- 配置大模型（核心依赖）---
# 确保环境变量中已设置GOOGLE_API_KEY（谷歌云平台的API密钥，用于调用Gemini）
try:# 初始化谷歌Gemini模型：# - temperature：温度参数（0=输出完全确定，无随机性；值越大越随机）llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0)print(f"语言模型初始化成功：{llm.model}")
except Exception as e:print(f"语言模型初始化失败：{e}")llm = None# --- 定义模拟子智能体处理器（业务逻辑核心）---
# 这些函数是实际处理不同类型请求的"工人"，此处为模拟逻辑（实际项目中可替换为真实业务代码）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}'。请补充说明。"# --- 定义协调者路由链（核心：判断请求该交给哪个处理器）---
# ChatPromptTemplate.from_messages()：创建多角色聊天提示模板
# 输入是一个列表，每个元素是（角色名, 提示内容）的元组
coordinator_router_prompt = ChatPromptTemplate.from_messages([# 1. system角色：给大模型设定"协调者"身份和判断规则（核心指令）("system", """分析用户请求，判断应由哪个专属处理器处理。- 若请求涉及预订机票或酒店，输出'booker'。- 其他一般信息问题，输出'info'。- 若请求不明确或不属于上述类别，输出'unclear'。只输出一个词：'booker'、'info'或'unclear'。"""), # 强制大模型输出固定关键词，方便后续判断# 2. user角色：占位符{request}，后续会被用户的真实请求替换("user", "{request}")
])# 只有大模型初始化成功后，才构建后续流程
if llm:# 构建"路由决策链"：提示模板 → 大模型 → 输出解析器# LangChain的"|"是"链式调用"语法，数据从左到右传递：# 1. 先将用户请求注入coordinator_router_prompt，生成完整提示# 2. 把完整提示传给llm（大模型），得到原始响应# 3. 用StrOutputParser把原始响应转为字符串（去除多余格式）coordinator_router_chain = coordinator_router_prompt | llm | StrOutputParser()# --- 定义委托逻辑（将请求路由到对应处理器）---# branches字典：key是大模型输出的决策关键词，value是对应的处理流程branches = {# 当决策是"booker"时：调用booking_handler处理"booker": RunnablePassthrough.assign(# assign()：给数据字典添加新字段（这里添加"output"字段存储处理结果）# lambda x: 匿名函数，x是当前数据字典（包含"request"和"decision"等字段））output=lambda x: booking_handler(x['user_input']['request'])),# 当决策是"info"时：调用info_handler处理"info": RunnablePassthrough.assign(output=lambda x: info_handler(x['user_input']['request'])),# 当决策是"unclear"时：调用unclear_handler处理"unclear": RunnablePassthrough.assign(output=lambda x: unclear_handler(x['user_input']['request'])),}# 构建"分支路由器"：根据决策结果选择对应的处理分支# RunnableBranch的参数是：(条件1, 分支1), (条件2, 分支2), ..., 默认分支delegation_branch = RunnableBranch(# 条件1：如果决策结果是"booker"，走booker分支(lambda x: x['decision'].strip() == 'booker', branches["booker"]),# 条件2：如果决策结果是"info"，走info分支(lambda x: x['decision'].strip() == 'info', branches["info"]),# 默认分支：以上条件都不满足时，走unclear分支（兜底）branches["unclear"])# --- 构建最终的"协调者智能体" ---# 整体流程：数据准备 → 分支路由 → 结果提取coordinator_agent = {# 第一步：并行处理两个任务，生成数据字典的两个字段：# 1. "decision"字段：通过coordinator_router_chain得到决策结果# 2. "user_input"字段：通过RunnablePassthrough透传原始请求（不修改）"decision": coordinator_router_chain,"user_input": RunnablePassthrough()} | delegation_branch | (lambda x: x['output'])# 第二步：把数据字典传给delegation_branch，根据"decision"路由到对应分支# 第三步：用lambda x: x['output']提取处理结果（只返回最终输出，隐藏中间数据）# --- 示例用法（测试不同类型请求）---
def main():# 先判断大模型是否初始化成功if not llm:print("\n因LLM初始化失败，跳过执行。")return# 测试1：预订类请求（预期路由到booker分支）print("--- 预订请求示例 ---")request_a = "帮我预订飞往伦敦的机票。"# invoke()：触发智能体执行，参数是输入数据字典result_a = coordinator_agent.invoke({"user_input": request_a})print(f"最终结果A: {result_a}")# 测试2：信息查询类请求（预期路由到info分支）print("\n--- 信息请求示例 ---")request_b = "意大利的首都是哪里？"result_b = coordinator_agent.invoke({"user_input": request_b})print(f"最终结果B: {result_b}")# 测试3：不明确请求（预期路由到unclear分支）print("\n--- 不明确请求示例 ---")request_c = "讲讲量子物理。"result_c = coordinator_agent.invoke({"user_input": request_c})print(f"最终结果C: {result_c}")# 程序入口：当脚本直接运行时，执行main函数
if __name__ == "__main__":main()代码说明
- 核心组件：
-
- 子智能体处理器：
booking_handler（预订）、info_handler（信息查询）、unclear_handler（不明确请求） - 协调者路由链：
coordinator_router_chain，通过提示让LLM对请求分类（输出booker/info/unclear） - 委托逻辑：
delegation_branch，基于分类结果将请求路由到对应处理器
- 子智能体处理器：
- 工作流程：
-
- 用户输入请求 → 协调者路由链判断意图 → 输出分类标签
- 根据标签将请求委托给对应子智能体
- 子智能体处理后返回结果，协调者输出最终响应
- 错误处理：
-
- 包含LLM初始化失败的捕获逻辑，确保系统鲁棒性
详细说明
要搞懂这段代码，核心是抓住 LangChain的“工作流组装”思想——把大模型、工具、逻辑判断像“乐高积木”一样拼起来，最终实现“用户请求 → 自动判断类型 → 交给对应处理器处理”的完整流程。
我们拆成 3个核心部分+1个整体流程 来讲，用通俗的例子类比，避免复杂概念：
先明确核心角色
把这段代码想象成一个 公司的“客户服务中心”：
coordinator_agent：整个服务中心（对外接收用户请求，最终返回结果）coordinator_router_chain：中心的“接线员”（判断用户请求类型）branches：3个专业部门（处理不同类型请求）delegation_branch：中心的“调度员”（根据接线员判断，把用户转给对应部门）booking_handler/info_handler/unclear_handler：部门里的“具体办事员”（执行实际处理逻辑）
1. 第一步：创建“接线员”——coordinator_router_chain
coordinator_router_chain = coordinator_router_prompt | llm | StrOutputParser()这行是创建一个 “请求类型判断工具”，作用是：接收用户请求，输出一个“决策关键词”（只能是 booker/info/unclear 三者之一）。
拆解每个部分：
coordinator_router_prompt：给大模型的“指令模板”（比如写着：“用户的请求是预约类（输出booker）、咨询类（输出info）、都不是（输出unclear），只返回关键词，别多写”）llm：大模型本身（根据上面的模板，分析用户请求）StrOutputParser()：“结果格式化器”（只提取大模型输出的文本字符串，去掉多余格式）
👉 举个例子：
用户说“我想预约明天的会议室” → 这个链会输出 booker；
用户说“你们公司几点下班” → 输出 info；
用户说“阿巴阿巴” → 输出 unclear。
2. 第二步：创建“专业部门”——branches字典
branches = {"booker": RunnablePassthrough.assign(output=lambda x: booking_handler(x['user_input']['request'])),"info": RunnablePassthrough.assign(output=lambda x: info_handler(x['user_input']['request'])),"unclear": RunnablePassthrough.assign(output=lambda x: unclear_handler(x['user_input']['request'])),
}这是一个“部门清单”，key是“接线员”输出的决策关键词，value是“这个部门怎么处理请求”。
重点拆解 RunnablePassthrough.assign(...)：
RunnablePassthrough()：“透传器”——把前面传来的所有数据（比如用户原始请求、决策结果）原封不动地传下去，不修改。.assign(output=...)：给“透传的数据”新增一个叫output的字段，字段值是“办事员处理后的结果”。lambda x: ...：匿名函数，接收“透传的数据x”，从中提取用户原始请求（x['user_input']['request']），传给对应的“办事员”（比如booking_handler）。
👉 通俗理解：
如果决策是 booker → 调用 booking_handler 处理用户的预约请求，把结果存到 output 字段；
如果决策是 info → 调用 info_handler 处理咨询请求，结果存到 output；
如果是 unclear → 调用 unclear_handler （比如返回“我没看懂你的请求”），结果存到 output。
3. 第三步：创建“调度员”——delegation_branch
delegation_branch = RunnableBranch((lambda x: x['decision'].strip() == 'booker', branches["booker"]),(lambda x: x['decision'].strip() == 'info', branches["info"]),branches["unclear"] # 默认分支
)这是“调度逻辑”：根据“接线员”的决策（x['decision']），把请求转给 branches 里对应的“部门”。
拆解：
RunnableBranch：LangChain的“分支选择器”，接收多个（条件+分支）对，再加上一个默认分支。- 第一个参数
(lambda x: x['decision'].strip() == 'booker', branches["booker"])：如果“决策结果”是booker，就走branches["booker"]这个部门。 - 第二个参数同理：决策是
info就走info部门。 - 最后一个
branches["unclear"]：兜底方案——如果决策既不是booker也不是info（比如大模型输出错了），就走“无法理解”部门。
👉 举个例子：
“接线员”输出 booker → 调度员把请求转给 booking_handler；
“接线员”输出 abc（乱码） → 调度员直接走默认的 unclear 部门。
4. 第四步：组装整个“服务中心”——coordinator_agent
coordinator_agent = {"decision": coordinator_router_chain,"user_input": RunnablePassthrough()
} | delegation_branch | (lambda x: x['output'])这是最终的“完整工作流”，把前面的“接线员、调度员、部门”串起来，形成一个可直接调用的“智能体”。
按执行顺序拆解（| 是LangChain的“管道符”，数据从左到右流动）：
第一步：并行准备数据（左边的字典）
这个字典的作用是“一次性生成两个关键数据”，传给后面的调度员：
- 键
decision：值是coordinator_router_chain→ 运行“接线员”，得到决策关键词（booker/info/unclear）。 - 键
user_input：值是RunnablePassthrough()→ 透传用户的原始请求（比如“我想预约会议室”），不让请求丢失。
👉 此时，字典会生成一个类似这样的数据：
{"decision": "booker", # 接线员的判断结果"user_input": {"request": "我想预约明天的会议室"} # 用户原始请求
}第二步：调度员处理（| delegation_branch）
把上面的字典传给“调度员”，调度员根据 decision 字段，调用对应的“部门”处理，最终返回一个包含 output 字段的数据：
{"decision": "booker","user_input": {"request": "我想预约明天的会议室"},"output": "预约成功！明天10点会议室A已为你预留" # booking_handler的处理结果
}第三步：提取最终结果（| (lambda x: x['output'])）
最后用一个简单的匿名函数，从上面的数据中只提取 output 字段的值，作为整个智能体的最终返回结果。
👉 最终给用户的响应就是："预约成功！明天10点会议室A已为你预留"