Agent with Memory：给 Agent 装上持久化记忆
Checkpointer · MemorySaver · Thread ID

1 问题：无状态 Agent 无法进行多轮对话

上一讲（1-4）我们构建了一个完整的 ReAct Agent：它能循环调用工具，直到问题解决。但它有一个致命缺陷——每次 invoke() 都是全新的对话，不记得之前说过任何内容。

为什么会这样？State 的生命周期

LangGraph 中的 State 是每次 invoke() 临时创建的对象。执行流程如下：

本质上，这是因为 LangGraph 的图执行是纯函数式的——输入进来，输出出去，不保留任何副作用。这在大多数情况下是好事（可预测、可测试），但多轮对话场景需要状态持久化。

2 解决方案：Checkpointer 检查点机制

Checkpointer（检查点器）是 LangGraph 的持久化层。它的职责是：在图的每个节点执行完毕后，自动把当前 State 序列化并保存到存储后端。

Checkpointer 做了什么（三件事）

MemorySaver：最简单的 Checkpointer 实现

from langgraph.checkpoint.memory import MemorySaver

# MemorySaver 把所有 checkpoint 存在 Python 进程内存里
# 本质上是一个字典：{thread_id: {step: state_snapshot}}
memory = MemorySaver()

# 编译时把 checkpointer 传进去
react_graph_memory = builder.compile(checkpointer=memory)

MemorySaver 使用内存中的字典存储所有检查点，零配置、开箱即用，适合开发阶段和单元测试。缺点是进程重启后数据全部丢失。

3 Thread ID：区分不同的对话会话

一个 Agent 可能同时服务多个用户，每个用户的对话历史必须隔离。Thread ID 就是"对话会话 ID"，相当于数据库里的主键。

三个线程的对话历史完全独立存储、互不干扰。Alice 的 "that" 是 7，Bob 的 "that" 是 25，Carol 的 "that" 是 42。

import uuid

# 方式一：固定字符串（测试/demo 时常用）
config = {"configurable": {"thread_id": "1"}}

# 方式二：UUID（生产环境，每个用户 session 一个）
thread_id = str(uuid.uuid4())
config = {"configurable": {"thread_id": thread_id}}

# 方式三：与业务 ID 绑定（最常见的实践）
config = {"configurable": {"thread_id": f"user_{user_id}_session_{session_id}"}}

4 核心代码：只差一个参数

给 Agent 加上记忆，代码改动极其微小——只需要在 compile() 时传入 checkpointer，其余所有代码保持不变。

from langgraph.graph import (
    StateGraph, MessagesState,
    START, END
)
from langgraph.prebuilt import (
    ToolNode, tools_condition
)

# 无 checkpointer
react_graph = builder.compile()

# 每次 invoke 都是全新对话
messages = react_graph.invoke(
    {"messages": messages}
    # 没有 config，无法区分对话
)

from langgraph.graph import (
    StateGraph, MessagesState,
    START, END
)
from langgraph.prebuilt import (
    ToolNode, tools_condition
)
from langgraph.checkpoint.memory import MemorySaver

memory = MemorySaver()

# 传入 checkpointer ← 唯一改动！
react_graph_memory = builder.compile(
    checkpointer=memory
)

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

# 必须传 config，带上 thread_id
messages = react_graph_memory.invoke(
    {"messages": messages},
    config   # ← 告诉图用哪个 thread
)

完整代码：两轮对话示例

from langchain_core.messages import HumanMessage
from langgraph.checkpoint.memory import MemorySaver

# ─── 初始化 Checkpointer ───
memory = MemorySaver()
react_graph_memory = builder.compile(checkpointer=memory)

# ─── 配置：使用 thread_id = "1" ───
config = {"configurable": {"thread_id": "1"}}

# ─── 第一轮对话 ───
messages = [HumanMessage(content="Add 3 and 4.")]
result = react_graph_memory.invoke({"messages": messages}, config)
# LLM 调用 add(3, 4) → ToolMessage("7") → AIMessage("The sum is 7.")
# ✅ Checkpointer 把这整个 State 保存到 thread "1" 的 checkpoint

