Skip to content

哪些操作必须让人确认?HITL 审批与接管设计 ​

内容摘要:删除、支付、发消息、改权限、执行生产命令、写业务数据——这些高风险操作必须经过人工确认。本章给出高风险工具的完整控制流程,并附一个可运行、可测试的审批闸门实现。 难度:高级 |前置知识:06 工具调用、09 状态

常见误区

界面上弹出一个“确认”按钮就完成了人工审批;模型可以自行生成确认摘要并决定执行参数。

先给结论 ​

人在回路(Human-in-the-Loop,HITL)必须绑定不可变的操作预览、审批主体、过期时间和幂等键。审批后执行的必须是人看到的同一份参数,高风险动作还需重新鉴权、审计和结果核验。

带着这三个问题阅读:

  1. 哪些副作用必须暂停任务并等待人工决策?
  2. 如何防止“审批的是 A,执行时参数变成 B”?
  3. 拒绝、超时、接管和恢复分别需要什么状态?

一、为什么需要 HITL ​

模型会犯错、会被注入、有非确定性。对不可逆或高影响的操作,正确的架构不是"让模型更聪明",而是"在执行前插入人"。人机协作(Human-in-the-Loop,HITL)就是把人放在关键决策点上。

哪些操作必须 HITL:删除数据、支付/退款、发送对外消息、发布内容、修改权限、执行生产命令、写入业务数据、任何标注为 high 风险的动作。

二、高风险操作的完整控制流程 ​

这八步缺一不可。注意顺序:校验在预览前,预览在确认前,确认在执行前,审计贯穿始终。

三、可运行的审批闸门 ​

python
"""approval_gate.py —— 高危操作审批闸门(可离线测试)。"""
from __future__ import annotations
import uuid
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Callable, Literal

class ApprovalRequired(Exception):
    """表示操作需人工审批,附预览信息。"""
    def __init__(self, preview: dict):
        super().__init__("需要人工审批")
        self.preview = preview

@dataclass
class AuditLog:
    entries: list[dict] = field(default_factory=list)
    def record(self, **kw) -> None:
        self.entries.append({"ts": datetime.now(timezone.utc).isoformat(), **kw})

@dataclass
class HighRiskTool:
    name: str
    risk: Literal["low", "high"]
    validator: Callable[[dict], dict]
    executor: Callable[..., dict]

def propose(tool: HighRiskTool, args: dict, *, actor: str, audit: AuditLog) -> dict:
    """步骤①-④:校验并生成预览。高危工具在此**中断**,不执行。"""
    valid = tool.validator(dict(args))                 # ② ③ 校验
    preview = {"tool": tool.name, "args": valid, "actor": actor,
               "idempotency_key": str(uuid.uuid4())}
    audit.record(event="propose", **preview)
    if tool.risk == "high":
        raise ApprovalRequired(preview)                # ④ 展示预览并中断
    return execute(tool, preview, audit=audit, approved=True)  # 低危可直接执行

def execute(tool: HighRiskTool, preview: dict, *, audit: AuditLog, approved: bool) -> dict:
    """步骤⑥-⑧:仅在获批后执行,用幂等键,记录审计并验证结果。"""
    if not approved:
        audit.record(event="rejected", **preview)
        return {"status": "rejected"}
    result = tool.executor(idempotency_key=preview["idempotency_key"], **preview["args"])
    audit.record(event="executed", result=result, key=preview["idempotency_key"])
    return {"status": "executed", "result": result}

使用(成功需要人显式批准,拒绝则不执行):

python
from approval_gate import HighRiskTool, propose, execute, ApprovalRequired, AuditLog

_KILLED: set[str] = set()
def _kill_query_impl(*, idempotency_key: str, instance_id: str, query_id: str) -> dict:
    if idempotency_key in _KILLED:            # 幂等:重复执行不产生新副作用
        return {"killed": False, "reason": "duplicate"}
    _KILLED.add(idempotency_key)
    return {"killed": True, "instance_id": instance_id, "query_id": query_id}

