Dynamic breakpoint: interrupt()
Let the graph decide whether to pause at runtime

1Limitations of static breakpoints: why you need dynamic breakpoints

In the first few lessons of Module 3, we learnedStatic Breakpoints: When compiling the graph, passcompile(interrupt_before=["node_name"])orinterrupt_after=to specify which nodes will trigger a pause.

Static breakpoints have a fundamental limitation:It is determined at "compile time" and has nothing to do with runtime data.。

this isDynamic BreakpointsReason for existence: Let the code inside the node function, based on the runtime data,self-determinationWhether an interrupt is triggered and the specific reason for the interrupt can be communicated to the outside world.

2The core of dynamic breakpoints: interrupt() function

LangGraph bylanggraph.types.interruptFunction implements dynamic breakpoints. When a node function is called during executioninterrupt(...)When LangGraph runs:

• 1

  Trigger a controlled pause instead of a crash

  interrupt()A LangGraph-aware controlled interrupt is triggered internally, and the runtime takes over without letting the program crash with an uncaught exception.

• 2

  Save current checkpoint status

  The complete state of the graph (including the output of executed nodes) is persisted toMemorySaveror other checkpointers.

• 3

  Write interrupt information to task record

  passed tointerrupt()The payload (it is recommended to use JSON serializable strings, dictionaries, etc.) will be exposed to the caller and recorded instate.tasksofinterruptsin the field.

• 4

  Figure pauses execution, waiting for external operations

  graph.stream()orgraph.invoke()When the call returns, the process stops at the node that triggered the interruption; it can be continued through manual input, status modification, or recovery commands.

Import method

from langgraph.types import interrupt

# 调用 interrupt() 会暂停当前图执行
# payload 会作为中断原因暴露给外部调用方
interrupt("需要人工审核：输入内容超出长度限制")
interrupt({"reason": "high_risk", "score": 0.95})

3Complete example: Building a graph with dynamic breakpoints

The example in the notebook builds a simple three-step diagram:step_1 → step_2 → step_3, among whichstep_2Will be called when the input string exceeds 5 charactersinterrupt()。

Step 1: Define State and node functions

from typing_extensions import TypedDict
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import START, END, StateGraph
from langgraph.types import interrupt   # ← 导入动态中断函数

# State 只有一个字段：input 字符串
class State(TypedDict):
    input: str

# ── step_1：普通节点，直接返回 state ──
def step_1(state: State) -> State:
    print("---Step 1---")
    return state

# ── step_2：包含动态断点逻辑的关键节点 ──
def step_2(state: State) -> State:
    # 运行时条件判断：如果输入超过 5 个字符，则中断
    if len(state['input']) > 5:
        # 调用 interrupt()，并附带说明原因的 payload
        interrupt(
            f"Received input that is longer than 5 characters: {state['input']}"
        )
    # 如果没有触发中断，则正常执行
    print("---Step 2---")
    return state

# ── step_3：普通节点 ──
def step_3(state: State) -> State:
    print("---Step 3---")
    return state

Step 2: Build the graph and compile

# 构建图结构
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)

# 准备 MemorySaver（动态断点依赖 checkpointer 保存状态）
memory = MemorySaver()

# 编译时 ★ 不需要 ★ 指定任何 interrupt_before / interrupt_after！
# 动态断点完全由节点内部的 interrupt() 调用控制
graph = builder.compile(checkpointer=memory)

Visual structure of graphs

There is no trace of "interruption" in the graph structure itself - dynamic breakpoints "appear" only at runtime. This is different from the visibility of static breakpoints (break nodes are fixed at compile time).

4Trigger an interrupt: check status and interrupt reason

Run the graph and trigger interrupt()

# 配置：使用 thread_id 追踪这次执行的状态
initial_input = {"input": "hello world"}  # 11 个字符，超过 5
thread_config = {"configurable": {"thread_id": "1"}}

# 使用 stream 运行图
for event in graph.stream(initial_input, thread_config, stream_mode="values"):
    print(event)

Observe the output:

• first line{'input': 'hello world'}: The initial State event when the graph starts
• ---Step 1---: step_1 executes normally and prints
• second line{'input': 'hello world'}: State event after step_1 is executed
• The third event contains__interrupt__:step_2 calledinterrupt(), the payload is returned to the caller
• There is no output of step_3 after that: the graph is paused,stream()Return

Check state.next: the next node waiting to be executed

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

