第 29 课 | Human-in-the-Loop关键步骤人工确认Agent 不是万能的。在关键决策节点SQL 执行、报告发送、数据写入加入人工确认步骤让人类掌控最终决策权——自动化与安全性的最优平衡。一、业务痛点Agent 自动执行的风险场景二的 Agent 已经能自动查数据、生成图表、分析趋势。但如果 Agent 生成的 SQL 有问题呢-- Agent 生成的 SQL幻觉表名不存在SELECT*FROMsales_2025WHEREregion华东;-- Agent 生成的 SQL危险删除操作DELETEFROMsalesWHEREdate2025-01-01;-- Agent 生成的 SQL逻辑错误JOIN 了错误的表SELECTa.*,b.priceFROMorders aJOINproducts cONa.idc.id;如果直接执行可能导致查询报错用户体验差数据被误删最坏情况分析结果错误导致错误决策Human-in-the-LoopHITL的核心思想在关键决策节点插入人工确认让 AI 负责「建议」人类负责「决策」。二、HITL 设计模式✅ 确认执行❌ 拒绝⏭️ 跳过 用户提问「删除3月之前的重复订单」 Agent 分析理解意图 → 生成 SQL 生成 SQLDELETE FROM ordersWHERE date 2025-03-01AND id IN (SELECT ...) 人工审核SQL 是否正确操作是否安全▶️ 执行 SQL安全执行 修改重试提供修改意见 记录日志跳过本次操作 返回结果删除成功23条记录 结束三、LangGraph HITL 实现3.1 状态定义fromtypingimportTypedDict,OptionalfromdatetimeimportdatetimeclassHITLState(TypedDict):question:str# 用户原始问题sql:str# Agent 生成的 SQLapproved:bool# 是否通过审核review_comment:str# 审核意见result:str# 执行结果answer:str# 最终回答risk_level:str# 风险等级low/medium/highexecuted_at:Optional[str]# 执行时间3.2 风险等级判断defassess_risk(sql:str)-str:评估 SQL 操作的风险等级sql_uppersql.upper().strip()ifany(kwinsql_upperforkwin[DELETE,DROP,TRUNCATE,ALTER]):returnhighelifany(kwinsql_upperforkwin[INSERT,UPDATE,CREATE]):returnmediumelifsql_upper.startswith(SELECT):returnlowreturnmedium3.3 审核节点defreview_sql(state:HITLState)-HITLState:人工审核 SQL 节点riskassess_risk(state[sql])state[risk_level]riskprint(\n*60)print(f ⚠️ 风险等级:{risk.upper()})print(f 用户问题:{state[question]})print(f 生成的 SQL:)print(f{state[sql]})print(*60)ifriskhigh:print( 高风险操作请仔细检查 SQL 是否正确。)print( 建议确认 WHERE 条件、LIMIT 限制、备份状态)print(\n 操作选项)print( [y] 确认执行)print( [n] 拒绝执行)print( [m] 提供修改意见后重试)responseinput( 请选择 (y/n/m): ).strip().lower()ifresponsey:state[approved]Truestate[review_comment]人工审核通过elifresponsem:state[approved]Falsecommentinput( 请输入修改意见: ).strip()state[review_comment]commentelse:state[approved]Falsestate[review_comment]人工审核拒绝returnstatedefroute_after_review(state:HITLState)-str:审核后的路由ifstate[approved]:returnexecuteelifstate.get(review_comment)andstate[review_comment]!人工审核拒绝:returnrevise# 有修改意见重新生成else:returncancel# 直接拒绝defexecute_sql(state:HITLState)-HITLState:安全执行 SQLtry:# 实际项目中这里应该用只读账号或事务resultdatabase.execute(state[sql])state[result]str(result)state[executed_at]datetime.now().isoformat()# 写入审计日志audit_log(state)exceptExceptionase:state[result]f执行失败:{str(e)}state[approved]Falsereturnstatedefaudit_log(state:HITLState):审计日志log_entry{timestamp:state[executed_at],sql:state[sql],reviewer:human,risk_level:state[risk_level],result:state[result],}# 写入日志文件或数据库withopen(audit.log,a,encodingutf-8)asf:f.write(json.dumps(log_entry,ensure_asciiFalse)\n)3.4 LangGraph 工作流fromlanggraph.graphimportStateGraph,ENDdefbuild_hitl_workflow():构建 HITL 工作流workflowStateGraph(HITLState)# 添加节点workflow.add_node(generate_sql,generate_sql_node)workflow.add_node(review,review_sql)workflow.add_node(execute,execute_sql)workflow.add_node(revise,revise_sql_node)workflow.add_node(generate_answer,generate_answer_node)# 设置入口workflow.set_entry_point(generate_sql)# 添加边workflow.add_edge(generate_sql,review)workflow.add_conditional_edges(review,route_after_review,{execute:execute,revise:revise,cancel:generate_answer,})workflow.add_edge(execute,generate_answer)workflow.add_edge(revise,review)# 修改后重新审核workflow.add_edge(generate_answer,END)returnworkflow.compile()四、HITL 适用场景操作类型风险等级是否需要 HITL审核方式SELECT 查询低不需要自动执行INSERT 数据中可选批量确认UPDATE 数据高必须逐条确认DELETE 数据极高必须逐条确认二次确认发送邮件/报告中建议预览确认调用付费 API中建议成本预估后确认修改系统配置高必须逐项确认生成最终决策建议高建议人工复核五、HITL 最佳实践分级审核低风险自动高风险人工审计日志所有操作可追溯超时处理审核超时自动拒绝防止流程卡死批量确认多条相似操作合并确认减少人工负担预览机制执行前展示影响范围如「将修改 23 条记录」六、运行cdcode/lesson-29-human-in-loop uvsyncuv run hitl_agent.py七、本课小结HITL 在关键决策点插入人工确认防止 Agent 犯错风险分级低风险自动执行高风险必须人工审核LangGraph 条件边实现审核 → 执行/修改/取消的路由审计日志是合规和安全的基础所有操作必须可追溯HITL 不是降低效率而是在自动化和安全性之间找到最优平衡配套代码code/lesson-29-human-in-loop/hitl_agent.py