Skip to content
🔗 分享本题
查看我的学习进度 →

Human-in-the-Loop 根据风险和置信度选择自动执行抽查显式审批或阻断,并提供可审计审批上下文

🧠 记忆锚点:风险和置信度共同决定介入;审批要看得懂影响,拒绝超时都安全收口。

💡 答案要点

Human-in-the-Loop = 在 Agent 自主决策链路中插入人工审核节点

何时需要人工介入:

风险矩阵:
高风险 + 低置信 → 强制人工审核
高风险 + 高置信 → 人工可选审核
低风险 + 低置信 → 提示用户确认
低风险 + 高置信 → Agent 自主执行

LangGraph 实现人工介入节点:

展开 Python 代码示例(82 行)
python
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# 定义状态
class AgentState(TypedDict):
    task: str
    plan: list
    current_step: int
    requires_approval: bool
    human_decision: str  # approved / rejected / modified

# 高风险动作检测
def check_requires_approval(state: AgentState) -> AgentState:
    current_action = state["plan"][state["current_step"]]

    # 定义高风险动作清单
    high_risk_actions = [
        "delete_data",
        "send_email_to_customer",
        "execute_payment",
        "modify_database",
    ]

    requires_approval = any(
        risk in current_action["type"]
        for risk in high_risk_actions
    )

    return {**state, "requires_approval": requires_approval}

# 等待人工审核节点
def human_review_node(state: AgentState) -> AgentState:
    """这个节点会暂停,等待人工输入"""
    # LangGraph 通过 interrupt 机制暂停
    # 实际触发:通过 Webhook/消息推送通知审核人员
    print(f"⚠️  需要人工审核: {state['plan'][state['current_step']]}")
    print(f"请输入决定 (approved/rejected/modified):")

    # 等待人工决定(通过 API 更新 state)
    # 在 LangGraph 中,通过恢复执行传入 human_decision
    return state

# 构建带人机协同的工作流
workflow = StateGraph(AgentState)
workflow.add_node("plan", planning_node)
workflow.add_node("check_risk", check_requires_approval)
workflow.add_node("human_review", human_review_node)
workflow.add_node("execute", execution_node)

# 条件路由
workflow.add_conditional_edges(
    "check_risk",
    lambda s: "human_review" if s["requires_approval"] else "execute",
    {"human_review": "human_review", "execute": "execute"}
)

workflow.add_conditional_edges(
    "human_review",
    lambda s: "execute" if s["human_decision"] == "approved" else END,
    {"execute": "execute", END: END}
)

# 使用 checkpointer 支持暂停恢复
app = workflow.compile(
    checkpointer=MemorySaver(),
    interrupt_before=["human_review"]  # 在此节点前暂停
)

# 执行到暂停点
thread_id = "task_001"
result = app.invoke(
    {"task": "删除3个月前的日志"},
    config={"configurable": {"thread_id": thread_id}}
)
# → Agent 在 human_review 节点暂停

# 人工审核后恢复
app.invoke(
    {"human_decision": "approved"},
    config={"configurable": {"thread_id": thread_id}}
)
# → Agent 继续执行

三级人机协同策略:

级别场景策略
全自动低风险、高置信任务Agent 直接执行,事后抽样审核
可选确认中等风险任务执行前展示计划,5秒内无反对则执行
强制审核高风险任务(删除/支付/发送)必须人工明确批准

面试话术:

"Human-in-the-Loop 不是'凡事都让人审',而是'关键节点卡人工'。我用风险矩阵划分:高风险操作(删数据/发邮件/支付)强制人工确认,用 LangGraph 的 interrupt 机制在节点前暂停;低风险操作 Agent 自主执行,事后5%抽样审核。这个设计让 Agent 自动化率达 85%,同时把高风险操作的人工审核率做到 100%,两头都不耽误。"