LangGraph 人审中断：interrupt 与 resume 怎么接得上

LangGraph人审中断AI Agent

TL;DR：interrupt() 不会把执行"冻"在原地等你输入。它抛出的是一个 GraphInterrupt 异常，运行时捕获它、存下当时的图状态，然后结束这一轮。等你带着 Command(resume=...) 再 invoke 一次，运行时是从被中断节点的开头重新执行的——中断点之前的每一行代码都会再跑一遍。这条规则决定了后面所有写法：循环里放 interrupt() 会让重放量按三角数增长；并行分支同时中断时只能用 Interrupt.id 映射，StateSnapshot.tasks[].id 那个带连字符的 UUID 会直接抛 RuntimeError。

一、先把一句话说清楚：中断不是"暂停"

大部分人第一次用 interrupt() 时，脑子里想的是 Python 的 input()：程序停在这一行，我敲点东西，它接着往下走。LangGraph 不是这样的。

官方文档 Interrupts 页的原话是：

"When an interrupt is triggered, LangGraph saves the graph state using its persistence layer and waits indefinitely until you resume execution."

关键在于后面这句：

"The node restarts from the beginning of the node where the interrupt was called when resumed, so any code before the interrupt runs again"

也就是说，中断机制由两个动作组成：

1. 抛出一个特殊异常（GraphInterrupt），让它一路穿过调用栈，被运行时接住；

2. 运行时把当前状态落盘，然后结束这次 invoke，把 payload 通过 __interrupt__ 返回给调用方。

第二次 invoke 时，运行时不会"跳回第 17 行"，而是重新调用整个节点函数，只是在遇到 interrupt() 时从 resume 列表里按位置取值返回。所以 interrupt() 之前的代码执行了两次，interrupt() 之后的代码只执行一次。

我实际验证这一点的方式很笨但有效：在节点函数开头给一个全局计数器加一。

RUNS = {"n": 0}

def node(state):
    RUNS["n"] += 1
    ans = interrupt({"q": "approve?"})
    return {"v": 10 + int(ans)}

结果：

阶段	RUNS["n"]	说明
首次 invoke 后	1	进入节点 → 撞上 interrupt → 挂起
`Command(resume=True)` 后	3	重放一遍（+1），interrupt 返回后再跑尾部（+1 是我第二次加的）

第一条铁律就出来了：interrupt() 之前的所有副作用必须是幂等的。官方文档把这条列为 "Rules of interrupts" 之一，原话是 "Side effects called before interrupt must be idempotent"。发邮件、扣款、写数据库这类动作，放在 interrupt() 前面就会执行两次。

二、最小闭环：从 interrupt 到 Command(resume=) 的信号链

下面这段代码是能跑起来的最小例子，本机 langgraph 1.2.16 + langgraph-checkpoint 4.3.0 实测通过。

from typing import Annotated, TypedDict
import operator
from langgraph.graph import StateGraph, START, END
from langgraph.types import interrupt, Command
from langgraph.checkpoint.memory import InMemorySaver

class S(TypedDict):
    v: int
    log: Annotated[list, operator.add]

def ask(state: S):
    ans = interrupt({"question": "approve?", "seen": state["v"]})
    return {"v": 10 + int(ans)}

g = StateGraph(S)
g.add_node("ask", ask)
g.add_edge(START, "ask")
g.add_edge("ask", END)

app = g.compile(checkpointer=InMemorySaver())
cfg = {"configurable": {"thread_id": "t1"}}

r = app.invoke({"v": 1, "log": []}, cfg)
print(r)

首次 invoke 的返回体里多了一个 __interrupt__ 键：

{'v': 1, 'log': [],
 '__interrupt__': [Interrupt(value={'question': 'approve?', 'seen': 1},
                             id='25d8aed8be448dfcebaa7f3bf3b4402c',
                             response_schema=None)]}

