LangGraph - “人机协同”（Human-in-the-Loop, HITL）

“人机协同”（Human-in-the-Loop, HITL）是 LangGraph 的核心能力之一。它的实现原理是：在图编译时或节点内部设置“中断点”，执行到此处时暂停并暴露当前状态，等待人类介入（审批、修改或提供信息）后，再从中断点恢复执行。

方式一：静态中断点 (interrupt_before) - 适用于“事前审批”

这种方式通过在编译图时指定 interrupt_before 参数，在特定节点执行之前强制中断。它最适合“高风险操作需要审批”的场景，比如在发送邮件、执行财务交易等动作前，强制要求人工确认。

📊 工作流示意图

graph TD
    START([开始]) --> generate[生成邮件草稿]
    generate -->|中断: 等待审批| human{👤 人工审批}
    human -->|✅ 批准| send[发送邮件]
    human -->|❌ 拒绝| END([结束])
    send --> END

💻 完整可运行代码

import uuid
from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

# 1. 定义状态
class EmailState(TypedDict):
    recipient: str
    subject: str
    body: str
    status: str  # "draft", "approved", "rejected"

# 2. 定义节点
def generate_email(state: EmailState):
    """生成邮件草稿"""
    print(f"📧 正在为 {state['recipient']} 生成邮件草稿...")
    # 模拟生成邮件内容
    return {
        "subject": f"关于 {state['recipient']} 的账户通知",
        "body": f"尊敬的 {state['recipient']}，您的账户存在异常活动，请及时核实。",
        "status": "draft"
    }

def send_email(state: EmailState):
    """发送邮件（仅当审批通过后执行）"""
    print(f"✅ 邮件已发送至 {state['recipient']}！")
    print(f"   主题: {state['subject']}")
    print(f"   内容: {state['body']}")
    return {"status": "sent"}

# 3. 构建图
builder = StateGraph(EmailState)
builder.add_node("generate_email", generate_email)
builder.add_node("send_email", send_email)

builder.add_edge(START, "generate_email")
builder.add_edge("generate_email", "send_email")  # 注意：这条边会被中断点拦截
builder.add_edge("send_email", END)

# 4. 【关键】配置检查点 + 中断点
checkpointer = InMemorySaver()
graph = builder.compile(
    checkpointer=checkpointer,
    interrupt_before=["send_email"]  # 在 send_email 节点执行前中断
)

# 5. 运行工作流并模拟人工介入
thread_config = {"configurable": {"thread_id": str(uuid.uuid4())}}

# 第一步：执行到中断点
print("🚀 启动工作流...")
for event in graph.stream(
    {"recipient": "user@example.com"},
    config=thread_config,
    stream_mode="values"
):
    print(f"当前状态: {event}")

# 第二步：获取中断状态并请求人工决策
state = graph.get_state(thread_config)
print("\n⏸️ 工作流已中断，等待人工审批...")
print(f"待审批的邮件: {state.values['subject']}")

# 模拟人工决策：这里可以替换为 input() 或 Web UI 交互
human_decision = input("请审批此邮件 (输入 'approve' 批准 / 'reject' 拒绝): ")

# 第三步：根据决策恢复工作流
if human_decision.lower() == 'approve':
    # 批准：直接恢复，执行 send_email
    print("\n✅ 人工已批准，继续执行...")
    for event in graph.stream(Command(resume=None), config=thread_config, stream_mode="values"):
        print(f"最终状态: {event}")
else:
    # 拒绝：可以更新状态后结束，或直接结束
    print("\n❌ 人工已拒绝，流程终止。")
    # 可以选择更新状态
    graph.update_state(thread_config, {"status": "rejected"})

运行效果

🚀 启动工作流...
当前状态: {'recipient': 'user@example.com'}
📧 正在为 user@example.com 生成邮件草稿...
当前状态: {'recipient': 'user@example.com', 'subject': '关于 user@example.com 的账户通知', 'body': '尊敬的 user@example.com，您的账户存在异常活动，请及时核实。', 'status': 'draft'}

⏸️ 工作流已中断，等待人工审批...
待审批的邮件: 关于 user@example.com 的账户通知
请审批此邮件 (输入 'approve' 批准 / 'reject' 拒绝): approve

