zhuangang/scenarios/approval-human-in-loop.md

id, title, type, domain, stack, tags, level, prerequisites, est_hours, related, week

id	title	type	domain	stack	tags	level	prerequisites	est_hours	related	week
scenario.approval-human-in-loop	审批流 + 人工介入的 Agent：高风险操作暂停审批、断点续跑、超时和幂等执行	scenario	cross	langgraph function-calling postgresql redis fastapi vue	人机协作 interrupt 检查点 审批 幂等 超时 审计	advanced	ai.langgraph.index ai.function-calling.index	3	ai.langgraph.scenario.refund-approval ai.function-calling.scenario.safe-business-tools ai.agent-design.index	11 12

审批流 + 人工介入的 Agent：高风险操作暂停审批、断点续跑、超时和幂等执行

业务背景与问题

AI 客服 Agent 可以办理退款、改地址、补发商品。业务要求：

1. 金额超过阈值或命中风险规则的操作必须人工审批；
2. 审批可能几小时后才完成，期间服务可能重启；
3. 审批通过后只能执行一次（不能重复退款）；
4. 审批超时自动拒绝并通知用户；
5. 全程可审计：谁发起、模型建议了什么、谁批准、实际执行了什么。

约束

• Agent 不能自己“决定”跳过审批，不能依赖提示词约束。
• 审批人在后台系统操作，不在对话中。

方案设计

1. 流程建模（LangGraph）：理解意图 → 查订单 → 计算方案 → 风险判断 → [需审批] interrupt 暂停 → 执行 → 通知。风险判断是代码规则（金额、频次、用户等级），不是模型判断。
2. 持久化：检查点存数据库（开发 SQLite、生产 PostgreSQL），thread_id 与审批单关联，重启后从检查点恢复。
3. 审批单：暂停时创建审批记录（方案、模型理由、风险命中项、截止时间）推送到后台；审批后用 Command(resume=...) 恢复对应线程。
4. 幂等执行：执行节点调用退款接口时带幂等键（thread_id + 步骤名），恢复或重试不会重复退款。
5. 超时：延迟任务到期检查审批状态，未处理则以“驳回（超时）”恢复。
6. 前端：用户端显示“已提交审核”，状态变化推送更新；审批后台展示对话摘要和方案。
7. 审计：每步写事件表，包括模型输出、工具参数和结果。

关键代码

审批节点（示意；可运行版本见 examples/ticket-flow，含 SqliteSaver、interrupt、崩溃恢复和幂等 Mailer 的测试）：

from langgraph.types import Command, interrupt

def risk_check(state: RefundState) -> dict:
    return {"need_approval": state["amount"] > 500 or state["user_refunds_30d"] >= 3}   # 规则写在代码里

def approval(state: RefundState) -> dict:
    decision = interrupt({"ticket": state["ticket_id"], "amount": state["amount"], "reason": state["plan_reason"]})
    return {"approved": decision["approved"], "approver": decision["by"]}

def execute(state: RefundState) -> dict:
    if state["need_approval"] and not state.get("approved"):
        return {"result": "rejected"}
    key = f"{state['thread_id']}:refund"                    # 幂等键
    return {"result": refund_api(state["order_id"], state["amount"], idempotency_key=key)}

def on_approval(thread_id: str, approved: bool, by: str) -> None:        # 审批后台回调
    graph.invoke(Command(resume={"approved": approved, "by": by}),
                 config={"configurable": {"thread_id": thread_id}})

取舍

决策	选择	原因
是否需要审批	代码规则	模型可能被注入或误判
流程引擎	LangGraph（interrupt + 检查点）	与 Agent 天然结合；Temporal 等工作流引擎更重
恢复方式	审批回调触发 resume	比轮询及时
执行保障	幂等键 + 状态检查	恢复和重试都可能重复执行节点

易错点

• 把“是否需要审批”交给模型判断。
• interrupt 所在节点恢复时会从头重新执行，节点内 interrupt 之前的副作用会重复——副作用放在 interrupt 之后的节点，或保证幂等。
• 审批超时没有处理，线程永远挂起。
• 审批人看到的信息不足，只能盲批。

面试怎么讲

“高风险操作我用 LangGraph 的 interrupt 做人工审批。是否需要审批由代码规则决定，不交给模型。暂停时创建审批单，检查点存数据库，服务重启不丢；审批回调用 Command(resume) 恢复线程。执行退款用 thread_id 加步骤名作为幂等键，恢复或重试都不会重复退款。超时由延迟任务以驳回恢复，每一步都写审计事件。”