LangChain 短期记忆实战：原理、步骤与两种持久化器

七崽爱吃小饼干2026/10/18阅读 0专栏 LangChain 开发实战

上一篇（demo-17）从概念上讲了记忆是什么、怎么分类、LangChain 如何管理。这一篇进入实操：用 LangChain 实现短期记忆。我们会先讲原理和步骤，再分别用「基于内存的持久化器」和「基于外部存储介质的持久化器」跑通同一段对话。

代码基于 JavaScript 版 langchain v1，模型用 DeepSeek。

一、短期记忆要解决什么

短期记忆（short-term memory）只服务同一段会话：让多轮对话能接得上。

如果没有它，每次 invoke 都是独立的：

await agent.invoke({ messages: [{ role: "user", content: "我叫小明，住在杭州。" }] });
await agent.invoke({ messages: [{ role: "user", content: "我叫什么？" }] });
// 第二次：我不知道你的名字……

因为第二次调用时，第一次的消息根本没出现在输入里。要让历史「自动接上」，就轮到 checkpointer 出场了。

二、实现原理：三块拼图

LangChain 的 Agent 建立在 LangGraph 之上，短期记忆由三部分组成：

2.1 Agent State：工作记忆

Agent 运行时有一份 state，最核心的字段就是 messages（对话历史）。模型每一步看到的内容，都来自这份 state。

2.2 Checkpointer：把状态存成检查点

checkpointer 会在 Agent 每执行完一步（一个 superstep）后，把当前 state 存成一个 checkpoint。下次调用时，它先按 thread 把最近的 checkpoint 读回来、恢复成 state，再继续跑。

你可以把它理解成「给 Agent 的 state 自动做存档 + 读档」。

2.3 thread_id：区分会话

checkpointer 用 thread_id 给存档分组：

同一个 thread_id = 同一段会话，历史持续累积；
换一个 thread_id = 另一段独立会话，互不干扰。

一次调用里，checkpoint 的定位通常是 thread_id + checkpoint_ns + checkpoint_id 三者共同决定（checkpoint_ns 用于子图，普通场景为空）。

2.4 整体流程

invoke(thread_id=A)
      ↓
从 checkpointer 读取 thread A 最近的 checkpoint，恢复 state
      ↓
把新消息并入 messages，跑 Agent（模型 / 工具…）
      ↓
每一步结束，checkpointer 再写入新的 checkpoint

关键点：你不需要自己维护 messages 数组，也不需要每次把历史全量传进去。 checkpointer 帮你做了。

三、实现步骤

只有三步：

创建一个 checkpointer；
createAgent 时把它传给 checkpointer；
每次 invoke 用 config.configurable.thread_id 指定会话。

import { createAgent } from "langchain";
import { MemorySaver } from "@langchain/langgraph";

const agent = createAgent({
  model,
  tools: [],
  systemPrompt: "你是简洁的中文助手。",
  checkpointer: new MemorySaver(), // 1) 挂上 checkpointer
});

const config = { configurable: { thread_id: "demo-user-1" } }; // 3) 会话 ID

const r1 = await agent.invoke(
  { messages: [{ role: "user", content: "我叫小明，住在杭州。" }] },
  config
);
console.log(r1.messages.length); // 2

const r2 = await agent.invoke(
  { messages: [{ role: "user", content: "我叫什么？住在哪？" }] }, // 只带新问题
  config
);
console.log(r2.messages.at(-1).content); // 你叫小明，住在杭州。
console.log(r2.messages.length); // 4 —— 历史被自动接上了

注意第 2 次 invoke 只发了新问题，没有手动回传第 1 轮的消息，但模型依然答得出来——这正是 checkpointer 的功劳。

线程隔离

同一个 checkpointer 可以服务多个 thread：

await agent.invoke({ messages: [{ role: "user", content: "我叫小李。" }] },
  { configurable: { thread_id: "thread-A" } });

await agent.invoke({ messages: [{ role: "user", content: "我叫什么？" }] },
  { configurable: { thread_id: "thread-B" } });
// thread-B：我不知道你的名字（看不到 thread-A 的历史）

await agent.invoke({ messages: [{ role: "user", content: "我叫什么？" }] },
  { configurable: { thread_id: "thread-A" } });
// thread-A：你叫小李

四、基于内存的持久化器：MemorySaver

MemorySaver 把 checkpoint 存在当前进程的内存里。它开箱即用，适合开发、测试、单进程 demo：

import { MemorySaver } from "@langchain/langgraph";

const agent = createAgent({ model, tools: [], checkpointer: new MemorySaver() });

它的局限也很明确：只到进程为止。换一个新实例（相当于重启进程），记忆就没了。

// 实例 A：写入
const agentA = createAgent({ model, tools: [], checkpointer: new MemorySaver() });
await agentA.invoke(
  { messages: [{ role: "user", content: "我叫小刚，是一名医生。" }] },
  { configurable: { thread_id: "mem-restart" } }
);

