尧图网站建设 
LangGraph 控制流详解一、控制流概述控制流Control Flow是 LangGraph 图结构的核心它决定了节点之间的执行顺序和跳转逻辑。LangGraph 支持多种控制流结构从简单的顺序执行到复杂的条件分支、并行处理和循环迭代为构建灵活的 AI 工作流提供了坚实基础。二、顺序结构顺序结构是最基础的控制流形式节点按照定义的顺序依次执行。2.1 add_edge基础边连接add_edge用于在两个节点之间添加一条有向边是最常用的控制流定义方式。fromlanggraph.graphimportStateGraph,START,END graphStateGraph(State)graph.add_node(step_a,step_a_func)graph.add_node(step_b,step_b_func)graph.add_node(step_c,step_c_func)# 定义顺序执行流graph.add_edge(START,step_a)graph.add_edge(step_a,step_b)graph.add_edge(step_b,step_c)graph.add_edge(step_c,END)执行流程START → step_a → step_b → step_c → END2.2 add_sequence批量添加顺序节点如果需要构建一组按顺序执行的节点可以使用add_sequence。它会按照列表顺序依次添加节点并在相邻节点之间自动添加边。graph.add_sequence([normalize_input,validate_input,process_data,save_result])等价于graph.add_edge(START,normalize_input)graph.add_edge(normalize_input,validate_input)graph.add_edge(validate_input,process_data)graph.add_edge(process_data,save_result)graph.add_edge(save_result,END)2.3 省略指向 END 的边在 LangGraph 中END并不是运行阶段真正执行的节点它更多用于表达图结构中的终止语义。# 在简单线性流程中可以省略指向 END 的边graph.add_edge(step_a,step_b)graph.add_edge(step_b,step_c)# step_c 执行完成后如果没有后续节点被触发图可以正常结束注意在分支、条件跳转、循环退出等场景中显式指向END通常更利于阅读和维护。而START通常不能省略因为它告诉 LangGraph 图运行时应当从哪些节点开始执行。三、分支结构分支结构允许图根据运行时状态选择不同的执行路径。3.1 静态分支Static Branch定义节点的下游候选节点在图编译阶段就完全确定只是运行时根据条件选择哪条边执行。核心判断编译期下游集合 → 静态分支3.1.1 并行节点并行节点是最简单的静态分支形式。当多个节点都从同一个上游节点触发时它们会在同一个超步SuperStep中被激活。# 从 prepare 节点同时触发三个并行任务graph.add_edge(prepare,task_a)graph.add_edge(prepare,task_b)graph.add_edge(prepare,task_c)# 三个任务完成后汇聚到 summary 节点graph.add_edge(task_a,summary)graph.add_edge(task_b,summary)graph.add_edge(task_c,summary)执行流程图┌──→ task_a ──┐ prepare ├──→ task_b ──┼──→ summary └──→ task_c ──┘3.1.2 条件分支StateGraph提供了add_conditional_edges方法用于从某个上游节点出发根据运行时状态选择下游节点。核心参数source条件分支的起始节点path路由规则是一个可执行对象通常是函数path_map路由规则的返回值到真实节点名之间的映射关系不使用 path_map路由函数的返回值应当直接是图中的节点名称defroute_by_status(state):ifstate[status]success:returnsuccess_handlerelse:returnerror_handlergraph.add_conditional_edges(check_status,route_by_status)使用 path_map如果不希望路由函数直接返回节点名而是返回业务语义更强的标识可以使用path_map进行映射defroute_by_symbol(state):ifstate[needs_processing]:returnrunelse:returnstopgraph.add_conditional_edges(START,route_by_symbol,{run:process_task,stop:END})同时路由至多个节点add_conditional_edges也支持一次路由到多个下游节点defroute_to_multiple(state):targets[]ifstate[condition_a]:targets.append(handler_a)ifstate[condition_b]:targets.append(handler_b)returntargets graph.add_conditional_edges(router,route_to_multiple)3.2 动态分支Dynamic Branch动态分支是指下游节点集合在运行时才能确定的分支。LangGraph 提供了SendAPI 来实现这种模式。fromlanggraph.typesimportSenddefdynamic_dispatch(state):# 根据状态动态创建并行任务tasks[]foriteminstate[items]:tasks.append(Send(process_item,{item:item}))returntasks graph.add_conditional_edges(dispatcher,dynamic_dispatch)四、循环结构循环结构允许图在满足特定条件时重复执行某些节点。4.1 基于条件边的循环通过让条件边返回上游节点可以实现循环defcheck_iteration(state):ifstate[iteration_count]3:returnprocess_node# 继续循环else:returnEND# 退出循环graph.add_conditional_edges(check,check_iteration)graph.add_edge(process_node,check)# 形成循环执行流程START → process_node → check ─┬→ process_node (iteration_count 3) └→ END (iteration_count 3)4.2 使用 Command API 实现循环CommandAPI 允许在节点内部同时更新状态和决定下一个节点可以更灵活地控制循环fromlanggraph.typesimportCommanddefiterative_processor(state):new_countstate[iteration_count]1ifnew_count3:returnCommand(update{iteration_count:new_count},gotoprocess_node)else:returnCommand(update{iteration_count:new_count,status:completed},gotoEND)五、高级控制流5.1 Send APIMap-Reduce 并行SendAPI 是实现动态并行任务的核心工具特别适合处理列表数据的 map-reduce 场景。fromlanggraph.typesimportSenddefmap_items(state):将列表中的每个元素分发到并行节点处理return[Send(process_item,{item:item,index:i})fori,iteminenumerate(state[items])]defreduce_results(state):汇总所有并行处理结果return{final_results:state[partial_results]}graph.add_node(process_item,process_single_item)graph.add_conditional_edges(start,map_items)graph.add_edge(process_item,aggregate)graph.add_node(aggregate,reduce_results)关键点Send允许动态创建多个并行任务每个任务可以有不同的输入参数所有任务完成后才能进入下一个节点5.2 Command API控制流与状态更新结合CommandAPI 将控制流边和状态更新节点结合起来提供了更灵活的控制方式。fromlanggraph.typesimportCommanddefsmart_router(state):根据条件同时更新状态和决定跳转ifstate[priority]high:returnCommand(update{status:escalated,handler:senior},gotosenior_handler)else:returnCommand(update{status:normal,handler:junior},gotojunior_handler)特点无需定义条件边控制流在节点内部定义可以同时更新状态和决定跳转目标适合复杂决策逻辑5.3 Command.PARENT跨图导航当使用子图时可以通过Command.PARENT导航到父图中的节点defsubgraph_node(state):ifstate[should_exit]:returnCommand(update{result:completed},gotoparent_node,graphCommand.PARENT)else:return{status:continuing}5.4 Deferred 节点执行使用deferTrue参数可以让节点在所有待执行任务完成后才执行# 延迟汇总节点等待所有并行分支完成graph.add_node(assemble_plan,assemble_plan,deferTrue)# 三个并行分支都会触发汇总节点graph.add_edge(build_a,assemble_plan)graph.add_edge(build_b,assemble_plan)graph.add_edge(build_c,assemble_plan)六、控制流对比总结控制流类型方法说明适用场景顺序执行add_edge(A, B)A 执行完后执行 B线性处理流程批量顺序add_sequence([A, B, C])自动添加节点和边简单流水线并行执行add_edge(X, A)add_edge(X, B)同一超步中同时执行独立任务并行处理条件分支add_conditional_edges(X, router)根据运行时状态选择下游if-else 逻辑动态并行SendAPI动态创建多个并行任务map-reduce 场景控制流结合CommandAPI同时更新状态和跳转复杂决策逻辑跨图导航Command.PARENT从子图跳转到父图子图复用七、最佳实践7.1 选择合适的控制流简单线性流程→ 使用add_edge或add_sequence条件分支→ 使用add_conditional_edges动态并行任务→ 使用SendAPI复杂决策逻辑→ 使用CommandAPI需要跨图跳转→ 使用Command.PARENT7.2 避免混合固定和条件后继不要从同一个来源节点同时添加普通边和条件边# 错误示范graph.add_edge(check,next_node)graph.add_conditional_edges(check,router_func)# 冲突# 正确做法只使用条件边graph.add_conditional_edges(check,router_func)7.3 使用 path_map 解耦路由逻辑# 推荐使用 path_map 分离业务逻辑和节点名defbusiness_router(state):ifstate[condition]:returnsuccessreturnfailuregraph.add_conditional_edges(decision_point,business_router,{success:handle_success,failure:handle_failure})7.4 处理并行任务的状态更新当多个并行节点更新同一状态字段时需要配置合适的 ReducerfromtypingimportAnnotatedfromoperatorimportaddclassState(TypedDict):results:Annotated[list,add]# 使用 add reducer 合并列表status:str# 默认覆盖八、实战示例智能审批流程fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportCommandfromtypingimportTypedDict,LiteralclassApprovalState(TypedDict):amount:floatstatus:strapprover:strdefvalidate_request(state):ifstate[amount]0:return{status:rejected,approver:system}return{status:validated}defroute_by_amount(state)-Literal[auto_approve,manager_approve,director_approve]:ifstate[amount]1000:returnauto_approveelifstate[amount]10000:returnmanager_approveelse:returndirector_approvedefauto_approve(state):return{status:approved,approver:auto}defmanager_approve(state):return{status:approved,approver:manager}defdirector_approve(state):return{status:approved,approver:director}# 构建图builderStateGraph(ApprovalState)# 添加节点builder.add_node(validate,validate_request)builder.add_node(auto_approve,auto_approve)builder.add_node(manager_approve,manager_approve)builder.add_node(director_approve,director_approve)# 定义控制流builder.add_edge(START,validate)builder.add_conditional_edges(validate,route_by_amount)builder.add_edge(auto_approve,END)builder.add_edge(manager_approve,END)builder.add_edge(director_approve,END)# 编译graphbuilder.compile()# 测试resultgraph.invoke({amount:5000})print(result)# {amount: 5000, status: approved, approver: manager}九、总结LangGraph 的控制流机制为构建复杂 AI 工作流提供了强大支持顺序结构通过add_edge和add_sequence实现线性执行分支结构通过add_conditional_edges实现条件跳转并行结构通过静态扇出和SendAPI 实现并行处理循环结构通过条件边返回上游节点实现迭代高级控制CommandAPI 结合状态更新与控制流跨图导航Command.PARENT实现子图到父图的跳转 LangGraph 控制流详解
LangGraph 控制流详解一、控制流概述控制流Control Flow是 LangGraph 图结构的核心它决定了节点之间的执行顺序和跳转逻辑。LangGraph 支持多种控制流结构从简单的顺序执行到复杂的条件分支、并行处理和循环迭代为构建灵活的 AI 工作流提供了坚实基础。二、顺序结构顺序结构是最基础的控制流形式节点按照定义的顺序依次执行。2.1 add_edge基础边连接add_edge用于在两个节点之间添加一条有向边是最常用的控制流定义方式。fromlanggraph.graphimportStateGraph,START,END graphStateGraph(State)graph.add_node(step_a,step_a_func)graph.add_node(step_b,step_b_func)graph.add_node(step_c,step_c_func)# 定义顺序执行流graph.add_edge(START,step_a)graph.add_edge(step_a,step_b)graph.add_edge(step_b,step_c)graph.add_edge(step_c,END)执行流程START → step_a → step_b → step_c → END2.2 add_sequence批量添加顺序节点如果需要构建一组按顺序执行的节点可以使用add_sequence。它会按照列表顺序依次添加节点并在相邻节点之间自动添加边。graph.add_sequence([normalize_input,validate_input,process_data,save_result])等价于graph.add_edge(START,normalize_input)graph.add_edge(normalize_input,validate_input)graph.add_edge(validate_input,process_data)graph.add_edge(process_data,save_result)graph.add_edge(save_result,END)2.3 省略指向 END 的边在 LangGraph 中END并不是运行阶段真正执行的节点它更多用于表达图结构中的终止语义。# 在简单线性流程中可以省略指向 END 的边graph.add_edge(step_a,step_b)graph.add_edge(step_b,step_c)# step_c 执行完成后如果没有后续节点被触发图可以正常结束注意在分支、条件跳转、循环退出等场景中显式指向END通常更利于阅读和维护。而START通常不能省略因为它告诉 LangGraph 图运行时应当从哪些节点开始执行。三、分支结构分支结构允许图根据运行时状态选择不同的执行路径。3.1 静态分支Static Branch定义节点的下游候选节点在图编译阶段就完全确定只是运行时根据条件选择哪条边执行。核心判断编译期下游集合 → 静态分支3.1.1 并行节点并行节点是最简单的静态分支形式。当多个节点都从同一个上游节点触发时它们会在同一个超步SuperStep中被激活。# 从 prepare 节点同时触发三个并行任务graph.add_edge(prepare,task_a)graph.add_edge(prepare,task_b)graph.add_edge(prepare,task_c)# 三个任务完成后汇聚到 summary 节点graph.add_edge(task_a,summary)graph.add_edge(task_b,summary)graph.add_edge(task_c,summary)执行流程图┌──→ task_a ──┐ prepare ├──→ task_b ──┼──→ summary └──→ task_c ──┘3.1.2 条件分支StateGraph提供了add_conditional_edges方法用于从某个上游节点出发根据运行时状态选择下游节点。核心参数source条件分支的起始节点path路由规则是一个可执行对象通常是函数path_map路由规则的返回值到真实节点名之间的映射关系不使用 path_map路由函数的返回值应当直接是图中的节点名称defroute_by_status(state):ifstate[status]success:returnsuccess_handlerelse:returnerror_handlergraph.add_conditional_edges(check_status,route_by_status)使用 path_map如果不希望路由函数直接返回节点名而是返回业务语义更强的标识可以使用path_map进行映射defroute_by_symbol(state):ifstate[needs_processing]:returnrunelse:returnstopgraph.add_conditional_edges(START,route_by_symbol,{run:process_task,stop:END})同时路由至多个节点add_conditional_edges也支持一次路由到多个下游节点defroute_to_multiple(state):targets[]ifstate[condition_a]:targets.append(handler_a)ifstate[condition_b]:targets.append(handler_b)returntargets graph.add_conditional_edges(router,route_to_multiple)3.2 动态分支Dynamic Branch动态分支是指下游节点集合在运行时才能确定的分支。LangGraph 提供了SendAPI 来实现这种模式。fromlanggraph.typesimportSenddefdynamic_dispatch(state):# 根据状态动态创建并行任务tasks[]foriteminstate[items]:tasks.append(Send(process_item,{item:item}))returntasks graph.add_conditional_edges(dispatcher,dynamic_dispatch)四、循环结构循环结构允许图在满足特定条件时重复执行某些节点。4.1 基于条件边的循环通过让条件边返回上游节点可以实现循环defcheck_iteration(state):ifstate[iteration_count]3:returnprocess_node# 继续循环else:returnEND# 退出循环graph.add_conditional_edges(check,check_iteration)graph.add_edge(process_node,check)# 形成循环执行流程START → process_node → check ─┬→ process_node (iteration_count 3) └→ END (iteration_count 3)4.2 使用 Command API 实现循环CommandAPI 允许在节点内部同时更新状态和决定下一个节点可以更灵活地控制循环fromlanggraph.typesimportCommanddefiterative_processor(state):new_countstate[iteration_count]1ifnew_count3:returnCommand(update{iteration_count:new_count},gotoprocess_node)else:returnCommand(update{iteration_count:new_count,status:completed},gotoEND)五、高级控制流5.1 Send APIMap-Reduce 并行SendAPI 是实现动态并行任务的核心工具特别适合处理列表数据的 map-reduce 场景。fromlanggraph.typesimportSenddefmap_items(state):将列表中的每个元素分发到并行节点处理return[Send(process_item,{item:item,index:i})fori,iteminenumerate(state[items])]defreduce_results(state):汇总所有并行处理结果return{final_results:state[partial_results]}graph.add_node(process_item,process_single_item)graph.add_conditional_edges(start,map_items)graph.add_edge(process_item,aggregate)graph.add_node(aggregate,reduce_results)关键点Send允许动态创建多个并行任务每个任务可以有不同的输入参数所有任务完成后才能进入下一个节点5.2 Command API控制流与状态更新结合CommandAPI 将控制流边和状态更新节点结合起来提供了更灵活的控制方式。fromlanggraph.typesimportCommanddefsmart_router(state):根据条件同时更新状态和决定跳转ifstate[priority]high:returnCommand(update{status:escalated,handler:senior},gotosenior_handler)else:returnCommand(update{status:normal,handler:junior},gotojunior_handler)特点无需定义条件边控制流在节点内部定义可以同时更新状态和决定跳转目标适合复杂决策逻辑5.3 Command.PARENT跨图导航当使用子图时可以通过Command.PARENT导航到父图中的节点defsubgraph_node(state):ifstate[should_exit]:returnCommand(update{result:completed},gotoparent_node,graphCommand.PARENT)else:return{status:continuing}5.4 Deferred 节点执行使用deferTrue参数可以让节点在所有待执行任务完成后才执行# 延迟汇总节点等待所有并行分支完成graph.add_node(assemble_plan,assemble_plan,deferTrue)# 三个并行分支都会触发汇总节点graph.add_edge(build_a,assemble_plan)graph.add_edge(build_b,assemble_plan)graph.add_edge(build_c,assemble_plan)六、控制流对比总结控制流类型方法说明适用场景顺序执行add_edge(A, B)A 执行完后执行 B线性处理流程批量顺序add_sequence([A, B, C])自动添加节点和边简单流水线并行执行add_edge(X, A)add_edge(X, B)同一超步中同时执行独立任务并行处理条件分支add_conditional_edges(X, router)根据运行时状态选择下游if-else 逻辑动态并行SendAPI动态创建多个并行任务map-reduce 场景控制流结合CommandAPI同时更新状态和跳转复杂决策逻辑跨图导航Command.PARENT从子图跳转到父图子图复用七、最佳实践7.1 选择合适的控制流简单线性流程→ 使用add_edge或add_sequence条件分支→ 使用add_conditional_edges动态并行任务→ 使用SendAPI复杂决策逻辑→ 使用CommandAPI需要跨图跳转→ 使用Command.PARENT7.2 避免混合固定和条件后继不要从同一个来源节点同时添加普通边和条件边# 错误示范graph.add_edge(check,next_node)graph.add_conditional_edges(check,router_func)# 冲突# 正确做法只使用条件边graph.add_conditional_edges(check,router_func)7.3 使用 path_map 解耦路由逻辑# 推荐使用 path_map 分离业务逻辑和节点名defbusiness_router(state):ifstate[condition]:returnsuccessreturnfailuregraph.add_conditional_edges(decision_point,business_router,{success:handle_success,failure:handle_failure})7.4 处理并行任务的状态更新当多个并行节点更新同一状态字段时需要配置合适的 ReducerfromtypingimportAnnotatedfromoperatorimportaddclassState(TypedDict):results:Annotated[list,add]# 使用 add reducer 合并列表status:str# 默认覆盖八、实战示例智能审批流程fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportCommandfromtypingimportTypedDict,LiteralclassApprovalState(TypedDict):amount:floatstatus:strapprover:strdefvalidate_request(state):ifstate[amount]0:return{status:rejected,approver:system}return{status:validated}defroute_by_amount(state)-Literal[auto_approve,manager_approve,director_approve]:ifstate[amount]1000:returnauto_approveelifstate[amount]10000:returnmanager_approveelse:returndirector_approvedefauto_approve(state):return{status:approved,approver:auto}defmanager_approve(state):return{status:approved,approver:manager}defdirector_approve(state):return{status:approved,approver:director}# 构建图builderStateGraph(ApprovalState)# 添加节点builder.add_node(validate,validate_request)builder.add_node(auto_approve,auto_approve)builder.add_node(manager_approve,manager_approve)builder.add_node(director_approve,director_approve)# 定义控制流builder.add_edge(START,validate)builder.add_conditional_edges(validate,route_by_amount)builder.add_edge(auto_approve,END)builder.add_edge(manager_approve,END)builder.add_edge(director_approve,END)# 编译graphbuilder.compile()# 测试resultgraph.invoke({amount:5000})print(result)# {amount: 5000, status: approved, approver: manager}九、总结LangGraph 的控制流机制为构建复杂 AI 工作流提供了强大支持顺序结构通过add_edge和add_sequence实现线性执行分支结构通过add_conditional_edges实现条件跳转并行结构通过静态扇出和SendAPI 实现并行处理循环结构通过条件边返回上游节点实现迭代高级控制CommandAPI 结合状态更新与控制流跨图导航Command.PARENT实现子图到父图的跳转