LangGraph Breakpoint Mechanism
Breakpoints & Human-in-the-Loop

1Why you need breakpoints: Three motivations for Human-in-the-Loop

In Module 1 and Module 2, the Agent we built runs completely automatically: from START to END, LLM makes decisions independently, and the tools are automatically called, all in one go. But in real production scenarios,It is not always safe to make the Agent completely autonomous。

LangGraph'sHuman-in-the-Loop(Human-machine collaboration) design solves three core problems:

Imagine a financial reimbursement agent: it can automatically analyze receipts, calculate amounts, and fill out reimbursement forms, butfinal commitThis step must be approved by a human treasurer. The breakpoint is the "waiting for approval" checkpoint you inserted.

2What are breakpoints: compile-time static breakpoints

The nature of breakpoints

Breakpoint (Breakpoint) is atWhen compiling the graph (compile phase)Just a definite, fixed interruption location. It tells LangGraph: "Every time execution reaches the front (or back) of this node, it will automatically pause. Do not continue going down and wait for external instructions."

breakpoint passedcompile()Two parameters to set:

# interrupt_before：在节点执行【之前】暂停
graph = builder.compile(
    interrupt_before=["tools"],   # 在 tools 节点运行前暂停
    checkpointer=memory
)

# interrupt_after：在节点执行【之后】暂停
graph = builder.compile(
    interrupt_after=["tools"],    # 在 tools 节点运行后暂停
    checkpointer=memory
)

Static breakpoints vs dynamic interrupts

"static" means the breakpoint is atcompile()It has been hard-coded - every time it is executed to that node,DefinitelyIt will be paused and there is no conditional judgment. This is consistent with the dynamic interruptions you will learn in Lesson 4 (NodeInterrupt), dynamic interrupts can decide whether to pause based on the runtime status.

3Core premise: MemorySaver checkpointer

Why must breakpoints be used with checkpointers?

The working principle of breakpoint is: when the graph execution reaches the breakpoint,Save the current state, and then stop execution. After humans issue the "continue" command, from the saved stateResume execution. This "save and restore" mechanism ischeckpointer。

How to use MemorySaver

MemorySaveris LangGraph's built-in memory checkpointer, which saves all state in process memory. Suitable for development, testing and stand-alone scenarios.

from langgraph.checkpoint.memory import MemorySaver

# 创建检查点器实例
memory = MemorySaver()

# 编译图时传入 checkpointer 和断点配置
graph = builder.compile(
    interrupt_before=["tools"],
    checkpointer=memory      # 必须提供！
)

The concept of Thread

For checkpointersthread_idto differentiate between different conversations/running sessions. Each time you run the graph, you need to provide a configuration dictionary:

# config：指定这次运行使用哪个 thread
thread = {"configurable": {"thread_id": "1"}}

# 同一个 thread_id = 同一个对话上下文
# 不同 thread_id = 独立的、互不干扰的对话

4Practical code: Constructing an Agent graph with breakpoints

Below is the complete example in Notebook. We build an Agent with math tools (addition, multiplication, division) and set a breakpoint before the tools node to allow humans to approve before the tool is called.

Step One: Define Tools and LLM

from langchain_deepseek import ChatDeepSeek

# 定义三个数学工具函数
def multiply(a: int, b: int) -> int:
    """Multiply a and b."""
    return a * b

def add(a: int, b: int) -> int:
    """Adds a and b."""
    return a + b

def divide(a: int, b: int) -> float:
    """Divide a by b."""
    return a / b

# 工具列表
tools = [add, multiply, divide]

# 绑定工具到 LLM（LLM 可以选择调用哪个工具）
llm = ChatDeepSeek(model="deepseek-v4-pro")
llm_with_tools = llm.bind_tools(tools)

Step 2: Build the graph and compile (with breakpoints)

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import MessagesState, StateGraph, START
from langgraph.prebuilt import tools_condition, ToolNode
from langchain_core.messages import SystemMessage

# 系统提示
sys_msg = SystemMessage(content="You are a helpful assistant tasked with performing arithmetic on a set of inputs.")

# 定义 assistant 节点：调用带工具的 LLM
def assistant(state: MessagesState):
    return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}

# 构建图
builder = StateGraph(MessagesState)

builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))