// 实例 B：全新的 MemorySaver，相当于重启后的新进程
const agentB = createAgent({ model, tools: [], checkpointer: new MemorySaver() });
const r = await agentB.invoke(
  { messages: [{ role: "user", content: "我叫什么？做什么工作？" }] },
  { configurable: { thread_id: "mem-restart" } }
);
// 我不知道你的名字和职业……

虽然 thread_id 一样，但两个 MemorySaver 是两份互不相干的内存，所以读不到。这就是「内存持久化器」的边界。

五、基于外部存储介质的持久化器：SqliteSaver

把 checkpoint 写到外部存储（文件 / 数据库），就能跨进程、跨重启保留记忆。LangChain 生态提供了 SQLite 版：

bash

npm install @langchain/langgraph-checkpoint-sqlite

用法只比 MemorySaver 多一个「从路径创建」：

import { SqliteSaver } from "@langchain/langgraph-checkpoint-sqlite";

const saver = SqliteSaver.fromConnString("./memory.db"); // 文件不存在会自动创建
const agent = createAgent({ model, tools: [], checkpointer: saver });

fromConnString 会自动建表、准备读写。写入一段对话后，可以直接查数据库：

saver.db.prepare("SELECT name FROM sqlite_master WHERE type='table'").all();
// [{ name: "checkpoints" }, { name: "writes" }]

saver.db.prepare("SELECT COUNT(*) AS c FROM checkpoints").get();
// { c: 3 }

checkpoints 存每个检查点（state 快照），writes 存中间写入。

验证「跨重启」

用两个独立的 SqliteSaver 实例、指向同一个文件，模拟「重启后的新进程」：

// 实例 A：写入磁盘
const saverA = SqliteSaver.fromConnString("./memory.db");
const agentA = createAgent({ model, tools: [], checkpointer: saverA });
await agentA.invoke(
  { messages: [{ role: "user", content: "我叫小红，是一名设计师。" }] },
  { configurable: { thread_id: "sqlite-user-1" } }
);
saverA.db.close();

// 实例 B：全新实例，同一个文件 —— 相当于重启后的新进程
const saverB = SqliteSaver.fromConnString("./memory.db");
const agentB = createAgent({ model, tools: [], checkpointer: saverB });
const r = await agentB.invoke(
  { messages: [{ role: "user", content: "我叫什么？做什么工作？" }] },
  { configurable: { thread_id: "sqlite-user-1" } }
);
// 你叫小红，是一名设计师。

saver B 和 saver A 没有任何内存共享，却能读到第 1 轮——因为数据在 SQLite 文件里。把 saver B 换成另一个进程，效果一样。

生产环境的同类选择还有 PostgresSaver、Redis 等数据库型 checkpointer；思路完全一致：把 checkpoint 落到可靠的外部存储。

六、两种持久化器对比

	`MemorySaver`	`SqliteSaver`
存储位置	当前进程内存	SQLite 文件（外部介质）
跨进程 / 重启	否	是
多实例共享	否	是（同一文件/库）
适用场景	开发、测试、单进程 demo	单机持久化、轻量生产
生产替换	—	PostgresSaver / Redis 等

七、实践要点

会话隔离靠 thread_id：同一个 ID 才共享历史；
checkpointer 会保存整个 state：所以 state 里的东西都会落盘，不要把敏感信息随便塞进 state；
生产用数据库型 checkpointer，并注意清理/归档旧 thread；
历史过长要控制上下文：配合 summarization / contextEditing 等中间件（见 demo-15、demo-16）；
MemorySaver 适合开发，不要拿它当生产存储。

八、Python 对应写法

Python 侧参数一致，只是类名和 import 不同：

python

from langchain.agents import create_agent
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model="deepseek:deepseek-chat",
    tools=[],
    checkpointer=InMemorySaver(),   # 基于内存
)

config = {"configurable": {"thread_id": "demo-user-1"}}
agent.invoke({"messages": [{"role": "user", "content": "我叫小明，住在杭州。"}]}, config)
result = agent.invoke({"messages": [{"role": "user", "content": "我叫什么？"}]}, config)
print(result["messages"][-1].content)

外部存储（SQLite）：

python

import sqlite3
from langgraph.checkpoint.sqlite import SqliteSaver

conn = sqlite3.connect("memory.db", check_same_thread=False)
checkpointer = SqliteSaver(conn)

agent = create_agent(model="deepseek:deepseek-chat", tools=[], checkpointer=checkpointer)

Postgres 对应 from langgraph.checkpoint.postgres import PostgresSaver。

九、小结

原理：Agent state 存对话；checkpointer 每步存检查点；thread_id 区分会话。
步骤：创建 checkpointer → 传给 createAgent → invoke 时带同一个 thread_id。
内存持久化器：MemorySaver，进程内有效，重启丢失，适合开发测试。
外部持久化器：SqliteSaver（SQLite 文件），跨进程/重启保留，生产可换 PostgresSaver / Redis。

一句话：短期记忆 = Agent state + checkpointer + thread_id；想让它活过进程重启，就把 checkpointer 从内存换成外部存储。

上一篇讲长期记忆的 store，后续还会继续展开长期记忆的具体读写与检索。