LangGraph 中断相关方法详解

LangGraph 中断相关方法详解1. 什么是中断中断允许图执行在特定节点暂停等待外部输入如人类反馈后再继续。这实现了“人机协同”模式图会自动保存状态直到被恢复。2. 中断类型2.1 静态中断在编译时配置对特定节点生效graph.compile(interrupt_before[node_a],# 在node_a执行前中断interrupt_after[node_b]# 在node_b执行后中断)2.2 动态中断在节点内部通过interrupt()函数动态触发更灵活。3. 核心方法与类3.1interrupt()函数作用暂停当前节点执行返回resume值当图恢复时。参数接受任何可JSON序列化的值作为中断信息。行为如果存在resume值来自Command直接返回该值。否则抛出GraphInterrupt异常图暂停。示例defmy_node(state):# 执行一些操作...valueinterrupt(请提供反馈)# 中断并等待输入# 使用value继续处理...return{result:value}3.2Interrupt类表示中断的元数据包含value中断时传递的值。id中断的唯一标识符默认自动生成。3.3Command类用于恢复图执行fromlanggraph.typesimportCommand# 恢复图传递resume值commandCommand(resume用户反馈内容)graph.invoke(command,config)4. 工作流程示例fromlanggraph.graphimportStateGraphfromlanggraph.typesimportCommand,interrupt# 定义图builderStateGraph(dict)builder.add_node(human_review,lambdas:{output:interrupt(等待审核)})builder.add_node(process,lambdas:{final:s[output] 处理完成})builder.add_edge(__start__,human_review)builder.add_edge(human_review,process)graphbuilder.compile()# 第一次调用在human_review节点中断resultgraph.invoke({})# result包含interrupt信息# 恢复图final_resultgraph.invoke(Command(resume审核通过))5. 注意事项节点重启恢复后节点会从头开始执行interrupt()之前的代码会重新运行。避免副作用在interrupt()前不要执行不可逆操作。路由一致性优先使用Command(goto...)进行路由避免静态边与动态路由冲突。嵌套中断子图的中断会自动使用父图的检查点器。6. 常见问题Q: 如何获取中断信息A: 图返回的状态中包含__interrupt__键存储中断值。Q: 可以多次中断吗A: 是的每个interrupt()调用都会按顺序处理。Q: 如何验证中断恢复A: 使用持久化存储如SQLite保存图状态确保恢复后状态正确。7. 最佳实践将中断逻辑放在单独节点中保持职责清晰。设计幂等节点确保重新执行不会产生副作用。使用Command统一管理路由和恢复。结合持久化层实现长期中断与恢复。