使用断点¶

要求¶

要使用断点，你需要

指定一个检查点器 来保存每一步后的图状态。
设置断点 来指定执行应暂停的位置。
使用 线程 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)!

断点在编译时设置。
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)!

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

注意

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

设置静态断点

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 参考：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 参考：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': ''}