路由模式Routing相比提示链模式的固定线性流程路由模式能根据输入动态选择不同的处理路径。它通过一个路由器对输入进行分类或评估然后将其导向最合适的专用处理器使系统从固定执行路径转变为动态决策系统。这种动态决策能力——即根据特定条件将控制流导向不同的专用函数、工具或子流程——就是通过路由机制实现的。路由为智能体的操作框架引入了条件逻辑使其从固定执行路径转变为动态决策系统。这种机制使智能体能够根据输入数据的特征、上下文信息或外部环境的变化选择最合适的工具或子流程来处理任务。核心思想条件分发—— 先判断这是什么类型的请求再决定交给谁处理。例如一个用于客户咨询的智能体在具备路由功能后可以对用户查询进行分类以判断意图。根据分类结果查询可以被导向专门的问答智能体、用于账户信息检索的数据库工具或者用于复杂问题升级的流程。与提示链的关系对比提示链模式路由模式流程固定线性A → B → C动态分支A → B₁ / B₂ / B₃决策无决策按顺序执行有分类/判断步骤适用步骤确定的流水线任务输入类型多样、需要不同处理方式的任务路由模式的核心组件路由模式的核心组件是执行评估并引导流程的机制其实现方式包括以下五种1. 基于 LLM 的路由通过提示语言模型分析输入并输出指示下一步或目标标识符或指令。示例 Prompt“分析用户查询仅输出类别‘订单状态’、‘产品信息’、‘技术支持’或’其他’。”智能体系统读取输出并据此引导工作流。优点灵活能处理复杂和新颖的输入缺点有延迟和成本结果可能不稳定2. 基于嵌入的路由利用嵌入向量RAG来表示输入数据并通过计算与不同工具或子流程的嵌入向量之间的相似度来决定路由。例如智能体可以将用户查询转换为嵌入向量并计算与预定义工具或子流程的嵌入向量之间的相似度以确定最相关的工具或子流程来处理查询。优点语义理解能力强适合模糊匹配缺点需要预先计算各路径的嵌入表示3. 基于规则的路由通过预定义的规则或条件如 if-else、switch-case来指导路由决策根据关键词或者结构化数据进行路由。优点比 LLM 路由更快、更确定无 LLM 调用成本缺点处理复杂或新颖输入的灵活性较低4. 基于机器学习模型的路由采用如分类器等判别模型在小规模标注数据集上专门训练以实现路由任务。与嵌入方法类似但其特点是监督微调过程路由逻辑编码在模型权重中。与 LLM 路由不同决策组件是一个独立的判别模型而非提示引导的语言模型输出。LLM 可用于生成合成训练数据但不参与实时路由决策。优点快速且准确适合高频场景缺点需要标注数据和训练过程5. 混合路由结合上述多种方法以实现更鲁棒的路由机制。例如系统可以首先使用基于规则的路由进行快速分类如果输入不符合任何规则则退回到基于 LLM 的路由进行更深入的分析。流程示意┌→ [处理器 A] → 结果 A 输入 → [路由器] ──→├→ [处理器 B] → 结果 B └→ [处理器 C] → 结果 C路由机制可以在智能体操作周期的多个阶段实现既可用于分配初始任务分类也可在处理链中间决定后续动作或在子流程中选择最合适的工具。适用场景路由模式是设计自适应智能体系统的关键控制机制使其能够根据输入和内部状态动态调整执行路径广泛应用于多个领域场景路由示例客服系统分类意图 → 订单查询 / 技术支持 / 投诉处理文档处理识别格式 → JSON 解析 / CSV 转换 / 纯文本提取AI 编程助手识别语言和意图 → 调试 / 解释 / 代码翻译研究系统判断任务类型 → 检索 / 摘要 / 深度分析人机交互场景在虚拟助手、AI 教师等场景中路由用于解析用户意图。系统对自然语言查询进行初步分析决定后续动作如调用信息检索工具、升级到人工、或根据用户表现选择下一个课程模块从而实现超越线性对话流的上下文响应。自动化数据与文档处理路由承担分类与分发功能。系统根据内容、元数据或格式分析如邮件、工单、API 数据并将其导向相应工作流如销售线索导入、针对 JSON/CSV 的数据转换、或紧急问题升级。多工具/多智能体协作系统路由充当高级调度器。例如研究系统由检索、摘要、分析等智能体组成路由器根据当前目标分配任务。同样AI 编程助手会先识别编程语言和用户意图调试、解释、翻译再将代码片段交给对应工具处理。总之路由为系统提供了逻辑仲裁能力是构建功能多样、具备上下文感知系统的基础。它将智能体从静态执行者转变为能根据变化条件做出决策的动态系统。代码示例示例 1LangChain 实现fromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnablePassthrough,RunnableBranch load_dotenv()# --- 配置 ---# 确保环境变量已设置 OPENAI_API_KEYtry:llmChatOpenAI(modelgpt-4o-mini,temperature0)print(f语言模型初始化成功{llm.model_name})exceptExceptionase:print(f语言模型初始化失败{e})llmNone# --- 定义模拟子智能体处理器等同于 ADK sub_agents---defbooking_handler(request:str)-str:模拟预订智能体处理请求。print(\n--- 委托给预订处理器 ---)returnf预订处理器已处理请求{request}。结果模拟预订动作。definfo_handler(request:str)-str:模拟信息智能体处理请求。print(\n--- 委托给信息处理器 ---)returnf信息处理器已处理请求{request}。结果模拟信息检索。defunclear_handler(request:str)-str:处理无法委托的请求。print(\n--- 处理不明确请求 ---)returnf协调者无法委托请求{request}。请补充说明。# --- 定义协调者路由链等同于 ADK 协调者指令---coordinator_router_promptChatPromptTemplate.from_messages([(system,分析用户请求判断应由哪个专属处理器处理。 - 若请求涉及预订机票或酒店输出 booker。 - 其他一般信息问题输出 info。 - 若请求不明确或不属于上述类别输出 unclear。 只输出一个词booker、info 或 unclear。),(user,{request})])ifllm:coordinator_router_chaincoordinator_router_prompt|llm|StrOutputParser()# --- 定义委托逻辑等同于 ADK 的 Auto-Flow---branches{booker:RunnablePassthrough.assign(outputlambdax:booking_handler(x[request][request])),info:RunnablePassthrough.assign(outputlambdax:info_handler(x[request][request])),unclear:RunnablePassthrough.assign(outputlambdax:unclear_handler(x[request][request])),}delegation_branchRunnableBranch((lambdax:x[decision].strip()booker,branches[booker]),(lambdax:x[decision].strip()info,branches[info]),branches[unclear]# 默认分支)coordinator_agent({decision:coordinator_router_chain,request:RunnablePassthrough()}|delegation_branch|(lambdax:x[output]))# --- 示例用法 ---defmain():ifnotllm:print(\n因 LLM 初始化失败跳过执行。)returnprint(--- 预订请求示例 ---)request_a帮我预订飞往伦敦的机票。result_acoordinator_agent.invoke({request:request_a})print(f最终结果 A:{result_a})print(\n--- 信息请求示例 ---)request_b意大利的首都是哪里result_bcoordinator_agent.invoke({request:request_b})print(f最终结果 B:{result_b})print(\n--- 不明确请求示例 ---)request_c讲讲量子物理。result_ccoordinator_agent.invoke({request:request_c})print(f最终结果 C:{result_c})if__name____main__:main()使用 LangChain OpenAI GPT 构建路由系统协调者coordinator通过 LLM 将请求分类为booker/info/unclear子处理器三个模拟函数分别处理预订、信息查询、不明确请求路由机制RunnableBranch根据分类结果将请求委托给对应处理器用户请求 → [LLM 分类] → booker / info / unclear → [对应处理器] → 结果代码利用 LangChain 和 GPT 构建了一个简单智能体系统。定义了booking_handler、info_handler、unclear_handler三个模拟子智能体处理器分别处理不同类型请求。核心组件coordinator_router_chain通过ChatPromptTemplate指示语言模型将用户请求分类为booker、info或unclear。RunnableBranch根据分类结果将原始请求委托给对应处理函数。coordinator_agent组合上述组件先路由请求再交由选定处理器最终输出处理结果。示例 2Google ADK 实现importuuidfromtypingimportDict,Any,Optionalfromdotenvimportload_dotenvfromgoogle.adk.agentsimportAgentfromgoogle.adk.runnersimportInMemoryRunnerfromgoogle.adk.toolsimportFunctionToolfromgoogle.genaiimporttypesfromgoogle.adk.eventsimportEvent load_dotenv()# --- 定义工具函数 ---defbooking_handler(request:str)-str: 处理机票和酒店预订请求。 Args: request: 用户的预订请求。 Returns: 预订处理确认信息。 print(-------------------------- 预订处理器已调用 ----------------------------)returnf已模拟处理预订请求{request}。definfo_handler(request:str)-str: 处理一般信息请求。 Args: request: 用户问题。 Returns: 信息检索处理结果。 print(-------------------------- 信息处理器已调用 ----------------------------)returnf信息请求{request}。结果模拟信息检索。defunclear_handler(request:str)-str:处理无法委托的请求。returnf协调者无法委托请求{request}。请补充说明。# --- 创建工具 ---booking_toolFunctionTool(booking_handler)info_toolFunctionTool(info_handler)# 定义配备工具的专用子智能体booking_agentAgent(nameBooker,modelgemini-2.0-flash,description专门处理机票和酒店预订请求通过 booking tool 实现。,tools[booking_tool])info_agentAgent(nameInfo,modelgemini-2.0-flash,description专门提供一般信息和答疑通过 info tool 实现。,tools[info_tool])# 定义父智能体协调者包含委托指令coordinatorAgent(nameCoordinator,modelgemini-2.0-flash,instruction(你是主协调者只负责分析用户请求并委托给合适的专用智能体。不要直接回答用户。\n- 任何涉及机票或酒店预订的请求委托给 Booker 智能体。\n- 其他一般信息问题委托给 Info 智能体。),description负责将用户请求路由到正确专用智能体的协调者。,sub_agents[booking_agent,info_agent])# --- 执行逻辑 ---asyncdefrun_coordinator(runner:InMemoryRunner,request:str):用给定请求运行协调者智能体并委托。print(f\n--- 协调者运行请求{request} ---)final_resulttry:user_iduser_123session_idstr(uuid.uuid4())awaitrunner.session_service.create_session(app_namerunner.app_name,user_iduser_id,session_idsession_id)foreventinrunner.run(user_iduser_id,session_idsession_id,new_messagetypes.Content(roleuser,parts[types.Part(textrequest)]),):ifevent.is_final_response()andevent.content:ifhasattr(event.content,text)andevent.content.text:final_resultevent.content.textelifevent.content.parts:text_parts[part.textforpartinevent.content.partsifpart.text]final_result.join(text_parts)breakprint(f协调者最终响应{final_result})returnfinal_resultexceptExceptionase:print(f处理请求时发生错误{e})returnf处理请求时发生错误{e}asyncdefmain():主函数运行 ADK 示例。print(--- Google ADK 路由示例ADK Auto-Flow 风格---)print(注意需安装并认证 Google ADK。)runnerInMemoryRunner(coordinator)result_aawaitrun_coordinator(runner,帮我预订巴黎的酒店。)print(f最终输出 A:{result_a})result_bawaitrun_coordinator(runner,世界最高的山峰是什么)print(f最终输出 B:{result_b})result_cawaitrun_coordinator(runner,说一个随机的事实。)print(f最终输出 C:{result_c})result_dawaitrun_coordinator(runner,查找下个月飞往东京的航班。)print(f最终输出 D:{result_d})if__name____main__:importnest_asyncioimportasyncio nest_asyncio.apply()asyncio.run(main())Agent Development KitADK是用于工程化智能体系统的框架提供结构化环境定义智能体能力与行为。与显式计算图架构不同ADK 路由通常通过定义一组代表智能体功能的工具实现。用户查询由底层模型匹配到合适工具框架内部自动完成路由。使用 Google ADK 构建多智能体路由系统协调者智能体Coordinator分析请求并自动委托子智能体Booker预订和 Info信息各自配备FunctionTool路由机制ADK 的 Auto-Flow 根据sub_agents的描述自动完成委托无需显式编写分支逻辑脚本包含一个主协调者智能体和两个专用子智能体。每个子智能体配备FunctionTool分别模拟预订和信息检索。协调者智能体的主要职责是分析用户消息并自动委托给 Booker 或 InfoADK 的 Auto-Flow 机制会根据sub_agents自动完成委托。run_coordinator函数设置InMemoryRunner创建用户和会话 ID通过 runner 处理请求并提取最终响应文本。两种实现的核心区别LangChain 需要显式编写路由分支逻辑RunnableBranch而 ADK 通过智能体描述 Auto-Flow 自动路由。局限性分类错误路由器判断失误会导致请求进入错误的处理路径维护成本路径增多时路由规则和处理器的维护复杂度上升单层限制简单路由只做一次分发复杂场景可能需要多层路由或与其他模式结合