LangGraph Checkpoint 入门：用 MemorySaver 为 Agent 装上「可恢复的记忆」

目录

一、为什么 Agent 需要 Checkpoint？

二、先建立正确的心智模型

三、Checkpoint、Thread 与版本链

（一）Checkpointer：状态保存后端

（二）Thread ID：一条状态链的业务标识

（三）Checkpoint 版本链：状态不只有“现在”

（四）一次 invoke 到底发生了什么？

四、代码演示详细说明

（一）代码结构分析

（二）详细演示说明分析

1. 演示 1：MemorySaver 基本用法

为什么 turn_count 是 1，而不是 3？

2. 演示 2：Thread ID 隔离

线上环境应该如何设计 thread_id？

3. 演示 3：状态恢复

一个必须讲清楚的细节

4. 演示 4：查看 Checkpoint 历史

为什么三次 invoke 会有六个以上的版本？

历史顺序也要注意

（三）从历史 Checkpoint 恢复，不等于读取缓存

1. 查看历史状态

2. 从历史状态重新执行

五、Checkpoint 与 Store 不要混为一谈

六、常见问题排查说明及线上考虑基本点说明

（一）常见问题与排查方式

1. 问题 1：忘记传入 thread_id

2. 问题 2：服务重启后 MemorySaver 状态消失

3. 问题 3：不同用户的对话发生串台

4. 问题 4：消息能够累积，普通字段却没有累积

5. 问题 5：修改 State 后，旧 Checkpoint 出现兼容问题

6. 问题 6：Checkpoint 越积越多

裁剪模型上下文

清理 Checkpoint 历史

（二）生产环境还需要考虑什么？

1. 持久化后端

2. 同一线程的并发调用

3. 节点幂等性

4. 敏感数据保护

5. 可观测性

七、如何读懂本次 Demo 的四个结果？

八、总结

一页速查

干货分享，感谢您的阅读！

这是「LangGraph Agent Engineering Mastery」系列 Stage 3「记忆」系列第 2 篇。 本文完整演示 MemorySaver、thread_id、状态恢复和 Checkpoint 历史版本，并重点解释一个容易被误解的问题：内存 Checkpoint 能恢复状态，但不等于真正的跨进程持久化。

一、为什么 Agent 需要 Checkpoint？

设想这样一个场景。用户正在和你的 Agent 交谈：

记住，我的密码提示是「蓝色的天空」。

紧接着，你发布了新版本，服务重新部署，原来的 Python 进程退出。几分钟后，用户回来问：

我的密码提示是什么？

Agent 却回答：

抱歉，我们好像刚刚开始聊天。

从模型的角度看，这个回答并没有错。因为大语言模型本身不会自动保存上一次调用的状态；如果应用也没有将消息历史和执行状态存下来，那么每一次请求都只是一次全新的推理。

问题的根源不是模型“记性差”，而是：

Agent 的状态默认只存在于当前执行上下文中。应用不主动保存，状态就不会自动延续。

LangGraph 的 Checkpoint 机制，就是为了解决这个问题。

它会在 Graph 执行过程中保存状态快照，并通过一个稳定的 thread_id 将这些快照组织成独立线程。后续请求再次使用相同的 thread_id 时，Graph 就能够先恢复已有状态，再继续执行。

官方文档将 Checkpointer 定位为线程范围内的状态持久化机制，主要用于多轮对话、人工审批、故障恢复和时间旅行；跨线程共享的长期事实，则更适合放入 Store。

二、先建立正确的心智模型

很多人第一次看到 Checkpoint，会把它理解成“自动保存聊天记录”。这个理解不算错，但还不够完整。

Checkpoint 保存的不只是 messages，而是当前 Graph State 中所有受管理的状态字段。例如：

class ChatState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
turn_count: int

这个 State 里包含两个字段：

• messages：完整的消息序列；

• turn_count：当前记录的对话轮次。

如果一个更复杂的 Agent 还保存了工具结果、审批状态、检索文档、执行计划、待处理任务或工作流进度，这些字段同样可以成为 Checkpoint 的一部分。

