跳至内容

断点

断点可以在特定点暂停图执行,并允许逐步执行。断点功能由 LangGraph 的持久化层提供支持,该层在每个图步骤后保存状态。断点也可用于启用人工参与 (human-in-the-loop)工作流程,但为此目的,我们建议使用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=..., # Specify a checkpointer
)

thread_config = {
    "configurable": {
        "thread_id": "some_thread"
    }
}

# Run the graph until the breakpoint
graph.invoke(inputs, config=thread_config)

# Optionally update the graph state based on user input
graph.update_state(update, config=thread_config)

# Resume the graph
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"
    }
}

# Run the graph until the breakpoint
graph.invoke(inputs, config=thread_config)

# Optionally update the graph state based on user input
graph.update_state(update, config=thread_config)

# Resume the graph
graph.invoke(None, config=thread_config)

注意

你不能在运行时为子图设置静态断点。如果你有子图,必须在编译时设置断点。

如果你想逐个节点地单步执行图,或者想在特定节点暂停图执行,静态断点会特别有用,尤其是在调试时。

NodeInterrupt 异常

如果你尝试实现人工参与 (human-in-the-loop)工作流程,我们建议你使用 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 作为输入来简单地恢复图执行。

# Attempt to continue the graph execution with no change to state after we hit the dynamic breakpoint 
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

图将再次中断,因为这个节点将使用相同的图状态重新运行。我们需要改变图状态,使得触发动态断点的条件不再满足。因此,我们可以简单地将图状态编辑为满足动态断点条件(小于 5 个字符)的输入,然后重新运行节点。

# Update the state to pass the dynamic breakpoint
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 的名义运行更新,从而有效地跳过该节点并绕过动态断点。

# This update will skip the node `my_node` altogether
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)

其他资源 📚

评论