人机协作HITL让Agent在关键决策时请示领导安全又高效 导读Agent再聪明也不是万能的。关键决策需要人类拍板Human-in-the-LoopHITL人在回路是Agent安全落地的最后一道防线。今天教你如何让Agent在关键时刻请示领导️一、为什么需要HITL1.1 Agent的翻车案例场景没有HITL有HITL财务审批Agent直接转账100万Agent提交审批等待确认✅邮件发送Agent群发错误邮件Agent起草邮件人工审核后发送✅医疗诊断Agent直接开处方Agent给出建议医生确认后执行✅️数据删除Agent删除生产数据库Agent列出待删数据人工确认✅没有HITL的Agent 没有刹车的汽车跑得越快越危险1.2 HITL的核心价值低风险中风险高风险 Agent自主处理风险等级?✅ 自动执行 通知人类⏸️ 等待审批人类确认二、HITL的三种模式 模式说明适用场景人类介入程度️监督模式Agent执行人类观察低风险、成熟场景⭐审批模式Agent建议人类批准中风险、重要决策⭐⭐⭐协作模式人机交替工作高风险、复杂任务⭐⭐⭐⭐⭐三、LangGraph实现HITL 3.1 基础版审批模式fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.checkpoint.memoryimportMemorySaverfromtypingimportAnnotatedfromlanggraph.graph.messageimportadd_messagesclassApprovalState(MessagesState):messages:Annotated[list,add_messages]pending_approval:boolFalseuser_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:returnexecutereturnend# 构建图workflowStateGraph(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 实现人工审批checkpointerMemorySaver()appworkflow.compile(checkpointercheckpointer,interrupt_before[approval]# 在审批节点前暂停)3.2 运行流程# 第一步Agent分析config{configurable:{thread_id:1}}resultapp.invoke({messages:[{role:user,content:帮我删除过期数据}]},config)print(Agent建议,result[messages][-1][content])# 输出Agent建议建议执行删除过期数据操作是否批准# 第二步人类审批通过 update_state 注入决策app.update_state(config,{user_decision:approved},as_nodeapproval)# 第三步继续执行resultapp.invoke(None,config)print(结果,result[messages][-1][content])# 输出结果✅ 操作已执行四、进阶分级审批系统 ️4.1 风险分级defassess_risk(action:str)-str:评估操作风险等级high_risk[删除数据,转账,发送邮件,修改配置,部署上线]medium_risk[生成报告,创建文件,修改草稿]forkeywordinhigh_risk:ifkeywordinaction:returnhighforkeywordinmedium_risk:ifkeywordinaction:returnmediumreturnlow# 路由决策defroute_by_risk(state):riskassess_risk(state[action])ifriskhigh:returnhuman_approval# 高风险等待审批elifriskmedium:returnhuman_notify# 中风险通知人类else:returnauto_execute# 低风险自动执行4.2 流程图高风险 中风险 低风险 批准拒绝 Agent决定执行操作风险评估⏸️ 暂停等待审批 通知人类✅ 自动执行人类审批❌ 取消操作五、HITL最佳实践 5.1 设计原则原则说明示例最小介入只在必要时让人类介入低风险操作自动执行⚡快速响应审批流程要快不阻塞Agent超时自动降级信息充分给人类足够的决策信息展示操作影响范围可回溯人类可以随时撤销支持undo操作逐步放权随着信任增加减少审批学习人类决策模式5.2 超时处理defhandle_timeout(state):审批超时处理timeoutstate.get(timeout_seconds,300)iftimeout_expired(timeout):# 超时策略保守处理return{user_decision:rejected,messages:[{role:assistant,content:f⏰ 审批超时{timeout}秒操作已取消。}]}returnstate六、HITL应用场景大全 场景风险等级HITL模式说明 财务转账 高审批模式必须人工确认 邮件发送 中审批模式重要邮件需审核 数据分析 低监督模式自动执行记录日志️ 数据删除 高审批模式必须人工确认 内容生成 低监督模式自动生成人工抽检 医疗建议 高审批模式必须医生确认 商品推荐 低自动模式全自动 系统配置 高审批模式必须运维确认七、本期小结 知识点核心内容为什么需要HITLAgent不是万能的关键决策需要人类三种模式监督、审批、协作实现方式LangGraph的interrupt_before机制风险分级高→审批中→通知低→自动最佳实践最小介入、快速响应、信息充分HITL是Agent安全落地的关键没有人类监督的Agent就像没有方向盘的赛车——跑得越快翻得越惨。学会HITL让你的Agent既智能又安全 下期预告《Agent可观测性如何监控、调试、追踪你的智能体》—— Agent上线了怎么知道它跑得怎么样三连走起HITL让AI学会请示汇报专栏第20/24期多Agent协作篇完结作者高炉炼铁智能化技术研究者专注钢铁冶金与人工智能 交叉领域。 如果觉得有帮助请点赞、收藏、转发版权归作者所有未经许可请勿抄袭套用商用(或其它具有利益性行为)。 关注专栏不错过后续精彩内容
AI Agent 从入门到封神：24 讲打造你的超级智能体~系列文章20：人机协作HITL：让Agent在关键决策时“请示领导“，安全又高效
人机协作HITL让Agent在关键决策时请示领导安全又高效 导读Agent再聪明也不是万能的。关键决策需要人类拍板Human-in-the-LoopHITL人在回路是Agent安全落地的最后一道防线。今天教你如何让Agent在关键时刻请示领导️一、为什么需要HITL1.1 Agent的翻车案例场景没有HITL有HITL财务审批Agent直接转账100万Agent提交审批等待确认✅邮件发送Agent群发错误邮件Agent起草邮件人工审核后发送✅医疗诊断Agent直接开处方Agent给出建议医生确认后执行✅️数据删除Agent删除生产数据库Agent列出待删数据人工确认✅没有HITL的Agent 没有刹车的汽车跑得越快越危险1.2 HITL的核心价值低风险中风险高风险 Agent自主处理风险等级?✅ 自动执行 通知人类⏸️ 等待审批人类确认二、HITL的三种模式 模式说明适用场景人类介入程度️监督模式Agent执行人类观察低风险、成熟场景⭐审批模式Agent建议人类批准中风险、重要决策⭐⭐⭐协作模式人机交替工作高风险、复杂任务⭐⭐⭐⭐⭐三、LangGraph实现HITL 3.1 基础版审批模式fromlanggraph.graphimportStateGraph,START,ENDfromlanggraph.checkpoint.memoryimportMemorySaverfromtypingimportAnnotatedfromlanggraph.graph.messageimportadd_messagesclassApprovalState(MessagesState):messages:Annotated[list,add_messages]pending_approval:boolFalseuser_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:returnexecutereturnend# 构建图workflowStateGraph(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 实现人工审批checkpointerMemorySaver()appworkflow.compile(checkpointercheckpointer,interrupt_before[approval]# 在审批节点前暂停)3.2 运行流程# 第一步Agent分析config{configurable:{thread_id:1}}resultapp.invoke({messages:[{role:user,content:帮我删除过期数据}]},config)print(Agent建议,result[messages][-1][content])# 输出Agent建议建议执行删除过期数据操作是否批准# 第二步人类审批通过 update_state 注入决策app.update_state(config,{user_decision:approved},as_nodeapproval)# 第三步继续执行resultapp.invoke(None,config)print(结果,result[messages][-1][content])# 输出结果✅ 操作已执行四、进阶分级审批系统 ️4.1 风险分级defassess_risk(action:str)-str:评估操作风险等级high_risk[删除数据,转账,发送邮件,修改配置,部署上线]medium_risk[生成报告,创建文件,修改草稿]forkeywordinhigh_risk:ifkeywordinaction:returnhighforkeywordinmedium_risk:ifkeywordinaction:returnmediumreturnlow# 路由决策defroute_by_risk(state):riskassess_risk(state[action])ifriskhigh:returnhuman_approval# 高风险等待审批elifriskmedium:returnhuman_notify# 中风险通知人类else:returnauto_execute# 低风险自动执行4.2 流程图高风险 中风险 低风险 批准拒绝 Agent决定执行操作风险评估⏸️ 暂停等待审批 通知人类✅ 自动执行人类审批❌ 取消操作五、HITL最佳实践 5.1 设计原则原则说明示例最小介入只在必要时让人类介入低风险操作自动执行⚡快速响应审批流程要快不阻塞Agent超时自动降级信息充分给人类足够的决策信息展示操作影响范围可回溯人类可以随时撤销支持undo操作逐步放权随着信任增加减少审批学习人类决策模式5.2 超时处理defhandle_timeout(state):审批超时处理timeoutstate.get(timeout_seconds,300)iftimeout_expired(timeout):# 超时策略保守处理return{user_decision:rejected,messages:[{role:assistant,content:f⏰ 审批超时{timeout}秒操作已取消。}]}returnstate六、HITL应用场景大全 场景风险等级HITL模式说明 财务转账 高审批模式必须人工确认 邮件发送 中审批模式重要邮件需审核 数据分析 低监督模式自动执行记录日志️ 数据删除 高审批模式必须人工确认 内容生成 低监督模式自动生成人工抽检 医疗建议 高审批模式必须医生确认 商品推荐 低自动模式全自动 系统配置 高审批模式必须运维确认七、本期小结 知识点核心内容为什么需要HITLAgent不是万能的关键决策需要人类三种模式监督、审批、协作实现方式LangGraph的interrupt_before机制风险分级高→审批中→通知低→自动最佳实践最小介入、快速响应、信息充分HITL是Agent安全落地的关键没有人类监督的Agent就像没有方向盘的赛车——跑得越快翻得越惨。学会HITL让你的Agent既智能又安全 下期预告《Agent可观测性如何监控、调试、追踪你的智能体》—— Agent上线了怎么知道它跑得怎么样三连走起HITL让AI学会请示汇报专栏第20/24期多Agent协作篇完结作者高炉炼铁智能化技术研究者专注钢铁冶金与人工智能 交叉领域。 如果觉得有帮助请点赞、收藏、转发版权归作者所有未经许可请勿抄袭套用商用(或其它具有利益性行为)。 关注专栏不错过后续精彩内容
