【LangGraph从小白到精通手把手实战教程】 007、Edge边与路由:条件路由、动态路由与固定流转
007、Edge边与路由条件路由、动态路由与固定流转从一次深夜调试说起上周三凌晨两点我在给一个智能客服系统做流程编排。需求很简单用户输入问题先判断是否属于业务咨询如果是就转业务模型否则走通用闲聊通道。我随手写了个条件判断跑起来却发现所有请求都卡在第一个节点——路由根本没按我预想的走。对着屏幕愣了三分钟才反应过来LangGraph 里的边Edge不是 if-else 那么简单它有自己的路由哲学。今天我们就来拆解 LangGraph 中边的三种核心玩法条件路由、动态路由和固定流转。这些概念听起来抽象但一旦理解透了你能把任何复杂流程编排得像流水线一样清晰。固定流转最朴素的连接固定流转是最基础的边类型相当于流程图里一根实线箭头。它不判断任何条件节点执行完就往下走。fromlanggraph.graphimportStateGraph,END# 构建一个简单三步流程builderStateGraph(dict)defstep_a(state):state[result]A完成returnstatedefstep_b(state):state[result] - B完成returnstatedefstep_c(state):state[result] - C完成returnstate# 添加节点builder.add_node(step_a,step_a)builder.add_node(step_b,step_b)builder.add_node(step_c,step_c)# 关键在这里用 add_edge 建立固定流转builder.add_edge(step_a,step_b)# A 完了一定走 Bbuilder.add_edge(step_b,step_c)# B 完了一定走 Cbuilder.add_edge(step_c,END)# C 完了就结束graphbuilder.compile()这种写法适合顺序明确的流水线比如“数据清洗→特征提取→模型预测”。但现实中的流程往往需要分支这时候就得请出条件路由。条件路由让流程学会“看情况”条件路由的核心是add_conditional_edges。它允许你根据当前状态决定下一步走哪个节点。fromlanggraph.graphimportStateGraph,ENDfromlanggraph.checkpointimportMemorySaver builderStateGraph(dict,checkpointerMemorySaver())defclassifier(state):判断用户意图querystate.get(query,)if价格inqueryor多少钱inquery:returnprice_queryelif售后inqueryor维修inquery:returnafter_saleselse:returngeneraldefprice_handler(state):state[response]正在查询价格...returnstatedefafter_sales_handler(state):state[response]转接售后专员...returnstatedefgeneral_handler(state):state[response]这是通用回答...returnstate builder.add_node(classifier,classifier)builder.add_node(price,price_handler)builder.add_node(after_sales,after_sales_handler)builder.add_node(general,general_handler)# 设置入口builder.set_entry_point(classifier)# 核心魔法在这里builder.add_conditional_edges(classifier,# 从哪个节点出发classifier,# 路由函数返回下一个节点的名称字符串{price_query:price,after_sales:after_sales,general:general})# 为业务节点添加固定出口builder.add_edge(price,END)builder.add_edge(after_sales,END)builder.add_edge(general,END)这里有个坑我踩过路由函数必须返回字典里存在的 key。如果你返回了unknown但没在映射表里定义LangGraph 会直接抛异常不会默默走默认路径。建议加个兜底逻辑像上面代码里的else: return general。动态路由运行时决定去向条件路由已经很强大了但有时候我们连下一步有多少种可能都不知道需要在运行时动态生成选项。这时候可以用动态路由——直接让路由函数返回下一个节点的名称不依赖预定义的映射表。defdynamic_router(state):根据查询复杂度决定路由querystate.get(query,)historystate.get(history,[])# 简单查询直接回答iflen(query)10:returnsimple_agent# 复杂查询需要拆解if并且inqueryor另外inquery:# 动态生成子任务节点sub_taskssplit_complex_query(query)fori,taskinenumerate(sub_tasks):node_namefsubtask_{i}ifnotbuilder.has_node(node_name):# 动态注册节点伪代码实际需调整builder.add_node(node_name,create_task_handler(task))returnparallel_dispatcher# 走并行分发器# 需要历史上下文的iflen(history)3:returncontext_aware_agent# 默认路径returndefault_agent# 动态路由的写法更简洁builder.add_conditional_edges(dynamic_router,dynamic_router# 函数直接返回节点名不用映射表)动态路由适合任务拆解、自适应流程这类场景。但要注意返回的节点必须已经注册到图中否则会报NodeNotFound。我一般会在初始化阶段预注册所有可能的节点或者像上面伪代码那样动态添加实际项目需要更精细的节点管理。混合使用一个真实案例去年我做了一个合同审核系统用到了三种边的混合编排# 1. 固定流转必须走的步骤builder.add_edge(contract_loader,format_checker)# 2. 条件路由根据合同类型分流builder.add_conditional_edges(format_checker,contract_type_detector,{NDA:nda_analyzer,SOW:sow_analyzer,MOU:mou_analyzer})# 3. 动态路由条款风险评级后动态跳转builder.add_conditional_edges(risk_evaluator,dynamic_risk_router# 可能返回 low_risk_approve, medium_risk_review, high_risk_legal)# 所有分支最终汇合到归档节点fornodein[nda_analyzer,sow_analyzer,mou_analyzer,low_risk_approve,medium_risk_review,high_risk_legal]:builder.add_edge(node,archiver)这种混合模式让流程既有主干道又有灵活分支最后还能收敛。关键技巧是在分支的尽头用固定边汇合到公共节点避免流程散开收不回来。调试心得可视化是王道LangGraph 自带graph.get_graph().draw_mermaid()一定要把图画出来看看边连对了没。我那个凌晨两点的问题画出来一眼就看到路由函数返回的值和映射表对不上。状态是唯一真相路由函数只能基于 state 做判断。确保你需要的信息比如state[intent]已经在前面节点写入了。经常有人路由时取不到值是因为忘了上游节点要return state。END 是个特殊节点想清楚流程在哪里结束。每个分支最后要么指向 END要么指向另一个汇聚节点。 dangling 的边会让流程卡住。测试时覆盖所有分支条件路由一定要测边界情况。特别是那些“理论上不会出现”的分支现实跑一个月总会遇到几次。个人经验建议LangGraph 的边设计其实反映了编排思维的本质不是写线性代码而是定义状态空间中的转移规则。刚开始我总想着“下一步该执行什么”后来才转变思维为“在当前状态下允许去哪些地方”。对于新项目我建议先全用固定流转把主干跑通再加条件路由处理主要分支最后在特别复杂的地方考虑动态路由。别一上来就搞动态——过度灵活的流程调试起来简直是噩梦。记住好的编排不是能处理所有情况而是让正常流程清晰异常流程有迹可循。那些“万一”的情况不妨用一个兜底节点统一处理比写一堆边缘逻辑更稳健。路由逻辑复杂到一定程度后考虑拆成多个子图。子图可以嵌套每个子图内部可以有自己的路由策略。这就像代码里的函数拆分保持每个单元的可理解性。最后说句实在的深夜调试路由问题时先别怀疑 LangGraph 有 bug大概率是你某个路由函数返回了None或者状态字段名拼错了。喝口水画个图从状态流转的角度再看一遍问题往往就浮现了。下一篇我们聊《008、状态持久化与检查点如何让流程随时暂停和恢复》。当你需要处理长对话、多轮审批这类中断再续的场景时检查点机制能救你的命。