断点¶

断点在特定点暂停图的执行，并允许逐步执行。断点由 LangGraph 的 持久化层 提供支持，该层在每个图步骤后保存状态。断点还可以用于启用 人机协作 工作流，尽管我们建议为此使用 interrupt 函数。

要求¶

要使用断点，您需要：

指定一个检查点，以在每个步骤后保存图状态。
设置断点，以指定执行应暂停的位置。
使用 线程 ID 运行图，以在断点处暂停执行。
使用 invoke/ainvoke/stream/astream 恢复执行（参见 Command 原语）。

设置断点¶

您可以在两个地方设置断点：

在**节点执行的**前**或**后，通过在**编译时**或**运行时**设置断点。我们称这些为 静态断点。
**在**节点内部使用 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)

注意

您无法在运行时为**子图**设置静态断点。如果您有子图，必须在编译时设置断点。

静态断点在调试时尤其有用，特别是当您想逐节点逐步执行图时，或希望在特定节点处暂停图执行时。

`NodeInterrupt` 异常¶

如果您试图实现人机协作工作流，我们建议您 使用 interrupt 函数而不是 NodeInterrupt 异常。interrupt 函数更易于使用且更灵活。

NodeInterrupt 异常

开发者可以定义一些*条件*，只有在这些条件满足时，断点才会被触发。这个动态断点的概念在开发者希望在*特定条件*下暂停图时很有用。这使用了 NodeInterrupt，这是可以根据某些条件从节点内部引发的特殊类型的异常。例如，我们可以定义一个当 input 长度超过 5 个字符时触发的动态断点。

def my_node(state: State) -> State:
    if len(state['input']) > 5:
        raise NodeInterrupt(f"收到的输入超过了 5 个字符: {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" 执行图更新，并将 None 传递给值。这将不会更新图状态，而是以 my_node 运行更新，从而有效地跳过节点并绕过动态断点。

# 此更新将完全跳过节点 `my_node`
graph.update_state(config=thread_config, values=None, as_node="my_node")
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

其他资源 📚¶

概念指南：持久化：阅读持久化指南以获取更多关于持久化的上下文。
概念指南：人机协作：阅读人机协作指南以获取有关在使用断点的 LangGraph 应用程序中集成人工反馈的更多上下文。
如何查看和更新过去的图状态：逐步说明如何处理图状态，演示重放和分叉操作。

断点¶

要求¶

设置断点¶

静态断点¶

NodeInterrupt 异常¶

其他资源 📚¶

`NodeInterrupt` 异常¶