注意这里返回的是入口状态（v 仍是 1），不是节点执行结果——节点根本没走完。要判断图是不是真的挂起了，看两个地方：

st = app.get_state(cfg)
st.next                      # ('ask',)   ← 还没走完
[t.interrupts for t in st.tasks]   # 每个待办 task 上挂着的 Interrupt 列表

__interrupt__ 只在你直接读 invoke 返回值时有用；在服务化场景里，真正可靠的是 get_state(cfg).next 和 st.tasks[].interrupts。原因见第五节——__interrupt__ 里那个列表的顺序会变。

补上恢复这一半：

r2 = app.invoke(Command(resume=5), cfg)
# {'v': 15, 'log': ['ask_done']}

Command(resume=5) 成为节点里 interrupt() 的返回值，于是 v = 10 + 5 = 15。

三、同一节点里多次 interrupt：靠下标匹配，不是靠次数

官方文档对"一个节点里多个 interrupt()"的描述是：

"Whenever execution resumes, it starts at the beginning of the node. For each interrupt encountered, LangGraph checks if a matching value exists in the task's resume list. Matching is strictly index-based, so the order of interrupt calls within the node is important."

实测行为（一个节点里连着两次 interrupt）：

def two(state):
    a = interrupt("first?")
    b = interrupt("second?")
    return {"log": [f"{a}/{b}"]}

操作	结果
首次 invoke	挂起，`__interrupt__` 只有 `first?` 一条
`Command(resume="A")`	挂起，`__interrupt__` 变成 `second?`
`Command(resume="B")`	完成，`log = ['A/B']`

关键：一个 resume 只能喂一个 interrupt。你不能一次把两个答案都交上去。我试过 Command(resume=[1, 2])，结果是整个 list 变成了第一个 interrupt() 的返回值，第二个照样挂起：

# 单 interrupt 节点，resume 传 list
app.invoke(Command(resume=["x", "y"]), cfg)
# log = ["['x', 'y']"]   ← 整个 list 被当成一个值

这是最容易踩的一个坑，因为"resume 值列表"这个说法在文档里出现过，但那个列表是运行时跨多轮累积的内部列表，不是让你一次性传进去的。

四、循环里放 interrupt：3 轮问答重放了 9 遍

把第三节的结论推广一步：如果一个节点的循环体里放了 interrupt()，那么每次 resume 都要把之前所有轮次重跑一遍。

官方原文：

"a loop that calls interrupt() multiple times causes each resume to replay all previous iterations: the first resume replays 1 iteration, the second replays 2, and so on."

我实测（3 轮循环，每轮都要答一次才继续）：

暂停后 body 累计执行 = 1
第 1 次 resume 后 body 累计执行 = 3
第 2 次 resume 后 body 累计执行 = 6
第 3 次 resume 后 body 累计执行 = 9

想收集 3 个答案，循环体一共跑了 9 次。官方用词是 "exponential re-execution"，本机实测的具体数字是三角数（1 → 3 → 6 → 9，增量 1/2/3），不是严格指数——但结论一致：成本随回答轮次超线性增长。

正确的写法是官方给出的那个模式：把问题放进状态，每轮只 interrupt() 一次，答案不合法就回写状态再路由回来。

def collect(state: F):
    q = state.get("pending") or "age?"
    a = interrupt(q)
    if isinstance(a, int) and a > 0:
        return {"age": a, "pending": None}
    return {"pending": "invalid, retry"}

def route(state: F):
    return END if state.get("age") is not None else "collect"

实测：3 次 resume 后节点只执行了 6 次，且每个答案只被消费一次，没有重复跑过的历史问答。

五、并行分支同时中断：只能按 interrupt id 映射

这是本篇最反直觉的一条。当两条并行分支各调用一次 interrupt() 时：

r = app.invoke({"v": 1, "log": []}, cfg)
# __interrupt__ = [Interrupt(value='left?', id='5e55830a...'),
#                  Interrupt(value='right?', id='c065a78a...')]