def _validate(args: dict) -> dict:
    if args.get("instance_id", "").startswith("prod"):
        raise ValueError("禁止对生产实例执行")
    if "query_id" not in args:
        raise ValueError("缺少 query_id")
    return args

kill_query = HighRiskTool("kill_query", "high", _validate, _kill_query_impl)
audit = AuditLog()

try:
    propose(kill_query, {"instance_id": "mysql-test-01", "query_id": "q-9"},
            actor="alice", audit=audit)
except ApprovalRequired as e:
    print("待审批预览:", e.preview)          # 人在此看到"将要做什么"
    # 人工批准后才执行:
    print(execute(kill_query, e.preview, audit=audit, approved=True))

预期输出:先打印待审批预览(含幂等键),人工批准后打印 {'status': 'executed', 'result': {'killed': True, ...}}。

四、失败路径 ​

  • 审批疲劳:什么都要批,人会盲目点"同意"。→ 只对真正高危操作审批,低危自动放行。
  • 预览不清:人看不懂"将要做什么",等于没审批。→ 预览要含影响范围、可回滚性。
  • 确认可被模型伪造:让模型自己说"用户已确认"。→ 确认必须来自带外的真实人工信号,不能由模型生成。
  • 绕过闸门:存在不经 propose 的执行路径。→ 高危 executor 只允许经闸门调用。

五、测试与验证 ​

python
"""test_approval_gate.py"""
import pytest
from approval_gate import propose, execute, ApprovalRequired, AuditLog, HighRiskTool

def _v(a): 
    if "query_id" not in a: raise ValueError("缺 query_id")
    return a
def _run(*, idempotency_key, **a): return {"ok": True, "key": idempotency_key}
tool = HighRiskTool("t", "high", _v, _run)

def test_high_risk_requires_approval():
    with pytest.raises(ApprovalRequired):
        propose(tool, {"query_id": "q1"}, actor="a", audit=AuditLog())

def test_rejected_not_executed():
    audit = AuditLog()
    try:
        propose(tool, {"query_id": "q1"}, actor="a", audit=audit)
    except ApprovalRequired as e:
        r = execute(tool, e.preview, audit=audit, approved=False)
    assert r["status"] == "rejected"
    assert any(x["event"] == "rejected" for x in audit.entries)

def test_validation_blocks_before_preview():
    with pytest.raises(ValueError):
        propose(tool, {}, actor="a", audit=AuditLog())   # 缺 query_id,校验先拦

六、接管(Takeover)与升级 ​

除了"批准/拒绝",HITL 还包括:人工接管(暂停 Agent,人接手操作)、升级(超阈值/低置信度时转人工)。这些都依赖状态持久化与暂停恢复(09)。

七、安全与权限 ​

  • 审批者身份要鉴权;审批动作要审计、防重放。
  • 确认信号必须带外、可信,不能由被审计的 Agent 自证。

八、性能与成本 ​

  • 人工审批引入延迟,异步化(发通知、等回调)可避免阻塞其他任务(15)。

九、框架差异 ​

  • LangGraph 用 interrupt/interrupt_before 在节点前暂停等待人工;LangChain 1.0 有 HumanInTheLoopMiddleware。机制不同,但"预览→带外确认→幂等执行→审计"这套流程通用。

十、关联章节 ​

面试复述 ​

HITL 不是一个按钮,而是一条可验证控制链:生成预览、冻结参数、鉴权审批、幂等执行、记录审计并核验结果。模型不能替人批准自己的高风险动作。

复述时先给结论,再补机制、适用边界和一个可验证指标;不要只罗列框架名称或产品功能。

迁移练习 ​

为删除数据或终止查询设计审批对象,包含参数摘要、风险、审批人、哈希、过期时间和幂等键,并测试篡改参数后执行会被拒绝。

检索关键词 ​

Human-in-the-Loop 人工审批流程 操作预览 确认 幂等执行 审计 审批疲劳 interrupt 暂停恢复

官方来源 ​

  • LangGraph human-in-the-loop / interrupt 文档;LangChain 1.0 中间件文档(verified_at: 2026-09-07)。

ICP 备案号待补充 · Gitee 主页