LangGraph Human-in-the-loop：人机协作设计

在构建生产级 Agent 时，一个核心问题是：哪些操作可以让 Agent 自动执行，哪些操作必须经过人工确认？

发邮件、写数据库、调用付款接口——这些有副作用的操作，在 AI 犯错的代价远大于多等几秒钟的情况下，必须有人工确认环节。这就是 Human-in-the-loop（人机协同：在 AI 自动执行的流程中插入人工审核或确认步骤）的价值所在。

1.1 为什么需要人工介入

LangGraph Human-in-the-Loop 流程——interrupt() 暂停执行，人工审批后通过 resume 恢复

LLM 会犯错。幻觉、误解用户意图、在边缘案例上做出奇怪的决策——这些不是 bug，是 LLM 的本质特性。

问题不在于 AI 会不会犯错，而在于犯错的代价。

查一个信息答错了，用户可以再问一遍，代价很小。但如果 Agent 自动发了一封措辞不当的邮件给 1000 个客户，或者自动删除了一批数据，代价就很大了。

因此生产级 Agent 的设计原则是：AI 负责生成方案，人负责确认高风险操作。

LangGraph 原生支持这种设计模式，通过 interrupt_before 和 interrupt_after 实现节点执行前后的暂停。

1.2 Checkpointer：让 Agent 能够暂停和恢复

要支持暂停等待人工输入，Agent 必须能把自己当前的状态保存下来，等人工响应后再从断点继续。这就是 Checkpointer（检查点存储器：负责把 Agent 每一步的运行状态持久化保存，使得任务可以随时中断、随时恢复）的作用。

LangGraph 提供了两种内置实现：

MemorySaver：把状态保存在内存里，进程重启后丢失。适合开发测试：

python

from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()

SqliteSaver：把状态保存在 SQLite 数据库文件里，进程重启后可以恢复。适合生产环境（需要单独安装 langgraph-checkpoint-sqlite 包）：

python

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

