langgraph之interrupt 事例篇

时间：2026-07-24 09:06:10 编辑：袖梨来源：一聚教程网

LangGraph 中的 interrupt 是一个人工介入（Human-in-the-loop）和流程暂停/恢复机制。它允许一个正在执行的 Agent 工作流在某个节点主动暂停，把当前状态暴露出来，等待外部输入，然后从暂停位置继续执行。

langgraph之interrupt（事例篇）

简单理解：

它类似：

调试器里的 breakpoint
工作流引擎里的人工审批节点
操作系统里的 sleep + resume
分布式任务里的 checkpoint + recovery

1. 为什么需要 interrupt？

传统 Agent：

 复制代码用户请求
   |
   v
LLM
   |
   v
调用工具
   |
   v
返回结果

问题：

如果 Agent 做了危险操作：

例如：

 复制代码用户:
帮我删除 AWS 上所有测试服务器Agent:
调用 delete_server()执行完成

没有机会人工确认。

有了 interrupt：

 复制代码用户请求
    |
    v
Agent 推理
    |
    v
准备调用 delete_server()
    |
    v
interrupt()
    |
    v
暂停
    |
人工确认:
"是否真的删除?"
    |
    v
继续执行
    |
    v
delete_server()

2. interrupt 的基本使用

例如：

 复制代码from langgraph.types import interrupt
def approval_node(state):    answer = interrupt(
        {
            "question": "是否允许执行删除操作?",
            "action": "delete_user"
        }
    )    if answer["approved"]:
        return {
            "result": "执行删除"
        }    return {
        "result": "取消删除"
    }

执行到：

 复制代码interrupt(...)

流程暂停。

 复制代码{
  "question":"是否允许执行删除操作?",
  "action":"delete_user"
}

外部系统展示：

 复制代码是否允许执行删除操作?[允许] [拒绝]

用户点击：

 复制代码{
 "approved":true
}

然后恢复执行。

3. interrupt 和普通 return 有什么区别？

普通 return：

 复制代码def node(state):
    return {
        "status":"done"
    }

生命周期：

 复制代码node
 |
 return
 |
结束

interrupt：

 复制代码def node(state):    value = interrupt(
        "需要确认"
    )    print(value)

生命周期：

第一次：

 复制代码node
 |
 interrupt
 |
保存状态
 |
暂停

第二次：

 复制代码恢复
 |
value = 用户输入
 |
继续执行

4. interrupt 的设计原理

核心思想：

"可持久化的暂停点"

LangGraph 本质是一个：

比如：

 复制代码        +-------+
        | start |
        +---+---+
            |
            v
       +----+----+
       |  Agent  |
       +----+----+
            |
            v
       +----+----+
       | Tool   |
       +----+----+
            |
            v
          END

每个节点执行：

 复制代码State
 |
Node
 |
New State

状态不断变化：

例如：

 复制代码{
 "messages":[
   "用户请求删除服务器"
 ],
 "current_step":"tool_call"
}

当执行：

 复制代码interrupt()

LangGraph 做：

1. 保存当前 state

类似：

 复制代码checkpoint:{
 state:{
   messages:[...],
   current_node:"approval"
 }
}

2. 保存执行位置

记录：

 复制代码graph position:node = approval_nodenext_step =
after_interrupt

3. 抛出特殊异常

内部类似：

 复制代码raise GraphInterrupt()

停止执行。

不是普通异常。

4. 等待 resume

外部调用：

 复制代码Command(
    resume={
        "approved":True
    }
)

5. 恢复执行

 复制代码checkpoint
     |
     v
approval_node
     |
     v
interrupt返回值
     |
     v
继续执行

5. interrupt 为什么需要 checkpoint？

因为 Agent 可能运行很久。

例如：

 复制代码Agent:搜索网页
 ↓
分析数据
 ↓
调用数据库
 ↓
生成报告
 ↓
等待审批
 ↓
发送邮件

审批可能：

5分钟后
1天后
一个月后

不能：

 复制代码while waiting:
    sleep()

因为：

服务可能重启
机器可能挂掉
内存会丢失

所以 LangGraph 使用：

 复制代码checkpoint storage

保存：

 复制代码State
+
执行位置
+
线程ID

例如：

 复制代码thread_id=abc123checkpoint:{
 node:"approval",
 state:{
    order_id:10001,
    amount:50000
 }
}

6. interrupt 和消息队列有什么区别？

很多人会混淆。

MQ

例如：

 复制代码Kafkaproducer
   |
   v
topic
   |
consumer

解决：

interrupt

解决：

例如：

 复制代码订单流程创建订单
 |
付款
 |
interrupt
 |
人工审核
 |
发货

7. interrupt 在 Agent 中的典型场景

① 人工审批

最常见：

 复制代码AI生成合同
        |
        |
 interrupt
        |
律师审核
        |
继续签署

② 高风险工具调用

例如：

 复制代码AI:准备执行:DELETE DATABASEinterrupt等待确认

③ 信息补充

例如：

AI：

 复制代码我要帮你订机票缺少:出发城市

interrupt:

 复制代码请输入出发城市

④ 多 Agent 协作

例如：

 复制代码Planner Agent      |
      vResearch Agent      |
      vinterrupt专家审核      |
      vWriter Agent

8. interrupt 的核心设计理念

可以总结为：

① 图执行模型

不是：

 复制代码函数调用链

而是：

 复制代码State Machine

② 可恢复计算

类似：

数据库事务：

 复制代码begin执行checkpointcommit

③ 长生命周期 Agent

普通 LLM：

 复制代码秒级

LangGraph：

 复制代码小时
天
月

④ 人机协同

不是：

 复制代码AI替代人

而是：

 复制代码AI执行80%人在关键节点决策

9. 和 LangChain Agent 的区别

传统 LangChain Agent：

 复制代码while True:  思考  action  observation

状态主要在内存。

LangGraph：

 复制代码State GraphNode
 |
Checkpoint
 |
Interrupt
 |
Resume

更适合：

企业 Agent
工作流自动化
审批流程
数据处理 Pipeline

推荐专题

最新下载

热门教程