LangGraph 断点机制
Breakpoints & Human-in-the-Loop

1 为什么需要断点：Human-in-the-Loop 的三大动机

在 Module 1 和 Module 2 中，我们构建的 Agent 是完全自动运行的：从 START 到 END，LLM 自主决策，工具自动调用，一气呵成。但在真实的生产场景中，让 Agent 完全自主并不总是安全的。

LangGraph 的 Human-in-the-Loop（人机协作）设计解决了三类核心问题：

想象一个财务报销 Agent：它可以自动分析收据、计算金额、填写报销单，但最终提交这一步必须由人类财务主管审批。断点就是你插入那个"等待审批"的卡口。

2 断点是什么：编译时静态中断点

断点的本质

断点（Breakpoint）是在 编译图时（compile 阶段）就确定的、固定的中断位置。它告诉 LangGraph："每次执行到这个节点的前面（或后面），自动暂停，不要继续往下走，等待外部指令。"

断点通过 compile() 的两个参数来设置：

# interrupt_before：在节点执行【之前】暂停
graph = builder.compile(
    interrupt_before=["tools"],   # 在 tools 节点运行前暂停
    checkpointer=memory
)

# interrupt_after：在节点执行【之后】暂停
graph = builder.compile(
    interrupt_after=["tools"],    # 在 tools 节点运行后暂停
    checkpointer=memory
)

静态断点 vs 动态中断

"静态"意味着断点在 compile() 时就已经硬编码——每次执行到那个节点，一定会暂停，没有条件判断。这与 Lesson 4 将学的动态中断（NodeInterrupt）不同，动态中断可以根据运行时状态决定是否暂停。

3 核心前提：MemorySaver 检查点器

为什么断点必须配合检查点器？

断点的工作原理是：图执行到断点时，把当前状态保存下来，然后停止执行。等人类发出"继续"的指令后，从保存的状态恢复执行。这个"保存和恢复"的机制就是 Checkpointer（检查点器）。

MemorySaver 的使用

MemorySaver 是 LangGraph 内置的内存检查点器，它将所有状态保存在进程内存中。适合开发、测试和单机场景。

from langgraph.checkpoint.memory import MemorySaver

# 创建检查点器实例
memory = MemorySaver()

# 编译图时传入 checkpointer 和断点配置
graph = builder.compile(
    interrupt_before=["tools"],
    checkpointer=memory      # 必须提供！
)

Thread（线程）的概念

检查点器用 thread_id 来区分不同的对话/运行会话。每次运行图时，你需要提供一个配置字典：

# config：指定这次运行使用哪个 thread
thread = {"configurable": {"thread_id": "1"}}

# 同一个 thread_id = 同一个对话上下文
# 不同 thread_id = 独立的、互不干扰的对话

4 实战代码：构建带断点的 Agent 图

以下是 Notebook 中的完整示例。我们构建一个带有数学工具（加法、乘法、除法）的 Agent，并在 tools 节点前设置断点，让人类在工具调用之前进行审批。

第一步：定义工具和 LLM

from langchain_deepseek import ChatDeepSeek

# 定义三个数学工具函数
def multiply(a: int, b: int) -> int:
    """Multiply a and b."""
    return a * b

def add(a: int, b: int) -> int:
    """Adds a and b."""
    return a + b

def divide(a: int, b: int) -> float:
    """Divide a by b."""
    return a / b

# 工具列表
tools = [add, multiply, divide]

# 绑定工具到 LLM（LLM 可以选择调用哪个工具）
llm = ChatDeepSeek(model="deepseek-v4-pro")
llm_with_tools = llm.bind_tools(tools)

第二步：构建图并编译（带断点）

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import MessagesState, StateGraph, START
from langgraph.prebuilt import tools_condition, ToolNode
from langchain_core.messages import SystemMessage

# 系统提示
sys_msg = SystemMessage(content="You are a helpful assistant tasked with performing arithmetic on a set of inputs.")

# 定义 assistant 节点：调用带工具的 LLM
def assistant(state: MessagesState):
    return {"messages": [llm_with_tools.invoke([sys_msg] + state["messages"])]}

# 构建图
builder = StateGraph(MessagesState)

builder.add_node("assistant", assistant)
builder.add_node("tools", ToolNode(tools))

builder.add_edge(START, "assistant")
builder.add_conditional_edges(
    "assistant",
    tools_condition,   # 如果 LLM 要调用工具 → tools；否则 → END
)
builder.add_edge("tools", "assistant")   # 工具结果返回给 LLM

