_{🧠 记忆锚点：高风险先暂停，带上下文审批；超时默认拒绝，可从检查点恢复。}

💡 答案要点

Human-in-the-Loop (HITL) = 在Agent关键决策节点插入人工确认，保证安全可控

为什么需要HITL

纯自动Agent的风险：
Agent决定删除数据库 → 直接执行 → 数据丢失 ❌
Agent误解用户意图 → 发错邮件 → 造成事故 ❌
Agent循环调用API → 账单暴涨  → 损失惨重 ❌

HITL的保障：
Agent决定删除数据库 → 暂停 → 人工确认 → 执行/拒绝 ✅

HITL触发条件设计

展开 Python 代码示例（139 行）

from enum import Enum
from typing import Callable

class RiskLevel(Enum):
    LOW = "low"       # 自动执行
    MEDIUM = "medium" # 警告但执行
    HIGH = "high"     # 必须人工确认
    CRITICAL = "critical" # 强制中断

class HITLAgent:
    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.pending_approvals = []

    def assess_risk(self, action: str, params: dict) -> RiskLevel:
        """评估操作风险等级"""

        # 规则1：不可逆操作 → HIGH
        irreversible_actions = ["delete", "drop", "truncate", "send_email", "post", "pay"]
        if any(kw in action.lower() for kw in irreversible_actions):
            return RiskLevel.HIGH

        # 规则2：涉及金额 → CRITICAL
        if "amount" in params and params.get("amount", 0) > 1000:
            return RiskLevel.CRITICAL

        # 规则3：批量操作 → HIGH
        if params.get("batch_size", 0) > 100:
            return RiskLevel.HIGH

        # 规则4：用LLM判断
        risk_prompt = f"""
        评估以下操作的风险等级（low/medium/high/critical）：
        操作：{action}
        参数：{params}

        考虑因素：是否可逆？影响范围？是否涉及敏感数据？

        输出JSON：{{"level": "风险等级", "reason": "原因"}}
        """
        result = json.loads(self.llm.generate(risk_prompt, temperature=0))
        return RiskLevel(result["level"])

    def execute_with_hitl(self, action: str, params: dict):
        """执行带HITL的操作"""
        risk = self.assess_risk(action, params)

        if risk == RiskLevel.LOW:
            # 直接执行
            return self.tools[action](**params)

        elif risk == RiskLevel.MEDIUM:
            # 执行但记录警告
            print(f"⚠️  中风险操作：{action}({params})")
            result = self.tools[action](**params)
            self.log_action(action, params, result)
            return result

        elif risk == RiskLevel.HIGH:
            # 暂停，请求人工确认
            return self.request_approval(action, params, risk)

        elif risk == RiskLevel.CRITICAL:
            # 强制中断，通知管理员
            self.notify_admin(action, params)
            raise Exception(f"🚨 危险操作已阻止：{action}")

    def request_approval(self, action: str, params: dict, risk: RiskLevel):
        """请求人工审批"""
        approval_id = f"approval_{int(time.time())}"

        # 保存待审批操作
        self.pending_approvals.append({
            "id": approval_id,
            "action": action,
            "params": params,
            "risk": risk.value,
            "status": "pending",
            "created_at": time.time()
        })

        # 通知审批人（实际场景：发钉钉/企微/邮件）
        self.send_approval_request(approval_id, action, params)

        # 等待审批（异步方式，这里简化为轮询）
        return self.wait_for_approval(approval_id)

    def wait_for_approval(self, approval_id: str, timeout=300):
        """等待人工审批，超时自动拒绝"""
        start = time.time()

        while time.time() - start < timeout:
            approval = self.get_approval(approval_id)

            if approval["status"] == "approved":
                # 获批，执行操作
                a = approval["action"]
                return self.tools[a](**approval["params"])

            elif approval["status"] == "rejected":
                return {"error": "操作被拒绝", "reason": approval.get("reason")}

            time.sleep(5)  # 每5秒轮询一次

        # 超时，自动拒绝
        return {"error": "审批超时，操作已取消"}

    def send_approval_request(self, approval_id, action, params):
        """发送审批通知（钉钉机器人示例）"""
        message = f"""
        🔔 Agent操作需要审批

        操作：{action}
        参数：{json.dumps(params, ensure_ascii=False)}

        审批链接：http://your-system/approve/{approval_id}
        超时时间：5分钟
        """
        # dingtalk_bot.send(message)
        print(message)

# 实战场景：邮件发送Agent
agent = HITLAgent(llm, tools={
    "search_web": search_tool,
    "send_email": email_tool,
    "delete_file": delete_tool,
})

# 低风险：自动执行
agent.execute_with_hitl("search_web", {"query": "最新AI新闻"})

# 高风险：暂停等待审批
agent.execute_with_hitl("send_email", {
    "to": "[email protected]",
    "subject": "重要通知",
    "body": "全员涨薪100%"
})
# → 触发HITL，发送钉钉审批消息 → 等待HR确认

LangGraph中实现HITL

展开 Python 代码示例（35 行）

from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver

# LangGraph原生支持HITL（interrupt_before）
workflow = StateGraph(AgentState)

workflow.add_node("agent", agent_node)
workflow.add_node("execute_tool", tool_node)
workflow.add_node("human_review", human_review_node)

# 在执行高风险工具前中断，等待人工确认
workflow.add_conditional_edges(
    "agent",
    route_after_agent,
    {
        "execute_safe": "execute_tool",    # 低风险直接执行
        "need_review": "human_review",     # 高风险转人工
        "done": END
    }
)

# 保存检查点，支持暂停恢复
checkpointer = MemorySaver()
app = workflow.compile(
    checkpointer=checkpointer,
    interrupt_before=["human_review"]  # 到达human_review节点时暂停
)

# 使用
thread = {"configurable": {"thread_id": "session_123"}}
result = app.invoke(initial_state, thread)

# Agent暂停在human_review节点
# 人工审批后恢复
app.invoke(None, thread)  # 从检查点恢复执行

面试话术：

示例表达（仅在能用本人经历或可复现实验佐证时使用）： "HITL是Agent安全的核心机制，关键是定义清晰的触发条件：不可逆操作（删除/发送）、涉及金额超阈值、批量操作超量。我设计了4级风险：LOW自动执行、MEDIUM记录警告、HIGH等待审批、CRITICAL强制中断。用LangGraph的interrupt_before做断点，配合钉钉审批通知，超时5分钟自动拒绝。生产上线后，高风险操作事故率降低了95%。"

📚 参考：LangGraph：Human-in-the-loop 概念文档

---