AI Agent 从入门到封神：24 讲打造你的超级智能体~系列文章20：人机协作HITL：让Agent在关键决策时“请示领导“，安全又高效-育师

人机协作HITL：让Agent在关键决策时"请示领导"，安全又高效 🤝

导读：Agent再聪明，也不是万能的。关键决策需要人类拍板！Human-in-the-Loop（HITL，人在回路）是Agent安全落地的最后一道防线。今天教你如何让Agent在关键时刻"请示领导"！🛡️

一、为什么需要HITL？🤔

1.1 Agent的"翻车"案例

场景	没有HITL	有HITL
💰财务审批	Agent直接转账100万💀	Agent提交审批，等待确认✅
📧邮件发送	Agent群发错误邮件💀	Agent起草邮件，人工审核后发送✅
🏥医疗诊断	Agent直接开处方💀	Agent给出建议，医生确认后执行✅
🗑️数据删除	Agent删除生产数据库💀	Agent列出待删数据，人工确认✅

💀没有HITL的Agent = 没有刹车的汽车，跑得越快越危险！

1.2 HITL的核心价值

二、HITL的三种模式 📊

模式	说明	适用场景	人类介入程度
👁️监督模式	Agent执行，人类观察	低风险、成熟场景	⭐
📋审批模式	Agent建议，人类批准	中风险、重要决策	⭐⭐⭐
🤝协作模式	人机交替工作	高风险、复杂任务	⭐⭐⭐⭐⭐

三、LangGraph实现HITL 💻

3.1 基础版：审批模式

fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.checkpoint.memoryimportMemorySaverfromtypingimportAnnotatedfromlanggraph.graph.messageimportadd_messagesclassApprovalState(MessagesState):messages:Annotated[list,add_messages]pending_approval:bool=Falseuser_decision:str=""# "approved" / "rejected"# Agent节点：执行任务defagent_node(state):"""Agent分析任务并给出建议"""return{"pending_approval":True,"messages":[{"role":"assistant","content":"建议执行XX操作，是否批准？"}]}# 人工审批节点defhuman_approval(state):"""等待人类审批"""# 在实际应用中，这里会暂停等待人类输入returnstate# 执行节点defexecute_node(state):"""执行已批准的操作"""ifstate["user_decision"]=="approved":return{"messages":[{"role":"assistant","content":"✅ 操作已执行！"}]}else:return{"messages":[{"role":"assistant","content":"❌ 操作已取消。"}]}# 路由defcheck_approval(state):ifstate["user_decision"]=="approved":return"execute"return"end"# 构建图workflow=StateGraph(ApprovalState)workflow.add_node("agent",agent_node)workflow.add_node("approval",human_approval)workflow.add_node("execute",execute_node)workflow.add_edge(START,"agent")workflow.add_edge("agent","approval")workflow.add_conditional_edges("approval",check_approval,{"execute":"execute","end":END})workflow.add_edge("execute",END)# 使用 interrupt_before 实现人工审批checkpointer=MemorySaver()app=workflow.compile(checkpointer=checkpointer,interrupt_before=["approval"]# 在审批节点前暂停)

3.2 运行流程

# 第一步：Agent分析config={"configurable":{"thread_id":"1"}}result=app.invoke({"messages":[{"role":"user","content":"帮我删除过期数据"}]},config)print("Agent建议：",result["messages"][-1]["content"])# 输出：Agent建议：建议执行删除过期数据操作，是否批准？# 第二步：人类审批（通过 update_state 注入决策）app.update_state(config,{"user_decision":"approved"},as_node="approval")# 第三步：继续执行result=app.invoke(None,config)print("结果：",result["messages"][-1]["content"])# 输出：结果：✅ 操作已执行！

四、进阶：分级审批系统 🏗️

4.1 风险分级

defassess_risk(action:str)->str:"""评估操作风险等级"""high_risk=["删除数据","转账","发送邮件","修改配置","部署上线"]medium_risk=["生成报告","创建文件","修改草稿"]forkeywordinhigh_risk:ifkeywordinaction:return"high"forkeywordinmedium_risk:ifkeywordinaction:return"medium"return"low"# 路由决策defroute_by_risk(state):risk=assess_risk(state["action"])ifrisk=="high":return"human_approval"# 高风险：等待审批elifrisk=="medium":return"human_notify"# 中风险：通知人类else:return"auto_execute"# 低风险：自动执行

4.2 流程图

五、HITL最佳实践 📋

5.1 设计原则

原则	说明	示例
🎯最小介入	只在必要时让人类介入	低风险操作自动执行
⚡快速响应	审批流程要快，不阻塞Agent	超时自动降级
📊信息充分	给人类足够的决策信息	展示操作影响范围
🔄可回溯	人类可以随时撤销	支持undo操作
📈逐步放权	随着信任增加，减少审批	学习人类决策模式

5.2 超时处理

defhandle_timeout(state):"""审批超时处理"""timeout=state.get("timeout_seconds",300)iftimeout_expired(timeout):# 超时策略：保守处理return{"user_decision":"rejected","messages":[{"role":"assistant","content":f"⏰ 审批超时（{timeout}秒），操作已取消。"}]}returnstate

六、HITL应用场景大全 🌟

场景	风险等级	HITL模式	说明
💰 财务转账	🔴 高	审批模式	必须人工确认
📧 邮件发送	🟡 中	审批模式	重要邮件需审核
📊 数据分析	🟢 低	监督模式	自动执行，记录日志
🗑️ 数据删除	🔴 高	审批模式	必须人工确认
📝 内容生成	🟢 低	监督模式	自动生成，人工抽检
🏥 医疗建议	🔴 高	审批模式	必须医生确认
🛒 商品推荐	🟢 低	自动模式	全自动
🔧 系统配置	🔴 高	审批模式	必须运维确认

七、本期小结 📝

知识点	核心内容
为什么需要HITL	Agent不是万能的，关键决策需要人类
三种模式	监督、审批、协作
实现方式	LangGraph的interrupt_before机制
风险分级	高→审批，中→通知，低→自动
最佳实践	最小介入、快速响应、信息充分