state.nextshow('step_2',), indicating that the graph stopped at the "about to be executed" position of step_2 - step_2 triggered an interrupt, but this execution was aborted, so it is still the "next node to be executed".

Check state.tasks: read interruption reason

print(state.tasks)

Interpretationstate.tasksKey fields of:

5Pitfall: Simple recovery retriggers the interrupt

Mistakes that beginners often make: Thinking that they will not modify the status or provideCommand(resume=...), directlygraph.stream(None, thread_config)can continue. Let's see what happens:

# 尝试直接恢复（不修改 state）
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

state = graph.get_state(thread_config)
print(state.next)  # 仍然是 step_2！

This differs from the behavior of static breakpoints:

• static breakpoint：interrupt_before=["step_2"]It is interrupted before the execution of step_2, and step_2 starts to execute after recovery. Whether it passes or not depends on the code logic of step_2.
• dynamic breakpoint: The interrupt condition is in step_2During executionTrigger, use directlyNoneRecovery will run step_2 again. If the status has not changed and the interrupt condition is still true, it will be interrupted again.

6Correct process: modify the status and then restore it

The correct way to handle it is: use it firstgraph.update_state()Modify the state so that the interrupt condition is no longer met, and then resume execution.

Step 1: Modify state and eliminate interruption conditions

# 将 input 改为短字符串（5 个字符以内），让 step_2 的条件不再触发
graph.update_state(
    thread_config,
    {"input": "hi"},  # "hi" 只有 2 个字符，len("hi") = 2 ≤ 5
)

update_stateReturns the new checkpoint configuration, indicating that the status was successfully written.

Step 2: Resume execution

# 传入 None 作为 input，表示"从上次中断处继续，不添加新输入"
for event in graph.stream(None, thread_config, stream_mode="values"):
    print(event)

This execution was successful! Execution process analysis:

• {'input': 'hi'}:Current State at the time of recovery (has been modified)
• ---Step 2---: step_2 is executed normally (len("hi") = 2 ≤ 5, does not trigger an interrupt)
• {'input': 'hi'}: State event after step_2 execution
• ---Step 3---: step_3 executes normally
• The last State event: the graph execution is completed

Summary of complete operation process

7Using dynamic breakpoints via the LangGraph API

In a production environment, graphs are typically deployed on the LangGraph server (vialanggraph devStart a local development server) and clients interact with it through the LangGraph SDK. Dynamic breakpoints are also applicable in API scenarios, and the operation logic is completely consistent with the local one.

Start local development server

in module/studioRun in the directory:

langgraph dev

After the service starts, it will display:

Connect and run graphs via SDK (trigger dynamic breakpoints)

from langgraph_sdk import get_client

# 连接到本地开发服务器
URL = "http://127.0.0.1:2024"
client = get_client(url=URL)

# 创建新线程（相当于一次独立的图执行会话）
thread = await client.threads.create()
input_dict = {"input": "hello world"}  # 超过 5 字符，会触发 interrupt()

async for chunk in client.runs.stream(
    thread["thread_id"],
    assistant_id="dynamic_breakpoints",  # 图的名称（在 langgraph.json 中定义）
    input=input_dict,
    stream_mode="values",
):
    print(f"Receiving new event of type: {chunk.event}...")
    print(chunk.data)

End of flow event - Figure in step_2interrupt()paused.

Check remote thread status

current_state = await client.threads.get_state(thread['thread_id'])
print(current_state['next'])

Modify status remotely and then restore

# 更新 state：将 input 改为短字符串
await client.threads.update_state(thread['thread_id'], {"input": "hi!"})

# 恢复执行：传入 input=None 表示从中断处继续
async for chunk in client.runs.stream(
    thread["thread_id"],
    assistant_id="dynamic_breakpoints",
    input=None,  # ← 关键：None 表示恢复，而非新的输入
    stream_mode="values",
):
    print(f"Receiving new event of type: {chunk.event}...")
    print(chunk.data)

The graph was successfully restored and executed (both step_2 and step_3 were executed normally).

8Static breakpoints vs dynamic breakpoints: a complete comparison

Both breakpoint mechanisms have their own applicable scenarios, and understanding their differences can help make the right choice in actual projects.

How to choose?

Examples of real engineering application scenarios

interrupt({
    "reason": "high_risk_content",
    "risk_score": 0.94,
    "flagged_text": state["generated_content"][:200],
    "suggested_action": "请人工审核后选择批准或拒绝"
})