因此，更准确的定义是：

Checkpoint 是 Graph State 在某个执行时刻的状态快照。

LangGraph 会在每个 super-step 边界保存完整的 StateSnapshot。在线性 Graph 中，一个 super-step 往往对应一次节点推进；在并行 Graph 中，则可能有多个节点处于同一个 super-step。

现在针对本次教学，我们完整的代码如下：

"""Demo 02: Checkpoint Basics — MemorySaver、Thread ID、状态恢复。

演示 LangGraph 内置的 Checkpoint 机制：
1. MemorySaver：纯内存的 Checkpoint 后端
2. Thread ID：线程隔离，不同对话互不干扰
3. 状态恢复：中断后从最近一次 Checkpoint 恢复执行

运行方式：
python stages/stage3_memory/02_checkpoint_basics/main.py
"""

from __future__ import annotations

import sys
from pathlib import Path
from typing import Annotated, TypedDict

from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import END, START, StateGraph
from langgraph.graph.message import add_messages

sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent.parent))

from shared import get_llm, get_logger, log_step, log_success

logger = get_logger("demo.02_checkpoint_basics")

class ChatState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
turn_count: int

def build_checkpoint_graph():
"""构建带 Checkpoint 支持的对话图。"""
llm = get_llm(fallback_to_mock=False)

def chat_node(state: ChatState) -> dict:
log_step(logger, "chat_node", f"第 {state.get('turn_count', 0) + 1} 轮对话")
response = llm.invoke(state["messages"])
return {
"messages": [response],
"turn_count": state.get("turn_count", 0) + 1,
}

graph = StateGraph(ChatState)
graph.add_node("chat", chat_node)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)
return graph

# ============================================================
# 演示 1：基本 Checkpoint 使用
# ============================================================
def demo_basic_checkpoint():
"""演示 MemorySaver 的基本用法。"""
print("\\n— 演示 1: MemorySaver 基本用法 —\\n")

checkpointer = MemorySaver()
graph = build_checkpoint_graph()
app = graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "demo-thread-1"}}

conversations = [
"你好，我叫李明",
"我是一名 Python 开发者",
"你还记得我叫什么吗？",
]

results = []
for user_msg in conversations:
print(f" 用户: {user_msg}")
result = app.invoke(
{"messages": [HumanMessage(content=user_msg)], "turn_count": 0},
config=config,
)
ai_msg = result["messages"][-1]
print(f" 助手: {ai_msg.content}")
print()
results.append(result)

snapshot = app.get_state(config)
print(" Checkpoint 状态:")
print(f" 消息数量: {len(snapshot.values.get('messages', []))}")
print(f" 对话轮次: {snapshot.values.get('turn_count', 'N/A')}")
print()

return {"results": results, "checkpointer": checkpointer, "config": config}

# ============================================================
# 演示 2：Thread ID 隔离
# ============================================================
def demo_thread_isolation():
"""演示不同 thread_id 之间的状态隔离。"""
print("\\n— 演示 2: Thread ID 隔离 —\\n")

checkpointer = MemorySaver()
graph = build_checkpoint_graph()
app = graph.compile(checkpointer=checkpointer)

config_alice = {"configurable": {"thread_id": "user-alice"}}
config_bob = {"configurable": {"thread_id": "user-bob"}}

print(" [Alice 的对话]")
app.invoke(
{"messages": [HumanMessage(content="我叫 Alice，我喜欢猫")], "turn_count": 0},
config=config_alice,
)
print(" Alice: 我叫 Alice，我喜欢猫")

print(" [Bob 的对话]")
app.invoke(
{"messages": [HumanMessage(content="我叫 Bob，我喜欢狗")], "turn_count": 0},
config=config_bob,
)
print(" Bob: 我叫 Bob，我喜欢狗")

print("\\n [验证隔离]")
alice_state = app.get_state(config_alice)
bob_state = app.get_state(config_bob)

