LangGraph 系列 14：使用静态断点实现工具执行审批

langgraph workflow human-in-the-loop interrupt
langgraph
发布日期:   2026-06-08
文章字数:   2.1k
阅读时长:   8 分
阅读次数:  

智能小秘书已经能够调用搜索和票务查询工具。这些工具只读取数据，即使参数不理想，通常也不会直接修改外部系统。

发送邮件、提交审批、删除数据或执行付款则不同。模型可以提出工具请求，但真正产生副作用前，应用应该让用户确认。

本文先使用 LangGraph 的静态断点 interrupt_before，让 Workflow 在进入 tools 节点前暂停。用户批准时继续执行；拒绝时跳过工具，并把拒绝结果写回消息状态。

1. 为什么不能只让模型判断是否执行

模型可以判断某个工具“看起来应该调用”，但它不能代替业务所有者承担后果。例如模型可能：

  • 选择了错误的收件人。
  • 从对话中提取了错误参数。
  • 忽略了最新的业务限制。
  • 对含糊指令做出过度推断。

工具调用分成两个阶段更合理：

模型提出 Tool Call
-> 应用检查或人工审批
-> 真正执行工具

2. 静态断点如何暂停流程

使用 interrupt_before 在工具节点前暂停

编译图时加入：

graph = builder.compile(
    checkpointer=InMemorySaver(),
    interrupt_before=["tools"],
)

执行路径到达 tools 前会停止。此时：

  • chatbot 已经生成 Tool Call。
  • Tool Call 已经保存在 messages。
  • ToolNode 尚未执行。
  • Checkpointer 已经保存暂停位置。

静态断点必须配合 Checkpointer 和 thread_id，否则没有可恢复的位置。

interrupt_before 不是一个真正的图节点，也不会自动弹出审批页面。它是编译后运行时的暂停规则：执行器准备调度 tools 时停止，并把控制权交还给调用方。读取 Tool Call、展示审批界面以及决定批准或拒绝，都属于 Workflow 外部的应用逻辑。

本文使用的 InMemorySaver 只适合同一进程内演示暂停和恢复。进程退出后内存状态会丢失；需要等待较长时间或跨进程审批时，应改用 PostgreSQL 等持久化 Checkpointer。

3. 使用确定性模型节点验证审批

本篇重点是断点和状态恢复，不需要把模型不确定性混进来。示例使用固定的 chatbot 节点模拟模型：

def chatbot(state: MessagesState) -> dict:
    """使用固定消息模拟模型提出工具调用和整理工具结果。"""

    last_message = state["messages"][-1]
    if isinstance(last_message, HumanMessage):
        return {
            "messages": [
                AIMessage(
                    content="",
                    tool_calls=[
                        {
                            "name": "send_email",
                            "args": {
                                "recipient": "demo@example.com",
                                "subject": "周末散步计划",
                            },
                            "id": "call-send-email",
                            "type": "tool_call",
                        }
                    ],
                )
            ]
        }
    if isinstance(last_message, ToolMessage):
        return {"messages": [AIMessage(content=f"处理结果：{last_message.content}")]}
    return {"messages": [AIMessage(content="没有需要处理的工具请求。")]}

邮件工具不会发送真实邮件，只把一行内容写入本地演示发件箱：

@tool
def send_email(recipient: str, subject: str) -> str:
    """把一封演示邮件写入本地发件箱。"""

    OUTBOX.parent.mkdir(parents=True, exist_ok=True)
    OUTBOX.write_text(f"收件人：{recipient}\n主题：{subject}\n", encoding="utf-8")
    return "演示邮件已经写入本地发件箱"

这样可以直接通过文件是否存在判断副作用有没有发生。

4. 检查暂停状态

图结构仍然是标准工具循环：

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

graph = builder.compile(
    checkpointer=InMemorySaver(),
    interrupt_before=["tools"],
)

第一次调用提供固定线程：

config = {"configurable": {"thread_id": "p14-inspect"}}

graph.invoke(
    {"messages": [{"role": "user", "content": "请发送周末散步计划邮件。"}]},
    config=config,
)
snapshot = graph.get_state(config)

print("下一节点：", snapshot.next)
print("工具调用：", snapshot.values["messages"][-1].tool_calls[0]["name"])
print("发件箱存在：", OUTBOX.exists())

运行：

cd source/_posts/llm_learning
source .venv_langgraph/bin/activate
python langgraph/p14_static_tool_approval/01_interrupt_before_tools.py

真实输出：

下一节点： ('tools',)
工具调用： send_email
发件箱存在： False

snapshot.next 指向 tools，说明下一步准备执行工具。发件箱不存在，证明副作用还没有发生。

静态断点暂停后的 StateSnapshot

5. 批准工具执行

静态断点通过传入 None 继续：

final_state = graph.invoke(None, config=config)