st = app.get_state(cfg)
st.next          # ('left', 'right')
[t.id for t in st.tasks]            # UUID，带连字符
[i.id for t in st.tasks for i in t.interrupts]   # 32 位 hex

用 t.id 做 key 会直接炸：

RuntimeError: When there are multiple pending interrupts, you must specify
the interrupt id when resuming.

用 i.id 做 key 才通过：

m = {i.id: "OK-" + t.name for t in st.tasks for i in t.interrupts}
app.invoke(Command(resume=m), cfg)
# {'v': 1, 'log': ["left->'OK-left'", "right->'OK-right'"]}

根因在源码里。langgraph/pregel/_loop.py 判断"这是不是一个 id 映射"的判据是：

if resume_is_map := (
    isinstance(resume, dict)
    and all(is_xxh3_128_hexdigest(k) for k in resume)
):

is_xxh3_128_hexdigest 只认 32 位十六进制。tasks[].id 是带连字符的 UUID，过不了这个检查，于是被当成"单个标量 resume 值"，再撞上"存在多个待处理中断"的分支抛错。

映射还支持部分恢复：只给其中一个 id 值，另一个保持挂起。

partial -> {'v': 1, 'log': ["left->'only-left'"],
            '__interrupt__': [Interrupt(value='right?', ...)]}
next after partial = ('right',)

这一点在做"批量审批"时非常有用——你可以先放过一个，把另一个继续挂在那里等人。

顺便记一个我踩到的坑：__interrupt__ 列表的顺序不是契约。首轮挂起时是 ['L', 'R']，但用 Command(update=...) 触发重跑后，同一批中断变成了 ['R', 'L']（连跑 8 次，8 次都是 R 在前）。所以永远不要用位置去猜"这是哪个节点在问"，只认 Interrupt.id。

六、response_schema 会真的校验，但布尔字段会被 "yes" 静默通过

1.2.x 的 interrupt() 多了个 response_schema 参数，签名是：

def interrupt(value: Any, *, response_schema: dict[str, Any] | type | None = None) -> Any:

传一个 TypedDict 进去，运行时会把 schema 挂到 Interrupt.response_schema 上：

response_schema = {'properties': {'approved': {'title': 'Approved', 'type': 'boolean'},
                                  'note': {'title': 'Note', 'type': 'string'}},
                   'required': ['approved', 'note'], 'title': 'Decision', 'type': 'object'}

恢复时的实测结果：

`Command(resume=...)`	结果
`{"approved": True}`（缺 note）	`ValidationError: 1 validation error for Decision / note / Field required`
`42`	`ValidationError: Input should be a valid dictionary`
`["a","b"]`	`ValidationError: Input should be a valid dictionary`
`{"approved": True, "note": "ok"}`	通过

但这里有个很危险的边界：{"approved": "yes", "note": "x"} 也通过了，节点拿到的是 {'approved': True, 'note': 'x'}——字符串 "yes" 被静默转成了布尔 True。

也就是说，response_schema 能挡住"缺字段"和"传错容器类型"，挡不住"类型近似的字符串"。如果你的审批流是"审批人从某个表单回传 JSON"，而那个表单把布尔序列化成了字符串，那 response_schema 不会帮你发现。要硬性约束，得在节点里自己再判一次 isinstance(ans, bool)。

七、子图与跨进程：中断的保质期比一个进程长

10-19 那篇做持久化时留了个尾巴没测：子图（checkpoint_ns 非空）里的中断会怎样。我这次补上了。

实测：子图单独 compile()（不带 checkpointer），作为节点挂到带 checkpointer 的父图上——

invoke#1 -> {'v': 1, 'log': [], '__interrupt__': [Interrupt(value='sub-approve?', ...)]}
next = ('child',)
Command(resume=7) -> {'v': 17, 'log': ['done']}

中断正常挂起在父图的 child 节点上，用父图的 Command(resume=...) 就能恢复。官方对这种"把子图当函数调用"的写法原话是：