builder.add_edge(START, "assistant")
builder.add_conditional_edges(
    "assistant",
    tools_condition,   # 如果 LLM 要调用工具 → tools；否则 → END
)
builder.add_edge("tools", "assistant")   # 工具结果返回给 LLM

# 关键：创建检查点器 + 编译时指定断点
memory = MemorySaver()
graph = builder.compile(
    interrupt_before=["tools"],   # 在 tools 节点执行前暂停
    checkpointer=memory             # 必须提供检查点器
)

Step 3: Run the graph until the first breakpoint

from langchain_core.messages import HumanMessage

# 初始输入：让 Agent 计算 2 乘以 3
initial_input = {"messages": HumanMessage(content="Multiply 2 and 3")}

# 指定 thread（对话上下文）
thread = {"configurable": {"thread_id": "1"}}

# 以流式模式运行图，直到遇到断点自动停止
for event in graph.stream(initial_input, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

The figure outputs Human Message and AI Message (AI decides to callmultiplytool) and thenauto-pause——The tools node execution is not continued. At this point, program control is returned to you, and you can view the status and make decisions.

5Check status after pause: graph.get_state(config)

Get the current graph status

After the picture is paused, you can use it at any timegraph.get_state(config)to see a complete current status snapshot. This method returns aStateSnapshotObject that contains the current State content and a lot of useful metadata.

# 获取当前状态快照
state = graph.get_state(thread)

# 查看下一个待执行的节点
print(state.next)

state.nextReturns a tuple containing the names of the nodes to be executed next in the graph. when it shows('tools',)When, it means: the graph has indeed been paused at the tools nodebefore execution, which is exactly what we setinterrupt_before=["tools"]effect.

The complete structure of StateSnapshot

# state 对象包含多个有用属性
state = graph.get_state(thread)

state.values        # 当前 State 的全部字段值（MessagesState 里就是 messages 列表）
state.next          # 下一步要执行的节点（元组），暂停时非空，结束时为空元组
state.config        # 当前检查点的配置信息（包含 checkpoint_id）
state.metadata      # 元数据（step 计数、source 等）
state.created_at    # 检查点创建时间
state.parent_config # 上一个检查点的配置（用于时间旅行）

# 查看所有消息
for msg in state.values["messages"]:
    msg.pretty_print()

Actual inspection of tool call content

After pausing, you can deeply check what tools the AI intends to call and what parameters it passes. This is the basis for making approval decisions:

# 获取最后一条消息（AI 的工具调用请求）
last_msg = state.values["messages"][-1]

# 查看工具调用详情
if hasattr(last_msg, 'tool_calls'):
    for tc in last_msg.tool_calls:
        print(f"工具名称: {tc['name']}")
        print(f"调用参数: {tc['args']}")
        print(f"调用 ID: {tc['id']}")

Now you know that the Agent intends to callmultiply(a=2, b=3). As a human approver, you can determine whether the call is reasonable and then decide whether to release it.

6Resume execution: graph.invoke(None, config)

Restore with None as input

This is one of the most elegant designs in the breakpoint mechanism:putNonePassed as inputinvokeorstream, the graph will continue execution from the previous checkpoint without re-entering any data.

# 人类审批后，继续执行图
# 传入 None 表示"从检查点恢复，不提供新输入"
for event in graph.stream(None, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

Pay attention to the order of output:

• The first Ai Message: LangGraph re-emitted the AI message (including tool call request) saved before the breakpoint. This is a normal phenomenon of state replay.
• Tool Message:tools node is actually executed,multiply(2, 3)returned6
• Article 2 Ai Message: LLM got the tool results and gave the final natural language answer

Refuse to execute (do not restore)

If the human approver decidesrejectThis time the tool is called, just don't callinvoke(None, ...), just end the process directly. The checkpoint for this conversation remains in the MemorySaver, but there will be no further execution.

# 完整的审批流程示例
initial_input = {"messages": HumanMessage(content="Multiply 2 and 3")}
thread = {"configurable": {"thread_id": "2"}}

# 运行直到断点
for event in graph.stream(initial_input, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

# 等待人类审批
user_approval = input("Do you want to call the tool? (yes/no): ")

if user_approval.lower() == "yes":
    # 审批通过：继续执行
    for event in graph.stream(None, thread, stream_mode="values"):
        event['messages'][-1].pretty_print()
else:
    # 审批拒绝：不继续执行
    print("Operation cancelled by user.")

7The difference between interrupt_before vs interrupt_after

Comparison of two breakpoint modes

interrupt_before: Approval tool call (Notebook example)

# 在 send_email 节点执行前暂停
graph = builder.compile(
    interrupt_before=["send_email"],
    checkpointer=memory
)

# 图暂停时，state 里包含 AI 拟好的邮件内容（工具调用参数）
# 人类可以查看收件人、主题、正文，决定是否发送
state = graph.get_state(thread)
tool_call = state.values["messages"][-1].tool_calls[0]
print(f"收件人: {tool_call['args']['to']}")
print(f"主题: {tool_call['args']['subject']}")
# → 人类确认后：graph.invoke(None, thread)

interrupt_after: Audit tool results

# 在 web_search 节点执行后暂停
graph = builder.compile(
    interrupt_after=["web_search"],
    checkpointer=memory
)

# 图暂停时，搜索已经完成，state 里有 ToolMessage（搜索结果）
# 人类可以检查来源、筛选掉不可信的结果
state = graph.get_state(thread)
# 最后一条是 ToolMessage（工具执行结果）
search_result = state.values["messages"][-1].content
print(f"搜索结果: {search_result}")
# → 确认结果可信后：graph.invoke(None, thread) 让 LLM 继续总结

Timing comparison of the two modes

8Complete user approval process integrated with LangGraph API

Complete local approval code mode

Integrate all the previous knowledge into a complete and usable approval process template:

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.messages import HumanMessage

# ─── 构建图（假设 tools、llm_with_tools、assistant 已定义）───
memory = MemorySaver()
graph = builder.compile(
    interrupt_before=["tools"],
    checkpointer=memory
)

def run_with_approval(user_question: str, thread_id: str):
    """完整的带审批的 Agent 执行函数"""
    thread = {"configurable": {"thread_id": thread_id}}
    initial_input = {"messages": HumanMessage(content=user_question)}

    print("=== Agent 开始执行 ===")

    # 第一阶段：运行到断点
    for event in graph.stream(initial_input, thread, stream_mode="values"):
        event['messages'][-1].pretty_print()

    # 检查是否暂停在断点（如果 next 非空，说明还有待执行的节点）
    state = graph.get_state(thread)
    if not state.next:
        print("=== 图已执行完毕（无需工具调用）===")
        return

    print(f"\n=== 图已暂停，下一步: {state.next} ===")

    # 显示工具调用详情供人类审查
    last_msg = state.values["messages"][-1]
    if hasattr(last_msg, 'tool_calls'):
        print("\n待审批的工具调用：")
        for tc in last_msg.tool_calls:
            print(f"  工具: {tc['name']}")
            print(f"  参数: {tc['args']}")

    # 等待人类审批
    approval = input("\n是否批准执行工具？(yes/no): ")

    if approval.lower() == "yes":
        print("\n=== 审批通过，继续执行 ===")
        for event in graph.stream(None, thread, stream_mode="values"):
            event['messages'][-1].pretty_print()
    else:
        print("=== 操作已被用户取消 ===")

# 使用示例
run_with_approval("Multiply 2 and 3", thread_id="session-001")

Using breakpoints via LangGraph API (SDK)

In production environments, graphs are typically deployed as LangGraph services (vialanggraph devStart a local server, or deploy to LangSmith Studio). At this point you passLangGraph SDKWhen interacting with graphs, the logic of breakpoints is exactly the same:

from langgraph_sdk import get_client
from langchain_core.messages import HumanMessage

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

# 创建新 thread
thread = await client.threads.create()

# 流式运行图，并在 API 层面指定断点
# 注意：也可以直接在 compile() 里设置，这里是另一种方式
initial_input = {"messages": HumanMessage(content="Multiply 2 and 3")}

async for chunk in client.runs.stream(
    thread["thread_id"],
    assistant_id="agent",
    input=initial_input,
    stream_mode="values",
    interrupt_before=["tools"],      # 在 API 层面设置断点
):
    print(f"Event type: {chunk.event}")
    messages = chunk.data.get('messages', [])
    if messages:
        print(messages[-1])

# 审批后继续（传入 None）
async for chunk in client.runs.stream(
    thread["thread_id"],
    "agent",
    input=None,                        # None = 从检查点恢复
    stream_mode="values",
    interrupt_before=["tools"],
):
    messages = chunk.data.get('messages', [])
    if messages:
        print(messages[-1])

Summary of core concepts