Day 34 Routing路由—— 让 Agent 学会“分流”欢迎来到第三十四天在构建复杂的 AI 应用时我们常常需要处理多种类型的用户请求闲聊、知识问答、调用外部 API、查询数据库等。如果用一个 Agent 处理所有请求不仅工具列表会变得庞大模型也容易选错工具。今天我们将学习Routing路由技术先判断用户意图再将请求分发给专门的子链或子 Agent。这就像医院的分诊台——先判断你该去哪个科室再由专科医生处理。一、今日学习目标理解 Routing 的核心思想将复杂问题分解为多个专门的处理器通过意图分类决定路由目标。掌握两种常见的路由实现方式基于 LLM 的路由用模型判断用户意图并输出路由标签。基于语义相似度的路由用 Embedding 匹配预定义意图。学会使用 LangChain 的RunnableBranch和自定义路由链。构建一个“多功能助手”根据用户输入分别路由到闲聊、数学计算、天气查询三条不同的处理链。了解 Routing 在多智能体系统中的作用为后续多 Agent 协作打下基础。二、详细实现步骤步骤 1理解 Routing 的架构一个典型的路由系统包含三个部分意图分类器Router接收用户输入输出一个路由标签如chat、math、weather。路由分发器Dispatcher根据标签将请求转发给对应的处理链。子处理器Handler每个处理链专门负责一类任务可以是简单的 LLM 链、RAG 链或完整的 Agent。用户输入 → 意图分类器 → 路由标签 → 分发器 → 子处理器 → 输出步骤 2环境准备确保已安装langchain、langchain-openai。新建routing_demo.py导入所需模块importosfromdotenvimportload_dotenvfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlangchain_core.runnablesimportRunnableBranch,RunnablePassthrough,RunnableLambdafromlangchain_core.messagesimportHumanMessage load_dotenv()llmChatOpenAI(modeldeepseek-chat,api_keyos.getenv(DEEPSEEK_API_KEY),base_urlhttps://api.deepseek.com,temperature0.1)步骤 3构建意图分类器意图分类器本质上是一个 LLM 链输入用户问题输出预定义的类别标签。我们定义三个类别chat闲聊、math数学计算、weather天气查询。router_promptChatPromptTemplate.from_messages([(system,你是一个意图分类器。请根据用户输入判断其意图并只输出以下类别之一 - chat闲聊、问候、与数学和天气无关的问题 - math数学计算、数学问题 - weather天气查询 只输出类别名称不要输出任何其他文字。),(human,{input})])router_chainrouter_prompt|llm|StrOutputParser()# 测试print(router_chain.invoke({input:你好今天过得怎么样}))# 期望输出 chatprint(router_chain.invoke({input:计算 25 * 4 等于多少}))# 期望输出 mathprint(router_chain.invoke({input:北京今天天气怎么样}))# 期望输出 weather如果输出不稳定例如带有多余文字可以加入 Few-shot 示例或降低温度。步骤 4构建三个子处理器每个子处理器是一个独立的链专门处理一类请求。4.1 闲聊处理器chat_promptChatPromptTemplate.from_messages([(system,你是一个友好的聊天助手请用轻松愉快的语气回复用户。),(human,{input})])chat_chainchat_prompt|llm|StrOutputParser()4.2 数学处理器数学处理器可以是一个简单的 LLM 链也可以是一个带计算器工具的 Agent。这里为了简洁我们用一个带 CoT 的链。math_promptChatPromptTemplate.from_messages([(system,你是一个数学专家。请一步步计算最后给出答案。),(human,{input})])math_chainmath_prompt|llm|StrOutputParser()4.3 天气处理器天气处理器需要一个工具来获取天气信息。我们定义一个简单的模拟函数并用 LLM 链调用它。defget_weather(city:str)-str:weather_data{北京:晴26°C湿度 40%,上海:多云28°C湿度 65%,广州:小雨30°C湿度 80%}returnweather_data.get(city,f未找到{city}的天气信息)weather_promptChatPromptTemplate.from_messages([(system,你是一个天气助手。用户会询问天气你需要从以下天气数据中提取信息回答。\n天气数据{weather_data}),(human,{input})])defweather_handler(input_dict):user_inputinput_dict[input]# 简单提取城市名实际项目中可以用 NER 或让 LLM 提取city北京# 默认forcin[北京,上海,广州]:ifcinuser_input:citycbreakweather_dataget_weather(city)returnweather_prompt|llm|StrOutputParser()weather_chainRunnableLambda(weather_handler)步骤 5使用 RunnableBranch 实现路由分发RunnableBranch是 LangChain 提供的路由组件它根据条件选择不同的分支。defroute_by_intent(input_dict):根据意图返回对应的链intentrouter_chain.invoke({input:input_dict[input]})intentintent.strip().lower()print(f[路由] 检测到意图{intent})ifmathinintent:returnmath_chainelifweatherinintent:returnweather_chainelse:returnchat_chain# 构建完整路由链full_chainRunnableLambda(route_by_intent)# 测试print(\n 测试路由 )forqueryin[你好呀,计算 (127)*3,上海天气怎么样]:print(f\n用户{query})resultfull_chain.invoke({input:query})print(f回答{result})运行后你会看到每个请求被正确路由到对应的处理器。步骤 6使用 RunnableBranch 的另一种写法如果你希望使用更声明式的方式可以用RunnableBranchbranchRunnableBranch((lambdax:mathinrouter_chain.invoke({input:x[input]}).lower(),math_chain),(lambdax:weatherinrouter_chain.invoke({input:x[input]}).lower(),weather_chain),chat_chain# 默认分支)# 测试resultbranch.invoke({input:计算 5*6})print(result)但注意这种方式会对每个条件都调用一次路由器效率较低。推荐使用步骤 5 的自定义函数方式只调用一次路由器。步骤 7基于语义相似度的路由备选方案除了 LLM 分类还可以用 Embedding 计算用户输入与预定义意图描述的相似度选择最接近的意图。这种方法更快但灵活性稍差。fromlangchain_openaiimportOpenAIEmbeddingsimportnumpyasnp embeddingsOpenAIEmbeddings(modelembedding-2,api_keyos.getenv(ZHIPU_API_KEY),base_urlhttps://open.bigmodel.cn/api/paas/v4)# 预定义意图描述intent_descriptions{chat:闲聊、问候、日常对话,math:数学计算、算术、数学问题,weather:天气查询、气温、下雨}# 预先计算意图向量intent_vectors{k:embeddings.embed_query(v)fork,vinintent_descriptions.items()}defcosine_similarity(v1,v2):v1,v2np.array(v1),np.array(v2)returnfloat(np.dot(v1,v2)/(np.linalg.norm(v1)*np.linalg.norm(v2)))defroute_by_similarity(user_input:str)-str:query_vecembeddings.embed_query(user_input)best_intentNonebest_sim-1forintent,vecinintent_vectors.items():simcosine_similarity(query_vec,vec)ifsimbest_sim:best_simsim best_intentintentreturnbest_intent# 测试print(route_by_similarity(今天天气如何))# 期望 weatherprint(route_by_similarity(3 加 5 等于几))# 期望 math这种方法的优点是不需要调用 LLM速度快、成本低缺点是无法处理复杂的意图边界且需要维护意图描述。步骤 8整合与测试将路由链封装成一个完整的应用测试多个不同意图的输入观察路由是否正确。defmulti_function_assistant(user_input:str)-str:intentrouter_chain.invoke({input:user_input}).strip().lower()print(f[意图]{intent})ifmathinintent:returnmath_chain.invoke({input:user_input})elifweatherinintent:returnweather_chain.invoke({input:user_input})else:returnchat_chain.invoke({input:user_input})# 测试test_queries[你好你叫什么名字,请帮我计算 100 / 4 20,广州今天热不热,讲个笑话吧,北京天气]forqintest_queries:print(f\n用户{q})print(f助手{multi_function_assistant(q)})三、常见问题与调试Q1意图分类器输出不稳定有时返回“chat”有时返回“闲聊”。→ 在 System Prompt 中明确要求“只输出类别名称不要输出其他文字”并给出确切的标签列表。加入 Few-shot 示例可以进一步提高稳定性。如果仍然不稳定可以在代码中用关键词匹配做二次校验。Q2用户的问题涉及多个意图例如“北京天气怎么样顺便算一下 25*4”。→ 简单的路由只能选择一个分支。要处理多意图可以先做意图拆分用 LLM 将问题分解为多个子问题然后分别路由。这属于更高级的 Agent 规划能力我们后续会涉及。Q3天气处理器中城市名提取太简单无法处理“今天首都的天气”。→ 可以改用 LLM 提取城市名或者使用专门的 NER 工具。在天气处理器中加入一个 LLM 调用从用户输入中提取城市名。Q4路由器的 LLM 调用增加了延迟和成本。→ 对于简单场景可以使用语义相似度路由步骤 7替代 LLM 路由。或者使用更小、更快的模型做分类。也可以缓存路由结果对于相同输入直接返回。Q5如何扩展到更多意图→ 只需在路由器的 System Prompt 中增加类别并添加对应的子处理器。当意图数量很多时可以考虑分层路由先粗分类如“技术类”、“生活类”再细分类。Q6路由和 Agent 的关系是什么→ 路由可以看作一种“预决策”决定用哪个 Agent 或链来处理。在多智能体系统中路由常用于将任务分发给不同角色的 Agent如“客服 Agent”、“技术 Agent”。这为后续的 AutoGen/CrewAI 协作奠定了基础。四、今日总结与作业今天你完成了✅ 理解了 Routing 的核心思想和架构。✅ 使用 LLM 构建了意图分类器并实现了三条子处理器链。✅ 掌握了RunnableBranch和自定义路由函数的用法。✅ 了解了基于语义相似度的路由方案。✅ 构建了一个多功能助手能根据意图自动分流。今日作业必做为多功能助手增加第四个意图translate翻译要求用户输入“把‘你好’翻译成英文”时路由到翻译链。翻译链可以使用简单的 LLM 提示词实现。修改意图分类器的 Prompt加入 Few-shot 示例每个类别给一个例子测试分类准确率是否提高。实现一个基于语义相似度的路由器使用智谱 Embedding 计算相似度并与 LLM 路由器的结果进行对比测试 10 个不同输入记录两者的一致性。思考题在构建一个企业级客服系统时你如何设计路由层次请画出路由流程图可以用文字描述并说明每个路由节点的判断依据。明日预告我们将进入第三阶段的最后一天进行项目复盘——完成一个完整的带记忆的客服 Agent。这个项目将整合 RAG、记忆管理、工具调用和路由技术是你简历上的又一个亮点。有任何问题欢迎随时提问
