文章目录第 7 章Command 与动态流程控制7.1 本章目标7.2 核心概念Command 是什么Command vs 条件边Command 跳转示意图7.3 实战实战 1动态路由跳转实战 2Command(update) 同时更新状态和导航实战 3子图用 Command 控制父图7.4 API 速查7.5 错误与避坑指南坑 1混淆 Command 和普通 dict坑 2Command(goto) 目标节点不存在坑 3在不需要动态路由时滥用 Command坑 4子图 Command 忘记 graphCommand.PARENT7.6 最佳实践总结第 7 章Command 与动态流程控制7.1 本章目标学完本章你将能够理解 Command 在运行时动态控制流程的作用掌握Command(goto)跳转节点和Command(update)更新状态学会 Send API 实现 map-reduce 并行模式理解 Command 与条件边的区别和使用场景7.2 核心概念Command 是什么Command是一个特殊的数据类节点可以返回它来同时更新状态和导航流程。它与普通 dict 返回值的关键区别# 普通 dict 返回值只更新 State流程按照边Edge走defnormal_node(state:State)-dict:return{counter:state[counter]1}# 流程由 add_edge 决定# Command 返回值更新 State 动态决定流程defcommand_node(state:State)-Command[Literal[node_a,node_b]]:returnCommand(update{counter:state[counter]1},# 更新状态gotonode_a,# 动态决定下一步去哪个节点)比喻普通 dict 就像完成工作后走固定传送带Command 就像完成工作后自己选择坐哪条传送带。Command vs 条件边特性条件边add_conditional_edgesCommandgoto决策时机编译时定义路由规则运行时动态决策决策逻辑位置独立的路由函数节点内部适用场景固定的分支逻辑如意图分类动态跳转如审批后跳转类型安全通过 Literal 标注通过 Command 的泛型参数Command 跳转示意图普通 dict 返回普通 dict 返回Command(gotoC)Command(gotoB)Command(gotoEND)START节点 A节点 B固定流程节点 CEND7.3 实战实战 1动态路由跳转fromtypingimportTypedDict,Literalfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportCommandclassState(TypedDict):step:strdata:strdefentry_node(state:State)-Command[Literal[process_a,process_b,END]]: 入口节点根据当前状态动态决定路由。 注意 Command 的泛型参数Command[Literal[...]] 提供类型安全 确保 goto 的目标是已注册的节点。 ifstate[step]start:print( → 路由到 process_a)returnCommand(gotoprocess_a,update{data:从入口进入})elifstate[step]skip:print( → 直接结束)returnCommand(gotoEND)else:print( → 路由到 process_b)returnCommand(gotoprocess_b,update{data:从入口进入B})defprocess_a(state:State)-Command[Literal[END]]:print( [process_a] 处理中...)returnCommand(gotoEND,update{data:f{state[data]}→ 经过A处理})defprocess_b(state:State)-Command[Literal[END]]:print( [process_b] 处理中...)returnCommand(gotoEND,update{data:f{state[data]}→ 经过B处理})builderStateGraph(State)builder.add_node(entry,entry_node)builder.add_node(process_a,process_a)builder.add_node(process_b,process_b)# 入口节点使用 Command 控制路由不需要条件边builder.add_edge(START,entry)# 注意process_a 和 process_b 都通过 Command(gotoEND) 结束# 不需要显式添加 process_a → END 的边graphbuilder.compile()print( 测试 1: stepstart )resultgraph.invoke({step:start,data:})print(f结果:{result[data]})print(\n 测试 2: stepother )resultgraph.invoke({step:other,data:})print(f结果:{result[data]})print(\n 测试 3: stepskip )resultgraph.invoke({step:skip,data:})print(f结果:{result[data]})实战 2Command(update) 同时更新状态和导航fromtypingimportTypedDict,Annotated,Literalimportoperatorfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportCommandclassState(TypedDict):messages:Annotated[list[str],operator.add]definitial_processor(state:State)-Command[Literal[finalizer]]: 同时更新状态添加消息和导航到 finalizer 节点。 returnCommand(update{messages:[处理完成]},gotofinalizer,)deffinalizer(state:State)-Command[Literal[END]]:returnCommand(update{messages:[✅ 最终确认]},gotoEND,)builderStateGraph(State)builder.add_node(process,initial_processor)builder.add_node(finalizer,finalizer)builder.add_edge(START,process)graphbuilder.compile()resultgraph.invoke({messages:[]})print(f消息:{result[messages]})实战 3子图用 Command 控制父图fromtypingimportTypedDict,Literalfromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.typesimportCommand# # 子图# classSubState(TypedDict):value:intdefsub_processor(state:SubState)-Command: 子图节点返回 Command(graphCommand.PARENT)控制父图流程。 ifstate[value]10:# 值太大让父图走 reject 路径returnCommand(graphCommand.PARENT,gotoparent_reject,update{value:-1},)return{value:state[value]*2}sub_builderStateGraph(SubState)sub_builder.add_node(process,sub_processor)sub_builder.add_edge(START,process)sub_builder.add_edge(process,END)sub_graphsub_builder.compile()# # 父图# classParentState(TypedDict):value:intdefparent_accept(state:ParentState)-dict:return{value:state[value]*10}defparent_reject(state:ParentState)-dict:return{value:0}parent_builderStateGraph(ParentState)parent_builder.add_node(sub,sub_graph)# 子图作为节点parent_builder.add_node(parent_accept,parent_accept)parent_builder.add_node(parent_reject,parent_reject)parent_builder.add_edge(START,sub)parent_builder.add_edge(parent_accept,END)parent_builder.add_edge(parent_reject,END)parent_graphparent_builder.compile()print( 测试: value5正常 )resultparent_graph.invoke({value:5})print(f结果:{result[value]})# 5*210, 然后 accept: 10*10100print(\n 测试: value15触发子图拒绝 )resultparent_graph.invoke({value:15})print(f结果:{result[value]})# 子图返回 reject父图执行 reject: 07.4 API 速查API完整签名入参说明返回值说明Command(gotonode)Command(goto: str)goto: 目标节点名Command对象跳转到指定节点Command(updatedict)Command(update: dict)update: 状态更新字典Command对象更新状态后按边继续Command(goto, update)Command(goto: str, update: dict)gotoupdateCommand对象更新状态并跳转Command(resumevalue)Command(resume: Any)resume: 恢复值Command对象恢复中断Command(graphCommand.PARENT)Command(graph: str)graph: 目标图Command对象子图控制父图Command.PARENT常量无__parent__导航到父图7.5 错误与避坑指南坑 1混淆 Command 和普通 dict# ❌ 错误写法期望跳转但实际不会defbad_node(state:State)-dict:return{goto:other_node}# 这只是更新 State 中的 goto 字段# ✅ 正确写法defgood_node(state:State)-Command[Literal[other_node]]:returnCommand(gotoother_node)# 真的跳转到 other_node坑 2Command(goto) 目标节点不存在# ❌ 错误写法defbad_node(state:State)-Command[Literal[nonexistent]]:returnCommand(gotononexistent)# 目标节点未注册 → 运行时报错# ✅ 正确写法defgood_node(state:State)-Command[Literal[node_a,node_b]]:ifcondition:returnCommand(gotonode_a)# 确保 node_a 已注册returnCommand(gotonode_b)坑 3在不需要动态路由时滥用 Command# ❌ 不推荐简单条件路由用 Commanddefbad_node(state:State)-Command[Literal[a,b]]:ifstate[intent]a:returnCommand(gotoa)returnCommand(gotob)# ✅ 推荐固定路由用 add_conditional_edgesdefroute(state:State)-Literal[a,b]:returnstate[intent]builder.add_conditional_edges(classifier,route)坑 4子图 Command 忘记 graphCommand.PARENT# ❌ 错误写法子图跳转被当作子图内部跳转defsub_node(state:SubState)-Command:returnCommand(gotoparent_node)# 在子图中找不到 parent_node# ✅ 正确写法明确指定跳转到父图defsub_node(state:SubState)-Command:returnCommand(graphCommand.PARENT,gotoparent_node)7.6 最佳实践总结编译时确定的路由用条件边运行时决策用 Command(goto)各司其职Command 返回类型用Literal泛型标注获得类型安全和 IDE 提示并行处理用 Send Reducer 实现 map-reduce不要用 Command 做并行子图控制父图时用Command.PARENT明确导航目标Command 适合带数据跳转的场景如审批后带着审批结果跳转到对应处理节点