使用断点¶

要求¶

要使用断点，你需要：

指定一个检查点器 以在每一步之后保存图的状态。
设置断点 来指定执行应暂停的位置。
使用线程 ID 运行图，以便在断点处暂停执行。
使用 invoke/ainvoke/stream/astream 并将输入参数设为 None 来**恢复执行**。

设置断点¶

你可以设置断点的两个位置是：

在节点执行**之前**或**之后**，通过在**编译时**或**运行时**设置断点。我们称这些为静态断点。
在节点内部使用 NodeInterrupt 异常。我们称这些为动态断点。

静态断点¶

静态断点会在节点执行**之前**或**之后**触发。你可以在“编译”时间或“运行”时间通过指定 interrupt_before 和 interrupt_after 来设置静态断点。

如果想逐步执行图的每个节点，或者希望在特定节点暂停图的执行，静态断点特别适用于调试。

编译时运行时

graph = graph_builder.compile( # (1)!
    interrupt_before=["node_a"], # (2)!
    interrupt_after=["node_b", "node_c"], # (3)!
    checkpointer=checkpointer, # (4)!
)

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

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

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

断点是在 compile 时间设置的。
interrupt_before 指定在节点执行前应该暂停的节点。
interrupt_after 指定在节点执行后应该暂停的节点。
需要一个检查点器来启用断点。
图会运行直到遇到第一个断点。
通过将输入设为 None 来恢复图的执行。这将运行图直到下一个断点被命中。

graph.invoke( # (1)!
    inputs, 
    interrupt_before=["node_a"], # (2)!
    interrupt_after=["node_b", "node_c"] # (3)!
    config={
        "configurable": {"thread_id": "some_thread"}
    }, 
)

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

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

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

通过 interrupt_before 和 interrupt_after 参数调用 graph.invoke。这是一个运行时配置，可以针对每次调用进行更改。
interrupt_before 指定在节点执行前应该暂停的节点。
interrupt_after 指定在节点执行后应该暂停的节点。
图会运行直到遇到第一个断点。
通过将输入设为 None 来恢复图的执行。这将运行图直到下一个断点被命中。

Note

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

设置静态断点

from IPython.display import Image, display
from typing_extensions import TypedDict

from langgraph.checkpoint.memory import InMemorySaver 
from langgraph.graph import StateGraph, START, END


class State(TypedDict):
    input: str


def step_1(state):
    print("---Step 1---")
    pass


def step_2(state):
    print("---Step 2---")
    pass


def step_3(state):
    print("---Step 3---")
    pass


builder = StateGraph(State)
builder.add_node("step_1", step_1)
builder.add_node("step_2", step_2)
builder.add_node("step_3", step_3)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")
builder.add_edge("step_3", END)

# Set up a checkpointer 
checkpointer = InMemorySaver() # (1)!

graph = builder.compile(
    checkpointer=checkpointer, # (2)!
    interrupt_before=["step_3"] # (3)!
)

# View
display(Image(graph.get_graph().draw_mermaid_png()))


# Input
initial_input = {"input": "hello world"}

# Thread
thread = {"configurable": {"thread_id": "1"}}

# Run the graph until the first interruption
for event in graph.stream(initial_input, thread, stream_mode="values"):
    print(event)

# This will run until the breakpoint
# You can get the state of the graph at this point
print(graph.get_state(config))

# You can continue the graph execution by passing in `None` for the input
for event in graph.stream(None, thread, stream_mode="values"):
    print(event)

动态断点¶

如果你需要根据条件从某个节点内部中断图的执行，请使用动态断点。

from langgraph.errors import NodeInterrupt

def step_2(state: State) -> State:
    if len(state["input"]) > 5:
        raise NodeInterrupt( # (1)!
            f"Received input that is longer than 5 characters: {state['foo']}"
        )
    return state

根据某些条件引发 NodeInterrupt 异常。在此示例中，如果属性 input 的长度超过 5 个字符，我们创建一个动态断点。

使用动态断点

^{API Reference: StateGraph | START | END | MemorySaver}

from typing_extensions import TypedDict
from IPython.display import Image, display

from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.errors import NodeInterrupt


class State(TypedDict):
    input: str


def step_1(state: State) -> State:
    print("---Step 1---")
    return state


def step_2(state: State) -> State:
    # Let's optionally raise a NodeInterrupt
    # if the length of the input is longer than 5 characters
    if len(state["input"]) > 5:
        raise NodeInterrupt(
            f"Received input that is longer than 5 characters: {state['input']}"
        )
    print("---Step 2---")
    return state


def step_3(state: State) -> State:
    print("---Step 3---")
    return state


builder = StateGraph(State)
builder.add_node("step_1", step_1)
builder.add_node("step_2", step_2)
builder.add_node("step_3", step_3)
builder.add_edge(START, "step_1")
builder.add_edge("step_1", "step_2")
builder.add_edge("step_2", "step_3")
builder.add_edge("step_3", END)

# Set up memory
memory = MemorySaver()

# Compile the graph with memory
graph = builder.compile(checkpointer=memory)

# View
display(Image(graph.get_graph().draw_mermaid_png()))

首先，让我们运行一个输入长度小于等于5个字符的图。这应该会安全地忽略我们定义的中断条件，并在图执行结束时返回原始输入。

