上一篇介绍 Persistence 时，我们已经看到，LangGraph 可以通过 Checkpointer 把执行过程中的 State 保存下来。

状态能够保存，并不等于任务已具备“暂停后继续执行”的能力。

Agent 开发中有一类场景，需要让 Agent 暂停下来，等待外部的决定。例如，发送邮件前先等确认，执行数据库变更前先等审批，或者生成内容后交由用户修改。这个等待可能是几秒，也可能是几个小时。

LangGraph 为这类场景提供了 Interrupt。它与 Persistence 配合，把一次原本必须连续完成的 Graph Run，拆成可以跨时间继续执行的多个阶段。

一、暂停执行的需求

假设一个 Agent 准备执行下面的操作：

代码片段Text
向生产环境发布版本 v2.0

生成发布命令之后直接执行，技术上没有问题。但生产发布通常需要增加一道人工确认：

代码片段Text
Agent 生成发布操作
        ↓
等待人工确认
        ↓
批准或拒绝
        ↓
继续执行

如果只是一个普通的命令行程序，可以使用 input() 等待用户输入。

然而，Agent 很难采用这种方式。

审批可能发生在十分钟以后，也可能发生在几个小时以后。最初的 HTTP 请求早已结束，运行任务的进程也可能已经重启。让线程一直阻塞在那里等待人工输入，同样不适合长时间运行的服务。

因此，Graph 暂停时需要留下几类信息：

代码片段Text
当前 State
执行到了哪个位置
正在等待什么输入
后续应该从哪里继续

这正是 Persistence 和 Interrupt 配合的地方。

Persistence 保存 Graph 的状态，Interrupt 让 Graph 在指定位置暂停。外部系统收到人工输入以后，再通过 Resume 将结果送回原来的执行过程。

二、中断与恢复机制

LangGraph 通过 interrupt() 触发动态中断。

它可以写在普通 Node 中：

代码片段Python
from langgraph.types import interrupt
 
approved = interrupt("是否批准这次操作？")

执行到这里时，Graph 会暂停，并把 "是否批准这次操作？" 交给调用 Graph 的程序。

调用方可以把这段信息显示在网页、后台管理系统或者审批界面中。

用户完成审批后，原程序中再次使用 invoke() 调用 Graph：

代码片段Python
from langgraph.types import Command
 
graph.invoke(
    Command(resume=True),
    config=config,
)

其中的resume 的 True 会返回给原来的 interrupt()：

代码片段Python
approved = interrupt("是否批准这次操作？")

恢复之后：

代码片段Python
approved == True

因此，一次完整的中断过程包含两个方向的数据传递：

代码片段Text
Graph
  │
  │ interrupt(payload)
  ▼
外部应用
  │
  │ Command(resume=value)
  ▼
Graph

interrupt() 中的参数描述“当前正在等待什么”，Command(resume=...) 中的参数则是外部系统给出的结果。

三、相关语法

由于 Interrupt 与 Persistence 结合起来实现暂停后继续执行这一机制，下面把所有相关的内容都一并列出来。它们各自承担的职责比较明确。

写法作用说明
interrupt(value)暂停 Graphvalue 会交给 Graph 调用方
Command(resume=value)恢复 Graphvalue 成为 interrupt() 的返回值
InMemorySaver()保存 Checkpoint只保存在当前进程内存中
thread_id标识一条 Thread恢复时需要继续使用原来的 ID
result["__interrupt__"]获取中断信息invoke() 模式下可读取中断 Payload

其中，thread_id 与 Checkpointer 的关系尤其重要。

配置方式与上一篇 Persistence 中使用的一样：

代码片段Python
config = {
    "configurable": {
        "thread_id": "deploy-001"
    }
}

Graph 第一次执行时使用：

代码片段Python
graph.invoke(
    initial_state,
    config=config,
)

恢复时仍然使用同一个配置：

代码片段Python
graph.invoke(
    Command(resume=True),
    config=config,
)

Checkpointer 会根据 thread_id 找到这条 Thread 已保存的执行状态。

如果第一次使用：

代码片段Text
deploy-001

恢复时改成：

代码片段Text
deploy-002

这已经是另一条 Thread，无法接着刚才的执行过程运行。

因此，可以把 thread_id 看成一条持久化执行上下文的标识。

本文示例还会使用：

代码片段Python
from langgraph.checkpoint.memory import InMemorySaver
 
