Skip to content

断点

断点可以在特定位置暂停图执行，并允许逐步执行每一步。断点由LangGraph的持久层提供支持，该层在每个图步骤后保存状态。断点也可以用于启用人机交互工作流，尽管我们推荐使用interrupt函数来实现这一目的。

使用要求

要使用断点功能，您需要：

  1. 指定一个检查点器来保存每次步骤后的图状态。
  2. 设置断点以指定执行应暂停的位置。
  3. 使用线程ID运行图，在遇到断点时暂停执行。
  4. 使用invoke/ainvoke/stream/astream恢复执行（参见命令原语）。

设置断点

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

  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 异常

我们建议您如果尝试实现人机交互工作流，应使用中断函数代替NodeInterrupt异常。中断函数更容易使用且更灵活。

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"执行图更新，并传递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)

额外资源 📚

Comments