智能体路由:让 AI 从 “按部就班” 到 “灵活决策” 的核心机制
路由机制是智能体系统的核心控制技术,通过动态决策实现上下文感知的智能响应。本文系统介绍了路由的实现方式(基于LLM、嵌入、规则和机器学习模型)及其在客服、数据分类、多智能体协作等场景的应用价值。包含LangChain和Google ADK两种框架的实战代码示例,分别展示了基于状态图和工具自动路由的实现路径。路由技术使智能体能够突破线性流程限制,根据输入语义和系统状态自动选择最优处理路径,是构建自适
🚂现实中的智能体系统通常需要根据环境状态、用户输入或前序操作结果等因素,在多个潜在动作之间进行仲裁。这种动态决策能力 —— 即根据特定条件将控制流导向不同的专用函数、工具或子流程 —— 就是通过 “路由” 机制实现的。
🚂路由为智能体的操作框架引入了条件逻辑,使其从固定执行路径转变为动态评估特定标准、从一组可能的后续动作中进行选择的模式,从而实现更灵活、具备上下文感知的系统行为。
🚂例如,一个用于客户咨询的智能体在具备路由功能后,可以首先对用户查询进行分类以判断意图。根据分类结果,查询可以被导向专门的问答智能体、用于账户信息检索的数据库工具,或用于复杂问题升级的流程,而不是始终采用单一、预设的响应路径。因此,一个更高级的路由智能体可以:
- 分析用户查询。
- 根据查询意图进行路由:
- 如果意图为 “查询订单状态”,则路由到与订单数据库交互的子智能体或工具链。
- 如果意图为 “产品信息”,则路由到检索产品目录的子智能体或链。
- 如果意图为 “技术支持”,则路由到访问故障排查指南或升级到人工的链。
- 如果意图不明确,则路由到澄清意图的子智能体或提示链。
🚂路由模式的核心组件是执行评估并引导流程的机制,其实现方式包括:
- 基于 LLM 的路由:通过提示语言模型分析输入,并输出指示下一步或目标的标识符或指令。例如,可以提示 LLM:“分析以下用户查询,仅输出类别:‘订单状态 '、‘产品信息 '、‘技术支持 ' 或‘其他 '”。智能体系统读取输出并据此引导工作流。这种方式的优势是能处理模糊、复杂的自然语言输入,无需预设所有关键词,但缺点是存在一定的响应延迟,且结果可能受提示词质量影响。
- 基于嵌入的路由:将输入查询转为向量嵌入(参见第 14 章 RAG),再与代表不同路由或能力的嵌入进行比对,将查询路由到最相似的路径。适用于语义路由,即决策基于输入的含义而非关键词。比如用户说 “我的东西啥时候能到”,即便没有 “订单” 关键词,其向量嵌入也能匹配 “订单状态” 的语义,精准路由。
- 基于规则的路由:使用预定义规则或逻辑(如 if‑else、switch case),根据关键词、模式或结构化数据进行路由。此方法比 LLM 路由更快、更确定,但处理复杂或新颖输入的灵活性较低。比如预设 “退款”“退货” 关键词对应售后流程,能快速响应,但无法识别 “想把买的东西退了” 这类无关键词但语义一致的表述。
- 基于机器学习模型的路由:采用如分类器等判别模型,在小规模标注数据集上专门训练以实现路由任务。与嵌入方法类似,但其特点是监督微调过程,路由逻辑编码在模型权重中。与 LLM 路由不同,决策组件不是推理时执行提示的生成模型,而是已微调的判别模型。LLM 可用于生成合成训练数据,但不参与实时路由决策。这种方式兼顾了速度和灵活性,适合对响应效率和准确率都有要求的场景,但需要一定量的标注数据支撑。
🚂路由机制可在智能体操作周期的多个阶段实现:既可用于初始任务分类(比如用户刚输入查询时的意图识别),也可在处理链中间决定后续动作(比如订单查询后发现订单异常,自动路由到售后核查流程),或在子流程中选择最合适的工具(比如数据分析智能体在处理数据时,根据数据格式路由到 CSV 解析工具或 JSON 处理工具)。
🚂LangChain、LangGraph 和 Google 的 Agent Developer Kit (ADK) 等计算框架提供了定义和管理此类条件逻辑的显式结构。LangGraph 以其基于状态的图架构,尤其适合需要根据系统累积状态进行决策的复杂路由场景 —— 比如智能体执行多步任务时,每一步的结果都会改变系统状态,🚂LangGraph 能基于最新状态动态调整路由方向。Google ADK 则为智能体能力和交互模型的结构化提供基础组件,便于开发者快速封装路由逻辑,无需从零搭建框架。在这些框架的执行环境中,开发者只需定义可能的操作路径,以及决定节点间转换的函数或模型评估规则,就能实现灵活的路由控制。
🚂路由的实现使系统超越了确定性顺序处理,能够开发出更具适应性的执行流,动态响应更广泛的输入和状态变化。
实践应用与场景:路由到底能解决哪些实际问题🗼യ˚𝓦𝓪𝓷𝓭𝓮𝓻*﹆
路由模式是设计自适应智能体系统的关键控制机制,使其能够根据输入和内部状态动态调整执行路径,广泛应用于多个领域,以下是几个贴近实际开发的典型场景:
1. 人机交互场景:让 AI 助手真正 “懂” 用户🗼യ˚𝓦𝓪𝓷𝓭𝓮𝓻*﹆
在虚拟助手、AI 教师、智能客服等场景中,路由是实现 “上下文感知对话” 的核心。比如校园智能助手,学生的输入可能是 “查一下我周一的课表”“助学金啥时候发”“宿舍灯坏了要修” 等完全不同的需求,路由机制会先解析用户意图:
- 识别到 “课表”“周一” 等关键词 / 语义,路由到 “教务系统查询子智能体”,调用课表接口返回结果;
- 识别到 “助学金”“发放”,路由到 “学生资助管理子智能体”,对接财务系统数据;
- 识别到 “报修”“灯坏了”,路由到 “后勤工单创建子智能体”,自动生成报修单并推送至后勤系统。没有路由的话,智能助手只能按固定流程回复,要么答非所问,要么需要用户手动选择功能入口,体验大打折扣。而路由让 AI 助手从 “被动回应” 变成 “主动判断”,实现超越线性对话流的自然交互。
2. 自动化数据与文档处理:给杂乱数据找 “专属处理通道”🗼യ˚𝓦𝓪𝓷𝓭𝓮𝓻*﹆
🚂在企业级自动化场景中,路由承担着 “分类分拣员” 的角色。比如某公司的客户工单处理系统,每天会收到上千条来自邮件、小程序、电话语音转文字的工单,内容涵盖销售咨询、售后投诉、技术问题、垃圾信息等:
- 基于规则 + 嵌入的混合路由:先通过规则路由筛选出含 “退款”“投诉” 的高优先级工单,直接路由到人工坐席;再通过嵌入路由将 “产品使用教程” 类工单,路由到自动回复子智能体,调用知识库生成标准化解答;
- 对于 API 接入的结构化数据(如 JSON 格式的订单反馈),路由会根据 “订单状态” 字段(已发货 / 未发货 / 退款中),分别导向物流跟踪、催单处理、退款审核等不同工作流;
- 甚至针对文档格式的差异,路由能判断输入是 PDF 合同还是 CSV 报表,分别路由到 OCR 解析工具或数据清洗工具,避免格式不兼容导致的处理失败。这种自动化路由不仅能减少 80% 的人工分拣工作量,还能保证工单处理的时效性和准确性。
3. 多工具 / 多智能体协作:做智能体系统的 “调度中枢”🗼യ˚𝓦𝓪𝓷𝓭𝓮𝓻*﹆
🚂在由多个专用智能体 / 工具组成的复杂系统中,路由是核心调度器。比如一个 AI 科研助手系统,整合了文献检索、论文摘要、数据分析、图表生成等多个子智能体:
- 用户输入 “分析近 5 年人工智能领域的顶会论文趋势”,路由首先判断核心需求是 “文献分析 + 数据可视化”;
- 先路由到 “文献检索子智能体”,调用 Google Scholar API 获取相关论文数据;
- 检索完成后,路由根据 “数据格式为 CSV” 的状态,将数据路由到 “数据分析子智能体”,进行趋势统计;
- 分析完成后,路由再根据 “生成可视化图表” 的目标,将结果路由到 “图表生成子智能体”,输出柱状图 / 折线图;
- 若检索过程中发现论文数量不足,路由会自动调整路径,转向 “补充检索子智能体”,扩大检索范围。再比如 AI 编程助手,路由会先识别用户输入的是 “Python 调试”“Java 代码翻译” 还是 “SQL 语句优化”,再将代码片段路由到对应的工具链,避免用 “通用代码解释” 逻辑处理所有问题,提升解决效率。
4. 工业级智能体:动态适配复杂环境状态🗼യ˚𝓦𝓪𝓷𝓭𝓮𝓻*﹆
🚂在智能制造、自动驾驶等场景中,路由更是不可或缺。比如工厂的智能巡检机器人,其路由逻辑会根据环境状态动态调整:
- 传感器检测到 “温度异常”,路由到 “故障预警子智能体”,立即停止对应设备并推送警报;
- 检测到 “巡检区域无人”,路由到 “快速巡检模式”,加快移动速度并简化检测流程;
- 检测到 “网络信号弱”,路由到 “本地缓存模式”,先存储数据,待信号恢复后再路由到 “数据上传子智能体”。这种基于环境状态的路由,让智能体能够适应复杂、动态变化的现实场景,避免因单一执行路径导致的故障或效率低下。
总之,路由为系统提供了逻辑仲裁能力,是构建功能多样、具备上下文感知系统的基础。它将智能体从静态执行者转变为能根据变化条件做出决策的动态系统,也是区分 “初级智能体” 和 “高级自适应智能体” 的核心特征之一。
实战代码示例(LangChain):从零实现一个简单的路由智能体🗼യ˚𝓦𝓪𝓷𝓭𝓮𝓻*﹆
🚂代码实现路由需定义可能路径及决策逻辑。LangChain 和 LangGraph 等框架提供了专用组件和结构,LangGraph 的状态图结构尤其适合路由逻辑的可视化和实现。以下代码演示了使用 LangChain 和 Google 生成式 AI 构建的简单智能体系统。系统设置一个 “协调者”,根据请求意图(订单查询、产品信息、技术支持、不明确)将用户请求路由到不同的 “子智能体” 处理器,模拟多智能体架构中的基本委托模式。
第一步:安装所需依赖🗼യ˚𝓦𝓪𝓷𝓭𝓮𝓻*﹆
🚂首先确保本地环境安装了必要的库,打开终端执行以下命令:
pip install langchain langgraph google‑cloud‑aiplatform langchain‑google‑genai google‑adk
第二步:路由器示例代码:🗼യ˚𝓦𝓪𝓷𝓭𝓮𝓻*﹆
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
# --- 配置 ---
# 确保环境变量已设置 API 密钥(如 GOOGLE_API_KEY)
try:
llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0)
print(f"语言模型初始化成功:{llm.model}")
except Exception as e:
print(f"语言模型初始化失败:{e}")
llm = None
# --- 定义模拟子智能体处理器(等同于 ADK sub_agents)---
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}'。请补充说明。"
# --- 定义协调者路由链(等同于 ADK 协调者指令)---
coordinator_router_prompt = ChatPromptTemplate.from_messages([
("system", """分析用户请求,判断应由哪个专属处理器处理。
- 若请求涉及预订机票或酒店,输出 'booker'。
- 其他一般信息问题,输出 'info'。
- 若请求不明确或不属于上述类别,输出 'unclear'。
只输出一个词:'booker'、'info' 或 'unclear'。"""),
("user", "{request}")
])
if llm:
coordinator_router_chain = coordinator_router_prompt | llm | StrOutputParser()
# --- 定义委托逻辑(等同于 ADK 的 Auto-Flow)---
branches = {
"booker": RunnablePassthrough.assign(output=lambda x: booking_handler(x['request'])),
"info": RunnablePassthrough.assign(output=lambda x: info_handler(x['request'])),
"unclear": RunnablePassthrough.assign(output=lambda x: unclear_handler(x['request'])),
}
delegation_branch = RunnableBranch(
(lambda x: x['decision'].strip() == 'booker', branches["booker"]),
(lambda x: x['decision'].strip() == 'info', branches["info"]),
branches["unclear"] # 默认分支
)
coordinator_agent = {
"decision": coordinator_router_chain,
"request": RunnablePassthrough()
} | delegation_branch | (lambda x: x['output'])
# --- 示例用法 ---
def main():
if not llm:
print("\n 因 LLM 初始化失败,跳过执行。")
return
print("--- 预订请求示例 ---")
request_a = "帮我预订飞往伦敦的机票。"
result_a = coordinator_agent.invoke({"request": request_a})
print(f"最终结果 A: {result_a}")
print("\n--- 信息请求示例 ---")
request_b = "意大利的首都是哪里?"
result_b = coordinator_agent.invoke({"request": request_b})
print(f"最终结果 B: {result_b}")
print("\n--- 不明确请求示例 ---")
request_c = "讲讲量子物理。"
result_c = coordinator_agent.invoke({"request": request_c})
print(f"最终结果 C: {result_c}")
if __name__ == "__main__":
main()
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
# --- 定义工具函数 ---
def booking_handler(request: str) -> str:
"""
处理机票和酒店预订请求。
Args:
request: 用户的预订请求。
Returns:
预订处理确认信息。
"""
print("------------------------------ 预订处理器已调用 ------------------------------")
return f"已模拟处理预订请求:'{request}'。"
def info_handler(request: str) -> str:
"""
处理一般信息请求。
Args:
request: 用户问题。
Returns:
信息检索处理结果。
"""
print("------------------------------ 信息处理器已调用 ------------------------------")
return f"信息请求:'{request}'。结果:模拟信息检索。"
def unclear_handler(request: str) -> str:
"""处理无法委托的请求。"""
return f"协调者无法委托请求:'{request}'。请补充说明。"
# --- 创建工具 ---
booking_tool = FunctionTool(booking_handler)
info_tool = FunctionTool(info_handler)
# 定义配备工具的专用子智能体
booking_agent = Agent(
name="Booker",
model="gemini-2.0-flash",
description="专门处理机票和酒店预订请求,通过 booking tool 实现。",
tools=[booking_tool]
)
info_agent = Agent(
name="Info",
model="gemini-2.0-flash",
description="专门提供一般信息和答疑,通过 info tool 实现。",
tools=[info_tool]
)
# 定义父智能体(协调者),包含委托指令
coordinator = Agent(
name="Coordinator",
model="gemini-2.0-flash",
instruction=(
"你是主协调者,只负责分析用户请求并委托给合适的专用智能体。"
"不要直接回答用户。\n"
"- 任何涉及机票或酒店预订的请求,委托给 'Booker' 智能体。\n"
"- 其他一般信息问题,委托给 'Info' 智能体。"
),
description="负责将用户请求路由到正确专用智能体的协调者。",
sub_agents=[booking_agent, info_agent]
)
# --- 执行逻辑 ---
async def run_coordinator(runner: InMemoryRunner, request: str):
"""用给定请求运行协调者智能体并委托。"""
print(f"\n--- 协调者运行请求:'{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"协调者最终响应:{final_result}")
return final_result
except Exception as e:
print(f"处理请求时发生错误:{e}")
return f"处理请求时发生错误:{e}"
async def main():
"""主函数,运行 ADK 示例。"""
print("--- Google ADK 路由示例(ADK Auto-Flow 风格)---")
print("注意:需安装并认证 Google ADK。")
runner = InMemoryRunner(coordinator)
# 示例用法
result_a = await run_coordinator(runner, "帮我预订巴黎的酒店。")
print(f"最终输出 A: {result_a}")
result_b = await run_coordinator(runner, "世界最高的山峰是什么?")
print(f"最终输出 B: {result_b}")
result_c = await run_coordinator(runner, "说一个随机的事实。") # 应委托给 Info
print(f"最终输出 C: {result_c}")
result_d = await run_coordinator(runner, "查找下个月飞往东京的航班。") # 应委托给 Booker
print(f"最终输出 D: {result_d}")
if __name__ == "__main__":
import nest_asyncio
import asyncio
nest_asyncio.apply()
asyncio.run(main())
🚂本脚本包含一个主协调者智能体和两个专用子智能体:Booker 和 Info。每个子智能体
配备 FunctionTool ,分别模拟预订和信息检索。 booking_handler 处理机票和酒店
预订, info_handler 处理一般信息请求。 unclear_handler 用于无法委托的请求(当
前逻辑未显式使用)。
协调者智能体的主要职责是分析用户消息并自动委托给 Booker 或 Info,ADK 的
Auto‑Flow 机制会根据 sub_agents 自动完成委托。 run_coordinator 函数设置
InMemoryRunner,创建用户和会话 ID,通过 runner 处理请求并提取最终响应文本。
main 函数演示了不同请求的处理流程,展示了预订请求委托给 Booker,信息请求委
托给 Info。
🚂是什么:智能体系统需应对多样输入和场景,单一线性流程无法根据上下文做决策。缺
乏选择合适工具或子流程的机制,系统将变得僵化,难以应对真实世界的复杂请求。
🚂为什么:路由模式通过引入条件逻辑,为智能体提供标准化解决方案。系统可先分析查
询意图,再动态将控制流导向最合适的工具、函数或子智能体。决策可由 LLM 提示、
规则、嵌入语义相似度等驱动。路由将静态执行路径转变为灵活、具备上下文感知的工
作流,能选择最佳动作。
🚂当智能体需根据用户输入或当前状态,在多个不同工作流、工具或子智能体
间做选择时,应采用路由模式。适用于需对请求进行分流或分类的应用,如客服机器人
区分销售、技术支持、账户管理等问题。

关键要点🗼യ˚𝓦𝓪𝓷𝓭𝓮𝓻*﹆
总结🗼യ˚𝓦𝓪𝓷𝓭𝓮𝓻*﹆
参考资料🗼യ˚𝓦𝓪𝓷𝓭𝓮𝓻*﹆
更多推荐


所有评论(0)