initial_input = {"input": "hello"}
thread_config = {"configurable": {"thread_id": "1"}}

for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

{'input': 'hello'}
---Step 1---
{'input': 'hello'}
---Step 2---
{'input': 'hello'}
---Step 3---
{'input': 'hello'}

如果我们此时检查图，可以看到没有剩余的任务需要运行，并且图确实已经完成了执行。

state = graph.get_state(thread_config)
print(state.next)
print(state.tasks)

()
()

现在，让我们运行一个输入长度超过5个字符的图。这应该会触发我们通过在`step_2`节点内引发`NodeInterrupt`错误所定义的动态中断。

initial_input = {"input": "hello world"}
thread_config = {"configurable": {"thread_id": "2"}}

# Run the graph until the first interruption
for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

{'input': 'hello world'}
---Step 1---
{'input': 'hello world'}
{'__interrupt__': (Interrupt(value='Received input that is longer than 5 characters: hello world', resumable=False, ns=None),)}

我们可以看到，现在图在执行 `step_2` 时停止了。如果我们检查此时的图状态，可以看到有关下一个要执行的节点的信息（`step_2`），以及引发中断的节点（也是 `step_2`），以及关于中断的其他信息。

state = graph.get_state(thread_config)
print(state.next)
print(state.tasks)

('step_2',)
(PregelTask(id='bfc767e3-a6c4-c5af-dbbf-0d20ea64501e', name='step_2', path=('__pregel_pull', 'step_2'), error=None, interrupts=(Interrupt(value='Received input that is longer than 5 characters: hello world', resumable=False, ns=None),), state=None, result=None),)

如果我们尝试从断点恢复图执行，由于输入和图状态没有发生变化，我们只会再次中断。

# NOTE: to resume the graph from a dynamic interrupt we use the same syntax as with regular interrupts -- we pass None as the input
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

{'input': 'hello world'}
{'__interrupt__': (Interrupt(value='Received input that is longer than 5 characters: hello world', resumable=False, ns=None),)}

state = graph.get_state(thread_config)
print(state.next)
print(state.tasks)

('step_2',)
(PregelTask(id='bfc767e3-a6c4-c5af-dbbf-0d20ea64501e', name='step_2', path=('__pregel_pull', 'step_2'), error=None, interrupts=(Interrupt(value='Received input that is longer than 5 characters: hello world', resumable=False, ns=None),), state=None, result=None),)

与子图一起使用¶

要向子图添加断点，可以采用以下方法之一：

在**编译**子图时指定它们，以定义静态断点。
定义动态断点。

向子图添加断点

^{API Reference: START | StateGraph | InMemorySaver | interrupt}

from typing_extensions import TypedDict

from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt


class State(TypedDict):
    foo: str


def subgraph_node_1(state: State):
    return {"foo": state["foo"]}


subgraph_builder = StateGraph(State)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_edge(START, "subgraph_node_1")

subgraph = subgraph_builder.compile(interrupt_before=["subgraph_node_1"])

builder = StateGraph(State)
builder.add_node("node_1", subgraph)  # directly include subgraph as a node
builder.add_edge(START, "node_1")

checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "1"}}

graph.invoke({"foo": ""}, config)

# Fetch state including subgraph state.
print(graph.get_state(config, subgraphs=True).tasks[0].state)

# resume the subgraph
graph.invoke(None, config)

StateSnapshot(values={'foo': ''}, next=('subgraph_node_1',), config={'configurable': {'thread_id': '1', 'checkpoint_ns': 'node_1:dfc321bb-7c91-ccfe-23b8-c2374ae3f1cc', 'checkpoint_id': '1f02a8d1-985a-6e2c-8000-77034088c0ce', 'checkpoint_map': {'': '1f02a8d1-9856-6264-8000-ed1534455427', 'node_1:dfc321bb-7c91-ccfe-23b8-c2374ae3f1cc': '1f02a8d1-985a-6e2c-8000-77034088c0ce'}}}, metadata={'source': 'loop', 'writes': None, 'step': 0, 'parents': {'': '1f02a8d1-9856-6264-8000-ed1534455427'}, 'thread_id': '1', 'langgraph_step': 1, 'langgraph_node': 'node_1', 'langgraph_triggers': ['branch:to:node_1'], 'langgraph_path': ['__pregel_pull', 'node_1'], 'langgraph_checkpoint_ns': 'node_1:dfc321bb-7c91-ccfe-23b8-c2374ae3f1cc'}, created_at='2025-05-06T15:16:35.543192+00:00', parent_config={'configurable': {'thread_id': '1', 'checkpoint_ns': 'node_1:dfc321bb-7c91-ccfe-23b8-c2374ae3f1cc', 'checkpoint_id': '1f02a8d1-9859-6d41-bfff-872b2e8f4db6', 'checkpoint_map': {'': '1f02a8d1-9856-6264-8000-ed1534455427', 'node_1:dfc321bb-7c91-ccfe-23b8-c2374ae3f1cc': '1f02a8d1-9859-6d41-bfff-872b2e8f4db6'}}}, tasks=(PregelTask(id='33218e09-8747-5161-12b1-5dc705d30b51', name='subgraph_node_1', path=('__pregel_pull', 'subgraph_node_1'), error=None, interrupts=(), state=None, result=None),), interrupts=())

{'foo': ''}