Langgraph 2. 路由 Routing

📅 发布时间:2026/7/9 14:15:08 👁️ 浏览次数:
Langgraph 2. 路由 Routing
摘要路由Routing为智能体引入条件分支先对输入或当前状态做判断再决定下一步走哪条处理路径从而从「单一路径执行」变为「按条件选路径」。本文说明路由的动机、典型实现方式LLM / 规则 / 嵌入 / 小模型并重点介绍在LangGraph中如何用状态图 条件边实现「先分类、再分发」的流程配套示例为基于 LLM 的意图分类booker / info / unclear驱动三条分支。关键词路由Routing条件分支LangGraph条件边StateGraph意图分类LLM 路由Agentic Design Patterns源代码链接link。1 为什么需要「路由」在只用顺序链Prompt Chaining时流程是固定的步骤 A → B → C每次请求都走同一条路。但现实里的智能体往往要根据当前情况做选择用户是在问订单、问产品还是在找技术支持不同意图应当走不同的处理流程而不是「一刀切」地走同一条流水线。路由Routing就是在智能体里引入条件分支先对输入或当前状态做判断再决定下一步走哪条路——是调用某个工具、交给某个子智能体还是进入某段子流程。这样系统就从「单一路径执行」变成「按条件选路径」更贴近真实业务需求。理解要点路由 先判断再选路没有路由时只有一条固定路径有了路由才能「因请求而异」地走不同分支。2 路由在解决什么问题想象一个客服场景用户问「订单到哪了」→ 应走订单查询流程查库、返回物流。用户问「这款有没有货」→ 应走商品信息流程查库存/目录。用户说「设备坏了要保修」→ 应走技术支持/升级流程工单或转人工。若没有路由要么所有话都进同一套逻辑难以兼顾三种需求要么写死大量 if-else难维护、难扩展。路由模式把「该走哪条路」抽象成一个独立决策步骤后续每条路可以各自实现、替换或扩展。实际例子就像公司前台先听你说「找谁/办什么事」再把你指到对应部门销售、技术、人事而不是所有人都进同一间办公室。3 路由的几种实现方式路由的「判断」可以用不同方式来做常见有四类方式简要说明特点基于 LLM 的路由用大模型分析输入输出一个类别或标签如booker/info/unclear再根据该标签选分支灵活、能理解语义适合意图复杂、表述多样的场景基于规则的路由用关键词、正则或结构化规则if-else / switch决定分支实现简单、可解释、确定性高但对表述变化敏感基于嵌入的路由把输入和若干「路由选项描述」都变成向量用相似度选最接近的那条路语义相似即可命中适合选项固定、表述多样的场景基于 ML 模型的路由用单独训练的分类器小模型做路由决策可离线优化、推理快需要标注数据与训练流程本系列以LangGraph为核心示例代码采用LLM 路由由 LLM 对用户请求做意图分类再用分类结果驱动图上的条件边把请求送到对应处理节点。其他方式只需把「路由节点」的实现换成规则/嵌入/小模型即可图结构不变。理解要点路由的「决策逻辑」可以换LLM / 规则 / 嵌入 / 小模型但「先决策、再按结果选分支」这一模式是统一的LangGraph 用条件边conditional edges把这一模式表达得很直观。4 LangGraph 中的路由状态图 条件边LangGraph 用状态图State Graph描述流程节点是处理步骤边是步骤之间的转移。条件边表示从某节点出来后根据当前状态决定下一个进入的节点而不是固定到唯一后继。状态State在图中流转的共享数据结构例如request、decision、response。路由节点负责根据输入或状态做分类并把结果写回状态如decision。路由函数供条件边使用读状态并返回「下一节点名」。条件边从路由节点出发根据路由函数的返回值映射到不同的下游节点。这样就把「分析请求 → 选路 → 进入对应处理器」清晰拆成一个路由节点 一条条件边 多个处理节点。实际例子本示例中的三类意图——预订booker、一般信息info、未识别unclear——分别映射到三个节点booking、info、unclear每个节点只做一件事结构一目了然。5 配套代码结构概览代码建议与源代码的 README 对照阅读。5.1 状态定义图的状态用TypedDict定义贯穿路由与各处理节点# 摘自 demo_codes/routing_graph.pyclassRoutingState(TypedDict):request:str# 用户原始请求decision:str# 路由节点输出的分类booker | info | unclearresponse:str# 当前分支处理节点的输出5.2 路由节点LLM 分类路由节点只做一件事用 LLM 对state[request]做意图分类并把结果写入state[decision]这里规范为booker/info/unclear# 摘自 demo_codes/routing_graph.py片段ROUTER_PROMPT分析用户请求判断应由哪个专门处理器处理。 - 若与预订机票/酒店相关只输出booker - 若为一般信息类问题只输出info - 若无法归类或不清楚只输出unclear 只输出一个词booker、info 或 unclear。 用户请求{request}defnode_router(state:RoutingState)-dict:chain_router_chain()# prompt | llm | StrOutputParserrawchain.invoke({request:state[request]})decision(rawor).strip().lower()ifdecisionnotin(booker,info,unclear):decisionunclearreturn{decision:decision}理解要点路由节点不直接「调下一个节点」只负责更新状态里的决策字段真正选路由后面的条件边 路由函数完成。5.3 路由函数与条件边路由函数读当前状态返回下一节点名条件边根据该返回值做映射# 摘自 demo_codes/routing_graph.pydefroute_by_decision(state:RoutingState)-Literal[booking,info,unclear]:d(state.get(decision)orunclear).strip().lower()ifdbooker:returnbookingifdinfo:returninforeturnunclear# 图中从 router 出发的条件边workflow.add_conditional_edges(router,route_by_decision,{booking:booking,info:info,unclear:unclear},)这样LLM 输出booker会进入booking节点info进入info节点其余含解析异常进入unclear节点。5.4 处理节点与图的入口/出口三个处理节点仅根据当前状态生成回复并写回response然后从各自节点连到END# 摘自 demo_codes/routing_graph.py片段defnode_booking(state:RoutingState)-dict:return{response:fBooking Handler 已处理请求{state[request]}。结果模拟完成预订操作。}# node_info、node_unclear 同理workflow.add_edge(START,router)workflow.add_edge(booking,END)workflow.add_edge(info,END)workflow.add_edge(unclear,END)整体流程START → router →条件边→ booking / info / unclear → END。5.5 完整代码如下也可以下载完整代码然后直接运行 LangGraph Routing 示例基于意图的路由图。 流程用户请求 → [路由节点LLM 分类] → 条件边 → [预订/信息/未识别] 处理节点 → 响应 运行前请配置环境变量 OPENAI_API_KEY 或 DASHSCOPE_API_KEY或在项目根目录放置 .env 文件。 importosfromtypingimportLiteral,TypedDictfromlangchain_openaiimportChatOpenAIfromlangchain_core.promptsimportChatPromptTemplatefromlangchain_core.output_parsersimportStrOutputParserfromlanggraph.graphimportStateGraph,START,ENDfromconfig_parserimportrouting_config# ---------------------------------------------------------------------------# 状态定义图的状态贯穿路由与各处理节点# ---------------------------------------------------------------------------classRoutingState(TypedDict):图的状态请求、路由决策与最终响应。request:str# 用户原始请求decision:str# 路由节点输出的分类booker | info | unclearresponse:str# 当前分支处理节点的输出# ---------------------------------------------------------------------------# LLM 与路由链# ---------------------------------------------------------------------------def_build_llm():构建 LLM从 config 读取 API Key 与 base_url。cfgrouting_config llmChatOpenAI(modelcfg.model,api_keycfg.api_key,base_urlcfg.base_urlifcfg.base_urlelseNone,temperature0,)returnllm# 路由提示让 LLM 仅输出一个分类词ROUTER_PROMPT分析用户请求判断应由哪个专门处理器处理。 - 若与预订机票/酒店相关只输出booker - 若为一般信息类问题只输出info - 若无法归类或不清楚只输出unclear 只输出一个词booker、info 或 unclear。 用户请求 {request}def_router_chain():路由链request - LLM - decision 字符串。promptChatPromptTemplate.from_messages([(user,ROUTER_PROMPT),])llm_build_llm()returnprompt|llm|StrOutputParser()# ---------------------------------------------------------------------------# 图节点# ---------------------------------------------------------------------------defnode_router(state:RoutingState)-dict: 路由节点调用 LLM 对用户请求做意图分类写入 state[decision]。 chain_router_chain()rawchain.invoke({request:state[request]})decision(rawor).strip().lower()ifdecisionnotin(booker,info,unclear):decisionunclearreturn{decision:decision}defnode_booking(state:RoutingState)-dict:预订处理器模拟处理机票/酒店预订类请求。msg(fBooking Handler 已处理请求{state[request]}。结果模拟完成预订操作。)return{response:msg}defnode_info(state:RoutingState)-dict:信息处理器模拟处理一般信息类请求。msg(fInfo Handler 已处理请求{state[request]}。结果模拟信息检索与回答。)return{response:msg}defnode_unclear(state:RoutingState)-dict:未识别处理器无法归类时的兜底回复。msg(f无法将请求归类{state[request]}。请补充说明是「预订」还是「一般信息」类问题。)return{response:msg}# ---------------------------------------------------------------------------# 路由函数供条件边使用根据 state[decision] 返回下一节点名# ---------------------------------------------------------------------------defroute_by_decision(state:RoutingState)-Literal[booking,info,unclear]:根据路由节点的分类结果返回下一节点名称。d(state.get(decision)orunclear).strip().lower()ifdbooker:returnbookingifdinfo:returninforeturnunclear# ---------------------------------------------------------------------------# 构建图START - router - [booking | info | unclear] - END# ---------------------------------------------------------------------------defbuild_routing_graph():构建并编译 Routing 图。workflowStateGraph(RoutingState)workflow.add_node(router,node_router)workflow.add_node(booking,node_booking)workflow.add_node(info,node_info)workflow.add_node(unclear,node_unclear)workflow.add_edge(START,router)workflow.add_conditional_edges(router,route_by_decision,{booking:booking,info:info,unclear:unclear,},)workflow.add_edge(booking,END)workflow.add_edge(info,END)workflow.add_edge(unclear,END)returnworkflow.compile()6 如何运行示例进入demo_codes目录创建并激活虚拟环境安装依赖pipinstall-rrequirements.txt在目录下配置.env如OPENAI_API_KEY或DASHSCOPE_API_KEY可选BASE_URL、MODEL。运行python main.py终端会依次对三条示例请求预订、信息、未识别做路由并打印决策与响应。7 小结与延伸路由让智能体根据输入或状态选择不同执行路径是超越「单一路径链式调用」的关键一步。实现方式可以是LLM、规则、嵌入或小模型示例采用LLM 条件边便于理解与修改。在LangGraph中状态承载数据路由节点写决策条件边 路由函数根据状态选下一节点结构清晰、易扩展。若要把路由节点改成规则如关键词匹配或嵌入相似度只需替换node_router的实现与decision的取值约定图的其余部分条件边、处理节点可保持不变。