checkpointer = InMemorySaver()

并在编译 Graph 时传入：

代码片段Python
graph = builder.compile(
    checkpointer=checkpointer
)

InMemorySaver 很适合本地学习和测试，但数据只存在于当前进程的内存中。进程结束后，其中的 Checkpoint 也随之丢失。生产环境需要换成真正的持久化 Checkpointer。

四、完整执行过程

下面用一个最小示例把这些对象连接起来。

这个 Graph 只有两个 Node：

代码片段Text
START
  ↓
review
  ↓
finish
  ↓
END

review 负责人工审核，finish 根据审核结果更新状态。

代码片段Python
from typing import TypedDict
 
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, START, END
from langgraph.types import Command, interrupt
 
 
class ApprovalState(TypedDict):
    action: str
    approved: bool | None
    status: str
 
 
def review(state: ApprovalState):
    print("进入 review 节点")
 
    approved = interrupt(
        {
            "type": "approval",
            "action": state["action"],
        }
    )
 
    return {"approved": approved}
 
 
def finish(state: ApprovalState):
    if state["approved"]:
        return {"status": "approved"}
 
    return {"status": "rejected"}
 
 
builder = StateGraph(ApprovalState)
 
builder.add_node("review", review)
builder.add_node("finish", finish)
 
builder.add_edge(START, "review")
builder.add_edge("review", "finish")
builder.add_edge("finish", END)
 
graph = builder.compile(
    checkpointer=InMemorySaver()
)
 
config = {
    "configurable": {
        "thread_id": "deploy-001"
    }
}
 
result = graph.invoke(
    {
        "action": "向生产环境发布版本 v2.0",
        "approved": None,
        "status": "pending",
    },
    config=config,
)
 
print(result["__interrupt__"][0].value)
 
result = graph.invoke(
    Command(resume=True),
    config=config,
)
 
print(result["status"])

运行结果如下：

代码片段Text
进入 review 节点
{'type': 'approval', 'action': '向生产环境发布版本 v2.0'}

进入 review 节点
approved

第一次进入 review 时，执行到 interrupt() 后暂停，因此 finish 此时还没有运行。

使用 invoke() 时，中断数据可以从 result["__interrupt__"] 获取到。其中：

代码片段Python
result["__interrupt__"][0].value

就是传给 interrupt() 的字典。

应用完成人工确认后，再传入：

代码片段Python
Command(resume=True)

approved 得到 True，review 返回：

代码片段Python
{"approved": True}

State 更新完成后，Graph 继续执行 finish，最终得到：

代码片段Text
approved

当前 invoke() API 仍然支持这种方式。对于需要同时处理模型流式输出、状态变化和人工中断的交互式应用，LangGraph 也提供 stream_events(..., version="v3")。本文重点放在 Interrupt 的执行机制上，因此继续使用更直观的 invoke()。

运行结果里还有一个很重要的现象：

代码片段Text
进入 review 节点

输出了两次。

理解这个细节，才能正确处理 Interrupt 前后的业务代码。

五、恢复时的重放

很多人第一次使用 Interrupt 时，容易把 Resume 当成普通程序里的“从暂停位置继续”。

LangGraph 的执行方式有所不同。

假设 Node 是：

代码片段Python
def review(state):
    print("A")
 
    approved = interrupt("是否批准？")
 
    print("B")
 
    return {"approved": approved}

第一次运行到 Interrupt：

代码片段Text
A

随后 Graph 暂停。

执行：

代码片段Python
Command(resume=True)

之后，输出变成：

代码片段Text
A
B

原因在于，Resume 时，中断所在的 整个 Node 会从头重新执行。

执行过程实际是：

代码片段Text
第一次运行

Node 开始
   ↓
执行 A
   ↓
interrupt
   ↓
暂停


恢复运行

Node 重新开始
   ↓
再次执行 A
   ↓
interrupt 获得 Resume Value
   ↓
继续执行 B

LangGraph 保存的是 Graph 的执行状态，并不会冻结一个 Python 函数的调用栈，然后从某一行指令恢复。

注意这一点尤为重要。比如：

代码片段Python
def review(state):
    create_order()
 
    approved = interrupt("是否批准订单？")
 
    return {"approved": approved}

create_order() 位于 Interrupt 前面。

Node 恢复时，这段代码还会再执行一次。如果 create_order() 每次都创建新记录，就可能产生两个订单。

