断点¶
断点可在特定点暂停图执行,并允许逐步执行。断点由 LangGraph 的持久化层提供支持,该层会在每个图步骤后保存状态。断点也可用于启用人工介入工作流,不过我们建议为此目的使用interrupt
函数。
要求¶
要使用断点,你需要:
- 指定一个检查点保存器 以在每一步之后保存图状态。
- 设置断点 以指定执行应在何处暂停。
- 使用 线程 ID 运行图 以在断点处暂停执行。
- 使用
invoke
/ainvoke
/stream
/astream
恢复执行(请参阅Command
原语)。
设置断点¶
有两个地方可以设置断点:
1. 在节点执行**之前**或**之后**,通过在**编译时**或**运行时**设置断点。我们称这些为静态断点。
2. 在节点**内部**使用 NodeInterrupt
异常。
静态断点¶
静态断点在节点执行**之前**或**之后**触发。你可以通过在**“编译”时**或**运行时**指定 interrupt_before
和 interrupt_after
来设置静态断点。
graph = graph_builder.compile(
interrupt_before=["node_a"],
interrupt_after=["node_b", "node_c"],
checkpointer=..., # 指定一个检查点器
)
thread_config = {
"configurable": {
"thread_id": "some_thread"
}
}
# 运行图直到断点
graph.invoke(inputs, config=thread_config)
# 可选:根据用户输入更新图状态
graph.update_state(update, config=thread_config)
# 恢复图的运行
graph.invoke(None, config=thread_config)
graph.invoke(
inputs,
config={"configurable": {"thread_id": "some_thread"}},
interrupt_before=["node_a"],
interrupt_after=["node_b", "node_c"]
)
thread_config = {
"configurable": {
"thread_id": "some_thread"
}
}
# 运行图直到断点
graph.invoke(inputs, config=thread_config)
# 可选:根据用户输入更新图状态
graph.update_state(update, config=thread_config)
# 恢复图的运行
graph.invoke(None, config=thread_config)
Note
对于**子图**,你不能在运行时设置静态断点。 如果你有一个子图,必须在编译时设置断点。
如果你想一次单步执行图中的一个节点,或者想在特定节点暂停图的执行,静态断点在调试时会特别有用。
NodeInterrupt
异常¶
如果你试图实现人在回路工作流,我们建议你使用 interrupt
函数 而不是 NodeInterrupt
异常。interrupt
函数更易于使用且更灵活。
NodeInterrupt
异常
开发者可以定义一些必须满足的*条件*,以触发断点。当开发者想在*特定条件*下暂停图的执行时,这种_动态断点_的概念很有用。这里使用了 NodeInterrupt
,它是一种特殊类型的异常,可以根据某些条件在节点内部抛出。例如,我们可以定义一个动态断点,当 input
的长度超过 5 个字符时触发。
def my_node(state: State) -> State:
if len(state['input']) > 5:
raise NodeInterrupt(f"Received input that is longer than 5 characters: {state['input']}")
return state
假设我们使用一个会触发动态断点的输入来运行图,然后尝试通过简单地传入 None
作为输入来恢复图的执行。
# 在触发动态断点后,尝试在不改变状态的情况下继续图的执行
for event in graph.stream(None, thread_config, stream_mode="values"):
print(event)
图会再次*中断*,因为这个节点会使用相同的图状态*重新运行*。我们需要更改图状态,使得触发动态断点的条件不再满足。因此,我们可以简单地将图状态编辑为满足动态断点条件(长度 < 5 个字符)的输入,然后重新运行节点。
# 更新状态以通过动态断点
graph.update_state(config=thread_config, values={"input": "foo"})
for event in graph.stream(None, thread_config, stream_mode="values"):
print(event)
或者,如果我们想保留当前输入并跳过执行检查的节点 (my_node
) 该怎么办?为此,我们可以简单地使用 as_node="my_node"
执行图更新,并为 values
传入 None
。这不会更新图状态,但会以 my_node
的身份运行更新,从而有效地跳过该节点并绕过动态断点。
额外资源 📚¶
- 概念指南:持久化:阅读持久化指南,以获取更多关于持久化的背景信息。
- 概念指南:人在回路:阅读人在回路指南,以获取更多关于如何使用断点将人工反馈集成到 LangGraph 应用程序中的背景信息。
- 如何查看和更新过往图状态:有关处理图状态的分步说明,展示了**重播**和**分叉**操作。