"the parent graph will resume execution from the beginning of the node where the subgraph was invoked and the interrupt was triggered. Similarly, the subgraph will also resume from the beginning of the node where interrupt was called."

注意"similarly"这个词：父子两层都会整段重放。子图里 interrupt() 前的代码，和父图节点里子图调用前的代码，恢复时都会再跑一遍。

跨进程这一半更有意思。我写了两个完全独立的 Python 进程，只用同一个 SQLite 文件和同一个 thread_id：

proc1（pause） : PAUSE_RESULT {'v': 5, 'log': [], '__interrupt__': [Interrupt(value={'q': 'cross-process?'}, ...)]}
proc2（resume）: BEFORE_NEXT ('ask',) {'v': 5, 'log': []}
                RESUME_RESULT {'v': 142, 'log': ['cross']}

进程 B 启动后读到 next=('ask',)、状态 v=5，直接把 A 挂起的那次执行接上了。这就是官方说的 "fault tolerance" 落地的样子：服务重启、容器重建，审批流不会断。

反过来也有坑：删掉 thread 后再 resume 不会报错。实测 delete_thread() 之后 get_state() 返回 values={}，再 invoke Command(resume="late") 得到的是图从头跑一遍并再次中断的结果。如果你依赖"resume 失败即报错"来做告警，这里是静默的。

八、踩坑录：六个我本机复现出来的报错

以下每条都是我在这台机器上真跑出来、原样抄下来的错误文本。

报错一：`RuntimeError: When there are multiple pending interrupts, you must specify the interrupt id when resuming.`

现象：两条并行分支各调用一次 interrupt()，用 get_state(cfg).tasks[].id 组装 dict 传进去就炸。

RuntimeError: When there are multiple pending interrupts, you must specify
the interrupt id when resuming. Docs: https://docs.langchain.com/oss/python/
langgraph/add-human-in-the-loop#resume-multiple-interrupts-with-one-invocation.

原因：resume 要被识别成"id → 值"的映射，必须满足 all(is_xxh3_128_hexdigest(k) for k in resume)，而 tasks[].id 是带连字符的 UUID。

解决：改取 task.interrupts[].id。

顺带一个发现：报错信息里那个文档链接是 404。我 curl 过 https://docs.langchain.com/oss/python/langgraph/add-human-in-the-loop，返回 404；现在能打开的是 https://docs.langchain.com/oss/python/langgraph/interrupts（后者会自动重定向到前者之外的新路径）。

报错二：`UnboundLocalError: cannot access local variable 'resume_is_map' where it is not associated with a value`

现象：Command(resume=None) 或干脆 Command() 什么都不传。

UnboundLocalError: cannot access local variable 'resume_is_map' where it is
not associated with a value

原因在 langgraph/pregel/_loop.py：

if (resume := cast(Command, self.input).resume) is not None:   # L1018
    if not self.checkpointer:
        raise RuntimeError("Cannot use Command(resume=...) without checkpointer")
    if resume_is_map := (...):                                  # L1026
        ...
...
if not writes and not resume_is_map:                            # L1038
    raise EmptyInputError("Received empty Command input")

resume 为 None 时，第 1018 行整块被跳过，resume_is_map 从未被赋值；到 1038 行读它就直接 UnboundLocalError。

连带后果：源码里那句 EmptyInputError("Received empty Command input") 在 resume=None 这条路径上永远到不了——它想在 1038 行给个友好提示，结果 1038 行自己先崩了。

解决：要继续空跑就传 None 给第一个参数（app.invoke(None, cfg)），不要传空 Command。

补充：这个 bug 在 1.2.16 和 1.2.17 上都能复现，不是已修的老问题。

报错三：`ValidationError: 1 validation error for Decision` / `note` / `Field required`

现象：给了 response_schema=Decision（含 approved: bool 与 note: str），resume 时只传了 {"approved": True}。

