LangChain Memory 完全指南：InMemorySaver、SQLite、Redis Stack 实战与避坑

LangChain Memory 完全指南：从内存到 Redis，让你的 AI 真正"记住"对话

---

📋 文章概览

解决什么问题：AI 对话没记忆？每次重启服务就失忆？

为什么值得读：从踩坑到源码，3 种方案 + 5 个避坑指南，生产环境可直接复制

你能获得什么：InMemorySaver → SQLite → Redis Stack 完整迁移路径，附 Docker 配置和代码示例

---

🎯 一句话总结

测试用 InMemorySaver，小项目用 SQLite，大项目用 Redis Stack —— 别再用内存存储撑生产环境了！

---

⚡ 开篇：一个让人崩溃的 Bug

最近做 LangChain 项目时，我遇到一个特别离谱的问题：

开发环境一切正常：

# 第一轮response=agent.invoke({"messages":[HumanMessage(content="她叫小美")]},{"configurable":{"thread_id":"1"}})# 第二轮：AI 记住了！✅response=agent.invoke({"messages":[HumanMessage(content="她叫什么？")]},{"configurable":{"thread_id":"1"}})# 输出：她叫小美 ✅

但是！服务一重启……

# 重启后再问response=agent.invoke({"messages":[HumanMessage(content="她还记得我吗？")]},{"configurable":{"thread_id":"1"}})# 输出：我不认识你... 😱

我当时就懵了：代码明明没改，配置也没变，怎么重启一下就不行了？

排查了两小时才发现：

真凶是 InMemorySaver —— 它只存内存，进程死了数据就没了！

这就好比你用记事本写日记，没点保存，电脑一关，全没了。

---

🔍 先说结论

✅ 核心结论

方案	存储位置	适用场景	推荐度
InMemorySaver	进程内存	本地测试	⭐⭐
SqliteSaver	SQLite 文件	单机小项目	⭐⭐⭐⭐
RedisSaver	Redis 服务器	分布式大项目	⭐⭐⭐⭐⭐

🎯 正确方案

# 生产环境推荐：Redis Stackfromlanggraph.checkpoint.redisimportRedisSaver checkpointer=RedisSaver.from_conn_string("redis://:password@host:6379")checkpointer.setup()# 首次需要初始化表结构agent=create_agent(model=your_model,checkpointer=checkpointer# 关键：传入 checkpointer)

⚠️ 避坑提醒

普通 Redis 不行！必须用 Redis Stack！因为需要 RediSearch 和 RedisJSON 模块。

---

💡 问题背景

为什么 AI 需要记忆？

想象一下这个场景：

第一轮对话：

👤 用户：“我喜欢一个女生，她叫小美”
🤖 AI：“哦~ 小美啊，那你可以这样表达…”

第二轮对话：

👤 用户：“那个女生叫什么名字？”
🤖 AI：（如果没记忆）“哪个女生？你没告诉我啊”

这就是没有记忆的 AI —— 每次对话都是全新的开始。

在真实项目中，这种体验简直是灾难：

• 😤客服机器人：用户说了三次问题，AI 还要问"您想咨询什么？"
• 😤智能助手：刚设置完偏好，下一句话就忘了
• 😤代码助手：前面定义的变量，后面全不认识

LangChain 的解决方案：Short-term Memory

LangChain 提供了短期记忆（Short-term Memory）机制：

在同一个线程（Thread）内，记住之前所有的对话历史。

核心组件叫Checkpointer（检查点保存器）：

