一文彻底搞懂 Human-in-the-Loop 的核心机制

前言

在构建 AI Agent 应用时，我们经常会遇到这样的场景：Agent 准备执行一个关键操作（如发送邮件、删除数据、调用付费 API）之前，我们希望停下来，等一等，让人类确认后再继续。

LangGraph 的 Interrupt（中断） 机制正是为此而生。它是实现 Human-in-the-Loop（人机协作） 的基石，让工作流能够在任意节点暂停，等待外部输入后再精确恢复执行。

本文将从零开始，带你彻底掌握 Interrupt 的所有知识点，包括：

• Interrupt 的核心概念与作用
• Checkpointer 为什么是 Interrupt 的前提
• 两次 graph.stream 流程的完整拆解
• config 与 thread_id 的底层原理
• Command 类的本质与用法
• 静态中断与动态中断的区别
• 完整可运行的代码示例

---

一、什么是 Interrupt？

Interrupt 是 LangGraph 内置的一个函数，用于在节点执行过程中主动暂停整个工作流。

1.1 核心语法

from langgraph.types import interrupt

def my_node(state: State):
    # 图会在这里暂停，并向外抛出 "请问您的年龄是？"
    user_input = interrupt("请问您的年龄是？")
    # 当用户输入恢复后，程序从这里继续执行
    return {"age": user_input}

1.2 Interrupt 的三大核心作用

作用	说明
暂停执行	立即停止当前图的运行，不继续执行后续节点
保存现场	与 Checkpointer 配合，将当前 State 和断点位置持久化
等待恢复	无限期等待外部输入，直到收到 Command(resume=...) 指令

---

二、Interrupt 的绝对前提：Checkpointer

2.1 为什么 Checkpointer 是必须的？

没有 Checkpointer，Interrupt 就无法工作！

Interrupt 依赖于 检查点（Checkpoint） 机制来保存和恢复状态：

• 中断时：Checkpointer 负责把当前的 State 和“停在哪个位置”保存下来。
• 恢复时：Checkpointer 负责把保存的状态读出来，让图“灵魂附体”般地继续运行。

如果省略了 Checkpointer，程序会在 interrupt() 处直接抛出一个致命异常（GraphInterrupt）并崩溃，无法恢复。

2.2 如何配置 Checkpointer？

from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import InMemorySaver

# 创建一个内存检查点保存器
checkpointer = InMemorySaver()

# 编译图时传入
graph = builder.compile(checkpointer=checkpointer)

2.3 三种常见的 Checkpointer

保存器	存储位置	持久化	适用场景
InMemorySaver	进程内存	❌ 重启丢失	本地开发测试
SqliteSaver	SQLite 文件	✅ 磁盘持久化	本地演示、小项目
PostgresSaver	PostgreSQL	✅ 磁盘持久化	生产环境

---

三、Interrupt 的核心流程：两次 graph.stream

3.1 为什么是两次？

Interrupt 需要两次 graph.stream 调用才能完成一个完整的“提问-回答”回合：

1. 第一次调用：执行图，遇到 interrupt 后暂停，并返回中断提示。
2. 第二次调用：传入 Command(resume=...)，恢复图的执行，直至结束。

3.2 完整流程拆解

from langgraph.types import interrupt, Command

# ==========================================
# 第一次调用：启动图，遇到 interrupt 暂停
# ==========================================
print("🚀 第一次调用：开始执行...")
for chunk in graph.stream({"user_name": "小明"}, config):
    print(f"📦 收到 chunk: {chunk}")
# 打印结果：{'__interrupt__': ('请问您的年龄是？',), 'user_name': '小明'}
print("⏸️ 第一次调用结束（图已暂停）\n")

# ==========================================
# 第二次调用：传入 Command 恢复执行
# ==========================================
print("🚀 第二次调用：开始执行...")
command = Command(resume=30)  # 把 30 传回去
for chunk in graph.stream(command, config):
    print(f"📦 收到 chunk: {chunk}")
# 打印结果：{'user_name': '小明', 'user_age': 30}
print("🏁 第二次调用结束（图执行完毕）")

3.3 两次调用的对比总结

对比维度	第一次调用	第二次调用
传入参数	初始业务数据（如 {"user_name": "小明"}）	Command(resume=值)
执行终点	停在 interrupt() 处	执行到 END
打印内容	__interrupt__ 提示 + 当前 State	节点更新后的最终 State
图的状态	暂停（Pending）	完成（Finished）
Checkpointer 角色	写入（保存现场）	读取（加载现场）

---

四、graph.stream 循环的本质

4.1 for chunk in graph.stream(...) 到底在循环什么？