# ─── 第二轮对话：注意只传了新的问题 ───
messages = [HumanMessage(content="Multiply that by 2.")]
result = react_graph_memory.invoke({"messages": messages}, config)
# ✅ LangGraph 先加载 thread "1" 的历史，再追加新消息
# LLM 看到完整历史，正确理解 "that" = 7
# → 调用 multiply(7, 2) → ToolMessage("14") → AIMessage("The result is 14.")

print(result["messages"][-1].content)
# 输出："The result is 14."

5 两次 invoke 如何"连接"——完整生命周期

要真正理解记忆机制，需要知道每次 invoke() 在内部经历了什么步骤。

6 对比演示：有记忆 vs 无记忆

下面直观展示第二轮 invoke() 时，LLM 实际收到的 messages 列表有何不同。

第二轮执行后的完整消息列表（有记忆版）

两轮对话执行完毕后，thread "1" 的 State 中 messages 共有 7 条：

# thread_id="1" 的最终 messages 列表：
[
    HumanMessage(content="Add 3 and 4."),
    AIMessage(content="", tool_calls=[{"name": "add", "args": {"a": 3, "b": 4}}]),
    ToolMessage(content="7", name="add"),
    AIMessage(content="The sum of 3 and 4 is 7."),
    HumanMessage(content="Multiply that by 2."),
    AIMessage(content="", tool_calls=[{"name": "multiply", "args": {"a": 7, "b": 2}}]),
    ToolMessage(content="14", name="multiply"),
    AIMessage(content="The result of multiplying 7 by 2 is 14."),
]

7 Checkpointer 工作流程图

下图展示了带 Checkpointer 的图在单次 invoke() 内部如何在每个节点后自动保存 State。

8 MemorySaver vs 生产级存储

MemorySaver 只适合开发阶段。生产环境需要能持久化到磁盘或外部数据库的 Checkpointer 实现。

所有 Checkpointer 实现都遵循相同的接口（BaseCheckpointSaver），因此从 MemorySaver 切换到 PostgresSaver 时，业务代码零改动，只需替换编译时传入的 checkpointer 对象。

# 开发阶段
from langgraph.checkpoint.memory import MemorySaver
graph = builder.compile(checkpointer=MemorySaver())

# 生产阶段（替换一行，其余代码不变）
from langgraph.checkpoint.postgres import PostgresSaver
with PostgresSaver.from_conn_string("postgresql://user:pass@host/db") as checkpointer:
    graph = builder.compile(checkpointer=checkpointer)
    # 同样的 invoke()，同样的 config，完全一致的行为

9 Config 参数的必要性

即使你传了 checkpointer，如果 invoke() 时没有提供 config（或没有 thread_id），LangGraph 会报错。这是设计上的强制约束。

# ❌ 错误：有 checkpointer 但没有 config
react_graph_memory.invoke({"messages": messages})
# 抛出 ValueError: "thread_id" is required in config["configurable"]
# when a checkpointer is used

# ✅ 正确：提供 config 和 thread_id
config = {"configurable": {"thread_id": "1"}}
react_graph_memory.invoke({"messages": messages}, config)

为什么是强制的？因为没有 thread_id，Checkpointer 根本不知道：

• 该从哪里加载历史（多个用户的数据混在一起，用哪个？）
• 该把新的 checkpoint 保存到哪里（保存到哪个对话会话？）

查询历史 State

有了 checkpointer，你还可以随时查询某个 thread 的历史 State，无需重新执行图：

# 获取某个 thread 的当前（最新）State 快照
snapshot = react_graph_memory.get_state(config)
print(snapshot.values["messages"])   # 查看完整消息历史
print(snapshot.next)               # 查看下一步将执行哪个节点

# 获取某个 thread 的全部历史 checkpoints（时间旅行用）
for checkpoint in react_graph_memory.get_state_history(config):
    print(checkpoint.config)          # 该 checkpoint 的 config（含 checkpoint_id）
    print(checkpoint.values["messages"])  # 该时刻的消息列表

10 总结：这一讲学到了什么

这一讲在 1-4 的完整 ReAct Agent 基础上，用最小的代码改动实现了持久化多轮对话记忆。