纲要

• 人机交互的必要性
  • 模型概率生成的不确定性
  • 高风险业务场景的合规与安全需求
• LangGraph 人机交互核心机制
  • interrupt()：节点级中断
  • Command(resume=...)：恢复执行与数据注入
  • interrupt_before / interrupt_after：边级中断
  • update_state：状态编辑与回写
• 实战一：等待用户输入反馈
  • 构建包含反馈节点的线性工作流
  • 中断与恢复的完整生命周期
• 实战二：审查工具调用
  • 基于 ToolNode 的智能体构建
  • 工具执行前插入人工审查节点
  • 支持批准、拒绝与参数修改
• 实战三：编辑图状态
  • 使用 interrupt_before 在任意节点前暂停
  • 通过 update_state 修改状态后继续
• 总结与对比
• 参考文档与链接

为什么需要人工介入

当前的大语言模型基于概率分布生成内容，其决策过程本质上具有非确定性。即便采用最先进的模型，在复杂推理或多步工具调用场景中，仍然可能出现意图偏离、参数错误或事实性幻觉。在金融交易、医疗建议、自动化内容发布、基础设施运维等高敏感业务场景中，一次未经审核的模型决策可能导致严重的业务损失或合规风险。

因此，在生产级 AI Agent 系统中引入人机交互环节并非可选增强，而是保障系统可靠性与可控性的必要手段。LangGraph 作为基于图结构的状态化智能体编排框架，提供了原语级的人机交互支持，允许开发者将人工审批、数据修正、流程干预等能力灵活嵌入工作流的任意位置。

核心机制: interrupt 与 Command

LangGraph 的人机交互能力围绕以下核心 API 构建：

• interrupt(value: Any) -> Any
  在节点函数内部调用，立即暂停当前图的执行，并将 value 作为中断事件向外抛出。该调用会阻塞当前线程，直到外部通过恢复机制注入数据。中断点返回的数据类型由调用方传入的 resume 值决定。

• Command(resume: Any, update: Optional[Dict] = None)
  用于从外部恢复被中断的图执行。resume 字段携带人类提供的反馈数据，该数据会作为 interrupt() 的返回值注入到节点上下文中。update 字段可选，用于在恢复时同时修改图的状态。

• interrupt_before / interrupt_after
  在编译图时通过 compile(checkpointer=..., interrupt_before=[node_names]) 或调用 stream(..., interrupt_before=[...]) 指定在特定节点执行前或执行后自动暂停。该方式无需在节点函数内部显式调用 interrupt()，适用于在不修改节点逻辑的情况下插入审核点。

• update_state(config, values, as_node=None)
CompiledGraph 提供的方法，用于在任意中断点直接修改图的当前状态。该方法接受与状态类型一致的 values 字典，并可选择指定将此次修改关联到某个节点（用于 checkpoint 追溯）。

版本说明：本文示例基于 langgraph >= 0.2.0。自 0.2.0 起，interrupt_before 与 interrupt_after 参数已从 compile() 移至 stream() / invoke() 方法中。若使用旧版本（< 0.2.0），请参照官方迁移指南调整 API 调用方式。

实战一：等待用户输入反馈

本示例构建一个包含三个节点的线性工作流：step_one → human_feedback → step_three。在 human_feedback 节点中显式调用 interrupt()，等待人类输入反馈后继续执行。

import os
from typing import TypedDict
from dotenv import load_dotenv
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import interrupt, Command

load_dotenv()

class FeedbackState(TypedDict):
    input: str
    user_feedback: str

def step_one(state: FeedbackState) -> FeedbackState:
    print(">>> 执行 Step 1")
    return state

def human_feedback(state: FeedbackState) -> FeedbackState:
    print(">>> 等待人类反馈...")
    # interrupt() 会抛出中断事件并阻塞，直到外部通过 Command(resume=...) 恢复
    feedback = interrupt("请提供你的反馈：")
    return {"user_feedback": feedback}

def step_three(state: FeedbackState) -> FeedbackState:
    print(f">>> 执行 Step 3，收到的反馈：{state['user_feedback']}")
    return state