ValidationError: 1 validation error for Decision
note
  Field required [type=missing, input_value={'approved': True}, input_type=dict]
    For further information visit https://errors.pydantic.dev/2.14/v/missing

原因：schema 里 note 是必填。

解决：resume 值必须满足完整 schema，或者把 note 声明成 Optional/给默认值。

报错四：`InvalidUpdateError: At key 'v': Can receive only one value per step`

现象：想在恢复的同时改状态，写成 Command(resume=good_map, update={"v": 77})。

InvalidUpdateError: At key 'v': Can receive only one value per step.
Use an Annotated key to handle multiple values.
For troubleshooting, visit: https://docs.langchain.com/oss/python/langgraph/
errors/INVALID_CONCURRENT_GRAPH_UPDATE

原因：resume 触发的节点回写与 update 写入落在同一步，v 这个非 reducer 字段收到了两个值。

解决：要么给 v 加 reducer（Annotated[list, operator.add] 之类），要么把状态修改推迟到恢复完成之后的下一轮。

报错五：`RuntimeError: Cannot use Command(resume=...) without checkpointer`

现象：图编译的时候忘了传 checkpointer。

RuntimeError: Cannot use Command(resume=...) without checkpointer

原因：中断依赖持久化层，没有 checkpointer 就没有"挂起"这回事。

最坑的地方在于前半段完全不报错：不带 checkpointer 的图里调 interrupt() 不抛异常，它把 Interrupt 对象塞进 invoke 返回值就结束了。你会以为流程是对的，直到恢复这一步才炸。

解决：编译阶段必须传 checkpointer 实例；判断有没有真的挂起，看 get_state(cfg).next，不能只看 invoke 有没有抛异常。

报错六：`GraphInterrupt` 被裸 `except Exception` 吞掉（没有报错的那一个）

这条不产生异常，但它是六个里面最危险的：

