哪些操作必须让人确认?HITL 审批与接管设计
内容摘要:删除、支付、发消息、改权限、执行生产命令、写业务数据——这些高风险操作必须经过人工确认。本章给出高风险工具的完整控制流程,并附一个可运行、可测试的审批闸门实现。 难度:高级 |前置知识:06 工具调用、09 状态
常见误区
界面上弹出一个“确认”按钮就完成了人工审批;模型可以自行生成确认摘要并决定执行参数。
先给结论
人在回路(Human-in-the-Loop,HITL)必须绑定不可变的操作预览、审批主体、过期时间和幂等键。审批后执行的必须是人看到的同一份参数,高风险动作还需重新鉴权、审计和结果核验。
带着这三个问题阅读:
- 哪些副作用必须暂停任务并等待人工决策?
- 如何防止“审批的是 A,执行时参数变成 B”?
- 拒绝、超时、接管和恢复分别需要什么状态?
一、为什么需要 HITL
模型会犯错、会被注入、有非确定性。对不可逆或高影响的操作,正确的架构不是"让模型更聪明",而是"在执行前插入人"。人机协作(Human-in-the-Loop,HITL)就是把人放在关键决策点上。
哪些操作必须 HITL:删除数据、支付/退款、发送对外消息、发布内容、修改权限、执行生产命令、写入业务数据、任何标注为 high 风险的动作。
二、高风险操作的完整控制流程
这八步缺一不可。注意顺序:校验在预览前,预览在确认前,确认在执行前,审计贯穿始终。
三、可运行的审批闸门
"""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}使用(成功需要人显式批准,拒绝则不执行):
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 只允许经闸门调用。
五、测试与验证
"""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)。
