使用断点

要求

要使用断点，你需要

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

设置断点

你可以在两个地方设置断点

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

静态断点

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

如果你想一次一个节点地逐步执行图，或者想在特定节点暂停图的执行，静态断点会特别有用。

<span id="__span-0-1">graph = graph_builder.compile( 
<span id="__span-0-2">    interrupt_before=["node_a"], 
<span id="__span-0-3">    interrupt_after=["node_b", "node_c"], 
<span id="__span-0-4">    checkpointer=checkpointer, 
<span id="__span-0-5">)
<span id="__span-0-6">
<span id="__span-0-7">config = {
<span id="__span-0-8">    "configurable": {
<span id="__span-0-9">        "thread_id": "some_thread"
<span id="__span-0-10">    }
<span id="__span-0-11">}
<span id="__span-0-12">
<span id="__span-0-13"># Run the graph until the breakpoint
<span id="__span-0-14">graph.invoke(inputs, config=thread_config) 
<span id="__span-0-15">
<span id="__span-0-16"># Resume the graph
<span id="__span-0-17">graph.invoke(None, config=thread_config)

设置静态断点

<span id="__span-2-1">from IPython.display import Image, display
<span id="__span-2-2">from typing_extensions import TypedDict
<span id="__span-2-3">
<span id="__span-2-4">from langgraph.checkpoint.memory import InMemorySaver 
<span id="__span-2-5">from langgraph.graph import StateGraph, START, END
<span id="__span-2-6">
<span id="__span-2-7">
<span id="__span-2-8">class State(TypedDict):
<span id="__span-2-9">    input: str
<span id="__span-2-10">
<span id="__span-2-11">
<span id="__span-2-12">def step_1(state):
<span id="__span-2-13">    print("---Step 1---")
<span id="__span-2-14">    pass
<span id="__span-2-15">
<span id="__span-2-16">
<span id="__span-2-17">def step_2(state):
<span id="__span-2-18">    print("---Step 2---")
<span id="__span-2-19">    pass
<span id="__span-2-20">
<span id="__span-2-21">
<span id="__span-2-22">def step_3(state):
<span id="__span-2-23">    print("---Step 3---")
<span id="__span-2-24">    pass
<span id="__span-2-25">
<span id="__span-2-26">
<span id="__span-2-27">builder = StateGraph(State)
<span id="__span-2-28">builder.add_node("step_1", step_1)
<span id="__span-2-29">builder.add_node("step_2", step_2)
<span id="__span-2-30">builder.add_node("step_3", step_3)
<span id="__span-2-31">builder.add_edge(START, "step_1")
<span id="__span-2-32">builder.add_edge("step_1", "step_2")
<span id="__span-2-33">builder.add_edge("step_2", "step_3")
<span id="__span-2-34">builder.add_edge("step_3", END)
<span id="__span-2-35">
<span id="__span-2-36"># Set up a checkpointer 
<span id="__span-2-37">checkpointer = InMemorySaver() # (1)!
<span id="__span-2-38">
<span id="__span-2-39">graph = builder.compile(
<span id="__span-2-40">    checkpointer=checkpointer, # (2)!
<span id="__span-2-41">    interrupt_before=["step_3"] # (3)!
<span id="__span-2-42">)
<span id="__span-2-43">
<span id="__span-2-44"># View
<span id="__span-2-45">display(Image(graph.get_graph().draw_mermaid_png()))
<span id="__span-2-46">
<span id="__span-2-47">
<span id="__span-2-48"># Input
<span id="__span-2-49">initial_input = {"input": "hello world"}
<span id="__span-2-50">
<span id="__span-2-51"># Thread
<span id="__span-2-52">thread = {"configurable": {"thread_id": "1"}}
<span id="__span-2-53">
<span id="__span-2-54"># Run the graph until the first interruption
<span id="__span-2-55">for event in graph.stream(initial_input, thread, stream_mode="values"):
<span id="__span-2-56">    print(event)
<span id="__span-2-57">
<span id="__span-2-58"># This will run until the breakpoint
<span id="__span-2-59"># You can get the state of the graph at this point
<span id="__span-2-60">print(graph.get_state(config))
<span id="__span-2-61">
<span id="__span-2-62"># You can continue the graph execution by passing in `None` for the input
<span id="__span-2-63">for event in graph.stream(None, thread, stream_mode="values"):
<span id="__span-2-64">    print(event)