alice_msgs = alice_state.values.get("messages", [])
bob_msgs = bob_state.values.get("messages", [])

print(f" Alice 消息数: {len(alice_msgs)}")
print(f" Bob 消息数: {len(bob_msgs)}")

alice_has_alice = any("Alice" in str(m.content) or "alice" in str(m.content) for m in alice_msgs)
bob_has_bob = any("Bob" in str(m.content) or "bob" in str(m.content) for m in bob_msgs)
print(f" Alice 对话包含 'Alice': {alice_has_alice}")
print(f" Bob 对话包含 'Bob': {bob_has_bob}")
print()

return {
"alice_msg_count": len(alice_msgs),
"bob_msg_count": len(bob_msgs),
"isolation_verified": True,
}

# ============================================================
# 演示 3：状态恢复
# ============================================================
def demo_state_recovery():
"""演示从 Checkpoint 恢复状态。"""
print("\\n— 演示 3: 状态恢复 —\\n")

checkpointer = MemorySaver()
graph = build_checkpoint_graph()

config = {"configurable": {"thread_id": "recovery-thread"}}

log_step(logger, "阶段 1", "创建对话并积累状态")
app1 = graph.compile(checkpointer=checkpointer)

app1.invoke(
{"messages": [HumanMessage(content="记住：我的密码提示是'蓝色的天空'")], "turn_count": 0},
config=config,
)
app1.invoke(
{"messages": [HumanMessage(content="我最喜欢的编程语言是 Rust")], "turn_count": 0},
config=config,
)
print(" 阶段 1: 已写入 2 轮对话")

state_before = app1.get_state(config)
msg_count_before = len(state_before.values.get("messages", []))
print(f" Checkpoint 中消息数: {msg_count_before}")

log_step(logger, "阶段 2", "模拟进程重启 — 用同一个 checkpointer 创建新的 app")
app2 = graph.compile(checkpointer=checkpointer)

state_after = app2.get_state(config)
msg_count_after = len(state_after.values.get("messages", []))
print(f" 恢复后消息数: {msg_count_after}")

result = app2.invoke(
{"messages": [HumanMessage(content="我最喜欢什么编程语言？")], "turn_count": 0},
config=config,
)
final_msg = result["messages"][-1]
print(" 用户: 我最喜欢什么编程语言？")
print(f" 助手: {final_msg.content}")

log_success(logger, f"状态恢复成功！恢复前 {msg_count_before} 条 → 恢复后 {msg_count_after} 条")
print()

return {
"msg_count_before": msg_count_before,
"msg_count_after": msg_count_after,
"recovery_success": msg_count_after == msg_count_before,
}

# ============================================================
# 演示 4：查看 Checkpoint 历史
# ============================================================
def demo_checkpoint_history():
"""演示查看 Checkpoint 的历史版本。"""
print("\\n— 演示 4: Checkpoint 历史版本 —\\n")

checkpointer = MemorySaver()
graph = build_checkpoint_graph()
app = graph.compile(checkpointer=checkpointer)

config = {"configurable": {"thread_id": "history-thread"}}

steps = ["第一步", "第二步", "第三步"]
for step in steps:
app.invoke(
{"messages": [HumanMessage(content=f"这是{step}")], "turn_count": 0},
config=config,
)

print(" Checkpoint 历史:")
history_count = 0
for state in app.get_state_history(config):
history_count += 1
msg_count = len(state.values.get("messages", []))
ts = state.config.get("configurable", {}).get("checkpoint_id", "N/A")
print(f" 版本 {history_count}: {msg_count} 条消息, checkpoint_id={ts[:20]}…")
if history_count >= 6:
print(" … (更多历史省略)")
break

print(f"\\n 总共 {history_count}+ 个 Checkpoint 版本")
print()

return {"history_count": history_count}

def run_demo() -> dict:
"""运行 Checkpoint Basics 全部演示。"""
print("=" * 60)
print(" Demo 02: Checkpoint Basics — 状态持久化基础")
print("=" * 60)

