Skip to content

断点

断点可在特定点暂停图执行,并允许逐步执行。断点由 LangGraph 的持久化层提供支持,该层会在每个图步骤后保存状态。断点也可用于启用人工介入工作流,不过我们建议为此目的使用interrupt 函数

要求

要使用断点,你需要:

  1. 指定一个检查点保存器 以在每一步之后保存图状态。
  2. 设置断点 以指定执行应在何处暂停。
  3. 使用 线程 ID 运行图 以在断点处暂停执行。
  4. 使用 invoke/ainvoke/stream/astream 恢复执行(请参阅 Command 原语)。

设置断点

有两个地方可以设置断点: 1. 在节点执行**之前**或**之后**,通过在**编译时**或**运行时**设置断点。我们称这些为静态断点。 2. 在节点**内部**使用 NodeInterrupt 异常

静态断点

静态断点在节点执行**之前**或**之后**触发。你可以通过在**“编译”时**或**运行时**指定 interrupt_beforeinterrupt_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 的身份运行更新,从而有效地跳过该节点并绕过动态断点。

# 此更新将完全跳过节点 `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