# 关键：创建检查点器 + 编译时指定断点
memory = MemorySaver()
graph = builder.compile(
    interrupt_before=["tools"],   # 在 tools 节点执行前暂停
    checkpointer=memory             # 必须提供检查点器
)

第三步：运行图直到第一个断点

from langchain_core.messages import HumanMessage

# 初始输入：让 Agent 计算 2 乘以 3
initial_input = {"messages": HumanMessage(content="Multiply 2 and 3")}

# 指定 thread（对话上下文）
thread = {"configurable": {"thread_id": "1"}}

# 以流式模式运行图，直到遇到断点自动停止
for event in graph.stream(initial_input, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

图输出了 Human Message 和 AI Message（AI 决定调用 multiply 工具），然后自动暂停——没有继续执行 tools 节点。此时程序控制权返回给你，你可以查看状态、做出决策。

5 暂停后检查状态：graph.get_state(config)

获取当前图状态

图暂停后，你可以随时用 graph.get_state(config) 来查看当前的完整状态快照。这个方法返回一个 StateSnapshot 对象，包含了当前的 State 内容以及许多有用的元数据。

# 获取当前状态快照
state = graph.get_state(thread)

# 查看下一个待执行的节点
print(state.next)

state.next 返回一个元组，包含图下一步准备执行的节点名称。当它显示 ('tools',) 时，意味着：图确实已经暂停在 tools 节点执行之前，这正是我们设置 interrupt_before=["tools"] 的效果。

StateSnapshot 的完整结构

# state 对象包含多个有用属性
state = graph.get_state(thread)

state.values        # 当前 State 的全部字段值（MessagesState 里就是 messages 列表）
state.next          # 下一步要执行的节点（元组），暂停时非空，结束时为空元组
state.config        # 当前检查点的配置信息（包含 checkpoint_id）
state.metadata      # 元数据（step 计数、source 等）
state.created_at    # 检查点创建时间
state.parent_config # 上一个检查点的配置（用于时间旅行）

# 查看所有消息
for msg in state.values["messages"]:
    msg.pretty_print()

实际检查工具调用内容

暂停后，你可以深入检查 AI 打算调用什么工具、传什么参数，这是做出审批决策的依据：

# 获取最后一条消息（AI 的工具调用请求）
last_msg = state.values["messages"][-1]

# 查看工具调用详情
if hasattr(last_msg, 'tool_calls'):
    for tc in last_msg.tool_calls:
        print(f"工具名称: {tc['name']}")
        print(f"调用参数: {tc['args']}")
        print(f"调用 ID: {tc['id']}")

现在你知道 Agent 打算调用 multiply(a=2, b=3)。作为人类审批者，你可以判断这个调用是否合理，然后决定是否放行。

6 恢复执行：graph.invoke(None, config)

用 None 作为输入来恢复

这是断点机制中最优雅的设计之一：把 None 作为 input 传给 invoke 或 stream，图就会从上一个检查点继续执行，不需要重新输入任何数据。

# 人类审批后，继续执行图
# 传入 None 表示"从检查点恢复，不提供新输入"
for event in graph.stream(None, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

注意输出的顺序：

• 第一条 Ai Message：LangGraph 重新发射了断点前保存的 AI 消息（包含工具调用请求），这是状态重放的正常现象
• Tool Message：tools 节点真正执行了，multiply(2, 3) 返回了 6
• 第二条 Ai Message：LLM 拿到工具结果，给出了最终的自然语言答案

拒绝执行（不恢复）

如果人类审批者决定拒绝这次工具调用，只需不调用 invoke(None, ...)，直接结束流程即可。这次对话的检查点仍然保留在 MemorySaver 中，但不会再有任何继续执行。

# 完整的审批流程示例
initial_input = {"messages": HumanMessage(content="Multiply 2 and 3")}
thread = {"configurable": {"thread_id": "2"}}

# 运行直到断点
for event in graph.stream(initial_input, thread, stream_mode="values"):
    event['messages'][-1].pretty_print()

# 等待人类审批
user_approval = input("Do you want to call the tool? (yes/no): ")

if user_approval.lower() == "yes":
    # 审批通过：继续执行
    for event in graph.stream(None, thread, stream_mode="values"):
        event['messages'][-1].pretty_print()
else:
    # 审批拒绝：不继续执行
    print("Operation cancelled by user.")

7 interrupt_before vs interrupt_after 的区别

两种断点模式对比

interrupt_before：审批工具调用（Notebook 示例）

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

# 图暂停时，state 里包含 AI 拟好的邮件内容（工具调用参数）
# 人类可以查看收件人、主题、正文，决定是否发送
state = graph.get_state(thread)
tool_call = state.values["messages"][-1].tool_calls[0]
print(f"收件人: {tool_call['args']['to']}")
print(f"主题: {tool_call['args']['subject']}")
# → 人类确认后：graph.invoke(None, thread)

interrupt_after：审核工具结果

# 在 web_search 节点执行后暂停
graph = builder.compile(
    interrupt_after=["web_search"],
    checkpointer=memory
)

# 图暂停时，搜索已经完成，state 里有 ToolMessage（搜索结果）
# 人类可以检查来源、筛选掉不可信的结果
state = graph.get_state(thread)
# 最后一条是 ToolMessage（工具执行结果）
search_result = state.values["messages"][-1].content
print(f"搜索结果: {search_result}")
# → 确认结果可信后：graph.invoke(None, thread) 让 LLM 继续总结

两种模式的时序对比

8 完整的用户审批流程与 LangGraph API 集成

完整的本地审批代码模式

将前面所有知识整合成一个完整可用的审批流程模板：

from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_core.messages import HumanMessage

# ─── 构建图（假设 tools、llm_with_tools、assistant 已定义）───
memory = MemorySaver()
graph = builder.compile(
    interrupt_before=["tools"],
    checkpointer=memory
)

def run_with_approval(user_question: str, thread_id: str):
    """完整的带审批的 Agent 执行函数"""
    thread = {"configurable": {"thread_id": thread_id}}
    initial_input = {"messages": HumanMessage(content=user_question)}

    print("=== Agent 开始执行 ===")

    # 第一阶段：运行到断点
    for event in graph.stream(initial_input, thread, stream_mode="values"):
        event['messages'][-1].pretty_print()

    # 检查是否暂停在断点（如果 next 非空，说明还有待执行的节点）
    state = graph.get_state(thread)
    if not state.next:
        print("=== 图已执行完毕（无需工具调用）===")
        return

    print(f"\n=== 图已暂停，下一步: {state.next} ===")

    # 显示工具调用详情供人类审查
    last_msg = state.values["messages"][-1]
    if hasattr(last_msg, 'tool_calls'):
        print("\n待审批的工具调用：")
        for tc in last_msg.tool_calls:
            print(f"  工具: {tc['name']}")
            print(f"  参数: {tc['args']}")

    # 等待人类审批
    approval = input("\n是否批准执行工具？(yes/no): ")

    if approval.lower() == "yes":
        print("\n=== 审批通过，继续执行 ===")
        for event in graph.stream(None, thread, stream_mode="values"):
            event['messages'][-1].pretty_print()
    else:
        print("=== 操作已被用户取消 ===")

# 使用示例
run_with_approval("Multiply 2 and 3", thread_id="session-001")

通过 LangGraph API（SDK）使用断点

在生产环境中，图通常部署为 LangGraph 服务（通过 langgraph dev 启动本地服务器，或部署到 LangSmith Studio）。此时你通过 LangGraph SDK 与图交互，断点的逻辑完全一致：

from langgraph_sdk import get_client
from langchain_core.messages import HumanMessage

# 连接本地 LangGraph 开发服务器
client = get_client(url="http://127.0.0.1:2024")

# 创建新 thread
thread = await client.threads.create()

# 流式运行图，并在 API 层面指定断点
# 注意：也可以直接在 compile() 里设置，这里是另一种方式
initial_input = {"messages": HumanMessage(content="Multiply 2 and 3")}

async for chunk in client.runs.stream(
    thread["thread_id"],
    assistant_id="agent",
    input=initial_input,
    stream_mode="values",
    interrupt_before=["tools"],      # 在 API 层面设置断点
):
    print(f"Event type: {chunk.event}")
    messages = chunk.data.get('messages', [])
    if messages:
        print(messages[-1])

# 审批后继续（传入 None）
async for chunk in client.runs.stream(
    thread["thread_id"],
    "agent",
    input=None,                        # None = 从检查点恢复
    stream_mode="values",
    interrupt_before=["tools"],
):
    messages = chunk.data.get('messages', [])
    if messages:
        print(messages[-1])

核心概念总结