Human-in-the-loop
Human-in-the-loop 是在 Agent 执行流程中引入人类判断点的模式,用于审批高风险操作、纠正错误决策、提供缺失信息。它是 Agent 安全性和可靠性的关键保障。
[!info] related notes
- 所属 MOC: AI Agent Application MOC, Agent MOC
- 相关概念: Agent Guardrails, Approval Checkpoints
- UI: Human Approval UI
- 运行时: Agent Runtime — 中断与恢复机制
Human-in-the-loop
一句话定义
Human-in-the-loop (HITL) 是在 Agent 自主执行流程中嵌入人类判断点的模式,让人类在关键节点介入:审批高风险操作、纠正错误决策、提供模型无法获取的信息。
它解决什么问题
完全自主的 Agent 有三个风险:
- 不可逆的错误: Agent 误删数据、误发邮件、误扣款项
- 缺乏领域判断: 某些决策需要人类的经验和直觉
- 用户信任缺失: 用户不放心让 Agent 自己做所有事
HITL 不是”让 Agent 变笨”,而是在自主性和安全性之间找到平衡。
核心原理
介入点类型
Agent 执行流程:
│
├─ [信息采集] → 用户提供缺失信息
│ 例: Agent 不确定用户要查哪个地区的数据
│
├─ [决策确认] → 用户确认 Agent 的计划
│ 例: Agent 说"我准备删除这 50 条记录,确认吗?"
│
├─ [审批检查] → 高风险操作必须审批
│ 例: 转账、删除、发布等不可逆操作
│
├─ [纠错介入] → 用户发现 Agent 方向错误
│ 例: Agent 搜索了错误的关键词
│
└─ [结果验收] → 用户确认最终结果
例: Agent 生成的报告是否满意
实现机制
需暂停等待决策的 HITL 常采用中断 (Interrupt) + 恢复 (Resume);监督、反馈和结果验收不一定都要暂停整个运行:
Agent 运行中
│
▼
遇到需要人类判断的节点
│
▼
保存当前状态 (Checkpoint)
│
▼
暂停执行,返回中间结果给前端
│
▼
前端展示审批界面
│
▼
用户做出决策 (批准/拒绝/修改)
│
▼
批准/拒绝后恢复;修改则生成新提案并重新校验
│
▼
继续 Agent 流程
LangGraph 的 Interrupt 机制
隔离环境安装 langgraph 后运行下面示例。它只验证中断与恢复,不执行真实工具,也不实现审批人鉴权。生产须用持久 checkpointer,并在服务端验证谁能恢复哪个 thread;InMemorySaver 仅用于本进程演示。节点恢复会从节点开头重跑,interrupt 之前不要放未保护的副作用。
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command
class State(TypedDict):
approved: bool
def review(state: State):
decision = interrupt({"action": "review demo", "choices": ["approve", "reject"]})
return {"approved": decision == "approve"}
builder = StateGraph(State)
builder.add_node("review", review)
builder.add_edge(START, "review")
builder.add_edge("review", END)
app = builder.compile(checkpointer=InMemorySaver())
config = {"configurable": {"thread_id": "demo-approval"}}
paused = app.invoke({"approved": False}, config=config)
assert "__interrupt__" in paused
result = app.invoke(Command(resume="approve"), config=config)
assert result["approved"] is True
在 React + Go + Python AI Service 架构中的位置
Python AI Service:
Agent 遇到高风险工具调用
→ 保存 Checkpoint
→ 返回 interrupt 事件
Go 后端:
收到 interrupt 事件
→ 通知前端需要审批
→ 等待用户决策
React 前端:
展示 Human Approval UI
→ 用户点击"批准"或"拒绝"
→ 发送审批结果到后端
Go 后端:
验证审批人身份、权限及请求版本,再转发审批结果到 AI Service
Python AI Service:
恢复执行,携带审批结果
典型工程实现
事件协议
以下是两个独立消息示例,不是完整鉴权协议;服务端需绑定请求版本、身份与截止时间。
Agent 返回的中断事件:
{
"event": "interrupt",
"data": {
"interrupt_id": "int_001",
"type": "approval_required",
"tool_call": {
"name": "delete_records",
"arguments": {"record_ids": ["id_1", "id_2", "id_3"]}
},
"message": "即将删除 3 条记录,此操作不可撤销。",
"options": ["approve", "reject", "modify"]
}
}
用户的审批响应:
{
"event": "resume",
"data": {
"interrupt_id": "int_001",
"decision": "approved",
"modifications": null
}
}
审批 UI 设计
组件与提交状态示例见 Human Approval UI。修改参数会产生新提案并重新检查策略,不把 modified 消息直接当作批准。
常见设计模式
1. 预定义审批点
可在受信工具定义中标记 requires_approval,再由 Runtime 实现拦截。该字段不是所有框架通用的内置开关;是否需要新增批准还取决于当前任务授权和策略。
2. 动态审批
根据参数动态判断是否需要审批(如金额 > 1000 需审批)。
3. 超时策略
使用服务端截止时间与持久状态标记 expired;浏览器倒计时只作提示。超时、拒绝与取消需区分,任何一种都不产生执行授权。
4. 批量审批
多个低风险操作打包成一个审批请求。
常见坑
- 审批点太多: 每个操作都要审批,用户体验极差
- 审批点太少: 高风险操作没有人类把关
- 中断后状态丢失: 没有 Checkpoint,恢复时上下文丢失
- 审批界面信息不足: 用户看不到完整参数,无法做出判断
- 不做超时处理: 用户永远不响应,Agent 永远卡住
和其他概念的关系
- vs Guardrails: Guardrails 是自动化的规则检查,HITL 是人类判断
- vs Approval Checkpoints: Checkpoints 是 HITL 的具体实现机制
- vs Checkpoint: Checkpoint 是 HITL 的状态持久化基础
- vs Human Approval UI: Approval UI 是 HITL 的前端实现
总结
HITL 的核心是让 Agent 知道什么时候该问人。好的 HITL 设计既不会让 Agent 太自由(风险),也不会让 Agent 太受限(效率)。