结论：循环的是“图在执行过程中，一步一步向外吐出的数据包（chunk）”。

graph.stream() 是一个生成器（Generator），它遵循“图执行一步，就 yield 一个 chunk”的规则。

4.2 执行细节

1. 图开始运行，进入某个节点。
2. 节点执行完毕（或遇到 interrupt），当前超级步结束。
3. 引擎将当前的 State（或中断信息）打包成一个 chunk，向外 yield。
4. for 循环接收这个 chunk 并打印。
5. 如果图还有下一步，继续执行并 yield 下一个 chunk；如果图已暂停或结束，循环终止。

4.3 一个形象的比喻

for chunk in graph.stream(...) 就像刷短视频时手指往下滑：

• 图每执行一小段（一个超级步），就给你推一个画面（chunk）。
• 如果视频卡住了（遇到 interrupt），你就滑不动了（循环结束）。
• 等你点了“继续播放”（传入 Command），画面才会继续往下滑。

---

五、config 与 thread_id：会话的“身份证”与“档案袋”

5.1 config 的作用

config 是本次执行的配置对象，必须包含 thread_id（线程ID）。它的核心作用有三点：

作用	解释
1. 打开正确的档案袋	Checkpointer 根据 thread_id 找到对应的历史 State，实现多轮对话记忆
2. 存盘时的标签	当 interrupt 触发时，Checkpointer 用 thread_id 做标签保存现场
3. 隔离不同会话	不同 thread_id 的 State 完全隔离，互不干扰

5.2 配置示例

config = {"configurable": {"thread_id": "user_123"}}

# 第一次调用（带初始数据）
for chunk in graph.stream({"query": "你好"}, config):
    print(chunk)

# 第二次调用（恢复中断，必须用同一个 thread_id）
command = Command(resume="30")
for chunk in graph.stream(command, config):
    print(chunk)

5.3 用类比彻底记住

• config = 医院的就诊卡（包含 thread_id）。
• thread_id = 就诊卡上的病历号。
• checkpointer = 医院的病历档案室。
• 第一次 stream = 挂号（创建新病历，或者调取旧病历）。
• interrupt = 医生问诊后，让你先去交费（暂停，等你回来）。
• 第二次 stream（带 Command）= 你交完费，把单子（resume 值）给医生，医生继续看病。

---

六、Command 类：万能遥控器

6.1 Command 是什么？

Command 是一个类，不是方法。 它像一个“指令封装器”，用来向 LangGraph 引擎下达控制命令。

6.2 Command 支持哪些命令？

参数名	类型	作用
resume	Any	“接着干”：给暂停的 interrupt() 传入值，让它继续运行
update	Dict	“改状态”：直接修改当前的 State，不经过节点函数
goto	str	“跳过去”：直接指定下一步去哪个节点

6.3 使用示例

from langgraph.types import Command

# 最常用：恢复中断，并传入用户输入
command = Command(resume="30")

# 进阶：同时修改状态并跳转
command = Command(
    update={"user_info": "张三"},
    goto="next_node"
)

6.4 为什么 Command 必须是独立的类？

因为 LangGraph 的设计哲学是 “执行与指令分离”。

• 如果直接传一个字典，引擎分不清“这是新的业务数据”还是“控制指令”。
• Command 是一个独立的类型，引擎一看到就知道：“哦，这不是业务数据，是给我的控制信号。”

一句话：Command 是遥控器，graph.stream() 是发射按钮，resume 是按键功能。

---

七、静态中断 vs 动态中断

LangGraph 提供了两种中断方式：

模式	实现方式	特点	适用场景
静态中断	编译时指定 interrupt_before / interrupt_after	在固定节点前/后暂停，简单直接	固定检查点（如执行工具前强制审批）
动态中断	节点内调用 interrupt()	根据逻辑动态决定是否暂停，极其灵活	条件触发（如遇到不确定情况主动求助）

静态中断示例

graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["tools"],  # 在执行 tools 节点前暂停
    interrupt_after=["chatbot"]  # 在执行 chatbot 节点后暂停
)

动态中断示例

def ask_human_node(state: State):
    # 当满足特定条件时才触发中断
    if state.get("need_approval"):
        confirm = interrupt("请确认是否继续执行？")
    else:
        confirm = "yes"
    return {"confirmed": confirm}

---

八、完整可运行代码示例

下面是一个完整的、可直接运行的 Interrupt 示例：

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

# 1. 定义状态
class State(TypedDict):
    user_name: str
    user_age: int