必须使用与暂停时相同的 thread_id。Checkpointer 根据它查找对应的 State 和下一节点。

完整批准逻辑：

OUTBOX.unlink(missing_ok=True)
config = {"configurable": {"thread_id": "p14-approve"}}

graph.invoke(
    {"messages": [{"role": "user", "content": "请发送周末散步计划邮件。"}]},
    config=config,
)
print("暂停后发件箱存在：", OUTBOX.exists())

final_state = graph.invoke(None, config=config)
lines = OUTBOX.read_text(encoding="utf-8").splitlines()

assert len(lines) == 1
print("批准后发件箱记录数：", len(lines))
print("最终回答：", final_state["messages"][-1].content)

真实输出：

暂停后发件箱存在： False
批准后发件箱记录数： 1
最终回答： 处理结果：演示邮件已经写入本地发件箱

记录数断言可以发现工具被重复执行的问题。

6. 拒绝工具执行

静态断点的 graph.invoke(None) 只表示继续，没有内置的“拒绝”参数。如果用户拒绝，需要自己构造一条工具结果，然后从工具节点之后继续。

首先取得模型生成的 Tool Call：

snapshot = graph.get_state(config)
tool_call = snapshot.values["messages"][-1].tool_calls[0]

然后写入匹配 ID 的 ToolMessage：

graph.update_state(
    config,
    {
        "messages": [
            ToolMessage(
                content="用户拒绝执行：收件人还没有确认。",
                name=tool_call["name"],
                tool_call_id=tool_call["id"],
            )
        ]
    },
    as_node="tools",
)

as_node=”tools” 表示这次更新相当于 tools 节点已经完成。LangGraph 会按照 tools -> chatbot 的边继续，而不会再次进入被暂停的 ToolNode。

这里的“相当于完成”只用于计算下一条边，并不会暗中执行 send_email。真正写入 State 的工具结果，是应用手动构造的拒绝 ToolMessage。

最后恢复流程：

final_state = graph.invoke(None, config=config)

assert not OUTBOX.exists()
print("拒绝后发件箱存在：", OUTBOX.exists())
print("最终回答：", final_state["messages"][-1].content)

真实输出：

拒绝后发件箱存在： False
最终回答： 处理结果：用户拒绝执行：收件人还没有确认。

拒绝也必须返回 ToolMessage。如果只删除 Tool Call 或直接添加普通 AI 消息，消息序列会失去“工具请求与工具结果”的配对关系。

7. 静态断点的限制

interrupt_before=[“tools”] 针对节点，而不是某个工具。如果一个 ToolNode 中有搜索、查询、发邮件和删除数据四个工具，它们都会在执行前暂停。

这会带来两个问题：

  • 只读查询也被迫等待人工批准。
  • 审批界面需要自行判断当前 Tool Call 的风险。

本文锁定版本的 LangGraph Interrupts 文档将静态断点主要定位为图调试手段。正式 Human-in-the-loop 更适合在业务代码中使用动态 interrupt()，只在满足条件时暂停。

8. 常见问题

8.1 没有配置 Checkpointer

静态断点需要保存暂停位置。没有 Checkpointer，流程无法可靠恢复。

8.2 恢复时换了 thread_id

新的 thread_id 会被视为另一条执行线程，找不到原来的暂停状态。

8.3 拒绝后仍然执行了工具

不能先调用 graph.invoke(None) 再写拒绝结果。None 会让流程进入真实 ToolNode。拒绝路径应该先 update_state(…, as_node=”tools”)。

8.4 ToolMessage 没有匹配调用 ID

tool_call_id 必须来自当前 AIMessage.tool_calls。不要自己生成另一个 ID。

8.5 把只读工具都加入审批

审批应该与风险相匹配。本文没有审批 Bing 和 12306 查询，而是使用本地模拟邮件验证副作用。

9. 总结

知识点 作用 本文验证结果
interrupt_before 在指定节点执行前暂停 成功暂停在 tools 前
Checkpointer 保存 State 和暂停位置 使用 InMemorySaver，仅验证同进程恢复
thread_id 定位同一条执行线程 相同 ID 可以恢复
StateSnapshot.next 查看接下来准备执行的节点 返回 (‘tools’,)
批准 允许 Workflow 继续 graph.invoke(None) 执行一次工具
拒绝 跳过真实工具执行 发件箱始终不存在
update_state() 在暂停后人工修改 State 写入拒绝 ToolMessage
as_node=”tools” 指定更新相当于哪个节点完成 从 tools 后续边继续
tool_call_id 配对工具请求与工具结果 使用原 Tool Call ID
静态断点限制 节点级暂停，无法按工具选择 更适合调试和粗粒度暂停

静态断点可以清楚展示“暂停、检查、批准、拒绝”的基本过程，但审批范围太粗。下一篇会把 interrupt() 放到高风险工具内部，并使用 Command(resume=…) 恢复流程。

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