basic = demo_basic_checkpoint()
isolation = demo_thread_isolation()
recovery = demo_state_recovery()
history = demo_checkpoint_history()

print()
print("=" * 60)
print(" 关键概念回顾")
print("=" * 60)
print(" 1. MemorySaver : LangGraph 内置的纯内存 Checkpoint 后端")
print(" 2. thread_id : 通过 config 传入，实现对话线程隔离")
print(" 3. 状态恢复 : 同一 thread_id 自动从最近 Checkpoint 恢复")
print(" 4. 历史版本 : 每次 invoke 都会创建新的 Checkpoint 版本")
print(" 5. get_state() : 查看当前线程的最新状态")
print(" 6. get_state_history(): 查看所有历史 Checkpoint")
print()

return {
"basic": basic,
"isolation": isolation,
"recovery": recovery,
"history": history,
}

if __name__ == "__main__":
run_demo()

三、Checkpoint、Thread 与版本链

整个 Checkpoint 系统可以拆成三个核心概念。

（一）Checkpointer：状态保存后端

Checkpointer 决定状态保存在哪里。本文代码使用）：

from langgraph.checkpoint.memory import MemorySaver

并创建一个纯内存后端：

checkpointer = MemorySaver()

MemorySaver 使用当前 Python 进程的内存保存 Checkpoint，适合：

• 本地学习；

• 单元测试；

• 快速验证；

• 开发阶段调试。

但它不适合真正的生产持久化，因为 Python 进程退出后，内存中的数据也会消失。官方文档同样明确指出，MemorySaver 和 InMemorySaver 的数据不会在进程重启后保留。

（二）Thread ID：一条状态链的业务标识

每次调用带 Checkpointer 的 Graph 时，都需要传入：

config = {"configurable": {"thread_id": "demo-thread-1"}}

thread_id 可以理解成一把钥匙。

• 相同的 thread_id：继续读取原有状态；

• 不同的 thread_id：创建相互隔离的新线程；

• 没有 thread_id：Checkpointer 不知道应该读取或写入哪条状态链。

官方文档将 thread_id 描述为 Checkpointer 保存和加载状态时使用的主要标识。

（三）Checkpoint 版本链：状态不只有“现在”

同一个线程不是只保存一个最终状态。随着 Graph 不断执行，Checkpointer 会持续产生新的状态版本：

checkpoint_1
↓
checkpoint_2
↓
checkpoint_3
↓
checkpoint_4

这意味着应用不仅能够读取“最新状态”，还可以查看历史状态，并从某个历史 Checkpoint 重新执行后续节点。

这也是 LangGraph 能够实现以下能力的基础：

• 对话状态恢复；

• Human-in-the-loop；

• 故障续跑；

• 历史调试；

• Replay；

• Fork；

• Time Travel。

（四）一次 invoke 到底发生了什么？

可以把一次调用理解成以下过程：

这里最容易被忽略的是 Reducer。

恢复状态并不意味着简单地用新输入覆盖旧状态。每一个 State 字段都有自己的更新规则：

• 声明了 Reducer 的字段，按照 Reducer 合并；

• 没有声明 Reducer 的字段，通常由新值直接覆盖旧值。

在本文代码中：

class ChatState(TypedDict):
messages: Annotated[list[BaseMessage], add_messages]
turn_count: int

messages 使用了 add_messages，因此新的消息会追加到已有消息列表中。

turn_count 没有自定义 Reducer，所以它采用默认覆盖行为。

这个差异，正是后面出现“消息数量是 6，但 turn_count 始终是 1”的原因。

四、代码演示详细说明

（一）代码结构分析

本文 Demo 的 Graph 非常简单：

START → chat → END

Graph 的构建代码如下，保持原样：

def build_checkpoint_graph():
"""构建带 Checkpoint 支持的对话图。"""
llm = get_llm(fallback_to_mock=False)

def chat_node(state: ChatState) -> dict:
log_step(logger, "chat_node", f"第 {state.get('turn_count', 0) + 1} 轮对话")
response = llm.invoke(state["messages"])
return {
"messages": [response],
"turn_count": state.get("turn_count", 0) + 1,
}

