主题
08 - 人机协同：审批通知与 HITL 闭环
一句话：审批被做成 Agent 执行链上的显式暂停点，HITL 事件生成卡片推到 IM，回调经绑定表确认操作人、用 request_id 做幂等，再由 RT 注回实例继续执行；这套闭环把安全押在“员工真的会看卡片”上，因此超时必须自动拒绝而不是无限挂起，代价是通知通道要三级降级，而高频低风险审批会把人训练成条件反射式批准。
一、设计目标
Agent 执行任务时，某些操作需要员工确认后才能执行——如删除文件、发送外部消息、调用高风险工具。这些操作不能由 Agent 自主决定，必须经过**人在回路中（Human-in-the-Loop）**的审批。
审批体验的设计目标：
- 审批即通知：审批请求以 IM 卡片形式推送到员工，点击即授权，无需打开 Web 控制台
- 超时安全：审批请求超时未处理自动拒绝，防止任务挂起
- 策略可配：不同操作的风险等级不同，审批策略可按会话/任务/工具粒度配置
二、审批点位
2.1 触发源
| 点位 | 场景 | 风险等级 | 默认策略 |
|---|---|---|---|
| 工具调用确认 | 执行敏感工具（写文件、外发消息、网络请求） | 中 | 需审批 |
| 委托执行 | 主 Agent 委托 ACP 执行体（opencode 等） | 中 | 需审批 |
| 数据删除 | 删除文件、数据库记录、外部资源 | 高 | 需审批 + 二次确认 |
| 外部通信 | 发送邮件、IM 消息、API 调用外部系统 | 高 | 需审批 |
| 批量操作 | 一次操作影响多条数据 | 高 | 需审批 + 数量确认 |
| 自动执行 | 低风险操作（读取、搜索、计算） | 低 | 自动放行 |
2.2 审批策略配置
json
{
"approval_policy": {
"default_level": "medium",
"rules": [
{
"tool_pattern": "file_delete",
"risk_level": "high",
"require_approval": true,
"timeout_seconds": 3600,
"timeout_action": "reject"
},
{
"tool_pattern": "web_search",
"risk_level": "low",
"require_approval": false
},
{
"tool_pattern": "send_email",
"risk_level": "high",
"require_approval": true,
"timeout_seconds": 1800,
"timeout_action": "reject"
}
]
}
}三、审批闭环流程
四、审批卡片协议
4.1 卡片数据结构
json
{
"card_type": "approval",
"request_id": "req_xyz789",
"trace_id": "550e8400-e29b-41d4-a716-446655440000",
"title": "需要你的授权",
"content": {
"action": "删除文件",
"target": "/data/reports/old/",
"risk_level": "high",
"instance_id": "inst_abc123",
"task_id": "task_abc123",
"tool_name": "file_delete",
"params_summary": "将删除 3 个文件，共 12MB"
},
"actions": [
{"id": "approve", "label": "批准", "style": "primary"},
{"id": "reject", "label": "拒绝", "style": "danger"},
{"id": "comment", "label": "附备注", "type": "input", "placeholder": "可选备注"}
],
"timeout_seconds": 3600,
"callback_url": "https://agent.example.com/api/channel/callback",
"created_at": "2026-10-25T10:30:00Z"
}4.2 卡片回调处理
python
async def handle_card_callback(platform: str, callback_data: dict):
"""处理 IM 卡片回调"""
# 1. 验签
adapter = get_adapter(platform)
if not adapter.verify_callback(callback_data):
raise AuthenticationError("Invalid callback signature")
# 2. 提取信息
request_id = callback_data["request_id"]
action = callback_data["action"] # "approve" / "reject"
comment = callback_data.get("comment", "")
platform_user_id = callback_data["sender_id"]
# 3. 身份验证
binding = await binding_store.get(platform, platform_user_id)
if not binding or binding.bind_status != "bound":
raise AuthenticationError("Unknown user")
# 4. 幂等检查
approval = await approval_store.get(request_id)
if approval is None:
raise NotFoundError("Approval request not found")
if approval.status != "pending":
return {"message": f"Already {approval.status}"} # 重复点击
# 5. 更新审批状态
approved = action == "approve"
await approval_store.update(request_id, {
"status": "approved" if approved else "rejected",
"approver": binding.user_id,
"comment": comment,
"decided_at": datetime.utcnow(),
})
# 6. 转发给实例
instance = await registry.get_by_user(binding.user_id)
rt = await vault.get_runtime_token(instance.instance_id)
await httpx.post(
f"http://{instance.endpoint}/api/hitl/{request_id}",
headers={"x-qwenpaw-runtime-token": rt},
json={"approved": approved, "comment": comment},
)
# 7. 更新卡片
await adapter.update_card(approval.chat_id, approval.card_id, {
"status": "已批准 ✓" if approved else "已拒绝 ✗",
"decided_by": binding.user_id,
"decided_at": datetime.utcnow().strftime("%H:%M:%S"),
})五、通知通道
5.1 通道优先级
| 优先级 | 通道 | 适用场景 | 触达率 |
|---|---|---|---|
| 1 | 企业 IM 卡片 | 日常审批（默认） | 高（员工在线即达） |
| 2 | 邮件 | IM 未绑定时的兜底 | 中（有延迟） |
| 3 | Web 控制台 | 员工在 Web 端时的内嵌通知 | 低（需主动查看） |
5.2 通道降级
审批事件产生
→ 查员工 IM 绑定状态
→ 已绑定 → 发送 IM 卡片（优先级 1）
→ 未绑定 → 发送邮件通知（优先级 2）
→ 邮件也失败 → 写入 Web 控制台通知中心（优先级 3）
→ 全部失败 → 审计告警（通知通道异常）六、审批超时策略
| 风险等级 | 超时时间 | 超时动作 | 通知策略 |
|---|---|---|---|
| 低 | 无超时 | 自动放行 | 不通知 |
| 中 | 1 小时 | 自动拒绝 | 超时前 10 分钟提醒一次 |
| 高 | 30 分钟 | 自动拒绝 | 超时前 5 分钟提醒 + 升级通知管理员 |
| 紧急 | 15 分钟 | 自动拒绝 | 超时前 5 分钟提醒 + IM + 邮件双通道 |
超时提醒实现：
python
async def approval_timeout_watcher():
"""后台任务：监控审批超时"""
while True:
pending = await approval_store.list_pending()
now = datetime.utcnow()
for approval in pending:
elapsed = (now - approval.created_at).total_seconds()
# 超时 → 自动拒绝
if elapsed > approval.timeout_seconds:
await handle_timeout(approval)
continue
# 即将超时 → 提醒
remaining = approval.timeout_seconds - elapsed
reminder_threshold = approval.timeout_seconds * 0.15 # 剩余 15% 时提醒
if remaining <= reminder_threshold and not approval.reminded:
await send_reminder(approval)
await approval_store.update(approval.request_id, {"reminded": True})
await asyncio.sleep(10) # 每 10 秒检查一次七、审批审计
所有审批操作全量留痕：
json
{
"event_type": "approval_decision",
"request_id": "req_xyz789",
"trace_id": "550e8400-e29b-41d4-a716-446655440000",
"user_id": "zhangsan",
"instance_id": "inst_abc123",
"tool_name": "file_delete",
"risk_level": "high",
"decision": "approved",
"comment": "确认删除旧报告",
"response_time_seconds": 120,
"channel": "feishu",
"timestamp": "2026-10-25T10:32:00Z"
}审计维度：
- 审批响应时间分布：识别审批瓶颈
- 自动拒绝率：过高说明超时阈值需要调整
- 高频审批工具：考虑将高频低风险工具改为自动放行
- 非工作时间审批率：异常模式检测