动态断点

如果你需要根据条件从给定节点内部中断图，请使用动态断点。

<span id="__span-3-1">from langgraph.errors import NodeInterrupt
<span id="__span-3-2">
<span id="__span-3-3">def step_2(state: State) -&gt; State:
<span id="__span-3-4">    if len(state["input"]) &gt; 5:
<span id="__span-3-5">        raise NodeInterrupt( 
<span id="__span-3-6">            f"Received input that is longer than 5 characters: {state['foo']}"
<span id="__span-3-7">        )
<span id="__span-3-8">    return state

使用动态断点 API 参考：StateGraph | START | END | MemorySaver

<span id="__span-4-1">from typing_extensions import TypedDict
<span id="__span-4-2">from IPython.display import Image, display
<span id="__span-4-3">
<span id="__span-4-4">from langgraph.graph import StateGraph, START, END
<span id="__span-4-5">from langgraph.checkpoint.memory import MemorySaver
<span id="__span-4-6">from langgraph.errors import NodeInterrupt
<span id="__span-4-7">
<span id="__span-4-8">
<span id="__span-4-9">class State(TypedDict):
<span id="__span-4-10">    input: str
<span id="__span-4-11">
<span id="__span-4-12">
<span id="__span-4-13">def step_1(state: State) -&gt; State:
<span id="__span-4-14">    print("---Step 1---")
<span id="__span-4-15">    return state
<span id="__span-4-16">
<span id="__span-4-17">
<span id="__span-4-18">def step_2(state: State) -&gt; State:
<span id="__span-4-19">    # Let's optionally raise a NodeInterrupt
<span id="__span-4-20">    # if the length of the input is longer than 5 characters
<span id="__span-4-21">    if len(state["input"]) &gt; 5:
<span id="__span-4-22">        raise NodeInterrupt(
<span id="__span-4-23">            f"Received input that is longer than 5 characters: {state['input']}"
<span id="__span-4-24">        )
<span id="__span-4-25">    print("---Step 2---")
<span id="__span-4-26">    return state
<span id="__span-4-27">
<span id="__span-4-28">
<span id="__span-4-29">def step_3(state: State) -&gt; State:
<span id="__span-4-30">    print("---Step 3---")
<span id="__span-4-31">    return state
<span id="__span-4-32">
<span id="__span-4-33">
<span id="__span-4-34">builder = StateGraph(State)
<span id="__span-4-35">builder.add_node("step_1", step_1)
<span id="__span-4-36">builder.add_node("step_2", step_2)
<span id="__span-4-37">builder.add_node("step_3", step_3)
<span id="__span-4-38">builder.add_edge(START, "step_1")
<span id="__span-4-39">builder.add_edge("step_1", "step_2")
<span id="__span-4-40">builder.add_edge("step_2", "step_3")
<span id="__span-4-41">builder.add_edge("step_3", END)
<span id="__span-4-42">
<span id="__span-4-43"># Set up memory
<span id="__span-4-44">memory = MemorySaver()
<span id="__span-4-45">
<span id="__span-4-46"># Compile the graph with memory
<span id="__span-4-47">graph = builder.compile(checkpointer=memory)
<span id="__span-4-48">
<span id="__span-4-49"># View
<span id="__span-4-50">display(Image(graph.get_graph().draw_mermaid_png()))

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

<span id="__span-5-1">initial_input = {"input": "hello"}
<span id="__span-5-2">thread_config = {"configurable": {"thread_id": "1"}}
<span id="__span-5-3">
<span id="__span-5-4">for event in graph.stream(initial_input, thread_config, stream_mode="values"):
<span id="__span-5-5">    print(event)

<span id="__span-6-1">{'input': 'hello'}
<span id="__span-6-2">---Step 1---
<span id="__span-6-3">{'input': 'hello'}
<span id="__span-6-4">---Step 2---
<span id="__span-6-5">{'input': 'hello'}
<span id="__span-6-6">---Step 3---
<span id="__span-6-7">{'input': 'hello'}

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

