前言：为什么我会学这个问题

上篇讲了 Checkpoint，它解决了"中断后恢复"的问题。但还有一种场景：不是机器崩溃了，而是我想让 Agent 停一下，等我确认。

比如：

• Agent 要删除用户数据——“等等，这数据真的要被删吗？”
• Agent 要转账 10 万——“让我确认一下收款人”
• Agent 要发邮件——“邮件内容对不对，我先看看”

这些场景下，Agent 不能闷头执行到底，必须停下来等待确认。Interrupt 就是来解决这个的。

前文关联：本文接续《Agent 的 Time Travel 是什么？》，Time Travel 是"回退到过去"，Interrupt 是"暂停等现在"——一个是时光倒流，一个是让时间暂停。

一、问题背景

想象一个客服 Agent 的流程，如果 Agent 不支持 Interrupt：

• 金额 1 万的退款，Agent 直接执行了
• 用户发现退错账户，但钱已经转走了

如果 Agent 支持 Interrupt：

• Agent 在执行退款前停住
• 人工确认后，Agent 才真正执行
• “让人类掌控关键决策点”
  

二、核心概念

Interrupt 是什么？

Interrupt = Agent 在某个点停下来，等确认后再继续。

就像汽车的安全带警报：检测到危险 → 停下来 → 你确认后才能继续。不是 AI 自己做主，而是 AI 提建议，操作者做决定。

# 伪代码：Interrupt 本质
def execute_refund(order_id, amount):
    if amount > 1000:
        # 停下来，等确认
        interrupt("退款金额较大，需要人工确认")
    
    # 确认后，这里才继续执行
    return process_refund(order_id, amount)

LangGraph 中的 Interrupt

LangGraph 用 interrupt() 函数实现暂停：

from langgraph.types import interrupt

def refund_node(state: OrderState) -> OrderState:
    if state["amount"] > 1000:
        # 触发中断，Agent 在这里停下
        interrupt("大额退款，需要人工确认")
    
    # 这行代码只有确认后才会执行
    state["refund_status"] = "processing"
    return state

两种中断模式

模式	说明	场景
Manual Interrupt	主动调用 interrupt()，代码停在这一点	需要人工审核的关键步骤
Auto Interrupt	满足某个条件自动触发	超时、异常、风险操作

三、错误理解 / 常见误区

误区1：Interrupt 是给 AI 用的

不对。Interrupt 是给人用的——让操作者有机会介入、确认、否决 AI 的决定。AI 停下后不做任何事，等确认指令。

误区2：Interrupt 后 AI 就停了，不会继续

不完全对。Interrupt 是暂停，不是终止。确认后，AI 会从中断点继续执行后续逻辑。

误区3：Interrupt 只能在 Node 里触发

不对。Interrupt 可以在 Node 执行前、执行后、边（Edge）判断时等多个位置触发，取决于你需要在哪里停下。

误区4：Interrupt 会破坏 Checkpoint

不对。Interrupt 保存了中断点的 State，和 Checkpoint 是两个独立的机制。恢复时可以从最近的 Checkpoint 恢复，包括中断点的状态。

四、正确实现思路

步骤1：识别需要中断的关键点

不是所有操作都需要中断，只有高风险操作才需要：

• 数据删除（不可逆）
• 金钱交易（财产损失）
• 外部发送（邮件、消息、API 调用）
• 权限变更（管理员操作）

步骤2：用 interrupt() 包装高风险操作

def high_risk_operation(state: AgentState) -> AgentState:
    # 检查是否是风险操作
    if state["action_type"] in ["delete", "refund", "send_email"]:
        # 中断，等确认
        result = interrupt({
            "action": state["action_type"],
            "details": state["action_details"],
            "confirm_prompt": f"确认执行 {state['action_type']} 吗？"
        })
        
        # 如果拒绝，直接返回
        if not result["confirmed"]:
            state["status"] = "rejected_by_human"
            return state
    
    # 确认后，继续执行
    state["status"] = "executing"
    return state

步骤3：确认后恢复执行

# 恢复执行（确认后）
config = {"configurable": {"thread_id": "user-123"}}

# 获取中断信息
interrupted_at = app.get_interrupt(config)
print(f"中断在: {interrupted_at}")

# 确认后，传入 confirmed=True 继续
resume_state = {"confirmed": True, "user_notes": "确认无误"}
result = app.invoke(resume_state, config)

五、代码示例

"""
LangGraph Interrupt 完整示例
演示：Agent 执行到风险操作 → 暂停 → 确认 → 继续/拒绝
"""

from typing import TypedDict, Literal
from langgraph.graph import StateGraph
from langgraph.types import interrupt
from langgraph.checkpoint import MemorySaver

class OrderState(TypedDict):
    """订单处理状态"""
    order_id: str
    amount: int
    action_type: str
    status: Literal["pending", "confirmed", "rejected", "completed"]
    message: str

def check_order(state: OrderState) -> OrderState:
    """检查订单"""
    print(f"  [check_order] 检查订单: {state['order_id']}")
    return {"status": "pending"}

def confirm_refund(state: OrderState) -> OrderState:
    """确认退款（高风险操作，需要中断）"""
    print(f"  [confirm_refund] 准备退款: ¥{state['amount']}")
    
    if state["amount"] > 1000:
        print(f"  [confirm_refund] 💰 大额退款，触发中断，等待人工确认...")
        
        # 触发中断，等确认
        confirmation = interrupt({
            "type": "refund_confirmation",
            "order_id": state["order_id"],
            "amount": state["amount"],
            "message": f"确认退款 ¥{state['amount']} 吗？"
        })
        
        # 检查确认结果
        if not confirmation.get("confirmed", False):
            print(f"  [confirm_refund] ❌ 拒绝退款")
            return {
                "status": "rejected",
                "message": "退款被人工拒绝"
            }
        
        print(f"  [confirm_refund] ✅ 确认退款，继续执行")
    
    # 确认后，这里才执行
    return {
        "status": "completed",
        "message": f"退款成功: ¥{state['amount']}"
    }

