在金融交易、数据库管理、医疗诊断等高危场景中,AI Agent的自主决策存在两类核心风险:
不可逆操作(如删除数据库记录、大额转账)
模糊决策场景(如医疗方案推荐)传统解决方案采用全流程人工审批,导致效率骤降50%以上。而HIL架构通过精准断点控制,仅在关键节点介入,实现效率与安全的动态平衡。
class State(TypedDict): user_input: str model_response: str user_approval: str # 人类决策状态容器
memory = MemorySaver() # 状态快照存储 graph = builder.compile(checkpointer=memory, interrupt_before=["execute_users"])
tool_node = ToolNode(tools) # 封装外部API workflow.add_conditional_edges("agent", should_continue, { "continue": "action", "run_tool": "action", # 高风险工具调用路径 "end": END })
# 动态断点检测逻辑 def should_continue(state): last_msg = state["messages"][-1] if last_msg.tool_calls and last_msg.tool_calls[0]["name"] == "bank_transfer": transfer_amount = parse_amount(last_msg.tool_calls[0]["args"]) # 解析交易金额 if transfer_amount > 100000: # 动态阈值检测 return "require_approval" # 触发审批断点 return "continue" # 审批节点绑定 workflow.add_conditional_edges("agent", should_continue, { "require_approval": "approval_node", # 转向人工审批 "continue": "action" })
用户输入:”向账户6217转账150万元” Agent解析请求,识别为bank_transfer工具调用 动态断点检测金额超阈值 → 暂停并保存状态 风控人员收到审批请求(含转账详情/风险评估) 人工决策后更新状态: snapshot.values["user_approval"] = "批准" # 或"拒绝" Agent从断点继续执行或终止
# 四眼原则审批实现 def quadruple_approval(state): approvals = state.get("approvals", []) if len(approvals) < 4: raise InterruptionRequired # 触发二次中断 return all(approval == "通过" for approval in approvals[-4:])