conn = sqlite3.connect("checkpoints.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)

编译图的时候把 checkpointer 传进去：

python

app = graph.compile(
    checkpointer=checkpointer,
    interrupt_before=["send_email"]  # 在 send_email 节点执行前暂停
)

1.3 thread_id：支持多个并发会话

每次调用 Agent 都需要指定一个 thread_id，作为这次会话的唯一标识。同一个 thread_id 的所有调用共享同一份状态，不同 thread_id 之间完全隔离。

python

config = {"configurable": {"thread_id": "user_123_complaint_456"}}

# 第一次调用，Agent 运行到 interrupt 点暂停
result = app.invoke(initial_state, config=config)

# 人工审核后，继续执行
result = app.invoke(None, config=config)  # 传 None 表示从断点继续

thread_id 的设计让 Agent 天然支持多用户并发：每个用户、每个任务都有自己的 thread_id，互不干扰。

1.4 三种人工介入场景

1.4.1 场景一：审批（Approve/Reject）

最常见的模式：Agent 准备好了要执行的操作，人工看了之后说批准还是拒绝。

python

# 人工审核后，把审批结果写入 State
app.update_state(
    config,
    {"approved": True}  # 或者 {"approved": False}
)
# 然后继续执行
result = app.invoke(None, config=config)

1.4.2 场景二：修改 Agent 的输入

Agent 起草了一封邮件，人工觉得措辞不太对，可以直接修改邮件内容，然后让 Agent 继续发送。

python

# 修改 Agent 生成的邮件草稿
app.update_state(
    config,
    {"email_draft": "修改后的邮件内容..."}
)
result = app.invoke(None, config=config)

1.4.3 场景三：完全接管

某些步骤 AI 完全搞不定，直接让人工替代 AI 完成那个节点的工作，然后把结果写入 State，让 Agent 继续执行后续步骤。

python

# 人工直接填写某个节点本该产出的字段
app.update_state(
    config,
    {"complex_analysis": "人工完成的分析结果..."},
    as_node="analysis_node"  # 告诉 LangGraph 这是 analysis_node 的输出
)
result = app.invoke(None, config=config)

1.5 完整示例：需要人工审批的邮件 Agent

下面是一个完整的邮件发送 Agent，在发送前会暂停等待人工审批。

python

from typing import TypedDict, Optional
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import MemorySaver


# 1. 定义 State
class EmailAgentState(TypedDict):
    complaint_text: str       # 用户投诉内容
    customer_email: str       # 客户邮箱
    email_draft: str          # Agent 起草的邮件
    approved: Optional[bool]  # 人工审批结果，None 表示待审批
    sent: bool                # 是否已发送
    rejection_reason: str     # 如果被拒绝，记录原因


# 2. 定义节点

def draft_email_node(state: EmailAgentState) -> dict:
    """起草邮件节点：根据投诉内容生成道歉邮件"""
    print("\n[起草节点] 正在根据投诉内容生成邮件...")

    # 实际项目里这里调用 LLM
    complaint = state["complaint_text"]
    draft = f"""尊敬的客户，

感谢您的反馈。针对您提到的「{complaint}」问题，我们深感抱歉。

我们已将您的问题记录在案，相关团队将在 24 小时内与您联系，提供解决方案。

如有任何疑问，请随时联系我们的客服团队。

此致
客户服务团队"""

    print(f"[起草节点] 邮件草稿已生成，目标邮箱：{state['customer_email']}")
    return {
        "email_draft": draft,
        "approved": None,  # 重置审批状态
    }


def review_node(state: EmailAgentState) -> dict:
    """
    审核节点：这个节点本身什么都不做
    它的存在只是为了让 interrupt_before 在这里暂停
    暂停后等待人工通过 update_state 写入 approved 字段
    """
    print("\n[审核节点] 等待人工审批...")
    return {}


def send_email_node(state: EmailAgentState) -> dict:
    """发送邮件节点：执行实际发送操作"""
    if not state.get("approved"):
        print("\n[发送节点] 未获批准，跳过发送")
        return {"sent": False}

    print(f"\n[发送节点] 正在发送邮件到 {state['customer_email']}...")
    # 实际项目里这里调用 SMTP 或邮件服务 API
    print("[发送节点] 邮件发送成功！")
    return {"sent": True}


def notify_node(state: EmailAgentState) -> dict:
    """通知节点：记录处理结果"""
    if state.get("sent"):
        print("\n[通知节点] 投诉处理完成，邮件已发送")
    else:
        reason = state.get("rejection_reason", "未说明原因")
        print(f"\n[通知节点] 邮件发送已取消，原因：{reason}")
    return {}


# 3. 路由函数

def route_after_review(state: EmailAgentState) -> str:
    """审批后的路由：批准则发送，拒绝则通知"""
    if state.get("approved") is True:
        return "send"
    elif state.get("approved") is False:
        return "skip"
    # approved 还是 None 说明人工还没操作，理论上不会到这里
    return "skip"


# 4. 构建图

def build_email_agent():
    checkpointer = MemorySaver()

    graph = StateGraph(EmailAgentState)

    graph.add_node("draft", draft_email_node)
    graph.add_node("review", review_node)
    graph.add_node("send", send_email_node)
    graph.add_node("notify", notify_node)

    graph.add_edge(START, "draft")
    graph.add_edge("draft", "review")

    graph.add_conditional_edges(
        "review",
        route_after_review,
        {
            "send": "send",
            "skip": "notify",
        }
    )

    graph.add_edge("send", "notify")
    graph.add_edge("notify", END)

    # 关键：在 review 节点执行前暂停，等待人工输入
    app = graph.compile(
        checkpointer=checkpointer,
        interrupt_before=["review"]
    )

    return app


# 5. 模拟完整的人机协作流程

def run_approval_workflow(complaint: str, customer_email: str, approve: bool):
    """
    模拟完整的审批流程
    approve=True 表示人工批准，approve=False 表示人工拒绝
    """
    app = build_email_agent()

    # 每个投诉处理任务有独立的 thread_id
    thread_id = f"complaint_{customer_email.split('@')[0]}"
    config = {"configurable": {"thread_id": thread_id}}

    initial_state = {
        "complaint_text": complaint,
        "customer_email": customer_email,
        "email_draft": "",
        "approved": None,
        "sent": False,
        "rejection_reason": "",
    }

    print(f"{'='*50}")
    print(f"开始处理投诉，thread_id: {thread_id}")
    print(f"{'='*50}")

    # 第一阶段：Agent 运行到 interrupt 点自动暂停
    result = app.invoke(initial_state, config=config)

    # 此时 Agent 已暂停，邮件草稿已生成
    current_state = app.get_state(config)
    draft = current_state.values.get("email_draft", "")

    print("\n" + "-"*50)
    print("Agent 已暂停，等待人工审批。邮件草稿如下：")
    print("-"*50)
    print(draft)
    print("-"*50)

    # 模拟人工审批操作
    if approve:
        print("\n[人工操作] 审批通过")
        app.update_state(config, {"approved": True})
    else:
        print("\n[人工操作] 审批拒绝，邮件措辞需要调整")
        app.update_state(config, {
            "approved": False,
            "rejection_reason": "措辞过于官方，缺乏温度"
        })

    # 第二阶段：从断点继续执行
    print("\n[系统] 继续执行 Agent...")
    final_result = app.invoke(None, config=config)

    print(f"\n{'='*50}")
    print(f"处理完成，邮件发送状态：{final_result.get('sent', False)}")
    print(f"{'='*50}")


if __name__ == "__main__":
    # 场景1：人工批准
    print("\n场景1：人工批准邮件发送")
    run_approval_workflow(
        complaint="购买的商品三天后才发货，和承诺的次日达不符",
        customer_email="zhang_san@example.com",
        approve=True
    )

    print("\n\n")

    # 场景2：人工拒绝
    print("场景2：人工拒绝邮件发送")
    run_approval_workflow(
        complaint="客服态度很差，问题半天没解决",
        customer_email="li_si@example.com",
        approve=False
    )

运行这段代码，可以清楚地看到两个阶段：Agent 生成草稿后暂停，人工操作后继续执行。

1.6 流程图

带人工节点的完整流程：

橙色的 interrupt 节点是暂停点，紫色的 review 节点是等待人工输入的节点。

1.7 interrupt_before vs interrupt_after

两者的区别如下：

interrupt_before=["node_name"]：在节点执行之前暂停。人工可以修改 State，然后让节点重新执行或者跳过。
interrupt_after=["node_name"]：在节点执行之后暂停。节点已经执行完了，人工审核输出结果，决定是否继续。

发邮件这种场景用 interrupt_before，因为需要在邮件发出去之前审核；生成报告这种场景可以用 interrupt_after，节点生成报告后，人工审核内容，觉得没问题再继续后续步骤。

1.8 一个容易踩的坑

用 interrupt_before 的时候，Agent 会在节点执行前暂停，也就是说那个节点根本没有运行。

如果把审批逻辑写在了被 interrupt 的节点里，而不是在外部通过 update_state 写入，那审批结果永远不会生效——因为那个节点从来没有运行过。

正确做法是：把需要人工填写的字段留在 State 里（比如 approved），通过 app.update_state() 从外部写入，让 Agent 在继续执行时自动读取这个字段。

1.9 小结

完全自主的 Agent 在演示里表现不错，在生产里风险很高。用户输入不可预测，LLM 判断存在偏差，触发的操作后果也难以回滚。把人放在关键节点上，不是不信任 AI，而是在代价超出可控范围之前加一道保险。

LangGraph 把这个模式做成了框架级别的原生支持，不需要在外面包临时方案。

下一步是把 State/Node/Edge、Conditional Edge、Human-in-the-loop 这三件事组合起来，搭一个真实的业务 Agent。