AI项目博客LangGraph 提供了两种中断机制：
- 动态中断：在图的任意节点中调用
interrupt()函数实现
它可以放在代码的任意位置，并且可以根据应用逻辑设置条件触发，所以是动态的。
动态中断提供了人机交互接口，使得调用者可以人为干预计算图的运行，是业务逻辑的一部分。
- 静态中断：在编译或调用状态图时通过
interrupt_before和interrupt_after参数设置断点
它是在运行前确定的，不能根据业务逻辑条件触发，所以是静态的。
静态中断主要用于调试，不是业务逻辑的一部分。
8.1. 动态中断
8.1.1. 概述
LangGraph 的动态中断机制允许用户在图执行过程中设置暂停点，等待外部输入后再继续执行，提供了人机交互接口。
当中断触发时，LangGraph 会通过持久化机制保存当前图状态，并无限期等待直到用户恢复执行。
中断通过在任意图节点中调用 interrupt() 函数实现，该函数可以接收任何可 JSON 序列化的值，并将该值暴露给调用方。
中断触发后，调用方根据 interrupt() 暴露的信息生成反馈，重新调用图时通过 Command 将反馈传递给计算图，恢复图的运行。
恢复运行后，调用方通过 Command 传递的反馈会作为 interrupt() 函数的返回值，参与后续计算。
8.1.2. 启用中断
-
配置检查点存储器
-
设置
thread_id
如果要确保中断后可以恢复执行，就需要保存运行图的完整状态，所以必须启用可恢复执行。
- 在需要中断的位置调用
interrupt()
8.1.3. 恢复中断
满足以下要求：
-
基于相同的配置再次调用计算图
-
将输入替换为
Command()实例即可恢复运行。
LangGraph 会从中断节点继续运行，该节点会被再次执行。
通过 Command 实例的 resume 属性将用户反馈传递给计算图，中断节点重新运行时，传递给 resume 属性的值将会作为 interrupt() 函数的返回值。
8.1.4. 常见使用模式
根据中断发生的位置、业务场景以及执行结构，动态中断可以衍生出多种常见的使用模式。本节将介绍以下模式：
- 基础 HITL 模式：状态图触发一次中断，获取人类输入后继续执行。
- 多个并行中断：多个并行任务分别产生中断，并根据 中断
ID接收各自的恢复数据。 - 审批模式：根据人类审批（批准、拒绝或其它处理方式），决定后续执行路径。
- 审核与编辑模式：将模型生成的内容交给人类检查，并允许人类直接修改后继续处理。
- 工具执行审批模式：在调用工具或执行具有副作用的操作之前，由人类确认是否允许执行。
- 单节点串行中断模式：在同一个节点内多次触发中断，上一个中断恢复后才能触发下一个。
- 人类输入验证模式：对人类输入进行校验；当输入不符合要求时，再次中断并要求重新输入。
需要注意，这些模式并不是完全互斥的。例如，工具执行审批本质上也是审批模式的一种，只是它专门应用于工具调用场景。
8.1.4.1. 基础HITL模式
8.1.4.1.1. 什么是HITL
HITL(Human In The Loop) 即人在环，是一种经典的人机协同设计模式，它允许计算图在运行过程中暂停，等待人工审批、修正意见或补充信息，然后从暂停位置恢复执行。
8.1.4.1.2. 基础HITL模式的实现
调用 interrupt() 可以暂停状态图的执行，并等待外部输入后恢复运行。因此它是实现 HITL 的核心机制。
不过，仅仅调用 interrupt() 并不一定构成 HITL。只有当中断后的决策、输入或修改确实由人类完成时，才能称为 HITL；如果恢复数据完全由程序自动提供，本质上仍然是普通的中断与恢复机制。
示例如下
from typing import TypedDict
from langgraph.graph import StateGraph,START,END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver
#1. 声明状态
class OverAllState(TypedDict):
username:str
#2. 声明节点
def node_a(state:OverAllState)->OverAllState:
username = interrupt("请输入您的姓名")
return {
"username":username
}
#3. 构建图
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a",node_a)
builder.add_edge(START,"node_a")
builder.add_edge("node_a",END)
#4. 想要使用中断 => 必须配置检查点
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
from IPython.display import display
display(graph)
config = {"configurable":{"thread_id":"123"}}
interrupt_res = graph.invoke({},config=config)
print(interrupt_res)
prompt = interrupt_res['__interrupt__'][0].value
print(prompt)
username = input(prompt)
# 5. 恢复图的中断执行
resume_res = graph.invoke(Command(resume=username),config=config)
print(resume_res)
运行结果如下
============================== -> interrupt_res <- ==============================
{'__interrupt__': [Interrupt(value='请输入您的姓名', id='3543beca616d469fe1ccd35250ef5acf')]}
============================== -> resumed_res <- ==============================
{'username': '小黄'}
8.1.4.2. 多个并行中断
示例如下
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import InMemorySaver
class OverAllState(TypedDict):
username: str
age: int
def node_a(state: OverAllState) -> OverAllState:
username = interrupt("请输入您的姓名")
return {
"username": username
}
def node_b(state: OverAllState) -> OverAllState:
age = interrupt("请输入您的年龄")
return {
"age": age
}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge(START, "node_a")
builder.add_edge(START, "node_b")
builder.add_edge("node_a", END)
builder.add_edge("node_b", END)
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
from IPython.display import display
display(graph)
config = {"configurable": {"thread_id": "123"}}
interrupted_res = graph.invoke({}, config=config)
print('=' * 30, '-> interrupt_res <-', '=' * 30)
print(interrupted_res)
# 构建 中断ID -> 恢复指令 的映射
resume_map = {}
for i in interrupted_res['__interrupt__']:
user_input = input(f"{i.value}: ")
if "年龄" in i.value:
resume_map[i.id] = int(user_input)
else:
resume_map[i.id] = user_input
resumed_res = graph.invoke(Command(resume=resume_map), config=config)
print('=' * 30, '-> resumed_res <-', '=' * 30)
print(resumed_res)
运行结果如下
============================== -> interrupt_res <- ==============================
{'__interrupt__': [Interrupt(value='请输入您的年龄', id='2770f00729006c756d481fffa712bb27'), Interrupt(value='请输入您的姓名', id='db7e81e1d4f5cab611f07e0469e99854')]}
============================== -> resumed_res <- ==============================
{'username': '小黄', 'age': 19}
8.1.4.3. 审批模式
示例如下
from time import sleep
from typing import TypedDict, Literal
from langchain_core.messages import HumanMessage
from langgraph.graph import StateGraph,START,END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver
from langchain_deepseek import ChatDeepSeek
from dotenv import load_dotenv
load_dotenv(override=True)
model =ChatDeepSeek(
model="deepseek-v4-flash",
extra_body={
"thinking":{
"type":"disabled"
}
}
)
#1. 声明状态
class OverAllState(TypedDict):
topic:str
poem:str
is_approved:bool
#2. 声明节点
def approve_node(state:OverAllState) -> Command[Literal["llm_node","default_node"]]:
is_approved = interrupt("是否同意调用模型?")
goto = "llm_node" if is_approved else "default_node"
return Command(
goto=goto,
update={"is_approved":is_approved}
)
def llm_node(state:OverAllState) -> OverAllState:
topic = state["topic"]
res = model.invoke([HumanMessage(content=f"帮我写一首关于{topic}主题的七言绝句,只写诗句,不需要赏析")]).content
return {
"poem":res
}
def default_node(state:OverAllState) -> OverAllState:
return {
"poem":"请求被拒绝"
}
#3. 构建图
builder = StateGraph(state_schema=OverAllState)
builder.add_node("approve_node",approve_node)
builder.add_node("llm_node",llm_node)
builder.add_node("default_node",default_node)
builder.add_edge(START,"approve_node")
builder.add_edge("llm_node",END)
builder.add_edge("default_node",END)
#4. 添加检查点后端
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
from IPython.display import display
display(graph)
#5. 首次执行 -> 触发中断
config = {"configurable":{"thread_id":"234"}}
interrupt_res = graph.invoke({"topic":"菊花"},config=config)
print(interrupt_res)
#6. 人工审核
user_approved = input("是否同意调用模型?(y/n)").strip().lower() == 'y'
# 重新调用图
approved_res = graph.invoke(Command(resume=user_approved),config=config)
print(approved_res)
#7. 审核不通过
config1 = {"configurable":{"thread_id":"456"}}
interrupt_res = graph.invoke({"topic":"牡丹花"},config=config1)
print(interrupt_res)
user_approved = input("是否同意调用模型?(y/n)").strip().lower() == 'y'
# 重新调用图
approved_res = graph.invoke(Command(resume=user_approved),config=config1)
print(approved_res)
运行结果如下
{'topic': '菊花', '__interrupt__': [Interrupt(value='是否同意调用模型?', id='91edf41ecf736a242c18fd61d40ad5b4')]}
{'topic': '菊花', 'poem': '《咏菊》\n西风猎猎卷霜华，独抱秋心隐士家。\n冷艳不争桃李色，寒香偏入野人茶。', 'is_approved': True}
{'topic': '牡丹花', '__interrupt__': [Interrupt(value='是否同意调用模型?', id='47958afc1277b7ce6bf850a5f27f2e46')]}
{'topic': '牡丹花', 'poem': '请求被拒绝', 'is_approved': False}
8.1.4.4. 审核与编辑模式
示例如下
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langchain.messages import HumanMessage
from langchain_deepseek import ChatDeepSeek
from langgraph.checkpoint.memory import InMemorySaver
from dotenv import load_dotenv
load_dotenv(override=True)
model = ChatDeepSeek(
model="deepseek-v4-flash",
extra_body={
"thinking": {
"type": "disabled"
}
}
)
class OverAllState(TypedDict):
topic: str
poem: str
reviewed_poem: str
def llm_node(state: OverAllState) -> OverAllState:
topic = state['topic']
res = model.invoke([HumanMessage(content=f"帮我写一首关于 {topic} 的七言绝句，只给出诗句，不要赏析")]).content
return {
"poem": res
}
def review_node(state: OverAllState) -> OverAllState:
reviewed_poem = interrupt({
"instruction": "请审核并修改大模型生成的七言绝句",
"poem": state['poem']
})
return {
"reviewed_poem": reviewed_poem
}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("llm_node", llm_node)
builder.add_node("review_node", review_node)
builder.add_edge(START, "llm_node")
builder.add_edge("llm_node", "review_node")
builder.add_edge("review_node", END)
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
from IPython.display import display
display(graph)
config = {"configurable": {"thread_id": "review_test"}}
interrupted_res = graph.invoke({"topic": "布偶猫"}, config=config)
print('=' * 30, '-> interrupt_res <-', '=' * 30)
print(interrupted_res)
print("原始诗句：")
print(interrupted_res['__interrupt__'][0].value['poem'])
user_review = input("请审核并修改诗句（直接回车保留原诗）: ")
if not user_review.strip():
user_review = interrupted_res['__interrupt__'][0].value['poem']
reviewed_res = graph.invoke(Command(resume=user_review), config=config)
print('=' * 30, '-> reviewed_res <-', '=' * 30)
print(reviewed_res)
运行结果如下
============================== -> interrupt_res <- ==============================
{'topic': '布偶猫', 'poem': '《布偶猫》\n冰眸玉骨雪团身，蓝影衔春卧月轮。\n偶启朱唇呼入梦，云鬟一枕醉红尘。', '__interrupt__': [Interrupt(value={'instruction': '请审核并修改大模型生成的七言绝句', 'poem': '《布偶猫》\n冰眸玉骨雪团身，蓝影衔春卧月轮。\n偶启朱唇呼入梦，云鬟一枕醉红尘。'}, id='de13581234c57fb8b5d9620295b8239e')]}
============================== -> reviewed_res <- ==============================
{'topic': '布偶猫', 'poem': '《布偶猫》\n冰眸玉骨雪团身，蓝影衔春卧月轮。\n偶启朱唇呼入梦，云鬟一枕醉红尘。', 'reviewed_poem': '已审核：《布偶猫》\n冰眸玉骨雪团身，蓝影衔春卧月轮。\n偶启朱唇呼入梦，云鬟一枕醉红尘。'}
8.1.4.5. 工具执行审批模式
示例如下
from typing import Literal
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.graph.message import MessagesState
from langgraph.checkpoint.memory import InMemorySaver
from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import tool
from langchain_deepseek import ChatDeepSeek
from loguru import logger
from dotenv import load_dotenv
load_dotenv(override=True)
@tool(parse_docstring=True)
def get_weather(city: str) -> str:
"""
查询指定城市的当日天气
Args:
city: 城市名称
"""
is_approved = interrupt({
"action": "get_weather",
"question": "是否同意查询天气？"
})
logger.info("is_approved: {}", is_approved)
if is_approved:
return f"{city} 今天天气不错"
else:
return "用户拒绝查询天气"
tools_by_name = {"get_weather": get_weather}
model = ChatDeepSeek(
model="deepseek-v4-flash",
extra_body={
"thinking": {
"type": "disabled"
}
}
)
model_with_tools = model.bind_tools([get_weather])
def llm_node(state: MessagesState) -> MessagesState:
messages = state['messages']
response = model_with_tools.invoke(messages)
return {
"messages": [response]
}
def tool_node(state: MessagesState) -> MessagesState:
last_msg = state['messages'][-1]
tool_msgs = []
for tool_call in last_msg.tool_calls:
tool = tools_by_name[tool_call["name"]]
logger.info("工具 {} 被调用, 对应的 tool_call: {}", tool_call["name"], tool_call)
tool_res = tool.invoke(tool_call["args"])
tool_msg = ToolMessage(
name = tool_call["name"],
content = tool_res,
tool_call_id = tool_call["id"]
)
tool_msgs.append(tool_msg)
return {
"messages": tool_msgs
}
def router(state: MessagesState) -> Literal["tool_node", END]:
if state['messages'][-1].tool_calls:
return "tool_node"
return END
builder = StateGraph(state_schema=MessagesState)
builder.add_node("llm_node", llm_node)
builder.add_node("tool_node", tool_node)
builder.add_edge(START, "llm_node")
builder.add_conditional_edges("llm_node", router, path_map=["tool_node", END])
builder.add_edge("tool_node", "llm_node")
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
from IPython.display import display
display(graph)
config = {"configurable": {"thread_id": "tool_test"}}
interrupted_res = graph.invoke({"messages": [HumanMessage("今天北京天气如何？")]}, config=config)
print('=' * 30, '-> interrupt_res <-', '=' * 30)
for msg in interrupted_res['messages']:
msg.pretty_print()
print('=' * 30, '-> interrupt_info <-', '=' * 30)
print(interrupted_res['__interrupt__'])
user_approved = input("是否同意查询天气？(y/n): ").strip().lower() == 'y'
approved_res = graph.invoke(Command(resume=user_approved), config=config)
print('=' * 30, '-> approved_res <-', '=' * 30)
for msg in approved_res['messages']:
msg.pretty_print()
运行结果如下
2026-06-16 14:51:50.662 | INFO | __main__:tool_node:59 - 工具 get_weather 被调用, 对应的 tool_call: {'name': 'get_weather', 'args': {'city': '北京'}, 'id': 'call_00_EGxsgveRV6mFkXsTtN4H9369', 'type': 'tool_call'}
============================== -> interrupt_res <- ==============================
================================ Human Message =================================
今天北京天气如何？
================================== Ai Message ==================================
好的，我来查询一下今天北京的天气情况。
Tool Calls:
get_weather (call_00_EGxsgveRV6mFkXsTtN4H9369)
Call ID: call_00_EGxsgveRV6mFkXsTtN4H9369
Args:
city: 北京
============================== -> interrupt_info <- ==============================
[Interrupt(value={'action': 'get_weather', 'question': '是否同意查询天气？'}, id='907128d6bef87a8473833ec0c90589ea')]
2026-06-16 14:51:50.675 | INFO | __main__:tool_node:59 - 工具 get_weather 被调用, 对应的 tool_call: {'name': 'get_weather', 'args': {'city': '北京'}, 'id': 'call_00_EGxsgveRV6mFkXsTtN4H9369', 'type': 'tool_call'}
2026-06-16 14:51:50.677 | INFO | __main__:get_weather:27 - is_approved: True
============================== -> approved_res <- ==============================
================================ Human Message =================================
今天北京天气如何？
================================== Ai Message ==================================
好的，我来查询一下今天北京的天气情况。
Tool Calls:
get_weather (call_00_EGxsgveRV6mFkXsTtN4H9369)
Call ID: call_00_EGxsgveRV6mFkXsTtN4H9369
Args:
city: 北京
================================= Tool Message =================================
Name: get_weather
北京 今天天气不错
================================== Ai Message ==================================
今天北京的天气**不错**哦！☀️ 是个适合出行的好天气。如果需要更详细的天气信息（如温度、风力等），可以随时告诉我，我帮你进一步查询！
8.1.4.6. 单节点串行中断模式
示例如下
from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
class OverAllState(TypedDict):
username: str # 姓名
age: int # 年龄
gender: Literal["male", "female"] # 性别
def get_info_node(state: OverAllState) -> OverAllState:
username = interrupt("请输入您的用户名：")
age = interrupt("请输入您的年龄：")
gender = interrupt("请输入您的性别：(male/female)")
return {
"username": username,
"age": age,
"gender": gender
}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("get_info_node", get_info_node)
builder.add_edge(START, "get_info_node")
builder.add_edge("get_info_node", END)
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "seq_interrupt_test"}}
username_interrupted_res = graph.invoke({}, config=config)
print('=' * 30, '-> username_interrupted_res <-', '=' * 30)
print(username_interrupted_res)
user_name = input("请输入您的用户名：")
age_interrupted_res = graph.invoke(Command(resume=user_name), config=config)
print('=' * 30, '-> age_interrupted_res <-', '=' * 30)
print(age_interrupted_res)
user_age = input("请输入您的年龄：")
gender_interrupted_res = graph.invoke(Command(resume=int(user_age)), config=config)
print('=' * 30, '-> gender_interrupted_res <-', '=' * 30)
print(gender_interrupted_res)
user_gender = input("请输入您的性别：(male/female): ")
resumed_res = graph.invoke(Command(resume=user_gender), config=config)
print('=' * 30, '-> resumed_res <-', '=' * 30)
print(resumed_res)
运行结果如下
============================== -> username_interrupted_res <- ==============================
{'__interrupt__': [Interrupt(value='请输入您的用户名：', id='0359f8a0b1ca357d2d092d5c55d80ec1')]}
============================== -> age_interrupted_res <- ==============================
{'__interrupt__': [Interrupt(value='请输入您的年龄：', id='0359f8a0b1ca357d2d092d5c55d80ec1')]}
============================== -> gender_interrupted_res <- ==============================
{'__interrupt__': [Interrupt(value='请输入您的性别：(male/female)', id='0359f8a0b1ca357d2d092d5c55d80ec1')]}
============================== -> resumed_res <- ==============================
{'username': '小黄', 'age': 15, 'gender': 'male'}
中断恢复时，整个被中断的节点函数都会重新运行。
单个节点中串行地多次调用 interrupt() 函数时，检查点存储器会记录历史的 resume 信息，LangGraph 运行时会读取这些信息，并在 interrupt() 函数中维护索引，按照节点内调用 interrupt() 函数的顺序，逐个取出历史 resume 的值，并将它们作为 interrupt() 函数的返回值。
所以，已经被恢复的 interrupt() 不会被重复触发。并且，由此可以推断，我们要保证历史 resume 可以被正确应用，就应保证中断恢复前后的多次 interrupt() 相对顺序保持不变。
当历史 resume 耗尽后，本次恢复运行时传入的 resume 会作为本次中断的返回值，然后节点函数继续运行。
所有中断都触发并恢复后，计算图正常结束。
8.1.5. 使用规范
8.1.5.1. 不要用try/catch包裹interrupt()调用
中断的触发是通过抛出 GraphInterrupt 异常实现的，如果用 try/catch 包裹，则底层运行时无法感知断点，不会中断计算图。
正确用法
def node_a(state: State):
# ✅ 正确用法：先打断点，再处理异常
interrupt("What's your name?")
try:
fetch_data() # 这一步可能失败
except Exception as e:
print(e)
return state
错误用法
def node_a(state: State):
# ❌ 错误用法，用 try/catch 包裹 interrupt()，这将会导致 GraphInterrupt 被捕获而无法触发中断
try:
interrupt("What's your name?")
fetch_data() # 这一步可能失败
except Exception as e:
print(e)
return state
8.1.5.2. 不要更改单个节点内interrupt的调用顺序
上文提到：
- 恢复运行时整个节点函数都会重新运行，而非精确地从断点继续
- 同一节点中存在多个断点时：历史恢复记录被记录在检查点中，并在恢复运行时按顺序加载
一旦中断恢复前后的断点顺序、数量不完全一致，将会导致语义混乱，程序执行结果无法满足预期。
这条规范就是为了避免这样的情况。
正确用法
def node_a(state: State):
# ✅ 正确用法，断点顺序固定
name = interrupt("What's your name?")
age = interrupt("What's your age?")
city = interrupt("What's your city?")
return {
"name": name,
"age": age,
"city": city
}
错误用法
1. 可能跳过其中部分断点
def node_a(state: State):
# ❌ 错误用法，以状态为条件，可能跳过部分断点
name = interrupt("What's your name?")
# 第一次运行到此处时，可能不会触发分支内的断点
# 恢复时，resume 本应作为 city，但如果分支内断点触发，将被错误地作为 age，导致语义混乱
if state.get("needs_age"):
age = interrupt("What's your age?")
city = interrupt("What's your city?")
return {"name": name, "city": city}
断点以状态为条件，如果中断恢复前后状态发生了变化，断点的顺序和数量可能发生变化，导致语义混乱。
2. 基于不确定的数据循环，并在其中设置断点
def node_a(state: State):
# ❌ 错误用法：基于不确定的数据循环
# 每次调用时的断点数量可能不同
results = []
for item in state.get("dynamic_list", []): # List might change between runs
result = interrupt(f"Approve {item}?")
results.append(result)
return {"results": results}
基于状态中可变的值循环，并在循环中设置断点，如果中断恢复前后状态发生变化，断点数量可能变化，导致语义混乱。
8.1.5.3. 不要在interrupt()中传递复杂类型
LangGraph 运行时会将 interrupt() 函数接收到的参数经过 JSON 序列化 之后传递给调用者。
如果传递不支持 JSON 序列化 的复杂类型，如函数，将会抛出异常，如下所示
TypeError: Type is not msgpack serializable: Interrupt
正确用法
1. 简单类型
def node_a(state: State):
# ✅ 正确用法：传递可以序列化的简单类型数据
name = interrupt("What's your name?")
count = interrupt(42)
approved = interrupt(True)
return {"name": name, "count": count, "approved": approved}
2. 结构化数据
def node_a(state: State):
# ✅ 正确用法：传递仅包含基本数据类型的可序列化字典
response = interrupt({
"question": "Enter user details",
"fields": ["name", "email", "age"],
"current_values": state.get("user", {})
})
return {"user": response}
错误用法
1. 函数
def validate_input(value):
return len(value) > 0
def node_a(state: State):
# ❌ 错误用法：传递函数
# 函数不支持 JSON 序列化
response = interrupt({
"question": "What's your name?",
"validator": validate_input # 此处会导致失败
})
return {"name": response}
2. 类的实例
class DataProcessor:
def __init__(self, config):
self.config = config
def node_a(state: State):
processor = DataProcessor({"mode": "strict"})
# ❌ 错误用法：传递类的实例
# 此处的自定义实例不支持 JSON 序列化
response = interrupt({
"question": "Enter data to process",
"processor": processor # 这将会导致失败
})
return {"result": response}
8.1.5.4. 断点之前的副作用操作必须是幂等的
- 副作用操作：修改了外部环境的操作，如
- 写数据库
- 写文件
- 发送邮件或消息
- 幂等性：多次执行的结果等价于一次执行。
我们知道，中断恢复时，断点所在的函数会被重复执行，所以，如果断点之前存在不满足幂等性的副作用操作，将会导致多次调用结果不一致。
如：写数据操作不满足幂等性，可能导致写入多条重复记录。
正确用法
1. 在断点之前使用满足幂等性的操作
def node_a(state: State):
# ✅ 正确用法：使用幂等的 upsert 操作
# 多次运行结果一致
db.upsert_user(
user_id=state["user_id"],
status="pending_approval"
)
approved = interrupt("Approve this change?")
return {"approved": approved}
upsert 的语义是：写入或更新，如果有相同记录则覆盖更新，否则新增。满足幂等性。
2. 将副作用操作放在断点之后
def node_a(state: State):
# ✅ 正确用法：将副作用操作放在断点之后
# 确保副作用操作只运行一次
approved = interrupt("Approve this change?")
if approved:
db.create_audit_log(
user_id=state["user_id"],
action="approved"
)
return {"approved": approved}
上述代码将副作用操作置于断点之后，之后中断触发并恢复运行之后，副作用操作才会被执行，后者只会被执行一次。
3. 将副作用操作和断点置于不同的节点
def approval_node(state: State):
# ✅ 正确用法：当前节点只做中断
approved = interrupt("Approve this change?")
return {"approved": approved}
def notification_node(state: State):
# ✅ 正确用法：副作用操作发生在独立的节点中
# 当前节点独立于中断，所以副作用操作，不会因为中断恢复而重复执行
if (state.approved):
send_notification(
user_id=state["user_id"],
status="approved"
)
return state
这种方式更彻底，将中断和副作用操作完全隔离。
错误用法
1. 在断点之前向数据库中新增数据
def node_a(state: State):
# ❌ 错误用法：在断点之前向数据库中新增数据
# 中断恢复会导致数据重复
audit_id = db.create_audit_log({
"user_id": state["user_id"],
"action": "pending_approval",
"timestamp": datetime.now()
})
approved = interrupt("Approve this change?")
return {"approved": approved, "audit_id": audit_id}
2. 将数据追加到已存在的列表
def node_a(state: State):
# ❌ 错误用法：在断点之前向列表追加数据
# 这将会在中断恢复时导致列表数据重复
db.append_to_history(state["user_id"], "approval_requested")
approved = interrupt("Approve this change?")
return {"approved": approved}
8.1.6. 中断触发和恢复前后的检查点
8.1.6.1. 基础HITL模式
8.1.6.1.1. 示例代码
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import InMemorySaver
class OverAllState(TypedDict):
username: str
age: int
def node_a(state: OverAllState) -> OverAllState:
username = interrupt("请输入您的姓名")
age = interrupt("请输入您的年龄")
return {
"username": username,
"age": age
}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", END)
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
config = {"configurable": {"thread_id": "123"}}
username_interrupted_res = graph.invoke({}, config=config)
print('=' * 30, '-> username_interrupted_res <-', '=' * 30)
print(username_interrupted_res)
username_interrupted_his = list(graph.get_state_history(config=config))
print('=' * 30, '-> username_interrupted_his <-', '=' * 30)
print(username_interrupted_his)
age_interrupted_res = graph.invoke(Command(resume="小黄"), config=config)
print('=' * 30, '-> age_interrupted_res <-', '=' * 30)
print(age_interrupted_res)
age_interrupted_his = list(graph.get_state_history(config=config))
print('=' * 30, '-> age_interrupted_his <-', '=' * 30)
print(age_interrupted_his)
resumed_res = graph.invoke(Command(resume=123), config=config)
print('=' * 30, '-> resumed_res <-', '=' * 30)
print(resumed_res)
resumed_his = list(graph.get_state_history(config=config))
print('=' * 30, '-> resumed_his <-', '=' * 30)
print(resumed_his)
8.1.6.1.2. 运行结果分析
8.1.6.1.2.1. 第一次中断
调用结果
============================== -> username_interrupted_res <- ==============================
{'__interrupt__': [Interrupt(value='请输入您的姓名', id='488d1fa95a4053d41ab09786f5260ae9')]}
历史检查点
============================== -> username_interrupted_his <- ==============================
[
StateSnapshot(
values={},
next=('node_a',),
config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a255-4aa0-631a-8000-12a45e5b02a0'
}
},
metadata={
'source': 'loop',
'step': 0,
'parents': {}
},
created_at='2026-06-17T08:19:59.195818+00:00',
parent_config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a255-4a9d-68e8-bfff-3573d795ed80'
}
},
tasks=(
PregelTask(
id='a7e05e86-30c1-a590-a7bf-e3542eb53d4b',
name='node_a',
path=('__pregel_pull', 'node_a'),
error=None,
interrupts=(
Interrupt(
value='请输入您的姓名',
id='488d1fa95a4053d41ab09786f5260ae9'
),
),
state=None,
result=None
),
),
interrupts=(
Interrupt(
value='请输入您的姓名',
id='488d1fa95a4053d41ab09786f5260ae9'
),
)
),
StateSnapshot(
values={},
next=('__start__',),
config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a255-4a9d-68e8-bfff-3573d795ed80'
}
},
metadata={
'source': 'input',
'step': -1,
'parents': {}
},
created_at='2026-06-17T08:19:59.194743+00:00',
parent_config=None,
tasks=(
PregelTask(
id='173d6621-cd7b-3edd-a3d5-8d1c3bb8dc53',
name='__start__',
path=('__pregel_pull', '__start__'),
error=None,
interrupts=(),
state=None,
result={}
),
),
interrupts=()
)
]
当前工作流只有一个节点，所以只有一个真正执行用户自定义节点的超步，对应的超步编号为 1
中断触发时，检查点存储器中记录的最新检查点超步为 0，中断信息只能和这个超步绑定
中断信息存了两份：
-
StateSnapshot的tasks属性下记录的PregelTask实例的interrupts字段记录了当前任务触发的中断信息。 -
StateSnapshot的interrupts属性记录了当前超步发生的所有中断
当存在多个并行中断时
- 中断信息被记录在各自所属任务的
PregelTask实例中 - 同时也会被汇总在
StateSnapshot的tasks属性中
8.1.6.1.2.2. 第一次恢复
调用结果
============================== -> age_interrupted_res <- ==============================
{'__interrupt__': [Interrupt(value='请输入您的年龄', id='488d1fa95a4053d41ab09786f5260ae9')]}
历史检查点
============================== -> age_interrupted_his <- ==============================
[
StateSnapshot(
values={},
next=('node_a',),
config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a255-4aa0-631a-8000-12a45e5b02a0'
}
},
metadata={
'source': 'loop',
'step': 0,
'parents': {}
},
created_at='2026-06-17T08:19:59.195818+00:00',
parent_config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a255-4a9d-68e8-bfff-3573d795ed80'
}
},
tasks=(
PregelTask(
id='a7e05e86-30c1-a590-a7bf-e3542eb53d4b',
name='node_a',
path=('__pregel_pull', 'node_a'),
error=None,
interrupts=(
Interrupt(
value='请输入您的年龄',
id='488d1fa95a4053d41ab09786f5260ae9'
),
),
state=None,
result={}
),
),
interrupts=(
Interrupt(
value='请输入您的年龄',
id='488d1fa95a4053d41ab09786f5260ae9'
),
)
),
StateSnapshot(
values={},
next=('__start__',),
config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a255-4a9d-68e8-bfff-3573d795ed80'
}
},
metadata={
'source': 'input',
'step': -1,
'parents': {}
},
created_at='2026-06-17T08:19:59.194743+00:00',
parent_config=None,
tasks=(
PregelTask(
id='173d6621-cd7b-3edd-a3d5-8d1c3bb8dc53',
name='__start__',
path=('__pregel_pull', '__start__'),
error=None,
interrupts=(),
state=None,
result={}
),
),
interrupts=()
)
]
checkpoint_id、任务id、中断id 和历史检查点完全一致，只是 Interrupt 实例的 value 属性变成了第二次中断的值。
由此可见，恢复运行时检查点的更新确实是覆盖写入，历史中断信息不会被保留。
并且，中断恢复时传递的 resume 在 graph.get_state_history() 时也不可见。
8.1.6.1.2.3. 第二次恢复
调用结果
============================== -> resumed_res <- ==============================
{'username': '小黄', 'age': 123}
历史检查点
============================== -> resumed_his <- ==============================
[
StateSnapshot(
values={
'username': '小黄',
'age': 123
},
next=(),
config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a255-4ab5-60bc-8001-df8746dc9400'
}
},
metadata={
'source': 'loop',
'step': 1,
'parents': {}
},
created_at='2026-06-17T08:19:59.204348+00:00',
parent_config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a255-4aa0-631a-8000-12a45e5b02a0'
}
},
tasks=(),
interrupts=()
),
StateSnapshot(
values={},
next=('node_a',),
config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a255-4aa0-631a-8000-12a45e5b02a0'
}
},
metadata={
'source': 'loop',
'step': 0,
'parents': {}
},
created_at='2026-06-17T08:19:59.195818+00:00',
parent_config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a255-4a9d-68e8-bfff-3573d795ed80'
}
},
tasks=(
PregelTask(
id='a7e05e86-30c1-a590-a7bf-e3542eb53d4b',
name='node_a',
path=('__pregel_pull', 'node_a'),
error=None,
interrupts=(
Interrupt(
value='请输入您的年龄',
id='488d1fa95a4053d41ab09786f5260ae9'
),
),
state=None,
result={
'username': '小黄',
'age': 123
}
),
),
interrupts=(
Interrupt(
value='请输入您的年龄',
id='488d1fa95a4053d41ab09786f5260ae9'
),
)
),
StateSnapshot(
values={},
next=('__start__',),
config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a255-4a9d-68e8-bfff-3573d795ed80'
}
},
metadata={
'source': 'input',
'step': -1,
'parents': {}
},
created_at='2026-06-17T08:19:59.194743+00:00',
parent_config=None,
tasks=(
PregelTask(
id='173d6621-cd7b-3edd-a3d5-8d1c3bb8dc53',
name='__start__',
path=('__pregel_pull', '__start__'),
error=None,
interrupts=(),
state=None,
result={}
),
),
interrupts=()
)
]
两个断点全部恢复运行后，编号为 1 的超步终于完成，任务实例的 result 被更新，状态被更新，最后一个超步的 values 是更新后的状态值，计算图运行完成。
8.1.6.2. 多个并行中断触发时的检查点
8.1.6.2.1. 示例代码
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import InMemorySaver
class OverAllState(TypedDict):
username: str
age: int
def node_a(state: OverAllState) -> OverAllState:
username = interrupt("请输入您的姓名")
return {
"username": username
}
def node_b(state: OverAllState) -> OverAllState:
age = interrupt("请输入您的年龄")
return {
"age": age
}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge(START, "node_a")
builder.add_edge(START, "node_b")
builder.add_edge("node_a", END)
builder.add_edge("node_b", END)
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
from IPython.display import display
display(graph)
config = {"configurable": {"thread_id": "123"}}
interrupted_res = graph.invoke({}, config=config)
print('=' * 30, '-> interrupt_res <-', '=' * 30)
print(interrupted_res)
list(graph.get_state_history(config=config))
# 6. 恢复执行
resume_map = {}
for i in interrupted_res['__interrupt__']:
user_input = input(f"{i.value}:")
if "年龄" in i.value:
resume_map[i.id] = int(user_input)
else:
resume_map[i.id] = user_input
resumed_res = graph.invoke(Command(resume=resume_map),config=config)
print(resumed_res)
list(graph.get_state_history(config=config))
8.1.6.2.2. 结果分析
计算图返回值
============================== -> interrupt_res <- ==============================
{
'__interrupt__': [
Interrupt(
value='请输入您的姓名',
id='0cae1932a74087f02711c6f1677b4b17'
),
Interrupt(
value='请输入您的年龄',
id='cd216e1ad6907c2432636ff5f32cfcf3'
)
]
}
同一个超步中并行触发的中断信息被放在列表中返回。
历史检查点列表
[
StateSnapshot(
values={},
next=(
'node_a',
'node_b'
),
config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a329-8b50-69e0-8000-7bdce1ba7379'
}
},
metadata={
'source': 'loop',
'step': 0,
'parents': {}
},
created_at='2026-06-17T09:54:56.810630+00:00',
parent_config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a329-8b4e-6085-bfff-3100dea319cc'
}
},
tasks=(
PregelTask(
id='689d8040-4b34-02a2-eb32-d901bbb3c67c',
name='node_a',
path=(
'__pregel_pull',
'node_a'
),
error=None,
interrupts=(
Interrupt(
value='请输入您的姓名',
id='0cae1932a74087f02711c6f1677b4b17'
),
),
state=None,
result=None
),
PregelTask(
id='034cbe3c-e0a6-025a-ce03-46436ce38fc3',
name='node_b',
path=(
'__pregel_pull',
'node_b'
),
error=None,
interrupts=(
Interrupt(
value='请输入您的年龄',
id='cd216e1ad6907c2432636ff5f32cfcf3'
),
),
state=None,
result=None
)
),
interrupts=(
Interrupt(
value='请输入您的姓名',
id='0cae1932a74087f02711c6f1677b4b17'
),
Interrupt(
value='请输入您的年龄',
id='cd216e1ad6907c2432636ff5f32cfcf3'
)
)
),
StateSnapshot(
values={},
next=(
'__start__',
),
config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a329-8b4e-6085-bfff-3100dea319cc'
}
},
metadata={
'source': 'input',
'step': -1,
'parents': {}
},
created_at='2026-06-17T09:54:56.809574+00:00',
parent_config=None,
tasks=(
PregelTask(
id='3df213dc-5fa0-a8aa-b1dc-428da9010be6',
name='__start__',
path=(
'__pregel_pull',
'__start__'
),
error=None,
interrupts=(),
state=None,
result={}
),
),
interrupts=()
)
]
如前所述，编号为 1 的超步执行时中断触发，此时检查点后端中存储的最新的超步编号为 0，因此中断信息记录在超步 0 的 StateSnapshot 实例中。
-
快照实例的
interrupts属性记录了并行触发的两个中断的信息。 -
并且，各自的中断信息被记录在对应的任务实例中。
8.1.6.3. 同一超步部分任务触发中断时的检查点
8.1.6.3.1. 示例代码
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
from langgraph.checkpoint.memory import InMemorySaver
class OverAllState(TypedDict):
username: str
age: int
def node_a(state: OverAllState) -> OverAllState:
username = interrupt("请输入您的姓名")
return {
"username": username
}
def node_b(state: OverAllState) -> OverAllState:
age = 12
return {
"age": age
}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_edge(START, "node_a")
builder.add_edge(START, "node_b")
builder.add_edge("node_a", END)
builder.add_edge("node_b", END)
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)
from IPython.display import display
display(graph)
config = {"configurable": {"thread_id": "123"}}
interrupted_res = graph.invoke({}, config=config)
print('=' * 30, '-> interrupt_res <-', '=' * 30)
print(interrupted_res)
list(graph.get_state_history(config=config))
8.1.6.3.2. 结果分析
计算图返回值
============================== -> interrupt_res <- ==============================
{
'age': 12,
'__interrupt__': [
Interrupt(
value='请输入您的姓名',
id='bbff88193340fd0ed196be702608e385'
)
]
}
历史检查点列表
============================== -> state_history <- ==============================
[
StateSnapshot(
values={},
next=(
'node_a',
'node_b'
),
config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a32c-b47b-6ca9-8000-79ee3c65e7ce'
}
},
metadata={
'source': 'loop',
'step': 0,
'parents': {}
},
created_at='2026-06-17T09:56:21.658113+00:00',
parent_config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a32c-b479-6766-bfff-901ed37f684d'
}
},
tasks=(
PregelTask(
id='e16690fc-c0f8-e9fc-e30d-706a64f24209',
name='node_a',
path=(
'__pregel_pull',
'node_a'
),
error=None,
interrupts=(
Interrupt(
value='请输入您的姓名',
id='bbff88193340fd0ed196be702608e385'
),
),
state=None,
result=None
),
PregelTask(
id='b628264b-d4ca-c09f-5197-6eb1c953b947',
name='node_b',
path=(
'__pregel_pull',
'node_b'
),
error=None,
interrupts=(),
state=None,
result={
'age': 12
}
)
),
interrupts=(
Interrupt(
value='请输入您的姓名',
id='bbff88193340fd0ed196be702608e385'
),
)
),
StateSnapshot(
values={},
next=(
'__start__',
),
config={
'configurable': {
'thread_id': '123',
'checkpoint_ns': '',
'checkpoint_id': '1f16a32c-b479-6766-bfff-901ed37f684d'
}
},
metadata={
'source': 'input',
'step': -1,
'parents': {}
},
created_at='2026-06-17T09:56:21.657164+00:00',
parent_config=None,
tasks=(
PregelTask(
id='5dc4ad26-1baa-7c18-2580-153a19d4c175',
name='__start__',
path=(
'__pregel_pull',
'__start__'
),
error=None,
interrupts=(),
state=None,
result={}
),
),
interrupts=()
)
]
由上可知，同一个超步中：
-
发生中断的任务，其中断信息被记录在任务实例和快照的
interrupts字段中 -
未触发中断的任务正常结束，其结果被记录在任务实例的
result字段中
底层负责执行节点逻辑的是超步的第二阶段，如果存在多个并行任务，则会将任务提交到线程池，在独立的线程中运行，负责执行第二阶段的 PregelRunner 实例会收集提交后的 future 对象，并等待所有任务完成。
当中断触发时，对应的线程会抛出 GraphInterrupt，对应的 future 对象会收集异常。
PregelRunner 实例在 future 对应的任务完成后，收集中断信息或任务执行结果，写入检查点存储器。
所以一个超步中某个任务触发中断，不会导致其它并行任务的终止，正常运行的任务结果、中断任务的中断信息都被记录在检查点后端。
8.2. 静态断点
LangGraph 提供了用于调试的 静态断点。
它是在状态图编译或调用时设置的，不会在运行时动态触发，不属于业务逻辑，所以称为 静态 断点。
动态断点提供了接收人类反馈的接口，而静态断点只是暂停计算图的运行，不能向调用者传递信息，也不能接收调用者的反馈。
8.2.1. 用法说明
-
静态断点也需要基于检查点恢复，所以必须配置检查点，启用可恢复运行机制。
-
计算图会在
interrupt_before指定的节点执行之前产生中断，暂停计算 -
计算图会在
interrupt_after指定的节点执行之后产生中断，暂停计算 -
计算图运行到断点位置会中断，和动态断点不同的是，它会在超步边界而非内部中断。
断点前的超步：三个阶段全部完成；断点后的超步：三个阶段都没有开始。
- 静态断点不会返回任何中断信息，只会将当前最新的状态返回。
因此，我们可以用静态断点查看每个超步边界的中间状态。
-
传入相同的配置并将
None作为输入，再次调用计算图，会从断点位置继续运行。 -
支持在两个阶段设置静态断点
-
状态图编译时
python
graph = builder.compile(
checkpointer=checkpointer,
interrupt_before=["node_a", "node_b"],
interrupt_after=["node_a", "node_b"]
) -
计算图调用时
python
first_res = graph.invoke(
{},
config=config,
interrupt_before=["node_a", "node_b"],
interrupt_after=["node_a", "node_b"]
)
断点都是在运行时生效，计算图调用时设置的断点优先级更高。
如果调用时传入的断点列表不为空，则会覆盖编译时配置。
8.2.2. 底层原理
- 编译时设置的静态断点保存在编译图的
interrupt_before_nodes和interrupt_after_nodes中。
运行时采用调用参数优先、编译配置兜底的规则。
调用时传入非空节点列表会完全覆盖编译配置，而 None 或空列表会回退到编译配置。
interrupt_before在超步的第一阶段：当前超步的任务列表计算完成后、节点任务开始执行前进行检查。
- 如果检查点中的状态和上一次静态断点或图的初始状态相比，产生了变化
- 并且当前超步的任务列表和
interrupt_before列表存在交集
则抛出 GraphInterrupt。
interrupt_after在超步的第三阶段：当前超步的任务全部执行完成、写入应用到通道并创建检查点之后进行检查。
- 如果检查点中的状态和上一次静态断点或图的初始状态相比，产生了变化
- 并且当前超步的任务列表和
interrupt_after列表存在交集
则抛出 GraphInterrupt。
- 如果存在这样的拓扑图
python
node_a -> node_b
那么在 node_a 之后 和 node_b 之前 设置断点只会中断一次。
因为二者对应同一个超步边界，如果同时配置：
-
node_a的after先触发； -
恢复运行时检查点中记录的上次断点可见状态会被更新，和最新状态保持一致
-
而产生中断之前已经保存了检查点，恢复运行时会从
node_b的超步开始 node_b的超步第一阶段，检查点和上次静态断点相比，没有变化，所以不会中断- 因此，
node_a之后 和node_b之前 设置断点只会中断一次
8.2.3. 示例
8.2.3.1. 编译时设置断点
8.2.3.1.1. 顺序结构设置断点
示例如下
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from loguru import logger
class OverAllState(TypedDict):
final_res: str
def node_a(state: OverAllState) -> OverAllState:
logger.info("node_a 被执行")
return {
"final_res": "node_a 运行的中间结果"
}
def node_b(state: OverAllState) -> OverAllState:
logger.info("node_b 被执行")
return {
"final_res": "node_b 运行的中间结果"
}
def node_c(state: OverAllState) -> OverAllState:
logger.info("node_c 被执行")
return {
"final_res": "node_c 运行的中间结果"
}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("node_c", node_c)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", "node_c")
builder.add_edge("node_c", END)
checkpointer = InMemorySaver()
graph = builder.compile(
checkpointer=checkpointer,
interrupt_before=["node_a", "node_b"],
interrupt_after=["node_a", "node_b"]
)
from IPython.display import display
display(graph)
config = {"configurable": {"thread_id": "123"}}
logger.info("{}-> 第一次执行 <-{}", "=" * 10, "=" * 10)
first_res = graph.invoke({}, config=config)
logger.info("第一次执行结果：{}", first_res)
logger.info("{}-> 第二次执行 <-{}", "=" * 10, "=" * 10)
second_res = graph.invoke(None, config=config)
logger.info("第二次执行结果：{}", second_res)
logger.info("{}-> 第三次执行 <-{}", "=" * 10, "=" * 10)
third_res = graph.invoke(None, config=config)
logger.info("第三次执行结果：{}", third_res)
logger.info("{}-> 第四次执行 <-{}", "=" * 10, "=" * 10)
final_res = graph.invoke(None, config=config)
logger.info("第四次执行结果：{}", final_res)
logger.info("{}-> 完毕 <-{}", "=" * 10, "=" * 10)
运行结果如下
编译时设置的断点在渲染拓扑结构时被感知，图中对应结点新增了断点标记：
__interrupt = before,after
这表示该节点之前和之后都设置了断点。
日志如下
2026-06-18 11:22:52.094 | INFO | __main__:<module>:51 - ==========-> 第一次执行 <-==========
2026-06-18 11:22:52.100 | INFO | __main__:<module>:53 - 第一次执行结果：None
2026-06-18 11:22:52.100 | INFO | __main__:<module>:55 - ==========-> 第二次执行 <-==========
2026-06-18 11:22:52.102 | INFO | __main__:node_a:11 - node_a 被执行
2026-06-18 11:22:52.104 | INFO | __main__:<module>:57 - 第二次执行结果：{'final_res': 'node_a 运行的中间结果'}
2026-06-18 11:22:52.104 | INFO | __main__:<module>:59 - ==========-> 第三次执行 <-==========
2026-06-18 11:22:52.105 | INFO | __main__:node_b:17 - node_b 被执行
2026-06-18 11:22:52.107 | INFO | __main__:<module>:61 - 第三次执行结果：{'final_res': 'node_b 运行的中间结果'}
2026-06-18 11:22:52.107 | INFO | __main__:<module>:63 - ==========-> 第四次执行 <-==========
2026-06-18 11:22:52.108 | INFO | __main__:node_c:23 - node_c 被执行
2026-06-18 11:22:52.110 | INFO | __main__:<module>:65 - 第四次执行结果：{'final_res': 'node_c 运行的中间结果'}
2026-06-18 11:22:52.111 | INFO | __main__:<module>:67 - ==========-> 完毕 <-==========
由日志可知，node_a 和 node_b 之间只产生了一次中断，和分析相符。
8.2.3.1.2. 存在并行节点时设置断点
为了确定断点是 以节点 为边界还是 以超步 为边界设置的，我们在存在并行节点的计算图中测试
示例如下
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from loguru import logger
class OverAllState(TypedDict):
final_res: str
def node_a(state: OverAllState) -> OverAllState:
logger.info("node_a 被执行")
return {
"final_res": "node_a 运行的中间结果"
}
def node_b(state: OverAllState) -> OverAllState:
logger.info("node_b 被执行")
return {
"final_res": "node_b 运行的中间结果"
}
def node_c(state: OverAllState) -> OverAllState:
logger.info("node_c 被执行")
return {
"final_res": "node_c 运行的中间结果"
}
def node_d(state: OverAllState) -> OverAllState:
logger.info("node_d 被执行")
return {}
def node_e(state: OverAllState) -> OverAllState:
logger.info("node_e 被执行")
return {}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("node_c", node_c)
builder.add_node("node_d", node_d)
builder.add_node("node_e", node_e)
builder.add_edge(START, "node_a")
builder.add_edge(START, "node_d")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_d", "node_e")
builder.add_edge(["node_b", "node_e"], "node_c")
builder.add_edge("node_c", END)
checkpointer = InMemorySaver()
graph = builder.compile(
checkpointer=checkpointer,
interrupt_before=["node_a", "node_b"],
interrupt_after=["node_a", "node_b"]
)
from IPython.display import display
display(graph)
config = {"configurable": {"thread_id": "123"}}
logger.info("{}-> 第一次执行 <-{}", "=" * 10, "=" * 10)
first_res = graph.invoke({}, config=config)
logger.info("第一次执行结果：{}", first_res)
logger.info("{}-> 第二次执行 <-{}", "=" * 10, "=" * 10)
second_res = graph.invoke(None, config=config)
logger.info("第二次执行结果：{}", second_res)
logger.info("{}-> 第三次执行 <-{}", "=" * 10, "=" * 10)
third_res = graph.invoke(None, config=config)
logger.info("第三次执行结果：{}", third_res)
logger.info("{}-> 第四次执行 <-{}", "=" * 10, "=" * 10)
final_res = graph.invoke(None, config=config)
logger.info("第四次执行结果：{}", final_res)
logger.info("{}-> 完毕 <-{}", "=" * 10, "=" * 10)
运行结果如下
2026-06-18 11:28:49.920 | INFO | __main__:<module>:62 - ==========-> 第一次执行 <-==========
2026-06-18 11:28:49.922 | INFO | __main__:<module>:64 - 第一次执行结果：None
2026-06-18 11:28:49.923 | INFO | __main__:<module>:66 - ==========-> 第二次执行 <-==========
2026-06-18 11:28:49.925 | INFO | __main__:node_a:11 - node_a 被执行
2026-06-18 11:28:49.927 | INFO | __main__:node_d:29 - node_d 被执行
2026-06-18 11:28:49.931 | INFO | __main__:<module>:68 - 第二次执行结果：{'final_res': 'node_a 运行的中间结果'}
2026-06-18 11:28:49.931 | INFO | __main__:<module>:70 - ==========-> 第三次执行 <-==========
2026-06-18 11:28:49.933 | INFO | __main__:node_b:17 - node_b 被执行
2026-06-18 11:28:49.935 | INFO | __main__:node_e:33 - node_e 被执行
2026-06-18 11:28:49.939 | INFO | __main__:<module>:72 - 第三次执行结果：{'final_res': 'node_b 运行的中间结果'}
2026-06-18 11:28:49.940 | INFO | __main__:<module>:74 - ==========-> 第四次执行 <-==========
2026-06-18 11:28:49.940 | INFO | __main__:node_c:23 - node_c 被执行
2026-06-18 11:28:49.942 | INFO | __main__:<module>:76 - 第四次执行结果：{'final_res': 'node_c 运行的中间结果'}
2026-06-18 11:28:49.943 | INFO | __main__:<module>:78 - ==========-> 完毕 <-==========
node_a 和 node_d 属于同一个超步
node_b 和 node_e 属于同一个超步
我们只在 node_a 和 node_b 前后设置了超步，但它们的并行节点也被中断了。
显然，中断发生在超步边界，而非节点边界，与分析相符。
8.2.3.2. 调用时设置断点
调用时的断点配置只对本次调用生效。
8.2.3.2.1. 正确用法：恢复调用时设置相同的断点
示例如下
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from loguru import logger
class OverAllState(TypedDict):
final_res: str
def node_a(state: OverAllState) -> OverAllState:
logger.info("node_a 被执行")
return {
"final_res": "node_a 运行的中间结果"
}
def node_b(state: OverAllState) -> OverAllState:
logger.info("node_b 被执行")
return {
"final_res": "node_b 运行的中间结果"
}
def node_c(state: OverAllState) -> OverAllState:
logger.info("node_c 被执行")
return {
"final_res": "node_c 运行的中间结果"
}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("node_c", node_c)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", "node_c")
builder.add_edge("node_c", END)
checkpointer = InMemorySaver()
graph = builder.compile(
checkpointer=checkpointer
)
from IPython.display import display
display(graph)
config = {"configurable": {"thread_id": "123"}}
logger.info("{}-> 第一次执行 <-{}", "=" * 10, "=" * 10)
first_res = graph.invoke(
{},
config=config,
interrupt_before=["node_a", "node_b"],
interrupt_after=["node_a", "node_b"]
)
logger.info("第一次执行结果：{}", first_res)
logger.info("{}-> 第二次执行 <-{}", "=" * 10, "=" * 10)
second_res = graph.invoke(
None,
config=config,
interrupt_before=["node_a", "node_b"],
interrupt_after=["node_a", "node_b"]
)
logger.info("第二次执行结果：{}", second_res)
logger.info("{}-> 第三次执行 <-{}", "=" * 10, "=" * 10)
third_res = graph.invoke(
None,
config=config,
interrupt_before=["node_a", "node_b"],
interrupt_after=["node_a", "node_b"]
)
logger.info("第三次执行结果：{}", third_res)
logger.info("{}-> 第四次执行 <-{}", "=" * 10, "=" * 10)
final_res = graph.invoke(
None,
config=config,
interrupt_before=["node_a", "node_b"],
interrupt_after=["node_a", "node_b"]
)
logger.info("第四次执行结果：{}", final_res)
logger.info("{}-> 完毕 <-{}", "=" * 10, "=" * 10)
运行结果如下
2026-06-18 11:32:55.457 | INFO | __main__:<module>:49 - ==========-> 第一次执行 <-==========
2026-06-18 11:32:55.460 | INFO | __main__:<module>:56 - 第一次执行结果：None
2026-06-18 11:32:55.460 | INFO | __main__:<module>:58 - ==========-> 第二次执行 <-==========
2026-06-18 11:32:55.461 | INFO | __main__:node_a:11 - node_a 被执行
2026-06-18 11:32:55.463 | INFO | __main__:<module>:65 - 第二次执行结果：{'final_res': 'node_a 运行的中间结果'}
2026-06-18 11:32:55.463 | INFO | __main__:<module>:67 - ==========-> 第三次执行 <-==========
2026-06-18 11:32:55.463 | INFO | __main__:node_b:17 - node_b 被执行
2026-06-18 11:32:55.465 | INFO | __main__:<module>:74 - 第三次执行结果：{'final_res': 'node_b 运行的中间结果'}
2026-06-18 11:32:55.466 | INFO | __main__:<module>:76 - ==========-> 第四次执行 <-==========
2026-06-18 11:32:55.467 | INFO | __main__:node_c:23 - node_c 被执行
2026-06-18 11:32:55.468 | INFO | __main__:<module>:83 - 第四次执行结果：{'final_res': 'node_c 运行的中间结果'}
2026-06-18 11:32:55.468 | INFO | __main__:<module>:85 - ==========-> 完毕 <-==========
我们期望在 node_a 和 node_b 前后中断，此时运行结果和预期相符。
8.2.3.2.2. 错误用法：恢复调用时不再设置断点
示例如下
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from loguru import logger
class OverAllState(TypedDict):
final_res: str
def node_a(state: OverAllState) -> OverAllState:
logger.info("node_a 被执行")
return {
"final_res": "node_a 运行的中间结果"
}
def node_b(state: OverAllState) -> OverAllState:
logger.info("node_b 被执行")
return {
"final_res": "node_b 运行的中间结果"
}
def node_c(state: OverAllState) -> OverAllState:
logger.info("node_c 被执行")
return {
"final_res": "node_c 运行的中间结果"
}
builder = StateGraph(state_schema=OverAllState)
builder.add_node("node_a", node_a)
builder.add_node("node_b", node_b)
builder.add_node("node_c", node_c)
builder.add_edge(START, "node_a")
builder.add_edge("node_a", "node_b")
builder.add_edge("node_b", "node_c")
builder.add_edge("node_c", END)
checkpointer = InMemorySaver()
graph = builder.compile(
checkpointer=checkpointer
)
from IPython.display import display
display(graph)
config = {"configurable": {"thread_id": "123"}}
logger.info("{}-> 第一次执行 <-{}", "=" * 10, "=" * 10)
first_res = graph.invoke(
{},
config=config,
interrupt_before=["node_a", "node_b"],
interrupt_after=["node_a", "node_b"]
)
logger.info("第一次执行结果：{}", first_res)
logger.info("{}-> 第二次执行 <-{}", "=" * 10, "=" * 10)
second_res = graph.invoke(
None,
config=config,
interrupt_before=["node_a", "node_b"],
interrupt_after=["node_a", "node_b"]
)
logger.info("第二次执行结果：{}", second_res)
logger.info("{}-> 第三次执行 <-{}", "=" * 10, "=" * 10)
third_res = graph.invoke(None, config=config)
logger.info("第三次执行结果：{}", third_res)
logger.info("{}-> 第四次执行 <-{}", "=" * 10, "=" * 10)
final_res = graph.invoke(None, config=config)
logger.info("第四次执行结果：{}", final_res)
logger.info("{}-> 完毕 <-{}", "=" * 10, "=" * 10)
运行结果如下
2026-06-18 11:35:16.236 | INFO | __main__:<module>:49 - ==========-> 第一次执行 <-==========
2026-06-18 11:35:16.259 | INFO | __main__:<module>:56 - 第一次执行结果：None
2026-06-18 11:35:16.260 | INFO | __main__:<module>:58 - ==========-> 第二次执行 <-==========
2026-06-18 11:35:16.261 | INFO | __main__:node_a:11 - node_a 被执行
2026-06-18 11:35:16.263 | INFO | __main__:<module>:65 - 第二次执行结果：{'final_res': 'node_a 运行的中间结果'}
2026-06-18 11:35:16.263 | INFO | __main__:<module>:67 - ==========-> 第三次执行 <-==========
2026-06-18 11:35:16.264 | INFO | __main__:node_b:17 - node_b 被执行
2026-06-18 11:35:16.265 | INFO | __main__:node_c:23 - node_c 被执行
2026-06-18 11:35:16.266 | INFO | __main__:<module>:69 - 第三次执行结果：{'final_res': 'node_c 运行的中间结果'}
2026-06-18 11:35:16.267 | INFO | __main__:<module>:71 - ==========-> 第四次执行 <-==========
2026-06-18 11:35:16.267 | INFO | __main__:<module>:73 - 第四次执行结果：{'final_res': 'node_c 运行的中间结果'}
2026-06-18 11:35:16.268 | INFO | __main__:<module>:75 - ==========-> 完毕 <-==========
本例中，我们只在初次调用和第一次恢复运行时设置断点。
由日志可知，按照期望，第二次恢复运行之后，node_b 执行完成后还应中断一次，然而，因为本次没有设置断点，因此计算图不再中断，node_b 和 node_c 都在本次调用完成，计算图结束。
在第四次调用时，计算图已经运行完毕，恢复运行实际上不会执行任何逻辑，只是将计算图的最终状态返回。
系列导航
上一篇：PostgreSQL 的部署与配置 ｜ 下一篇：项目部署
系列目录：AI框架