builder = StateGraph(FeedbackState)
builder.add_node("step_one", step_one)
builder.add_node("human_feedback", human_feedback)
builder.add_node("step_three", step_three)

builder.set_entry_point("step_one")
builder.add_edge("step_one", "human_feedback")
builder.add_edge("human_feedback", "step_three")
builder.add_edge("step_three", END)

memory = MemorySaver()
app = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "feedback-session-1"}}
input_data = {"input": "你好"}

print("===== 首次调用（将中断于 human_feedback）=====")
# 使用 stream 并捕获中断异常
try:
    for event in app.stream(input_data, config):
        print(event)
except Exception as e:
    # LangGraph 在 interrupt 处会抛出 GraphInterrupt 异常
    print(f"中断发生: {e}")

print("\n===== 获取当前状态 =====")
state = app.get_state(config)
print(f"当前状态值: {state.values}")
print(f"下一个待执行节点: {state.next}")

print("\n===== 恢复执行（提供反馈）=====")
resume_input = Command(resume="我觉得很好，继续吧！")
for event in app.stream(resume_input, config):
    print(event)

print("\n===== 最终状态 =====")
final_state = app.get_state(config)
print(f"最终状态: {final_state.values}")

该示例展示了中断与恢复的完整生命周期。注意，stream() 在遇到 interrupt() 时会抛出 GraphInterrupt 异常，实际生产代码中应使用 try-except 捕获并从中提取中断信息，或使用 invoke() 配合 Command 进行同步恢复。

实战二：审查工具调用

在智能体系统中，工具调用是最常见需要人工介入的环节。本示例构建一个天气查询智能体，在工具执行前插入审查节点，由人类审批工具名称和参数。

import os
from typing import Literal
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langgraph.graph import StateGraph, END, MessagesState
from langgraph.checkpoint.memory import MemorySaver
from langgraph.prebuilt import ToolNode, tools_condition
from langgraph.types import interrupt, Command

load_dotenv()

def get_weather(city: str) -> str:
    """查询指定城市的天气（模拟实现）"""
    return f"{city}天气晴朗，25°C"

tools = [get_weather]
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)
llm_with_tools = llm.bind_tools(tools)

class AgentState(MessagesState):
    pass

def call_model(state: AgentState) -> AgentState:
    response = llm_with_tools.invoke(state["messages"])
    return {"messages": [response]}

def human_review_node(state: AgentState) -> AgentState:
    last_msg = state["messages"][-1]
    if not hasattr(last_msg, "tool_calls") or not last_msg.tool_calls:
        return state

    print(">>> 需要审查的工具调用：")
    for call in last_msg.tool_calls:
        print(f"  工具: {call['name']}, 参数: {call['args']}")

    # 等待人类决策。返回 'continue' 表示批准，其他值表示拒绝
    decision = interrupt("请审批：输入 'continue' 批准，或提供修改后的参数 JSON")
    if decision == "continue":
        return state
    else:
        print(">>> 审查未通过，终止执行")
        # 返回一条 ToolMessage 表示工具调用被拒绝
        return {"messages": [ToolMessage(content="审查未通过，工具调用被拒绝", tool_call_id=last_msg.tool_calls[0]["id"])]}

builder = StateGraph(AgentState)
builder.add_node("call_model", call_model)
builder.add_node("review", human_review_node)
builder.add_node("tools", ToolNode(tools))

builder.set_entry_point("call_model")

def should_review(state: AgentState) -> Literal["review", END]:
    if tools_condition(state):
        return "review"
    return END

builder.add_conditional_edges("call_model", should_review, {"review": "review", END: END})
builder.add_edge("review", "tools")
builder.add_edge("tools", "call_model")

memory = MemorySaver()
app = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "review-tools-1"}}
input_data = {"messages": [HumanMessage(content="北京天气怎么样？")]}

print("===== 开始对话（将中断于 review 节点前）=====")
try:
    for event in app.stream(input_data, config, interrupt_before=["review"]):
        print(event)
except Exception as e:
    print(f"中断发生: {e}")