graph = StateGraph(ChatState)
graph.add_node("chat", chat_node)
graph.add_edge(START, "chat")
graph.add_edge("chat", END)
return graph

chat_node 做了两件事：

需要注意，build_checkpoint_graph() 返回的只是 Graph Builder。

真正让 Graph 获得状态保存能力的，是编译时传入的 Checkpointer：

checkpointer = MemorySaver()
graph = build_checkpoint_graph()
app = graph.compile(checkpointer=checkpointer)

同一份 Graph 结构，可以根据需要挂载不同后端：

Graph Builder
├── compile(checkpointer=MemorySaver)
├── compile(checkpointer=SQLite Saver)
├── compile(checkpointer=PostgreSQL Saver)
└── compile(checkpointer=Redis Saver)

Graph 负责定义执行逻辑，Checkpointer 负责保存执行状态。两者是解耦的。

（二）详细演示说明分析

1. 演示 1：MemorySaver 基本用法

第一个演示连续发起三轮对话：

conversations = [
"你好，我叫李明",
"我是一名 Python 开发者",
"你还记得我叫什么吗？",
]

所有请求都使用相同的配置：

config = {"configurable": {"thread_id": "demo-thread-1"}}

调用代码保持原样：

result = app.invoke(
{"messages": [HumanMessage(content=user_msg)], "turn_count": 0},
config=config,
)

三次调用的状态变化可以表示为：

第三次调用前，Graph 会先恢复前两次调用留下的消息：

Human: 你好，我叫李明
AI: 你好，李明……
Human: 我是一名 Python 开发者
AI: 很高兴认识你……
Human: 你还记得我叫什么吗？

模型收到的并不只是最后一个问题，而是恢复并合并后的完整消息序列，因此它能够回答“李明”。

最后通过：

snapshot = app.get_state(config)

读取最新状态：

Checkpoint 状态:
消息数量: 6
对话轮次: 1

为什么 turn_count 是 1，而不是 3？

关键在于每次调用都显式传入了：

"turn_count": 0

而 turn_count 没有 Reducer。

因此每一次 invoke 恢复旧状态之后，本次输入中的 0 又会覆盖旧值，随后 chat_node 执行：

state.get("turn_count", 0) + 1

最终重新得到 1。

执行过程实际上是：

上次 Checkpoint：turn_count = 1
↓
本次输入覆盖：turn_count = 0
↓
chat_node 加一：turn_count = 1

这个现象非常有教学价值：

Checkpoint 负责恢复状态，但状态如何合并，最终由每个字段的 Reducer 决定。

2. 演示 2：Thread ID 隔离

第二个演示创建了两个独立线程：

config_alice = {"configurable": {"thread_id": "user-alice"}}
config_bob = {"configurable": {"thread_id": "user-bob"}}

Alice 线程写入：

HumanMessage(content="我叫 Alice，我喜欢猫")

Bob 线程写入：

HumanMessage(content="我叫 Bob，我喜欢狗")

由于两次调用使用不同的 thread_id，Checkpointer 会为它们分别维护独立的状态链。

代码随后分别读取两个线程的最新状态：

alice_state = app.get_state(config_alice)
bob_state = app.get_state(config_bob)

运行结果：

Alice 消息数: 2
Bob 消息数: 2
Alice 对话包含 'Alice': True
Bob 对话包含 'Bob': True

每条线程中都只有：

1 条 HumanMessage
1 条 AIMessage

状态没有互相混入。

线上环境应该如何设计 thread_id？

通常不要直接把 thread_id 写死成：

"thread_id": "default"

更稳妥的方式是将用户标识和会话标识组合起来：

thread_id = f"{user_id}:{conversation_id}"

其中：

• user_id 用于区分用户；

• conversation_id 用于区分同一用户的多个会话。

例如：

u1001:conversation-001
u1001:conversation-002
u2048:conversation-001