发送邮件、扣减库存、调用支付接口、插入数据库记录，也有类似问题。

工程中常见的处理方式有两种。

一种是让 Interrupt 前面的操作具备幂等性。例如：

代码片段Python
db.upsert_order(
    order_id=state["order_id"],
    status="pending",
)

同一个 order_id 重复执行不会产生多条订单。

另一种方式更直观：把多次执行受影响的操作放到审核后的 Node。

代码片段Text
review
   ↓
Interrupt
   ↓
人工批准
   ↓
execute
   ↓
调用外部系统

对于审批类流程，这种设计通常更清楚。review 只负责取得决定，execute 才负责执行真正的业务操作。

六、常见人工介入

Interrupt 并不只用于“批准或拒绝”。

从 Graph 的角度看，它所做的是：运行到某个位置后，需要等待外部输入。

以下我们看几种常见的场景。

批准或拒绝

例如数据库变更、生产发布和高风险 Tool 调用：

代码片段Python
approved = interrupt(
    {
        "type": "approval",
        "action": state["action"],
    }
)

批准：

代码片段Python
Command(resume=True)

拒绝：

代码片段Python
Command(resume=False)

Node 根据布尔值决定后续处理。

审核并修改

例如 Agent 已经生成了一封邮件：

代码片段Python
edited_email = interrupt(
    {
        "type": "email_review",
        "email": state["email"],
    }
)

审核人员可以修改邮件，再把完整内容传回：

代码片段Python
Command(
    resume={
        "to": "user@example.com",
        "subject": "项目进度调整通知",
        "body": "修改后的正文",
    }
)

此时 edited_email 得到人工修改后的字典，再由 Node 写入 State。

补充缺失信息

Graph 运行过程中也可能缺少某个业务参数：

代码片段Python
environment = interrupt(
    {
        "type": "select_environment",
        "options": [
            "development",
            "staging",
            "production",
        ],
    }
)

用户选择生产环境：

代码片段Python
Command(resume="production")

恢复后：

代码片段Python
environment == "production"

因此，Human-in-the-loop 的范围比审批更广。审核、修改、补充参数以及需要人工判断的业务决策，都可以通过相同的中断机制实现。

七、工程中的边界

Interrupt 的 API 很少，实际项目中的问题更多来自执行语义。

除了前面提到中断所在的整个 Node 会从头重新执行，还有几条规则应当提前了解。

异常处理

不要使用宽泛的 try/except 包住 interrupt()：

代码片段Python
try:
    approved = interrupt("是否批准？")
except Exception:
    ...

interrupt() 会通过内部的特殊异常把暂停信号交给 LangGraph Runtime。宽泛捕获异常可能把这个信号一起截获。

业务代码需要异常处理时，可以缩小 try 的范围，或者只捕获明确的异常类型。

多个 Interrupt

一个 Node 可以包含多个 Interrupt：

代码片段Python
name = interrupt("姓名")
age = interrupt("年龄")
city = interrupt("城市")

恢复时，LangGraph 会按照 Interrupt 的调用顺序匹配 Resume Value，因此这些调用的顺序应保持稳定。

如果 Node 根据不稳定条件跳过其中某个 Interrupt，就容易造成恢复值与调用位置不一致。

为了简单清晰，一次 Node 调用仅包含一个 Interrupt，执行关系会更容易维护。

输入校验

需要反复要求用户重新输入时，也不适合在 Node 中写：

代码片段Python
while True:
    value = interrupt(...)

更稳妥的方式是把校验结果写入 State，再通过 Conditional Edge 回到收集输入的 Node：

代码片段Text
collect_input
      ↓
   validate
    ↙    ↘
 无效     有效
  ↓       ↓
重新输入   后续流程

这样每一次输入都对应一个明确的 Graph Step，状态历史也更清楚。

总结

本文介绍了 LangGraph 中 Interrupt、Resume 与 Persistence 的配合方式。Graph 可以在关键步骤通过 interrupt() 暂停，由 Checkpointer 保存状态，再通过相同的 thread_id 和 Command(resume=...) 恢复执行，因此适合人工审批、参数补充和高风险操作确认等场景。

实际使用时，应把人工介入放在高风险或不可逆的操作前，同时注意 Resume 会重新执行中断所在的 Node。Interrupt 前应尽量避免执行可能重复产生结果的外部操作，必要时采用幂等设计，或把真正的业务执行放到确认后的 Node 中。