def notify_user(state: OrderState) -> OrderState:
    """通知用户"""
    print(f"  [notify_user] 发送通知: {state['message']}")
    return state

# 构建图
graph = StateGraph(OrderState)
graph.add_node("check_order", check_order)
graph.add_node("confirm_refund", confirm_refund)
graph.add_node("notify_user", notify_user)

graph.set_entry_point("check_order")
graph.add_edge("check_order", "confirm_refund")
graph.add_edge("confirm_refund", "notify_user")
graph.set_finish_point("notify_user")

# 编译（启用 Checkpoint + Interrupt）
checkpointer = MemorySaver()
app = graph.compile(checkpointer=checkpointer)

# ========== 模拟执行 ==========
print("=" * 60)
print("场景1：小额退款（<1000，不需要中断）")
print("=" * 60)

config1 = {"configurable": {"thread_id": "order-001"}}
result1 = app.invoke({
    "order_id": "ORD-001",
    "amount": 500,  # 小额，不需要确认
    "action_type": "refund",
    "status": "pending",
    "message": ""
}, config1)
print(f"最终状态: {result1['status']}")
print()

print("=" * 60)
print("场景2：大额退款（>1000，触发中断）")
print("=" * 60)

config2 = {"configurable": {"thread_id": "order-002"}}

# 第一次调用：执行到中断点停下
print("第一次调用：执行到大额退款，中断等待确认")
result2 = app.invoke({
    "order_id": "ORD-002",
    "amount": 5000,  # 大额，触发中断
    "action_type": "refund",
    "status": "pending",
    "message": ""
}, config2)
print(f"中断时状态: {result2.get('status', 'interrupted')}")
print()

# 第二次调用：模拟确认后继续
print("第二次调用：模拟确认（confirmed=True）")
result2_continued = app.invoke({
    "confirmed": True,
    "user_notes": "确认无误"
}, config2)
print(f"最终状态: {result2_continued['status']}")
print()

print("=" * 60)
print("场景3：大额退款，拒绝")
print("=" * 60)

config3 = {"configurable": {"thread_id": "order-003"}}

# 第一次调用：中断
result3 = app.invoke({
    "order_id": "ORD-003",
    "amount": 10000,
    "action_type": "refund",
    "status": "pending",
    "message": ""
}, config3)
print(f"中断时状态: {result3.get('status', 'interrupted')}")

# 第二次调用：模拟拒绝
result3_rejected = app.invoke({
    "confirmed": False,
    "user_notes": "收款账户有风险，暂缓"
}, config3)
print(f"最终状态: {result3_rejected['status']}")
print(f"消息: {result3_rejected['message']}")

六、运行结果 / 流程图

运行输出：

============================================================
场景1：小额退款（<1000，不需要中断）
============================================================
  [check_order] 检查订单: ORD-001
  [confirm_refund] 准备退款: ¥500
  [notify_user] 发送通知: 退款成功: ¥500
最终状态: completed

============================================================
场景2：大额退款（>1000，触发中断）
============================================================
第一次调用：执行到大额退款，中断等待确认
  [check_order] 检查订单: ORD-002
  [confirm_refund] 准备退款: ¥5000
  [confirm_refund] 💰 大额退款，触发中断，等待人工确认...
中断时状态: interrupted

第二次调用：模拟确认（confirmed=True）
  [confirm_refund] ✅ 确认退款，继续执行
  [notify_user] 发送通知: 退款成功: ¥5000
最终状态: completed

============================================================
场景3：大额退款，拒绝
============================================================
  [check_order] 检查订单: ORD-003
  [confirm_refund] 准备退款: ¥10000
  [confirm_refund] 💰 大额退款，触发中断，等待人工确认...
  [confirm_refund] ❌ 拒绝退款
最终状态: rejected
消息: 退款被人工拒绝

七、项目中怎么用

实际场景：

1. 金融交易风控：转账、提现、购买等操作前增加人工确认环节。
2. 数据删除保护：删除用户数据、清除日志等不可逆操作，必须人工确认。
3. 外部通知审核：发送邮件、短信、推送前，让人先看一遍内容对不对。
4. 权限变更控制：管理员变更用户权限、角色分配等需要二次确认。

注意点：

1. 中断点要少而精：只在真正高风险的地方中断，全是中断等于没中断。
2. 中断信息要清晰：让审核人员一眼看懂要确认什么、风险是什么。
3. 设置超时机制：长时间不确认要自动取消或降级处理。
4. 记录中断历史：方便审计，谁批准了什么、什么时候批准的。

系列关联：本文讲了 Interrupt 让 Agent"等等，需要手动确认"。下篇《如何设计一个能调用工具的 AI 助手？》会讲，Interrupt 配合工具调用，可以让 AI 在执行危险工具前先请求授权，实现"工具级别的安全控制"。

八、总结 + 下篇预告

核心要点：

• Interrupt = 在关键点暂停，等确认后再继续
• 触发条件：高风险操作（删除、交易、外部发送等）
• interrupt() 函数让 Agent 停在某一行代码，等指令
• Human-in-the-loop：让人掌控关键决策，AI 负责执行

下篇预告：下一篇我们将探讨《如何设计一个能调用工具的 AI 助手？》，理解如何设计工具描述、调用逻辑、结果处理，让 AI 能够正确、稳定地调用各种外部工具。

延伸阅读：

• LangGraph Interrupt 文档：
• Human-in-the-loop AI 设计模式