def node(state):
    try:
        ans = interrupt("trapped?")
    except Exception as e:
        return {"log": [f"swallowed:{type(e).__name__}"]
    return {"log": [f"got:{ans!r}"]}

实测输出：

caught: GraphInterrupt (Interrupt(value='trapped?', id='4f62a0b8...'))
result -> {'v': 1, 'log': ['swallowed:GraphInterrupt']}
next = ()

图"正常结束"了，next 是空的，人工环节被整个跳过——你的审批流看起来跑通了，实际上人从头到尾没有参与过。

官方文档把这条列为第一条规则："Do not wrap interrupt calls in try/except"，原因是 interrupt() 就是靠抛异常来暂停的，你 catch 了就没有中断了。

解决：要么把 interrupt() 挪到 try 块外面，要么只捕获具体异常类型（except (ValueError, TypeError)）。我实测后者是安全的——改成具体类型后中断正常挂起，next 为 ('ok',)。

九、什么时候不该用 interrupt

需要长时间挂起的跨系统流程。中断依赖 checkpointer 里有那条 thread，如果你们的 thread 生命周期是按"会话结束即清理"来管的，中断会在人还没审的时候就丢上下文。
把 interrupt 当重试队列用。见第四节，循环 + 中断的重放代价是超线性的。要重试就用专门的编排（参考 10-24 那篇 urllib3/tenacity 的四处静默失效）。
节点里有非幂等副作用。发通知、扣库存这类动作要么挪到 interrupt() 之后，要么自己加去重键。
在多轮对话里用 Command(update=...) 当输入接上去。官方明示 Command(resume=...) 是唯一设计给 invoke() / stream() 输入的 Command 模式，update / goto / graph 是给节点函数返回用的。

十、本文的边界

全部在本机 Windows 11 + Python 3.13 上跑，langgraph 1.2.16（2026-10-16 发布）为主测环境，另用 --target 隔离装 1.2.17（2026-10-24 发布，版本口径以 GitHub Releases 为准）复跑了关键场景，行为一致。
只测了 InMemorySaver 与 SqliteSaver 两个后端。没有测 Postgres 后端的中断行为，也没有测托管运行时的自动持久化。
没有测异步路径（ainvoke / astream / astream_events）下的中断，官方推荐的新接口是 graph.stream_events(..., version="v3")，本机只浅尝了一下，本文不对它下断言。
没有测 prebuilt 的 ToolNode / 工具函数内部的 interrupt()，也没有测 Send 做 map-reduce 扇出时的多中断恢复。
没有做并发压力测试：多个审批人同时 resume 同一 thread_id 会怎样，本文不断言（参考 10-19 那篇 16 线程并发写导致 93.75% 更新丢失的结论，可以推断这里同样不安全）。

FAQ

Q：第二次 invoke 一定要传 Command(resume=...) 吗？

分情况。如果是 interrupt() 造成的动态中断，必须传 Command(resume=...)，中断点才有值可以返回。如果是 compile(interrupt_before=["b"]) 这种静态断点，传 None（app.invoke(None, cfg)）就能继续，因为断点只是"暂停执行"，不等任何输入。

Q：静态断点和 interrupt() 该用哪个？

官方给的说法很直接："Static interrupts are not recommended for human-in-the-loop workflows. Use the interrupt function instead." 静态断点更适合调试时一步一步走。

Q：怎么知道现在有几个中断在等人？

遍历 app.get_state(cfg).tasks，每个 task 上有 interrupts 元组。__interrupt__ 里的条数等于待处理中断总数，但顺序不可靠（见第五节）。

Q：response_schema 能挡住所有错误输入吗？

不能。实测缺字段、传标量、传列表都会被 ValidationError 挡住，但 approved: "yes" 会被静默转成 True。这是 pydantic 宽松模式的既有行为。

Q：状态存下来之后，人多久之内响应都行？

取决于 checkpointer 与你们的清理策略。官方说"waits indefinitely"，但那指的是运行时不主动超时，不代表 thread 会永远留着。SqliteSaver 的 prune() 实测直接 raise NotImplementedError（10-19 已记录），要清理得用 delete_thread()。

Q：能不能在审批前先改一下状态再让人看？

可以，用 app.update_state(cfg, values, as_node=...)。但注意恢复时节点会整段重放，所以你改的是"重放之后"的输入，不是中断那一刻的快照。

---

如果你在搭的是"状态存下来之后怎么把人接进去"这一层，可以先看站内这两篇：一篇讲 checkpointer 本身怎么落盘（LangGraph 1.2.16 持久化实测），一篇讲跨服务传数据时格式选型会付什么代价（数据序列化实测：JSON、MessagePack 与 Protobuf）。本篇答的是两者中间那一层——存下来了，怎么让人安全地接上，接上之后哪几行代码会被再跑一遍。

👉 立即参与小米 MiMo 开放平台：国内领先的 AI 大模型开放平台，高性价比推理服务

👉 立即参与阿里云 AI：汇集爆款 AI 产品，热门模型专属权益优惠券助力企业创新加速

📌 本文由 AI 辅助生成并经人工审核发布 | TechPassive — AI 驱动的内容测试站点，专注于效率工具与 SaaS 真实评测

🔗 精选推荐工具

使用以下链接支持我们持续产出高质量内容（点击可直接前往购买）：

☁️ DigitalOcean 云服务器 ⚡ Vultr 高性能 VPS 🤖 QoderWork 中国版（推荐有奖） ☁️ 阿里云爆款 AI 产品 📚 WordPress 实用书单 🔍 WordPress SEO 书单 🌐 虚拟主机书单 🐳 Docker 书单 🐧 Linux 书单 🐍 Python 书单 💰 联盟营销书单 💵 被动收入书单 🖥️ 服务器书单 ☁️ 云计算书单 🚀 DevOps 书单 🤖 小米 MiMo 开放平台

← 返回首页