LangGraph 系列 15：使用 interrupt 与 Command 实现可恢复人工介入

langgraph human-in-the-loop interrupt command
langgraph
发布日期:   2026-06-11
文章字数:   2.6k
阅读时长:   10 分
阅读次数:  

上一篇使用 interrupt_before=[“tools”] 在整个工具节点前暂停。它能实现基本审批，但同一个 ToolNode 中的安全查询和高风险操作都会被拦截。

更实用的方式是在真正需要审批的位置调用 interrupt()：查询城市资料可以直接执行，发送邮件才暂停。用户提交批准、拒绝或修改参数后，应用使用 Command(resume=…) 恢复原线程。

本文完成三件事：

  1. 根据工具风险选择是否中断。
  2. 验证批准、拒绝和修改参数三种反馈。
  3. 使用 PostgreSQL 保存暂停状态，在新进程中继续执行。

1. 动态中断与静态断点的区别

静态断点在编译时指定节点：

builder.compile(interrupt_before=["tools"])

动态中断直接写在业务代码中：

decision = interrupt(
    {
        "action": "review_tool_call",
        "tool_name": "send_email",
        "arguments": {"recipient": recipient, "subject": subject},
        "allowed_actions": ["approve", "reject", "edit"],
    }
)

只有执行到这行代码时才暂停，因此可以精确放在高风险工具内部。

按工具风险选择是否人工审批

2. interrupt() 实际做了什么

调用 interrupt(payload) 时，LangGraph 会：

  1. 暂停当前节点的本轮执行。
  2. 通过 Checkpointer 保存图状态。
  3. 把可序列化的 payload 返回调用方。
  4. 等待相同线程使用 Command(resume=…) 恢复。

恢复值会成为 interrupt() 的返回值：

final_state = graph.invoke(
    Command(
        resume={
            "action": "approve",
            "arguments": {},
            "reason": "内容已经确认",
        }
    ),
    config=config,
)

interrupt 与 Command 的可恢复审批

3. 定义安全工具和高风险工具

安全工具只返回固定城市资料，不产生外部副作用：

@tool
def lookup_city_info(city: str) -> str:
    """查询固定的演示城市资料。"""

    return f"{city}：浙江省省会，著名景点是西湖。"

邮件工具先中断，审批通过后才写入本地发件箱：

@tool
def send_email(recipient: str, subject: str) -> str:
    """审批通过后，把演示邮件写入本地发件箱。"""

    decision = interrupt(
        {
            "action": "review_tool_call",
            "tool_name": "send_email",
            "arguments": {"recipient": recipient, "subject": subject},
            "allowed_actions": ["approve", "reject", "edit"],
        }
    )
    action = decision.get("action")
    if action == "reject":
        return f"用户拒绝发送邮件：{decision.get('reason', '未提供原因')}"
    if action not in {"approve", "edit"}:
        return f"未执行邮件：不支持的审批操作 {action}"

    final_arguments = decision.get("arguments", {})
    final_recipient = final_arguments.get("recipient", recipient)
    final_subject = final_arguments.get("subject", subject)

    OUTBOX.parent.mkdir(parents=True, exist_ok=True)
    with OUTBOX.open("a", encoding="utf-8") as file:
        file.write(f"收件人：{final_recipient}；主题：{final_subject}\n")
    return "演示邮件已经写入本地发件箱"

真正产生副作用的 OUTBOX.open() 位于 interrupt() 和审批动作校验之后，这是整个实现中最重要的顺序。未知动作必须按拒绝执行处理，不能落入默认发送分支。

4. 验证选择性中断

两个工具放入同一个 ToolNode：

builder = StateGraph(MessagesState)
builder.add_node("chatbot", chatbot)
builder.add_node("tools", ToolNode([lookup_city_info, send_email]))
builder.add_edge(START, "chatbot")
builder.add_conditional_edges("chatbot", tools_condition)
builder.add_edge("tools", "chatbot")
graph = builder.compile(checkpointer=InMemorySaver())

分别运行安全查询和邮件操作：

cd source/_posts/llm_learning
source .venv_langgraph/bin/activate
python langgraph/p15_dynamic_human_in_the_loop/01_selective_interrupt.py

真实输出：