✅ 人工已批准，继续执行...
✅ 邮件已发送至 user@example.com！
   主题: 关于 user@example.com 的账户通知
   内容: 尊敬的 user@example.com，您的账户存在异常活动，请及时核实。
最终状态: {...}

方式二：动态中断 (interrupt()) - 适用于“请求信息”

这种方式在节点内部调用 interrupt() 函数，在节点执行过程中动态暂停。它更灵活，适用于需要向用户提问、收集信息或进行复杂交互的场景。

📊 工作流示意图

graph TD
    START([开始]) --> human{👤 请求人工输入}
    human -->|提供信息| process[处理信息]
    process --> END([结束])

💻 完整可运行代码

import uuid
from typing import Optional, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import interrupt, Command

# 1. 定义状态
class HumanInputState(TypedDict):
    user_query: str
    human_feedback: Optional[str]  # 存储人工输入
    final_response: Optional[str]

# 2. 定义节点（内部包含中断）
def ask_human_node(state: HumanInputState):
    """向人类请求额外信息"""
    print(f"🤖 我需要更多信息来处理: '{state['user_query']}'")
    
    # 【核心】调用 interrupt()，暂停执行并返回一个值（这里返回问题）
    # 该值会通过中断事件传递给调用者
    feedback = interrupt(f"请提供关于 '{state['user_query']}' 的更多细节：")
    
    # 当恢复执行时，interrupt() 的返回值就是人工提供的信息
    print(f"👤 收到人工反馈: {feedback}")
    return {"human_feedback": feedback}

def process_node(state: HumanInputState):
    """根据用户输入和人工反馈生成最终响应"""
    response = f"根据您的查询 '{state['user_query']}' 和补充信息 '{state['human_feedback']}'，我已处理完成。"
    print(f"✅ 最终响应: {response}")
    return {"final_response": response}

# 3. 构建图
builder = StateGraph(HumanInputState)
builder.add_node("ask_human", ask_human_node)
builder.add_node("process", process_node)

builder.add_edge(START, "ask_human")
builder.add_edge("ask_human", "process")
builder.add_edge("process", END)

# 4. 【关键】配置检查点（中断依赖持久化）
checkpointer = InMemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 5. 运行工作流并模拟人工介入
thread_config = {"configurable": {"thread_id": str(uuid.uuid4())}}

print("🚀 启动工作流...\n")

# 第一步：执行直到遇到 interrupt()
for event in graph.stream(
    {"user_query": "帮我查询一下我的订单状态"},
    config=thread_config,
    stream_mode="values"
):
    print(event)

# 第二步：捕获中断并获取需要反馈的信息
state = graph.get_state(thread_config)
# 中断信息存储在 __interrupt__ 中
interrupt_info = state.tasks[0].interrupts[0] if state.tasks else None

if interrupt_info:
    print(f"\n⏸️ 工作流已中断，需要人工输入:")
    print(f"问题: {interrupt_info.value}")  # 即 interrupt() 中传入的值

    # 模拟人工输入
    human_response = input("👤 请输入您的反馈: ")

    # 第三步：使用 Command(resume=...) 恢复执行，并传入人工输入
    print("\n▶️ 恢复工作流...")
    for event in graph.stream(
        Command(resume=human_response),  # 这里 resume 的值会作为 interrupt() 的返回值
        config=thread_config,
        stream_mode="values"
    ):
        print(event)
else:
    print("工作流未中断，直接结束。")

运行效果

🚀 启动工作流...

🤖 我需要更多信息来处理: '帮我查询一下我的订单状态'
{'user_query': '帮我查询一下我的订单状态'}

⏸️ 工作流已中断，需要人工输入:
问题: 请提供关于 '帮我查询一下我的订单状态' 的更多细节：
👤 请输入您的反馈: 我的订单号是 12345

▶️ 恢复工作流...
👤 收到人工反馈: 我的订单号是 12345
✅ 最终响应: 根据您的查询 '帮我查询一下我的订单状态' 和补充信息 '我的订单号是 12345'，我已处理完成。
{'user_query': '帮我查询一下我的订单状态', 'human_feedback': '我的订单号是 12345', 'final_response': '根据您的查询 ...'}

两种方式对比