用动态 interrupt 实现人工审批

暂停之后继续

本章目标

暂停退款流程、持久化上下文，并根据人工决定进入执行或拒绝分支。

生产审批应使用动态 interrupt()：

from langgraph.types import Command, interrupt


def approval(state: SupportState) -> dict[str, bool]:
    approved = interrupt(
        {
            "ticket_id": state["ticket_id"],
            "question": "是否批准执行退款？",
            "proposal": state["proposal"],
        }
    )
    return {"approved": bool(approved)}

首次运行遇到中断后返回审批 payload。恢复时使用同一个 thread_id：

审批暂停与恢复

result = await graph.ainvoke(
    Command(resume=True),
    config=config,
    version="v2",
)

interrupt_before 和 interrupt_after 是静态断点，适合调试，不作为本文的生产审批方案。

恢复安全规则

副作用放在审批后

包含 interrupt() 的节点在恢复时从函数开头重新执行。因此：

  • 中断之前只能放纯计算或幂等操作。
  • 支付、退款、发送消息放在审批之后的独立节点。
  • 中断 payload 必须可序列化。
  • 不要用普通 try/except 捕获 interrupt()。

审批超时

interrupt() 默认无限等待。不要在 Web Worker 中 sleep() 或循环轮询。

生产做法是：

  1. 业务数据库记录审批请求和 deadline。
  2. 调度器扫描过期审批。
  3. 调度器使用原 thread_id 调用 Command(resume=False)。
  4. 拒绝分支记录“超时拒绝”原因。

这部分依赖企业现有调度平台，示例不伪造一个进程内定时器来冒充分布式调度。

审批记录必须独立存在

interrupt() 保存执行位置，但企业审批还需要可查询、可审计的业务记录。建议至少保存：

approval_id, tenant_id, ticket_id, thread_id,
status, proposal_hash, requested_at, deadline,
decided_at, decided_by, decision_reason, version

proposal_hash 防止审批人在看到方案 A 后，系统恢复时执行了方案 B。恢复前重新计算待执行参数并与审批记录比对；金额、收款方或动作类型变化时必须重新审批。

授权、审计和原子状态转换

审批接口不能只接收一个布尔值。生产入口应从身份系统获得 decided_by，检查审批人角色、租户、金额权限和职责分离规则，并把业务审批记录从 PENDING 原子更新为 APPROVED 或 REJECTED。

UPDATE approval_request
SET status = 'APPROVED', decided_by = :actor, version = version + 1
WHERE approval_id = :id AND status = 'PENDING' AND version = :expected_version;

受影响行数为零表示已经处理或版本冲突。只有成功完成该状态转换的请求可以提交 Command(resume=...)。当前示例通过读取 snapshot.next 返回 409，能阻止普通重复提交，但读状态和恢复之间不是跨实例原子操作，因此仍需要业务数据库或运行队列串行化。

超时任务也要幂等

调度器扫描到期记录后，应先原子地把 PENDING 改为 TIMED_OUT，再恢复图。多个调度器同时扫描时只有一个更新成功。恢复失败可以重试，因为业务记录已经固定为超时拒绝，Command(resume=False) 的后续副作用仍遵循幂等约束。

审批故障演练

至少覆盖：批准、拒绝、重复批准、批准与超时同时发生、无权限审批、方案被修改、恢复时数据库短暂不可用、审批完成后进程崩溃。每条路径都要验证“退款服务调用次数”，不能只验证 HTTP 状态码。

本章验收

  • 审批记录包含操作者、原因、截止时间和方案摘要。
  • 重复或并发审批只有一次业务状态转换成功。
  • 恢复前会重新授权并校验批准内容没有变化。
  • 超时处理由持久调度任务驱动，不占用 Web Worker。