安全工具发生中断： False
安全工具最终回答： 处理结果：杭州：浙江省省会，著名景点是西湖。
高风险工具发生中断： True
待审批工具： send_email
允许的操作： ['approve', 'reject', 'edit']
发件箱存在： False

中断发生在工具内部。ToolNode 只负责按照 Tool Call 调用对应工具，不负责在执行前判断风险；是否暂停由每个工具自己的实现决定。

5. 批准工具执行

批准时可以不修改原参数：

Command(
    resume={
        "action": "approve",
        "arguments": {},
        "reason": "内容已经确认",
    }
)

节点恢复后，interrupt() 返回这个字典。代码读取 action，然后使用原始收件人和主题执行工具。

6. 拒绝工具执行

拒绝时传入：

Command(
    resume={
        "action": "reject",
        "arguments": {},
        "reason": "收件人还没有确认",
    }
)

工具不会写发件箱，而是返回一段普通结果：

if action == "reject":
    return f"用户拒绝发送邮件：{decision.get('reason', '未提供原因')}"

ToolNode 会把字符串包装成与原 Tool Call ID 匹配的 ToolMessage。模型仍然能读取拒绝结果并生成最终回答。

7. 修改工具参数

审批界面还可以允许用户修正模型生成的参数：

Command(
    resume={
        "action": "edit",
        "arguments": {
            "recipient": "friend@example.com",
            "subject": "修改后的周末计划",
        },
        "reason": "修正收件人与主题",
    }
)

工具使用人工提交的字段覆盖原参数：

edited_arguments = decision.get("arguments", {})
final_recipient = edited_arguments.get("recipient", recipient)
final_subject = edited_arguments.get("subject", subject)

运行三个审批案例：

python langgraph/p15_dynamic_human_in_the_loop/02_approve_reject_and_edit.py

真实输出：

批准结果： 处理结果：演示邮件已经写入本地发件箱
拒绝结果： 处理结果：用户拒绝发送邮件：收件人还没有确认
修改结果： 处理结果：演示邮件已经写入本地发件箱
实际发送记录数： 2
修改后的记录： 收件人：friend@example.com；主题：修改后的周末计划

三次请求只有批准和修改产生记录，因此发件箱中正好有两行。

8. 查看中断信息

使用普通 invoke() 时，中断信息出现在结果的 interrupt 中：

paused = graph.invoke(inputs, config=config)
payload = paused["__interrupt__"][0].value

也可以从状态快照读取：

snapshot = graph.get_state(config)
payload = snapshot.interrupts[0].value

完整脚本会比较两处内容：

invoke 返回的待审批工具： send_email
StateSnapshot 中的待审批工具： send_email
暂停节点： ('tools',)
恢复后的最终回答： 处理结果：用户拒绝发送邮件

不要把中断对象中的动态 ID 写进教程日志。调用方真正需要的是 value 中的业务信息。

9. 为什么恢复时节点会重新执行

interrupt() 不是普通函数的阻塞等待。恢复后，LangGraph 会从当前节点开头重新执行，并按照同一节点中的中断顺序，让对应的 interrupt() 返回 Command(resume=…) 的值。本文每个工具只有一个中断点，因此恢复关系非常直接。

因此下面的写法有风险：

write_database()       # 恢复时会再次执行
decision = interrupt(...)

应该改成：

decision = interrupt(...)
if decision["action"] == "approve":
    write_database()

中断之前如果必须执行某些操作，它们必须是幂等的，例如读取配置、构造显示数据或使用固定键执行可重复写入。

10. 不要捕获 interrupt()

LangGraph 通过内部特殊异常暂停节点。下面的宽泛捕获可能破坏中断传播：

try:
    decision = interrupt(...)
except Exception:
    return "执行失败"

不要把 interrupt() 放进宽泛的 try/except。真正的业务异常应该在中断前后分别处理。

11. 使用 PostgreSQL 跨进程恢复

InMemorySaver 只在当前 Python 进程中保存状态。真实人工审批可能等待数分钟甚至数天，需要持久化 Checkpointer。

本文复用 LangGraph 系列 9 的独立 PostgreSQL：

docker compose -f langgraph/p09_agent_memory/docker-compose.yml up -d

