Human-in-the-Loop：人工介入

6.1 Human-in-the-Loop 模式

人机交互是 LangGraph 的杀手级特性。它让你可以在 Agent 执行过程中插入人工干预。

6.2 NodeInterrupt 用法

python
from langgraph.errors import NodeInterrupt

class OrderState(TypedDict):
    order_id: str
    amount: float
    status: str
    human_decision: str | None

def check_order(state: OrderState) -> dict:
    # 大额订单需要人工审批
    if state["amount"] > 5000:
        raise NodeInterrupt(
            f"订单 {state['order_id']} 金额 {state['amount']} 元，需要人工审批"
        )
    return {"status": "auto_approved"}

def process_after_approval(state: OrderState) -> dict:
    decision = state["human_decision"]
    if decision == "approve":
        return {"status": "approved"}
    else:
        return {"status": "rejected"}

中断恢复流程

python
# 1. 执行（触发中断）
try:
    result = app.invoke({"order_id": "O123", "amount": 10000, ...})
except NodeInterrupt as e:
    print(f"等待审批: {e.message}")
    thread_id = result["configurable"]["thread_id"]

# 2. 人工审批（外部系统）
human_decision = "approve"  # 假设人工批准

# 3. 恢复执行
result = app.invoke(
    {"human_decision": human_decision},
    config={"configurable": {"thread_id": thread_id}}
)

6.3 典型应用场景

Human-in-the-Loop 适用于需要人工介入的关键节点：

6.4 审批工作流实战

6.3.1 内容审核

python
class ContentModerationState(TypedDict):
    content: str
    flag_words: list[str]
    needs_review: bool
    final_decision: str | None

def detect_sensitive_content(state: ContentModerationState) -> dict:
    content = state["content"]
    flag_words = [w for w in SENSITIVE_WORDS if w in content]
    
    if flag_words:
        raise NodeInterrupt(
            f"检测到敏感词: {flag_words}",
            context={"flag_words": flag_words}
        )
    return {"needs_review": False}

def publish_content(state: ContentModerationState) -> dict:
    return {"final_decision": "published"}

6.3.2 邮件发送审批

python
class EmailState(TypedDict):
    to: str
    subject: str
    body: str
    is_approved: bool

def draft_email(state: EmailState) -> dict:
    # 草稿完成，等待审批
    raise NodeInterrupt(
        f"邮件发送给 {state['to']}: {state['subject']}",
        context={"preview": state["body"][:100]}
    )

def send_email(state: EmailState) -> dict:
    # 调用邮件 API
    send_via_smtp(state["to"], state["subject"], state["body"])
    return {"status": "sent"}

6.4 状态更新与恢复

python
# 从检查点恢复并更新状态
def resume_with_update():
    # 获取之前的检查点
    checkpoint = app.get_state({"configurable": {"thread_id": "tx_123"}})
    
    # 更新状态
    app.update_state(
        {"configurable": {"thread_id": "tx_123"}},
        {"human_feedback": "Looks good!", "approved": True}
    )
    
    # 继续执行
    return app.invoke(None, {"configurable": {"thread_id": "tx_123"}})

6.6 本章小结

• Human-in-the-Loop 让 Agent 可以暂停等待人工决策
• NodeInterrupt 触发中断，可携带上下文信息
• 使用 thread_id 恢复中断的任务
• 典型场景：审批、审核、确认
• 可结合 Checkpointer 实现持久化中断