# 2. 定义节点（包含 interrupt）
def ask_age_node(state: State):
    print(f"👋 你好，{state['user_name']}。我现在要问你的年龄。")
    age = interrupt("请问您的年龄是？")
    print(f"✅ 收到年龄回复：{age} 岁")
    return {"user_age": age}

# 3. 构建图
builder = StateGraph(State)
builder.add_node("ask_age", ask_age_node)
builder.add_edge(START, "ask_age")

# 4. 编译（必须传入 checkpointer）
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 5. 配置（thread_id 是会话标识）
config = {"configurable": {"thread_id": "user_123"}}

# ==========================================
# 第一次调用：启动图并暂停
# ==========================================
print("\n🚀 第一次调用开始...")
for chunk in graph.stream({"user_name": "小明"}, config):
    print(f"📦 chunk: {chunk}")
print("⏸️ 图已暂停，等待用户输入...\n")

# ==========================================
# 第二次调用：恢复执行
# ==========================================
print("🚀 第二次调用开始...")
command = Command(resume=25)
for chunk in graph.stream(command, config):
    print(f"📦 chunk: {chunk}")
print("🏁 图执行完毕！")

运行结果

🚀 第一次调用开始...
👋 你好，小明。我现在要问你的年龄。
📦 chunk: {'__interrupt__': ('请问您的年龄是？',), 'user_name': '小明'}
⏸️ 图已暂停，等待用户输入...

🚀 第二次调用开始...
✅ 收到年龄回复：25 岁
📦 chunk: {'user_name': '小明', 'user_age': 25}
🏁 图执行完毕！

---

九、常见误区与澄清

误区 1：interrupt 可以在没有 Checkpointer 的情况下工作

❌ 错误。没有 Checkpointer，interrupt 会直接抛出 GraphInterrupt 异常导致程序崩溃。

✅ 正确：interrupt 必须与 checkpointer 配合使用。

---

误区 2：Command 是一个方法

❌ 错误。Command 是一个类，Command(resume=...) 是在创建它的实例。

✅ 正确：Command 是一个类，用来封装控制指令。

---

误区 3：invoke 也能处理中断

❌ 错误。graph.invoke() 遇到中断会直接抛出异常。

✅ 正确：必须使用 graph.stream() 来优雅地处理中断。

---

误区 4：中断后只能用同一个程序恢复

❌ 错误。如果使用 PostgresSaver 或 SqliteSaver，即使程序重启，只要传入相同的 thread_id，就能从数据库读取存档并恢复执行。

✅ 正确：持久化的 Checkpointer 支持跨进程、跨时间的状态恢复。

---

误区 5：__interrupt__ 中的数字（如 [reference:23]）是代码的一部分

❌ 错误。这通常是 AI 聊天界面自动添加的引用标记，与代码逻辑完全无关。

✅ 正确：忽略这些数字，它们不影响代码运行。

---

十、知识点速查表

概念	一句话解释
interrupt()	一个函数，在节点内调用，暂停图执行并向外抛出信息
Checkpointer	Interrupt 的绝对前提，负责保存和加载状态快照
第一次 stream	启动图，遇到 interrupt 暂停，返回 __interrupt__ 信息
第二次 stream	传入 Command(resume=...)，恢复图并执行到 END
chunk	图每执行一个超级步就向外吐出的数据包
config	执行的配置，包含 thread_id（会话标识）
thread_id	会话的“病历号”，用于查找和隔离状态
Command	一个类，用于封装控制指令（恢复、跳转、更新状态）
静态中断	编译时指定 interrupt_before / interrupt_after
动态中断	节点内调用 interrupt()，根据逻辑动态触发

---

十一、总结

Interrupt 机制是 LangGraph 构建安全、可控、可交互的 AI Agent 应用的基石。它的核心设计可以浓缩为三句话：

1. 暂停与恢复：interrupt() 负责暂停，Command(resume=...) 负责恢复，两者通过 Checkpointer 无缝衔接。
2. 两次 Stream：第一次启动并暂停，第二次输入答案并继续，中间可以等待任意时长。
3. 会话隔离：thread_id 是唯一标识，不同会话的 State 完全隔离，互不干扰。

掌握了这些知识点，你就可以构建出真正具备 Human-in-the-Loop 能力的智能体——让 AI 在需要人类决策时主动“停下来问一问”，从而做出更准确、更可靠的响应。

下一步可以尝试：

• 在复杂工作流中嵌入多个 interrupt 节点
• 使用 PostgresSaver 实现跨程序恢复
• 结合 Command.update 在恢复时直接修改 State

---

本文基于 LangGraph 官方文档及实战经验总结，如有疑问欢迎交流。