> ## Documentation Index
> Fetch the complete documentation index at: https://www.yuan111.asia/doc/llms.txt
> Use this file to discover all available pages before exploring further.

# 13 · 记忆、暂停与进度：checkpointer、interrupt 与 stream

> checkpointer 持久化记忆 + interrupt 人审暂停 + stream 进度展示。

# 13 · 记忆、暂停与进度：checkpointer、interrupt 与 stream

## 开头的现象

小林用手搓的图做出了 agent 循环，但老板连环三问：

1. "**它能记住每个用户聊到哪吗？重启不丢？**" —— 需要 checkpointer
2. "**转账、删数据之前，能不能先停下来等我确认？**" —— 需要 interrupt
3. "**转三个工具，能不能一步步展示进度？**" —— 需要 stream

小林一查，发现这三样 LangGraph 全内置了。

## 第五幕：checkpointer——让记忆变成"图的一部分"

老板的需求终于来了："让机器人记住每个用户聊到哪了，重启不丢。"

小林本来想用第 4 篇的手写历史，但他现在知道：**LangGraph 内置了记忆——checkpointer（检查点保存器）。** 他用 SQLite 存状态：

```python theme={null}
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3, tempfile, os

db_path = os.path.join(tempfile.gettempdir(), "lg_memory.db")
conn = sqlite3.connect(db_path, check_same_thread=False)
checkpointer = SqliteSaver(conn)   # 检查点存 SQLite

g4 = StateGraph(MessagesState)
g4.add_node("chat", lambda s: {"messages": [model.invoke(s["messages"])]})
g4.add_edge(START, "chat")
g4.add_edge("chat", END)
app4 = g4.compile(checkpointer=checkpointer)   # ← 挂上 checkpointer

# 第 1 轮：自我介绍（thread_id 区分会话）
app4.invoke({"messages": [HumanMessage(content="我叫小林，后端工程师")]},
            config={"configurable": {"thread_id": "user1"}})
# 第 2 轮：同一 thread 接着问 → checkpointer 自动带上历史
r4 = app4.invoke({"messages": [HumanMessage(content="我叫什么？")]},
                 config={"configurable": {"thread_id": "user1"}})
print(r4["messages"][-1].content)   # "小林" —— 记住了！
```

小林看着输出，脑子里第 10 篇的 `session_id` 和这里的 `thread_id` 对上了：**"同一个概念——thread\_id 区分会话，checkpointer 把状态（包括历史）存进 SQLite。第 10 篇我手写的 InMemoryChatMessageHistory，LangGraph 用 checkpointer 全内置了，还能持久化到磁盘。"**

他叹了口气：**"我第 4 篇手写的滚雪球、第 10 篇手写的会话隔离——LangGraph 全都有，还带持久化。早用 LangGraph 是不是能少走一半弯路？"** 他想了想，又摇头：**"不手写一遍，我根本看不懂 checkpointer 在干嘛。"**

> 想验证"重启不丢"吗？跑 `11_langgraph.py` 第四部分——SQLite 文件存了检查点，第 1 轮自我介绍、第 2 轮同 thread 问"我叫什么"它记得。你把 `thread_id` 换成新的，它立刻失忆——隔离就是这么来的。

## 第六幕：interrupt——让图"暂停"等老板点头（人审）

小林做完 checkpointer 的记忆版，老板又提新需求："**机器人要转账、要删数据之前，先停下来问我一句，我确认了它才能动手。**"

小林第一反应："那我写个 if 判断？用户输入'确认'才继续？" 他写着写着发现不对劲：**模型调用是异步的、状态是分散的，你要"在模型调用前停住"，得手动控制整个流程——这跟第 10 篇手写 agent 循环一样，又丑又容易错。**

他翻 LangGraph 文档，看到 `interrupt_before`——**"让图在指定节点之前停下来，等你确认后从断点继续。"**

```python theme={null}
from langgraph.checkpoint.memory import MemorySaver
from langgraph.types import Command

# 关键：编译时 interrupt_before=["chat"]——在 chat 节点前暂停
checkpointer = MemorySaver()
app = g.compile(checkpointer=checkpointer, interrupt_before=["chat"])

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

# 第 1 次 invoke：图停在 chat 前，模型【没被调用】
r1 = app.invoke({"messages": [HumanMessage(content="你好")]}, config)
print(app.get_state(config).next)   # ('chat',) ← 暂停在这里！
print(len(r1["messages"]))          # 1 —— 只有你的问题，没有 AI 回复

# 用户确认后：从断点恢复（resume）
r2 = app.invoke(Command(resume="确认了，继续"), config)
print(r2["messages"][-1].content)   # 模型这才被调用，给出回复
print(app.get_state(config).next)   # () —— 空了，图跑完了
```

小林盯着 `app.get_state(config).next` 那行，悟了：**"interrupt 就是给图加了一个'闸门'——`interrupt_before=["chat"]` 意味着'在 chat 节点前必须等我放行'。checkpointer 记住断点位置，`Command(resume=...)` 就是'放行指令'。老板确认 → resume → 模型才继续干活。"**