这样，同一个用户可以拥有多个彼此隔离的对话。还需要注意：

thread_id 是状态隔离标识，不应直接承担权限校验。

服务端仍然需要确认当前请求用户是否有权访问这个 thread_id，否则攻击者只要猜到别人的线程标识，就可能尝试读取不属于自己的状态。

3. 演示 3：状态恢复

第三个演示先使用 app1 写入两轮对话：

app1.invoke(
{"messages": [HumanMessage(content="记住：我的密码提示是'蓝色的天空'")], "turn_count": 0},
config=config,
)
app1.invoke(
{"messages": [HumanMessage(content="我最喜欢的编程语言是 Rust")], "turn_count": 0},
config=config,
)

这时，Checkpoint 中共有 4 条消息：

Human：记住，我的密码提示是“蓝色的天空”
AI：……
Human：我最喜欢的编程语言是 Rust
AI：……

接着，代码重新编译出一个新的 app2：

app2 = graph.compile(checkpointer=checkpointer)

因为 app1 和 app2 使用的是同一个 checkpointer 实例，所以 app2 能够读取 app1 已经保存的状态。

随后提问：

result = app2.invoke(
{"messages": [HumanMessage(content="我最喜欢什么编程语言？")], "turn_count": 0},
config=config,
)

模型能够从恢复后的消息历史中找到答案：

助手：你最喜欢的是 Rust。

一个必须讲清楚的细节

原代码中的日志描述是：

log_step(logger, "阶段 2", "模拟进程重启 — 用同一个 checkpointer 创建新的 app")

这里的“模拟进程重启”不能按字面理解成真正杀掉并重启 Python 进程。

代码实际模拟的是：

重新创建编译后的 Graph 应用，但仍然复用原来的 MemorySaver 对象。

只要下面这个对象还在内存中：

checkpointer = MemorySaver()

状态就还在。

但真正结束 Python 进程后：

Python 进程退出
↓
MemorySaver 对象销毁
↓
所有内存 Checkpoint 消失

所以这个 Demo 证明的是：

只要新的 Graph 应用能够访问同一个 Checkpointer 后端，就可以恢复原有状态。

它没有证明：

MemorySaver 可以跨进程重启保存数据。

真正需要跨进程、跨机器或跨部署恢复时，必须使用外部持久化后端，例如 SQLite、PostgreSQL 或 Redis。

4. 演示 4：查看 Checkpoint 历史

代码连续调用三次 Graph：

steps = ["第一步", "第二步", "第三步"]
for step in steps:
app.invoke(
{"messages": [HumanMessage(content=f"这是{step}")], "turn_count": 0},
config=config,
)

随后遍历历史状态：

for state in app.get_state_history(config):
history_count += 1
msg_count = len(state.values.get("messages", []))
ts = state.config.get("configurable", {}).get("checkpoint_id", "N/A")
print(f" 版本 {history_count}: {msg_count} 条消息, checkpoint_id={ts[:20]}…")

输出类似：

版本 1: 6 条消息
版本 2: 5 条消息
版本 3: 4 条消息
版本 4: 4 条消息
版本 5: 3 条消息
版本 6: 2 条消息

为什么三次 invoke 会有六个以上的版本？

因为 Checkpoint 并不是“每次 invoke 只保存一次”。LangGraph 会在 super-step 边界保存状态。对于本文这个只有一个 chat 节点的线性 Graph，一次调用通常至少会经历：

输入被写入状态
↓
chat 节点执行完成

因此历史中可能同时出现：

5 条消息：新 HumanMessage 已写入，AIMessage 尚未生成
6 条消息：chat_node 已完成，AIMessage 已写入

这正好解释了历史输出中的：

6 → 5 → 4 → 4 → 3 → 2

其中重复的 4 并不表示重复保存了完全相同的执行位置，而可能分别对应：

• 上一轮执行完成后的状态；

• 下一轮输入进入之前或之后的状态边界。

历史顺序也要注意

get_state_history(config) 默认按倒序返回状态：

