HITL：高风险动作交给人确认
HITL 是 Human-in-the-Loop 的缩写。意思很简单：
Agent 可以准备动作，但高风险动作必须人批准了才执行。
旅游助手可以自动查天气、搜景点、估路线。但下面这些事不能默默干：
- 订票扣款
- 发邮件给酒店
- 取消已有预订
- 提交签证信息
Policy 能挡住"不允许的调用"，Guardrails 能挡住"危险的内容"。但它们都不能替用户做决定。"这张 680 块的票要不要买？"——只有人能确认。
没有 HITL 会怎么坏
| 问题 | 表面看起来 | 实际风险 |
|---|---|---|
| Agent 自动下单 | 用户说"帮我看看"，Agent 理解成"帮我订" | 扣了钱、退不了 |
| Agent 发邮件确认 | 回复很快 | 内容有误，发给了错误的收件人 |
| Agent 取消预订"帮用户整理" | 看起来贴心 | 不可逆操作，用户并没有这个意图 |
流程
flowchart TD
A["Agent 提出高风险动作"] --> D["检测：是否需要审批？"]
D -->|不需要| E["直接执行"]
D -->|需要| R["创建审批请求"]
R --> H["展示给人：工具、参数、原因、影响"]
H -->|批准| T["执行工具"]
H -->|拒绝| P["Agent 换计划或通知用户"]
T --> O["记录结果"]
P --> O2["记录拒绝原因"]
完整代码
完整的 detect → pause → confirm → continue/cancel 流程。模拟了旅游助手订票、发邮件、查天气三种调用。
from __future__ import annotations
import json
from dataclasses import dataclass, field
from datetime import datetime
# ========== 审批请求 ==========
@dataclass
class ApprovalRequest:
request_id: str
tool: str
args: dict
reason: str # 为什么需要审批
impact: str # 执行后会发生什么
deny_impact: str # 拒绝后会发生什么
created_at: str = field(default_factory=lambda: datetime.now().strftime("%Y-%m-%d %H:%M:%S"))
class NeedsApproval(Exception):
"""工具调用需要人工审批时抛出。"""
def __init__(self, request: ApprovalRequest):
self.request = request
# ========== 审批规则 ==========
# 哪些工具需要审批，以及审批的描述
APPROVAL_RULES: dict[str, dict] = {
"book_ticket": {
"reason": "订票会产生实际扣款",
"impact": "将从绑定的支付方式扣款，生成不可撤销的订单",
"deny_impact": "不会扣款，Agent 会提供替代方案或等待用户指示",
},
"send_email": {
"reason": "发邮件不可撤回",
"impact": "邮件将发送给指定收件人，发送后无法撤回",
"deny_impact": "邮件不会发送，Agent 会展示邮件草稿供用户修改",
},
"cancel_order": {
"reason": "取消订单可能不可逆",
"impact": "订单将被取消，退款按平台规则处理",
"deny_impact": "订单保持不变",
},
}
# 已批准的请求（真实系统应该持久化）
approved_requests: set[str] = set()
denied_requests: set[str] = set()
def make_request_id(tool: str, args: dict) -> str:
"""为一次工具调用生成唯一的请求 ID。"""
args_key = json.dumps(args, sort_keys=True, ensure_ascii=False)
return f"{tool}:{hash(args_key)}"
def check_approval(tool: str, args: dict) -> ApprovalRequest | None:
"""检查工具调用是否需要审批。如果需要且未审批，抛出 NeedsApproval。"""
rule = APPROVAL_RULES.get(tool)
if rule is None:
return None # 不需要审批
request_id = make_request_id(tool, args)
if request_id in approved_requests:
return None # 已经批准过
if request_id in denied_requests:
raise NeedsApproval(ApprovalRequest(
request_id=request_id,
tool=tool,
args=args,
reason="此请求之前被拒绝过",
impact=rule["impact"],
deny_impact=rule["deny_impact"],
))
request = ApprovalRequest(
request_id=request_id,
tool=tool,
args=args,
reason=rule["reason"],
impact=rule["impact"],
deny_impact=rule["deny_impact"],
)
raise NeedsApproval(request)
def show_approval_request(request: ApprovalRequest) -> None:
"""展示审批请求给人看。"""
print(f"\n {'='*50}")
print(f" 需要人工确认")
print(f" {'='*50}")
print(f" 请求 ID: {request.request_id}")
print(f" 工具: {request.tool}")
print(f" 参数: {json.dumps(request.args, ensure_ascii=False)}")
print(f" 原因: {request.reason}")
print(f" 执行影响: {request.impact}")
print(f" 拒绝影响: {request.deny_impact}")
print(f" 时间: {request.created_at}")
def human_decision(request: ApprovalRequest, approve: bool) -> None:
"""模拟人做出决定。"""
if approve:
approved_requests.add(request.request_id)
print(f" 决定: 批准")
else:
denied_requests.add(request.request_id)
print(f" 决定: 拒绝")
# ========== 模拟工具 ==========
def get_weather(city: str) -> dict:
return {"city": city, "temperature": "22°C", "condition": "多云"}
def book_ticket(city: str, ticket_type: str, price: float) -> dict:
return {"status": "已预订", "city": city, "ticket_type": ticket_type, "price": price, "order_id": "ORD-20250601-001"}
def send_email(to: str, subject: str, body: str) -> dict:
return {"status": "已发送", "to": to, "subject": subject}
TOOL_MAP = {
"get_weather": get_weather,
"book_ticket": book_ticket,
"send_email": send_email,
}
# ========== Agent 执行循环（带 HITL）==========
def execute_with_hitl(tool: str, args: dict, simulated_approval: bool | None = None) -> dict:
"""
尝试执行工具。如果需要审批，暂停并展示审批请求。
simulated_approval: 模拟人的决定（True=批准, False=拒绝, None=不需要审批）
"""
print(f"\nAgent 请求调用: {tool}({json.dumps(args, ensure_ascii=False)})")
try:
check_approval(tool, args)
except NeedsApproval as e:
show_approval_request(e.request)
if simulated_approval is None:
print(f" 等待人工决定...")
return {"status": "pending_approval", "request_id": e.request.request_id}
human_decision(e.request, simulated_approval)
if not simulated_approval:
return {"status": "denied", "reason": e.request.deny_impact}
# 批准后重新检查（这次会通过）
try:
check_approval(tool, args)
except NeedsApproval:
return {"status": "denied", "reason": "审批状态异常"}
# 执行工具
func = TOOL_MAP.get(tool)
if func is None:
return {"error": f"未知工具: {tool}"}
result = func(**args)
print(f" 执行结果: {json.dumps(result, ensure_ascii=False)}")
return result
# ========== 演示 ==========
def demo():
print("=" * 60)
print("HITL 人工确认流程演示")
print("=" * 60)
# 场景 1：查天气——不需要审批，直接执行
print("\n--- 场景 1：查天气（无需审批）---")
execute_with_hitl("get_weather", {"city": "杭州"})
# 场景 2：订票——需要审批，人批准
print("\n--- 场景 2：订票（需要审批 → 批准）---")
execute_with_hitl(
"book_ticket",
{"city": "杭州", "ticket_type": "standard", "price": 280},
simulated_approval=True,
)
# 场景 3：发邮件——需要审批，人拒绝
print("\n--- 场景 3：发邮件（需要审批 → 拒绝）---")
execute_with_hitl(
"send_email",
{"to": "hotel@example.com", "subject": "预订确认", "body": "确认入住..."},
simulated_approval=False,
)
# 场景 4：同样的订票参数再调一次——已经批准过，直接执行
print("\n--- 场景 4：同参数订票重试（已批准，直接执行）---")
execute_with_hitl(
"book_ticket",
{"city": "杭州", "ticket_type": "standard", "price": 280},
)
if __name__ == "__main__":
demo()
运行输出：
============================================================
HITL 人工确认流程演示
============================================================
--- 场景 1：查天气（无需审批）---
Agent 请求调用: get_weather({"city": "杭州"})
执行结果: {"city": "杭州", "temperature": "22°C", "condition": "多云"}
--- 场景 2：订票（需要审批 → 批准）---
Agent 请求调用: book_ticket({"city": "杭州", "ticket_type": "standard", "price": 280})
==================================================
需要人工确认
==================================================
请求 ID: book_ticket:-6915588706498002498
工具: book_ticket
参数: {"city": "杭州", "ticket_type": "standard", "price": 280}
原因: 订票会产生实际扣款
执行影响: 将从绑定的支付方式扣款，生成不可撤销的订单
拒绝影响: 不会扣款，Agent 会提供替代方案或等待用户指示
时间: 2025-06-01 14:30:00
决定: 批准
执行结果: {"status": "已预订", "city": "杭州", "ticket_type": "standard", "price": 280, "order_id": "ORD-20250601-001"}
--- 场景 3：发邮件（需要审批 → 拒绝）---
Agent 请求调用: send_email({"to": "hotel@example.com", "subject": "预订确认", "body": "确认入住..."})
==================================================
需要人工确认
==================================================
请求 ID: send_email:4471429671498920498
工具: send_email
参数: {"to": "hotel@example.com", "subject": "预订确认", "body": "确认入住..."}
原因: 发邮件不可撤回
执行影响: 邮件将发送给指定收件人，发送后无法撤回
拒绝影响: 邮件不会发送，Agent 会展示邮件草稿供用户修改
时间: 2025-06-01 14:30:00
决定: 拒绝
--- 场景 4：同参数订票重试（已批准，直接执行）---
Agent 请求调用: book_ticket({"city": "杭州", "ticket_type": "standard", "price": 280})
执行结果: {"status": "已预订", "city": "杭州", "ticket_type": "standard", "price": 280, "order_id": "ORD-20250601-001"}
代码走读
核心机制：detect → pause → confirm → continue/cancel
APPROVAL_RULES定义哪些工具需要审批。不在这个表里的工具（如get_weather）直接放行。check_approval检查两件事：这个工具需不需要审批？如果需要，之前批没批过？- 没批过就抛
NeedsApproval异常。异常里带着完整的ApprovalRequest——包括工具、参数、原因、影响。 - 人看到这些信息后做决定。批准了就存进
approved_requests，下次同样的调用不再问。
审批请求里必须包含什么：
不要只展示"批准/拒绝"两个按钮。人需要看到：
- 要执行什么工具、什么参数
- 为什么需要审批
- 执行了会怎样
- 拒绝了会怎样
信息不够，审批就变成橡皮图章。人会习惯性点批准，然后出事。
重试逻辑：
场景 4 展示了一个重要细节：同样参数的 book_ticket 第二次调用时，因为已经在 approved_requests 里了，直接放行。真实系统里这个"已批准"状态需要持久化，否则 Agent 重启后又会卡住。
什么时候用
- 工具会付款、下单、取消、删除、发消息。
- 合规要求人工确认（比如金融、医疗）。
- Agent 还在灰度阶段，需要限制自主权。
- Guardrail 触发后需要人来判断是不是误报。
什么时候不需要
不要给所有操作都加审批。查天气、搜景点这种只读操作不需要人确认。审批太多会导致审批疲劳——人习惯了点"批准"，真正危险的操作也批过去了。
只给不可逆的、有成本的、有外部影响的操作加审批。
常见坑
| 坑 | 现象 | 修法 |
|---|---|---|
| 每一步都要审批 | 用户烦了，无脑点批准 | 只拦高风险操作 |
| 审批请求信息不足 | 人看不懂在批什么 | 包含工具、参数、原因、影响 |
| 批准后 Agent 重试失败 | 中断→恢复流程没设计好 | 持久化审批状态，设计 resume 机制 |
| 假设审批一定会通过 | 拒绝后 Agent 不知道怎么继续 | 处理 deny 分支：换计划或告知用户 |
| 不记录审批历史 | 出事以后查不到谁批的 | 日志记录：谁、什么时候、批了什么 |
读完以后
Policy、Guardrails、HITL 让单次运行更安全。但你改了 prompt、改了工具描述、加了新规则之后，怎么知道有没有破坏之前能跑通的任务？
下一节 Eval Harness 讲怎么用固定任务集做回归检查。