他立刻想到了实际场景：**"转账工具调用前 interrupt → 等用户输'确认' → resume → 工具才真的转账。这就是'人审'（human-in-the-loop）——机器提议，人拍板。"**

**interrupt 的两种暂停方式：**

* `interrupt_before=["节点"]`：在指定节点**前**停（做事前审核，如转账前）
* `interrupt_after=["节点"]`：在指定节点**后**停（做事后检查，如转账后看结果）

> 想验证"闸门"吗？跑上面的代码——第一次 `invoke` 后 `get_state().next` 显示 `('chat',)`（暂停），模型没回复；`Command(resume=...)` 后才出现回复，`next` 变空。这就是"人审"的完整闭环。

***

## 第七幕：stream——让图"边跑边报进度"

小林做完人审，又想要"进度感"：**机器人转 3 个工具 + 1 次回答，能不能一步步展示给用户看？** 而不是闷头转半天最后"啪"一下出结果。

LangGraph 的 `stream()` 提供了三种"看进度"的模式（实测）：

```python theme={null}
# ① stream_mode="updates"：每个节点完成后，只吐"这个节点改了什么"
for chunk in app.stream(input, stream_mode="updates"):
    print(list(chunk.keys()))   # ['chat'] → "chat 节点刚跑完"

# ② stream_mode="values"：每次状态变化，吐【全量状态】
for chunk in app.stream(input, stream_mode="values"):
    print(chunk["messages"][-1])   # 每次都给你完整的最新状态

# ③ stream_mode="messages"：token 级流式（打字机效果，需真实模型）
for chunk in app.stream(input, stream_mode="messages"):
    print(chunk[0].content)   # 一个 token 一个 token 地蹦
```

**三种模式的区别（实测数据）：**

| 模式         | 吐什么          | 用途            |
| ---------- | ------------ | ------------- |
| `updates`  | 只吐增量（哪个节点动了） | 看流程到哪一步了      |
| `values`   | 吐全量状态        | 每个阶段都要完整数据    |
| `messages` | token 级（打字机） | 前端流式输出（聊天 UI） |

小林给机器人接上了 `updates` 模式，用户能看到："正在查天气 → 正在算数 → 正在汇总……"——**进度感有了，体验上一个台阶。**

> 想验证吗？用假模型（不调 API）跑 `app.stream(input, stream_mode="updates")`——每次吐 `['chat']`（节点名）；换 `stream_mode="values"`——每次都吐完整状态列表。两种模式肉眼可见的区别。

### 🔧 技术细节：这一章涉及的关键类和签名

**④ `SqliteSaver`——checkpointer（持久化记忆）**

```python theme={null}
from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

conn = sqlite3.connect("lg_memory.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)
app = g.compile(checkpointer=checkpointer)
# 调用时按 thread_id 隔离会话：
app.invoke(input, config={"configurable": {"thread_id": "user1"}})
```

**⑤ `interrupt`——人审暂停（第六幕）**

```python theme={null}
app = g.compile(
    checkpointer=checkpointer,        # 必须配 checkpointer，中断状态才能存
    interrupt_before=["chat"],        # 在 chat 节点前暂停（做事前审核）
    interrupt_after=["tools"],        # 在 tools 节点后暂停（做事后检查）
)
# 调用后检查暂停位置：
app.get_state(config).next            # ('chat',) → 停在这；() → 跑完了
# 用户确认后从断点恢复：
from langgraph.types import Command
app.invoke(Command(resume="确认了"), config)
```

**⑥ `stream()`——三种进度模式（第七幕）**

```python theme={null}
app.stream(input, stream_mode="updates")    # 每节点完成吐增量：{'chat': {...}}
app.stream(input, stream_mode="values")     # 每状态变化吐全量：{'messages': [...]}
app.stream(input, stream_mode="messages")   # token 级流式（打字机，需真实模型）
```

> 想验证吗？`app.get_state(config).next` 在 interrupt 后显示 `('chat',)`；`Command(resume=...)` 后变空。`app.stream(input, stream_mode="values")` 每次都吐完整状态。

## 结论（小林用两天换来的）

**LangGraph 三大进阶能力**：checkpointer + thread\_id（持久化记忆，重启不丢）、interrupt + Command(resume)（人审暂停，机器提议人拍板）、stream() 三种模式（边跑边报进度）。**这三样就是"生产级 Agent"的标配**——能记住、能等人、能给进度。

## 复现信号：什么时候你会想起这一章

1. **想让记忆"持久化、重启不丢"**——checkpointer + thread\_id。
2. **想"中途插一手"（人审确认后再执行）**——`compile(interrupt_before=[...])` + `Command(resume=...)`。
3. **想"边跑边报进度"**——`app.stream(input, stream_mode=...)`（updates/values/messages 三种粒度）。

小林把 LangGraph 的核心能力学完了。但他还有最后一个疑问：**LangGraph 跟 langchain\_core 到底是什么关系？它是不是踩着 core 的肩膀？** 他决定先扒一扒引擎盖：LangChain 的并发到底怎么调度？同步异步各走什么路？网络请求用哪个库？——[翻到第 14 篇：并发与网络模型](/doc/doc/narrative-course/14-并发与网络模型)。