最新 Checkpoint
↓
更早的 Checkpoint
↓
最初的 Checkpoint

因此输出中的“版本 1”实际上是最新状态，而不是最老状态。官方时间旅行文档也明确说明，get_state_history() 返回的是逆时间顺序。

（三）从历史 Checkpoint 恢复，不等于读取缓存

Checkpoint 历史不仅可以查看，还可以用于 Replay 和 Fork。但这里有一个重要区别：

1. 查看历史状态

history = list(app.get_state_history(config))

只是读取历史 StateSnapshot，不会重新调用模型或工具。

2. 从历史状态重新执行

如果使用某个历史 Checkpoint 的 config 再次调用 Graph，Checkpoint 之前的步骤不会重新运行，但其后的节点会再次执行。

这意味着：

• LLM 可能产生不同回答；

• 外部 API 会再次请求；

• 工具调用可能产生新的副作用；

• Human-in-the-loop 中断可能再次触发。

官方文档特别提醒：Replay 会重新执行 Checkpoint 后面的节点，而不是简单地从缓存中读取旧结果。

因此，支持时间旅行的生产 Agent 必须考虑节点的幂等性。例如，下面这些操作不能在 Replay 时无条件重复：

发送付款
发送邮件
创建订单
删除数据
调用不可逆的第三方接口

更稳妥的做法是为副作用操作增加业务幂等键。

五、Checkpoint 与 Store 不要混为一谈

很多文章会笼统地把两者都叫作“Agent 记忆”，但它们解决的问题不同。

例如：

“这轮对话里，用户刚上传了一个文件”

适合放入当前线程的 State。而：

“用户长期偏好使用深色主题”

更适合放入 Store。

官方文档将 Checkpointer 定义为 thread-scoped short-term memory，将 Store 定义为跨线程的长期数据存储。

六、常见问题排查说明及线上考虑基本点说明

（一）常见问题与排查方式

1. 问题 1：忘记传入 thread_id

错误调用：

app.invoke({"messages": [HumanMessage(content="你好")]})

挂载 Checkpointer 后，每次调用都应该带上：

config = {
"configurable": {
"thread_id": "some-thread-id"
}
}

否则 Checkpointer 无法确定状态属于哪个线程。

2. 问题 2：服务重启后 MemorySaver 状态消失

这是正常行为。MemorySaver 的数据只存在于当前 Python 进程中。

适用场景：

学习 / Demo / 测试 / 本地调试

不适用场景：

生产服务 / 多进程部署 / 容器滚动发布 / 跨机器恢复

生产环境应使用外部持久化 Checkpointer。

3. 问题 3：不同用户的对话发生串台

最常见原因是所有请求使用了同一个固定值：

"thread_id": "default"

结果是所有用户都在读写同一条状态链。推荐至少包含：

user_id + conversation_id

并在服务端验证线程归属关系。

4. 问题 4：消息能够累积，普通字段却没有累积

Checkpoint 只负责恢复状态。字段究竟是追加、求和、合并还是覆盖，由 Reducer 决定。

本文中：

messages: Annotated[list[BaseMessage], add_messages]

会追加消息。而：

turn_count: int

默认会被新输入覆盖。因此，设计 State 时不能只关注字段类型，还要关注字段更新语义。

5. 问题 5：修改 State 后，旧 Checkpoint 出现兼容问题

生产系统中的 State Schema 会随着业务迭代发生变化，例如：

v1:
messages
turn_count

升级后变成：

v2:
messages
turn_count
user_profile
workflow_status

如果新代码直接假设所有历史状态都有新字段，就可能在恢复旧线程时报错。

可以采用以下策略：

• 新字段提供默认值；

• 读取时使用 state.get()；

• 在 State 中增加 schema_version；

• 编写迁移逻辑；

• 重大不兼容升级时创建新的线程命名空间。

6. 问题 6：Checkpoint 越积越多

长时间运行的 Agent 可能产生大量版本。而且要区分两件事：

裁剪模型上下文

减少发送给 LLM 的消息数量，控制 Token 成本。

