Human-in-the-loop 中断与审核

Human-in-the-loop 中断与审核

学习目标

本章要解决什么问题：

• 如何在 Agent 执行中途触发中断，等待人类审核
• 如何通过 Command 恢复执行并注入人类决策
• 如何配置中断的交互选项（accept、edit、response、ignore）
• 中断点与 checkpoint 的协同机制

前置知识

本章涉及 HITL 中断的通用原理，建议先阅读：

• Human-in-the-loop 中断机制 — 中断的四种响应类型、中断与异常的区别

下文假设你已理解上述概念，直接聚焦 LangGraph 的具体实现。

---

项目实践

触发中断

LangGraph 通过 interrupt() 函数在执行中暂停：

from langgraph.types import interrupt, Command

def review_node(state):
    response = interrupt({
        "question": "请审核以下内容",
        "content": state["draft"]
    })
    return {"approved": response.get("approved", False)}

interrupt() 的参数是一个字典，包含需要向人类展示的信息。返回值是人类通过 Command(resume=...) 注入的数据。

恢复执行

中断后，使用相同的 thread_id 和 Command 恢复：

config = {"configurable": {"thread_id": "session-1"}}

# 第一次调用：执行到 interrupt() 暂停
result = app.invoke({"messages": ["写一篇关于 AI 的文章"]}, config)

# 人类审核后恢复
result = app.invoke(
    Command(resume={"approved": True, "feedback": "很好"}),
    config
)

中断配置：控制人类可执行的操作

通过 interrupt_before / interrupt_after 参数在编译时配置中断点：

# 在 "review" 节点执行前中断
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["review"]
)

# 在多个节点执行后中断
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_after=["model", "tools"]
)

# 在所有节点都中断
from langgraph.types import All
app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["*"]
)

获取中断状态

中断后，可以通过 get_state() 查看当前状态：

snapshot = app.get_state(config)
print(snapshot.values)  # 当前状态
print(snapshot.next)    # 下一个要执行的节点

修改状态后恢复

人类不仅可以审核，还可以直接修改状态：

# 获取当前状态
snapshot = app.get_state(config)
current_state = snapshot.values

# 修改状态
app.update_state(config, {"draft": "修改后的内容"})

# 恢复执行
app.invoke(Command(resume=None), config)

问题与规避

中断后必须使用相同 thread_id

问题：恢复时如果使用了不同的 thread_id，系统会加载错误的 checkpoint，导致中断上下文丢失。

规避：中断时将 thread_id 与中断信息一起返回给调用方，恢复时原样传入。

中断前未持久化的数据丢失

问题：如果 Durability 设置为 async 或 exit，中断前的状态可能尚未持久化。

规避：对于需要中断的场景，将 Durability 设置为 sync，确保每次写入先持久化。

中断期间的状态过期

问题：人类审核可能耗时较长，期间外部依赖（API、数据库）可能已变化。

规避：

• 在中断前保存外部依赖的快照到状态中
• 恢复后重新验证外部依赖
• 设置中断超时时间，超时自动取消

设计取舍

优势

• 灵活的中断点：可以在任意节点前后设置中断
• 状态可修改：人类不仅可以审核，还可以直接修改状态
• 与 checkpoint 深度集成：中断后自动保存，恢复时自动加载

代价

• 人类等待延迟：审核时间不可控
• 外部系统依赖：需要外部界面展示中断信息和收集人类决策
• 状态一致性：中断期间外部世界可能已变化

替代方案

• 全自动化 + 事后审核：执行完成后审核结果
• 沙箱预览：在沙箱中执行并展示结果，确认后再应用

参考来源

• 源码验证: libs/langgraph/langgraph/types.py — Command 和 interrupt 定义
• 源码验证: libs/prebuilt/langgraph/prebuilt/interrupt.py — 中断封装
• 文档: LangGraph Interrupts, https://docs.langchain.com/oss/python/langgraph/interrupts