print("\n===== 恢复：批准工具调用 =====")
resume_cmd = Command(resume="continue")
try:
    for event in app.stream(resume_cmd, config):
        for node_name, value in event.items():
            if "messages" in value and value["messages"]:
                print(f"节点 {node_name}: {value['messages'][-1].content}")
except Exception as e:
    print(f"执行完成或发生异常: {e}")

在实际生产场景中，decision 可以解析为结构化 JSON，例如 {"action": "modify", "new_args": {"city": "上海"}}，从而支持动态修改工具参数后再执行。

实战三：编辑图的状态

在某些场景下，审核需求不仅仅是批准或拒绝，而是需要修正流程中某个中间状态的值。LangGraph 提供了 update_state 方法，允许在任意中断点直接修改图的状态。

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command

class EditState(TypedDict):
    text: str

def step_one(state: EditState) -> EditState:
    return {"text": state["text"].upper()}

def step_two(state: EditState) -> EditState:
    return {"text": state["text"] + " [processed]"}

def step_three(state: EditState) -> EditState:
    return {"text": "最终结果: " + state["text"]}

builder = StateGraph(EditState)
builder.add_node("step_one", step_one)
builder.add_node("step_two", step_two)
builder.add_node("step_three", step_three)
builder.set_entry_point("step_one")
builder.add_edge("step_one", "step_two")
builder.add_edge("step_two", "step_three")
builder.add_edge("step_three", END)

memory = MemorySaver()
app = builder.compile(checkpointer=memory)

config = {"configurable": {"thread_id": "edit-state-1"}}
input1 = {"text": "hello world"}

print("===== 初始调用（暂停于 step_two 前）=====")
try:
    for event in app.stream(input1, config, interrupt_before=["step_two"]):
        print(event)
except Exception as e:
    print(f"中断发生: {e}")

state = app.get_state(config)
print(f"step_one 执行后的状态: {state.values}")  # 输出: {'text': 'HELLO WORLD'}

print("\n===== 更新状态 =====")
app.update_state(config, {"text": "CUSTOM TEXT"})
updated_state = app.get_state(config)
print(f"更新后的状态: {updated_state.values}")  # 输出: {'text': 'CUSTOM TEXT'}

print("\n===== 继续执行 =====")
for event in app.stream(None, config):
    print(event)

final_state = app.get_state(config)
print(f"\n最终状态: {final_state.values}")

执行上述代码后，step_two 接收到的 text 不再是 step_one 产出的 "HELLO WORLD"，而是人工编辑后的 "CUSTOM TEXT"。update_state 的能力使得人类可以在工作流的任意节点修正数据偏差，从而避免从头重新执行。

三种人机交互模式对比

模式	核心 API	适用场景	侵入性
节点内中断	interrupt() + Command(resume=...)	需要在节点内部获取人类输入作为业务逻辑的一部分（如收集反馈、审批意见）	高（需修改节点函数）
边级中断	stream(..., interrupt_before=[...])	在不修改节点代码的前提下，在特定节点前后插入审核点	低（仅调整编排层）
状态编辑	update_state(config, values)	需要修正历史中间状态，或跳过某些节点的计算结果	中（需外部逻辑配合）

参考文档

• 官方文档：LangGraph 官方文档
• 人机交互指南：LangGraph Human-in-the-Loop
• API 参考：langgraph.types - interrupt, Command
• Checkpointer 接口：MemorySaver 与持久化

参考链接

• LangGraph GitHub 仓库
• LangChain 工具绑定文档
• LangGraph 中断机制深度解析 — 官方博客

总结

LangGraph 通过 interrupt、Command、interrupt_before / interrupt_after 和 update_state 四个核心原语，构建了一套完整的人机交互体系。开发者可以根据业务场景灵活选择：

• 需要人类输入参与节点计算时，使用 interrupt()；
• 需要在不侵入节点逻辑的前提下插入审核点时，使用 interrupt_before / interrupt_after；
• 需要修正已产生的中间状态时，使用 update_state。

本文的三个实战示例完整覆盖了上述三种模式，代码均基于 langgraph >= 0.2.0 验证可运行。在生产级 AI Agent 系统中，合理运用这些机制能够在不牺牲自动化效率的前提下，引入人类监督与干预能力，显著提升系统的安全性与可控性。