fromlanggraph.checkpoint.memoryimportInMemorySaver agent=create_agent(model=model,checkpointer=InMemorySaver()# ← 就是这个东西)

工作原理：

1. ✅ 每次调用agent.invoke()时，自动保存当前状态
2. ✅ 下次调用时，根据thread_id恢复历史消息
3. ✅ AI 可以看到完整的对话上下文

我遇到的问题

我在开发环境用InMemorySaver测试得很好：

# 第一轮response=agent.invoke({"messages":[HumanMessage(content="她叫小美")]},{"configurable":{"thread_id":"1"}})# 第二轮：AI 记住了"小美"response=agent.invoke({"messages":[HumanMessage(content="她叫什么？")]},{"configurable":{"thread_id":"1"}})# 输出：她叫小美 ✅

但是！

服务一重启，或者换个终端，thread_id="1"的记忆就没了！

因为 InMemorySaver 是存内存的，进程死了数据就没了。

这就好比你用记事本写日记，没点保存，电脑一关，全没了。

---

🔬 问题排查过程

❌ 第一次怀疑：是不是 thread_id 配置错了？

怀疑原因：不同对话用了相同的thread_id，导致串台？

验证方法：

# 打印每次调用的 messagesprint(response['messages'])# 发现：确实记录了历史消息

结论：❌ 不是thread_id的问题，配置正确。

---

❌ 第二次怀疑：是不是模型不支持多轮对话？

怀疑原因：Qwen-Turbo 不支持长上下文？

验证方法：

# 手动拼接历史消息测试messages=[SystemMessage(content="你是幽默大师"),HumanMessage(content="她叫小美"),AIMessage(content="好的，我知道了"),HumanMessage(content="她叫什么？")]model.invoke(messages)# 输出：她叫小美 ✅

结论：❌ 模型支持多轮对话，不是模型的锅。

---

❌ 第三次怀疑：是不是代码逻辑有问题？

怀疑原因：create_agent的参数传错了？

验证方法：

# 检查 agent 配置print(agent.config)# 发现：checkpointer 确实是 InMemorySaver

结论：❌ 代码没问题，Checkpointer 配置正确。

---

✅ 第四次发现真相：InMemorySaver 只存内存！

关键线索：

# 重启服务后response=agent.invoke({"messages":[HumanMessage(content="她还记得我吗？")]},{"configurable":{"thread_id":"1"}})# 输出：我不认识你... 😱

恍然大悟：

InMemorySaver = 内存存储 = 进程重启 = 数据清空

这就好比：

• 你跟朋友聊天，朋友记得你说的话 ✅
• 但朋友失忆了（重启），全忘了 ❌

最终结论：

测试环境可以用 InMemorySaver，但生产环境必须用持久化存储！

---

📚 原理解析（通俗版）

7.1 什么是 Checkpointer？

人话版本：

Checkpointer 就是一个自动保存器。

每次对话结束，它自动把聊天记录存起来。
下次对话开始，它自动把记录加载出来。

技术版本：

classCheckpointer:defsave(self,thread_id,state):"""保存状态到存储"""passdefload(self,thread_id):"""从存储加载状态"""pass

7.2 三种 Checkpointer 对比

方案	存储位置	适用场景	优点	缺点
InMemorySaver	进程内存	本地测试	简单快速	重启丢失
SqliteSaver	SQLite 文件	单机小项目	无需安装数据库	不支持高并发
RedisSaver	Redis 服务器	分布式大项目	高性能、支持集群	需要 Redis Stack

7.3 数据存储结构（以 Redis 为例）

当你使用 RedisSaver 时，Redis 中会存储这些 Key：

thread_id = 1 │ ▼ checkpoint_latest ← 当前最新状态 │ ▼ checkpoint_版本号 ← 历史快照 │ ┌────┴────┐ ▼ ▼ checkpoint checkpoint_write (完整快照) (执行日志) │ ▼ write_keys_zset ← 日志索引（有序集合）

具体说明：

• checkpoint_latest：最新的对话状态（快速读取）
• checkpoint_xxx：每个版本的完整快照（支持回滚）
• checkpoint_write：写入操作的详细日志
• write_keys_zset：按时间排序的日志索引

为什么要这么设计？

1. 快速读取：直接读checkpoint_latest，不用遍历历史
2. 版本回滚：出问题时可以恢复到任意版本
3. 调试方便：可以查看完整的操作日志

7.4 工作流程图

用户发送消息 │ ▼ ┌─────────────────┐ │ 1. 加载历史 │ ← 从 Checkpointer 读取 │ (load) │ └────────┬────────┘ │ ▼ ┌─────────────────┐ │ 2. 拼接消息 │ ← 历史消息 + 新消息 │ │ └────────┬────────┘ │ ▼ ┌─────────────────┐ │ 3. 调用模型 │ ← 发送给 LLM │ │ └────────┬────────┘ │ ▼ ┌─────────────────┐ │ 4. 保存状态 │ → 写入 Checkpointer │ (save) │ └─────────────────┘

---

💻 正确解决方案

方案一：InMemorySaver（仅测试环境）⚠️

适用场景：本地开发、单元测试、Demo 演示

📦 代码示例

fromlangchain.chat_modelsimportinit_chat_modelfromlangchain.agentsimportcreate_agentfromlangchain.messagesimportSystemMessage,HumanMessagefromlanggraph.checkpoint.memoryimportInMemorySaver# 1. 初始化模型model=init_chat_model(model="qwen-turbo",model_provider="openai",base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")# 2. 创建 Agent（带内存记忆）agent=create_agent(model=model,system_prompt=SystemMessage(content="你是一个幽默大师"),checkpointer=InMemorySaver()# ⚠️ 仅用于测试)# 3. 第一轮对话config={"configurable":{"thread_id":"1"}}response=agent.invoke({"messages":[HumanMessage(content="我喜欢一个女生，她叫小美")]},config)print(response['messages'][-1].content)# 4. 第二轮对话（AI 会记住"小美"）response=agent.invoke({"messages":[HumanMessage(content="那个女生叫什么？")]},config)print(response['messages'][-1].content)# 输出：她叫小美 ✅

⚠️ 缺点提醒

• ❌ 进程重启后数据丢失
• ❌ 不支持多进程/分布式
• ❌ 不能用于生产环境

---

方案二：SqliteSaver（单机小项目）✅ 推荐

适用场景：个人项目、小型应用、单机部署

📦 前置依赖

uvaddlanggraph-checkpoint-sqlite

💻 代码示例

fromlangchain.chat_modelsimportinit_chat_modelfromlangchain.agentsimportcreate_agentfromlangchain.messagesimportSystemMessage,HumanMessagefromlanggraph.checkpoint.sqliteimportSqliteSaverimportos# 1. 初始化模型model=init_chat_model(model="qwen-turbo",model_provider="openai",base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")# 2. 创建数据目录os.makedirs("data",exist_ok=True)# 3. 使用 SQLite 存储（持久化到文件）withSqliteSaver.from_conn_string("data/checkpoints.sqlite"# 数据库文件路径)asmemory:# 4. 创建 Agentagent=create_agent(model=model,system_prompt=SystemMessage(content="你是一个幽默大师"),checkpointer=memory)# 5. 会话配置config={"configurable":{"thread_id":"1"}}# 6. 调用response=agent.invoke({"messages":[HumanMessage(content="我喜欢一个女生，她叫小美")]},config)print(response['messages'][-1].content)# 重启服务后，记忆依然存在 ✅

✅ 优点

• ✅ 数据持久化到文件
• ✅ 无需安装额外数据库
• ✅ 适合小型项目

⚠️ 注意事项

• ❌ 不支持高并发写入
• ❌ 不适合分布式部署
• ❌ 大量数据时性能下降

---

方案三：RedisSaver（生产环境）✅✅ 强烈推荐

适用场景：企业级应用、高并发场景、分布式系统

📦 前置依赖

uvaddlanggraph-checkpoint-redis

⚠️ 重要提示

必须使用 Redis Stack！普通 Redis 不行！

因为需要：

• RediSearch：用于搜索和索引
• RedisJSON：用于存储 JSON 数据

🐳 Docker Compose 配置

version:"3.8"services:redis:image:redis/redis-stack:latest# 必须用 redis-stackcontainer_name:redis7restart:alwaysports:-"6379:6379"# Redis 端口-"8001:8001"# Redis Insight 管理界面environment:REDIS_ARGS:>--requirepass 123456 # 密码 --appendonly yes # 开启 AOF 持久化 --maxmemory 512mb # 最大内存 --maxmemory-policy allkeys-lru # 内存淘汰策略volumes:-./data:/data# 数据持久化networks:default:name:redis-network

🚀 启动命令

docker-composeup-d

💻 Python 代码示例

fromlangchain.chat_modelsimportinit_chat_modelfromlangchain.agentsimportcreate_agentfromlangchain.messagesimportHumanMessagefromlanggraph.checkpoint.redisimportRedisSaver# 1. 初始化模型model=init_chat_model(model="qwen-turbo",model_provider="openai",base_url="https://dashscope.aliyuncs.com/compatible-mode/v1")# 2. 初始化 Redis Checkpointerredis_cm=RedisSaver.from_conn_string("redis://:123456@192.168.191.128:6379"# 格式：redis://:密码@主机:端口)# 3. 手动获取连接对象checkpointer=redis_cm.__enter__()# 4. 首次执行需要 setup（创建必要的索引）checkpointer.setup()# 5. 创建 Agentagent=create_agent(model=model,system_prompt="你是一个幽默大师，能够用幽默的语言回答用户的问题。",checkpointer=checkpointer)# 6. 会话配置config={"configurable":{"thread_id":"1"# 每个 user 用不同的 thread_id}}# 7. 第一轮对话response=agent.invoke({"messages":[HumanMessage(content="我喜欢一个女生，她叫小美")]},config)print(response['messages'][-1].content)# 8. 第二轮对话（即使重启服务，记忆依然存在）response=agent.invoke({"messages":[HumanMessage(content="那个女生叫什么名字？")]},config)print(response['messages'][-1].content)# 输出：她叫小美 ✅✅✅

✅ 优点

• ✅ 支持高并发（每秒万级读写）
• ✅ 数据持久化（AOF + RDB）
• ✅ 支持分布式部署
• ✅ 支持集群模式
• ✅ 适合生产环境

⚠️ 注意事项

• ❌ 需要额外的 Redis 服务
• ❌ 必须用 Redis Stack（不能是普通 Redis）

---

🚨 避坑指南

🔥 坑 1：普通 Redis 存储会报错

错误现象：

Error: Redis command 'FT.CREATE' is not available

原因：
LangChain 的 RedisSaver 内部使用了RediSearch和RedisJSON模块，普通 Redis 没有这两个模块。

正确做法：

# ❌ 错误：普通 Redisdockerrun-d--nameredis redis:latest# ✅ 正确：Redis Stackdockerrun-d--nameredis redis/redis-stack:latest

如何验证是否安装成功：

# 进入 Redis 容器dockerexec-itredis7 redis-cli-a123456# 检查模块MODULE LIST# 应该输出：# 1) 1) "name"# 2) "ReJSON"# 3) "ver"# 4) "99999"# 2) 1) "name"# 2) "search"# 3) "ver"# 4) "99999"

---

🔥 坑 2：忘记调用 setup() 方法

错误现象：

Error: Index not found

原因：
首次使用 RedisSaver 时，需要调用setup()方法来创建必要的索引结构。

正确做法：

checkpointer=RedisSaver.from_conn_string("redis://:password@host:6379")checkpointer=checkpointer.__enter__()checkpointer.setup()# ⚠️ 这一步不能少！

注意：

• setup()只需要执行一次
• 后续启动不需要再次调用（除非清空了 Redis 数据）

---

🔥 坑 3：thread_id 混用导致串台

错误现象：

用户 A 说"我叫张三"，用户 B 问"我叫什么"，AI 回答"你叫张三"

原因：
多个用户使用了相同的thread_id。

正确做法：

# ❌ 错误：所有用户共用一个 thread_idconfig={"configurable":{"thread_id":"1"}}# ✅ 正确：每个用户使用唯一的 thread_idimportuuid user_id=get_current_user_id()# 从 session/token 获取config={"configurable":{"thread_id":str(user_id)}}

最佳实践：

defget_user_config(user_id):return{"configurable":{"thread_id":f"user_{user_id}",# 加前缀避免冲突"user_id":user_id}}

---

🔥 坑 4：消息过长导致上下文溢出

错误现象：

Error: This model's maximum context length is XXX tokens

原因：
长时间对话导致消息历史超过模型的上下文窗口限制。

解决方案（三种）：

方案 A：裁剪消息（Trim Messages）

fromlangchain.agents.middlewareimportbefore_modelfromlangchain.messagesimportRemoveMessagefromlanggraph.graph.messageimportREMOVE_ALL_MESSAGES@before_modeldeftrim_messages(state,runtime):"""只保留最近的 N 条消息"""messages=state["messages"]iflen(messages)<=10:returnNone# 不需要裁剪# 保留第一条系统消息 + 最近 9 条first_msg=messages[0]recent_messages=messages[-9:]return{"messages":[RemoveMessage(id=REMOVE_ALL_MESSAGES),# 清空所有first_msg,*recent_messages]}agent=create_agent(model=model,middleware=[trim_messages],# 注册中间件checkpointer=checkpointer)

方案 B：删除旧消息（Delete Messages）

@after_modeldefdelete_old_messages(state,runtime):"""删除最早的消息"""messages=state["messages"]iflen(messages)>20:# 删除最早的 10 条return{"messages":[RemoveMessage(id=m.id)forminmessages[:10]]}returnNone

方案 C：消息摘要（SummarizationMiddleware）✅ 推荐

fromlangchain.agents.middlewareimportSummarizationMiddleware agent=create_agent(model="gpt-4",middleware=[SummarizationMiddleware(model="gpt-4-mini",# 用于摘要的小模型trigger=("tokens",4000),# 超过 4000 token 时触发keep=("messages",20)# 至少保留 20 条消息)],checkpointer=checkpointer)

工作原理：

1. 当消息超过阈值时，自动调用小模型生成摘要
2. 用摘要替换早期的详细消息
3. 既保留关键信息，又控制 token 数量

---

🔥 坑 5：SQLite 并发写入锁冲突

错误现象：

Error: database is locked

原因：
SQLite 同一时间只允许一个写操作，高并发时会锁表。

解决方案：

# 方案 1：降低并发（简单粗暴）fromqueueimportQueueimportthreading task_queue=Queue()lock=threading.Lock()defsafe_invoke(agent,message,config):withlock:returnagent.invoke(message,config)# 方案 2：改用 PostgreSQL 或 Redis（推荐）fromlanggraph.checkpoint.postgresimportPostgresSaver checkpointer=PostgresSaver.from_conn_string("postgresql://user:pass@localhost:5432/langchain")

---

📊 最后总结

🎯 核心原因

AI 失忆不是因为模型笨，而是因为你没给它配"笔记本"（Checkpointer）。

✅ 正确方案

测试用 InMemorySaver，小项目用 SQLite，生产环境用 Redis Stack。

⚠️ 避坑提醒

普通 Redis 存不了 LangChain 记忆，必须用 Redis Stack！这个坑我替你踩过了。

💡 一句话经验

很多时候 Bug 不是代码写错了，而是选错了工具。

就像你不能用记事本去管理企业级数据库一样，

你也不能用 InMemorySaver 去支撑生产环境。

---

📋 附录：快速选型指南

你的场景	推荐方案	理由
本地学习/Demo	InMemorySaver	零配置，开箱即用
个人项目/工具	SqliteSaver	无需装数据库，文件即存储
企业应用/SAAS	RedisSaver (Stack)	高性能、支持集群
已有 PostgreSQL	PostgresSaver	复用现有基础设施
需要审计日志	PostgreSQL + Redis	PG 存日志，Redis 存热数据

---

🔗 参考资源

• 官方文档：https://docs.langchain.com/oss/python/langchain/short-term-memory
• Checkpoint 列表：https://docs.langchain.com/oss/python/integrations/checkpointers/index
• Redis Stack：https://redis.io/docs/stack/

---

如果这篇文章帮到了你，欢迎点赞、收藏、转发！🎉

有问题评论区见，我看到都会回复~ 💬

🙏 作者介绍

📌写文不易，Bug 更不易。

如果这篇文章对你有帮助，可以搜一搜：空门技术栈

这里分享：

• ✅ Java / Spring AI / 企业级项目实战
• ✅ Docker / RAG知识库 / 微服务踩坑
• ✅ Python、前端、AI应用落地
• ✅ 偶尔分享一些「头发保卫战」经验 😆

---

一个热爱技术、持续填坑的开发者，
陪你一起少踩坑，少加班，多写优雅代码。

📖 推荐阅读

• 看了 3 天官方文档后，我决定自己写一篇 LangChain 人话教程

• GPT-5.5 变强、Spring AI 更新、Ollama 爆漏洞｜今天值得看的技术热点

• 还在复制粘贴 if-else？模板方法模式，专治重复代码！

• CSDN：LangChain 入门实战指南