企业官网建设流程全解析

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序列化的值，作为中断信息。

• 行为：

  1. 如果存在resume值（来自Command），直接返回该值。
  2. 否则抛出GraphInterrupt异常，图暂停。

• 示例：

  defmy_node(state):# 执行一些操作...value=interrupt("请提供反馈")# 中断并等待输入# 使用value继续处理...return{"result":value}

3.2Interrupt类

表示中断的元数据，包含：

• value：中断时传递的值。
• id：中断的唯一标识符（默认自动生成）。

3.3Command类

用于恢复图执行：

fromlanggraph.typesimportCommand# 恢复图，传递resume值command=Command(resume="用户反馈内容")graph.invoke(command,config)

4. 工作流程示例

fromlanggraph.graphimportStateGraphfromlanggraph.typesimportCommand,interrupt# 定义图builder=StateGraph(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")graph=builder.compile()# 第一次调用：在human_review节点中断result=graph.invoke({})# result包含interrupt信息# 恢复图final_result=graph.invoke(Command(resume="审核通过"))

5. 注意事项

1. 节点重启：恢复后，节点会从头开始执行（interrupt()之前的代码会重新运行）。
2. 避免副作用：在interrupt()前不要执行不可逆操作。
3. 路由一致性：优先使用Command(goto=...)进行路由，避免静态边与动态路由冲突。
4. 嵌套中断：子图的中断会自动使用父图的检查点器。

6. 常见问题

• Q: 如何获取中断信息？
  A: 图返回的状态中包含__interrupt__键，存储中断值。
• Q: 可以多次中断吗？
  A: 是的，每个interrupt()调用都会按顺序处理。
• Q: 如何验证中断恢复？
  A: 使用持久化存储（如SQLite）保存图状态，确保恢复后状态正确。

7. 最佳实践

• 将中断逻辑放在单独节点中，保持职责清晰。
• 设计幂等节点，确保重新执行不会产生副作用。
• 使用Command统一管理路由和恢复。
• 结合持久化层实现长期中断与恢复。