暂停脚本使用固定线程并先清理旧演示状态：

config = {"configurable": {"thread_id": "p15-postgres-email"}}

with PostgresSaver.from_conn_string(DB_URI) as postgres_checkpointer:
    postgres_checkpointer.setup()
    postgres_checkpointer.delete_thread("p15-postgres-email")
    graph = build_graph(postgres_checkpointer)
    paused = graph.invoke(
        {"messages": [{"role": "user", "content": "请发送跨进程测试邮件。"}]},
        config=config,
    )

运行：

python langgraph/p15_dynamic_human_in_the_loop/04_pause_with_postgres.py

输出：

流程已经暂停： True
待审批工具： send_email
发件箱存在： False
可以退出当前进程，再运行 05_resume_with_postgres.py。

第一个 Python 进程已经结束。新进程重新编译相同图，通过同一 thread_id 读取 Checkpoint：

with PostgresSaver.from_conn_string(DB_URI) as postgres_checkpointer:
    graph = build_graph(postgres_checkpointer)
    snapshot = graph.get_state(config)
    if not snapshot.interrupts:
        raise RuntimeError("没有找到待恢复的中断")

    final_state = graph.invoke(
        Command(
            resume={
                "action": "approve",
                "arguments": {},
                "reason": "跨进程恢复测试通过",
            }
        ),
        config=config,
    )

运行：

python langgraph/p15_dynamic_human_in_the_loop/05_resume_with_postgres.py

真实输出：

恢复前待审批工具： send_email
恢复后发件箱存在： True
最终回答： 处理结果：演示邮件已经写入持久化案例发件箱

这证明暂停位置不依赖原 Python 进程。数据库保存的是 Workflow State 和 Checkpoint，不是一个仍在阻塞的函数调用。

12. 常见问题

12.1 没有 Checkpointer

动态中断同样需要 Checkpointer。没有持久化状态，LangGraph 不知道恢复哪一次运行。

12.2 恢复时使用新的 thread_id

新的线程不会看到原中断。初始执行和 Command(resume=…) 必须使用相同 ID。

12.3 interrupt 之前已经产生副作用

恢复会重新执行节点，导致邮件、数据库写入或付款重复发生。把副作用放到审批之后，并增加业务幂等键。

12.4 审批参数结构不稳定

调用方和工具应该约定固定的 action、arguments、reason 字段。未知操作不能默认为批准。

12.5 拒绝后模型无法继续

拒绝也要作为工具结果返回。使用 ToolNode 时返回字符串即可，由节点构造匹配 ID 的 ToolMessage。

12.6 使用 InMemorySaver 等待长期审批

进程退出后内存状态会丢失。需要跨进程等待时使用 PostgreSQL 等持久化 Checkpointer。

13. 总结

知识点 作用 本文验证结果
interrupt() 在业务代码的指定位置暂停 只有邮件工具触发中断
中断 Payload 把待审批工具和参数交给调用方 包含工具名、参数和允许动作
Command(resume=…) 将人工反馈交回原节点 批准、拒绝、修改均成功
选择性审批 按工具风险决定是否暂停 城市查询直接执行
参数修改 人工修正模型生成的 Tool Call 收件人和主题被替换
拒绝结果 不执行副作用但保持消息闭环 最终模型能够说明拒绝原因
节点重放 恢复时从节点开头重新执行 副作用放在中断之后
interrupt invoke() 返回的中断信息 可读取稳定业务 Payload
StateSnapshot.interrupts 从线程状态检查待处理中断 与 invoke 结果一致
InMemorySaver 当前进程内保存暂停状态 适合教学和短暂等待
PostgresSaver 持久化 Checkpoint 新 Python 进程恢复成功
thread_id 定位同一条暂停线程 跨进程使用固定 ID

从异步工具执行、双 MCP 小秘书到静态和动态人工介入，一条完整的工具 Workflow 已经形成。关键不是让模型获得更多权限，而是在工具真正执行前建立清晰、可恢复、可审计的控制边界。

文章作者: hnbian
文章链接: https://www.hnbian.cn/posts/7cd6846.html
版权声明: 本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 hnbian !
langgraph human-in-the-loop interrupt command
评论
  目录