<span id="__span-7-1">state = graph.get_state(thread_config)
<span id="__span-7-2">print(state.next)
<span id="__span-7-3">print(state.tasks)

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

<span id="__span-9-1">initial_input = {"input": "hello world"}
<span id="__span-9-2">thread_config = {"configurable": {"thread_id": "2"}}
<span id="__span-9-3">
<span id="__span-9-4"># Run the graph until the first interruption
<span id="__span-9-5">for event in graph.stream(initial_input, thread_config, stream_mode="values"):
<span id="__span-9-6">    print(event)

<code tabindex="0"><span id="__span-10-1">{'input': 'hello world'}
<span id="__span-10-2">---Step 1---
<span id="__span-10-3">{'input': 'hello world'}
<span id="__span-10-4">{'__interrupt__': (Interrupt(value='Received input that is longer than 5 characters: hello world', resumable=False, ns=None),)}

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

<span id="__span-11-1">state = graph.get_state(thread_config)
<span id="__span-11-2">print(state.next)
<span id="__span-11-3">print(state.tasks)

<code tabindex="0"><span id="__span-12-1">('step_2',)
<span id="__span-12-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),)

如果从断点恢复图，我们将再次中断，因为我们的输入和图状态没有改变。

<code tabindex="0"><span id="__span-13-1"># NOTE: to resume the graph from a dynamic interrupt we use the same syntax as with regular interrupts -- we pass None as the input
<span id="__span-13-2">for event in graph.stream(None, thread_config, stream_mode="values"):
<span id="__span-13-3">    print(event)

<code tabindex="0"><span id="__span-14-1">{'input': 'hello world'}
<span id="__span-14-2">{'__interrupt__': (Interrupt(value='Received input that is longer than 5 characters: hello world', resumable=False, ns=None),)}

<span id="__span-15-1">state = graph.get_state(thread_config)
<span id="__span-15-2">print(state.next)
<span id="__span-15-3">print(state.tasks)

<code tabindex="0"><span id="__span-16-1">('step_2',)
<span id="__span-16-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

<span id="__span-17-1">from typing_extensions import TypedDict
<span id="__span-17-2">
<span id="__span-17-3">from langgraph.graph import START, StateGraph
<span id="__span-17-4">from langgraph.checkpoint.memory import InMemorySaver
<span id="__span-17-5">from langgraph.types import interrupt
<span id="__span-17-6">
<span id="__span-17-7">
<span id="__span-17-8">class State(TypedDict):
<span id="__span-17-9">    foo: str
<span id="__span-17-10">
<span id="__span-17-11">
<span id="__span-17-12">def subgraph_node_1(state: State):
<span id="__span-17-13">    return {"foo": state["foo"]}
<span id="__span-17-14">
<span id="__span-17-15">
<span id="__span-17-16">subgraph_builder = StateGraph(State)
<span id="__span-17-17">subgraph_builder.add_node(subgraph_node_1)
<span id="__span-17-18">subgraph_builder.add_edge(START, "subgraph_node_1")
<span id="__span-17-19">
<span id="__span-17-20">subgraph = subgraph_builder.compile(interrupt_before=["subgraph_node_1"])
<span id="__span-17-21">
<span id="__span-17-22">builder = StateGraph(State)
<span id="__span-17-23">builder.add_node("node_1", subgraph)  # directly include subgraph as a node
<span id="__span-17-24">builder.add_edge(START, "node_1")
<span id="__span-17-25">
<span id="__span-17-26">checkpointer = InMemorySaver()
<span id="__span-17-27">graph = builder.compile(checkpointer=checkpointer)
<span id="__span-17-28">
<span id="__span-17-29">config = {"configurable": {"thread_id": "1"}}
<span id="__span-17-30">
<span id="__span-17-31">graph.invoke({"foo": ""}, config)
<span id="__span-17-32">
<span id="__span-17-33"># Fetch state including subgraph state.
<span id="__span-17-34">print(graph.get_state(config, subgraphs=True).tasks[0].state)
<span id="__span-17-35">
<span id="__span-17-36"># resume the subgraph
<span id="__span-17-37">graph.invoke(None, config)

<code tabindex="0"><span id="__span-18-1">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=())