清理 Checkpoint 历史

减少持久化后端中的历史状态版本，控制存储成本。只裁剪上下文，不一定会删除已经保存的历史 Checkpoint。

生产环境需要单独设计：

• 数据保留周期；

• 最近版本保留数量；

• 冷热数据分层；

• 会话归档；

• 用户删除数据流程。

（二）生产环境还需要考虑什么？

1. 持久化后端

生产环境的关键要求不是“对象能复用”，而是：

进程退出后，状态仍然存在

还要支持：

• 多实例访问；

• 失败重试；

• 连接池；

• 备份恢复；

• 数据清理；

• 权限控制；

• 监控告警。

2. 同一线程的并发调用

假设同一个 thread_id 同时收到两个请求：

请求 A：用户上传了一份合同
请求 B：用户询问上一份合同的结论

如果两个请求同时恢复同一个旧版本，再各自写回新状态，就可能产生并发冲突。

常见解决方式包括：

• 同一 thread_id 串行执行；

• 使用分布式锁；

• 使用任务队列；

• 使用乐观并发控制；

• 检测 Checkpoint 版本冲突。

3. 节点幂等性

有了 Checkpoint 之后，节点可能因为以下原因重新执行：

• 故障恢复；

• Retry；

• Replay；

• Time Travel；

• 人工修改状态后续跑。

纯计算节点通常没有问题。涉及真实副作用的节点需要格外谨慎：

支付、发信、建单、删除、发布、修改外部系统

推荐为这些操作保存业务执行标识，避免重复执行。

4. 敏感数据保护

本文为了演示状态记忆，使用了：

密码提示：蓝色的天空

这只是教学示例。生产环境不应在消息历史或 Checkpoint 中保存：

• 明文密码；

• API Key；

• Access Token；

• 私钥；

• 完整支付信息；

• 不必要的身份证明数据。

需要保存的敏感字段应进行：

• 最小化采集；

• 脱敏；

• 加密；

• 权限隔离；

• 生命周期管理；

• 审计记录。

5. 可观测性

建议在日志和链路追踪中记录：

thread_id
checkpoint_id
checkpoint_ns
执行节点
恢复来源
写入耗时
状态大小
历史版本数量

但不要直接将完整消息或敏感 State 无限制输出到日志。

七、如何读懂本次 Demo 的四个结果？

最重要的四个结论是：

八、总结

这一篇，我们用 MemorySaver 给 LangGraph Agent 接入了最基础的 Checkpoint 能力。

它带来的不只是“记住聊天记录”，而是完整的线程级状态管理：

compile(checkpointer=…)
↓
保存 Graph State
↓
thread_id 隔离线程
↓
后续 invoke 自动恢复
↓
get_state 查看最新状态
↓
get_state_history 查看历史版本

同时，也要牢牢记住它的能力边界：

MemorySaver 能让新编译的 app 从同一个内存后端恢复状态，但无法在 Python 进程真正退出后保留数据。

因此，它最适合用来学习 Checkpoint 的工作方式，而不是直接作为生产持久化方案。

下一篇进入 Redis Persistence：把 Checkpoint 从 Python 内存迁移到外部存储，实现真正的跨进程恢复，并进一步讨论连接失败时的降级策略、生产配置和状态治理。

一页速查

checkpointer = MemorySaver()

app = graph.compile(checkpointer=checkpointer)

config = {
"configurable": {
"thread_id": "user-id:conversation-id"
}
}

result = app.invoke(
{"messages": [HumanMessage(content="你好")], "turn_count": 0},
config=config,
)

latest_state = app.get_state(config)

history = app.get_state_history(config)

MemorySaver → 当前进程内保存
thread_id → 线程隔离与状态定位
get_state → 读取最新 StateSnapshot
get_state_history → 查看历史版本链
Reducer → 决定状态如何合并
持久化 Checkpointer → 实现真正的跨进程恢复

 系列导航：LangGraph从零构建生产级 AI Agent 平台的递进式学习项目-CSDN博客