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

LangChain Memory 完全指南从内存到 Redis让你的 AI 真正记住对话 文章概览解决什么问题AI 对话没记忆每次重启服务就失忆为什么值得读从踩坑到源码3 种方案 5 个避坑指南生产环境可直接复制你能获得什么InMemorySaver → SQLite → Redis Stack 完整迁移路径附 Docker 配置和代码示例 一句话总结测试用 InMemorySaver小项目用 SQLite大项目用 Redis Stack —— 别再用内存存储撑生产环境了⚡ 开篇一个让人崩溃的 Bug最近做 LangChain 项目时我遇到一个特别离谱的问题开发环境一切正常# 第一轮responseagent.invoke({messages:[HumanMessage(content她叫小美)]},{configurable:{thread_id:1}})# 第二轮AI 记住了✅responseagent.invoke({messages:[HumanMessage(content她叫什么)]},{configurable:{thread_id:1}})# 输出她叫小美 ✅但是服务一重启……# 重启后再问responseagent.invoke({messages:[HumanMessage(content她还记得我吗)]},{configurable:{thread_id:1}})# 输出我不认识你... 我当时就懵了代码明明没改配置也没变怎么重启一下就不行了排查了两小时才发现真凶是 InMemorySaver —— 它只存内存进程死了数据就没了这就好比你用记事本写日记没点保存电脑一关全没了。 先说结论✅ 核心结论方案存储位置适用场景推荐度InMemorySaver进程内存本地测试⭐⭐SqliteSaverSQLite 文件单机小项目⭐⭐⭐⭐RedisSaverRedis 服务器分布式大项目⭐⭐⭐⭐⭐ 正确方案# 生产环境推荐Redis Stackfromlanggraph.checkpoint.redisimportRedisSaver checkpointerRedisSaver.from_conn_string(redis://:passwordhost:6379)checkpointer.setup()# 首次需要初始化表结构agentcreate_agent(modelyour_model,checkpointercheckpointer# 关键传入 checkpointer)⚠️ 避坑提醒普通 Redis 不行必须用 Redis Stack因为需要 RediSearch 和 RedisJSON 模块。 问题背景为什么 AI 需要记忆想象一下这个场景第一轮对话 用户“我喜欢一个女生她叫小美” AI“哦~ 小美啊那你可以这样表达…”第二轮对话 用户“那个女生叫什么名字” AI如果没记忆“哪个女生你没告诉我啊”这就是没有记忆的 AI —— 每次对话都是全新的开始。在真实项目中这种体验简直是灾难客服机器人用户说了三次问题AI 还要问您想咨询什么智能助手刚设置完偏好下一句话就忘了代码助手前面定义的变量后面全不认识LangChain 的解决方案Short-term MemoryLangChain 提供了短期记忆Short-term Memory机制在同一个线程Thread内记住之前所有的对话历史。核心组件叫Checkpointer检查点保存器fromlanggraph.checkpoint.memoryimportInMemorySaver agentcreate_agent(modelmodel,checkpointerInMemorySaver()# ← 就是这个东西)工作原理✅ 每次调用agent.invoke()时自动保存当前状态✅ 下次调用时根据thread_id恢复历史消息✅ AI 可以看到完整的对话上下文我遇到的问题我在开发环境用InMemorySaver测试得很好# 第一轮responseagent.invoke({messages:[HumanMessage(content她叫小美)]},{configurable:{thread_id:1}})# 第二轮AI 记住了小美responseagent.invoke({messages:[HumanMessage(content她叫什么)]},{configurable:{thread_id:1}})# 输出她叫小美 ✅但是服务一重启或者换个终端thread_id1的记忆就没了因为 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 只存内存关键线索# 重启服务后responseagent.invoke({messages:[HumanMessage(content她还记得我吗)]},{configurable:{thread_id:1}})# 输出我不认识你... 恍然大悟InMemorySaver 内存存储 进程重启 数据清空这就好比你跟朋友聊天朋友记得你说的话 ✅但朋友失忆了重启全忘了 ❌最终结论测试环境可以用 InMemorySaver但生产环境必须用持久化存储 原理解析通俗版7.1 什么是 Checkpointer人话版本Checkpointer 就是一个自动保存器。每次对话结束它自动把聊天记录存起来。下次对话开始它自动把记录加载出来。技术版本classCheckpointer:defsave(self,thread_id,state):保存状态到存储passdefload(self,thread_id):从存储加载状态pass7.2 三种 Checkpointer 对比方案存储位置适用场景优点缺点InMemorySaver进程内存本地测试简单快速重启丢失SqliteSaverSQLite 文件单机小项目无需安装数据库不支持高并发RedisSaverRedis 服务器分布式大项目高性能、支持集群需要 Redis Stack7.3 数据存储结构以 Redis 为例当你使用 RedisSaver 时Redis 中会存储这些 Keythread_id 1 │ ▼ checkpoint_latest ← 当前最新状态 │ ▼ checkpoint_版本号 ← 历史快照 │ ┌────┴────┐ ▼ ▼ checkpoint checkpoint_write (完整快照) (执行日志) │ ▼ write_keys_zset ← 日志索引有序集合具体说明checkpoint_latest最新的对话状态快速读取checkpoint_xxx每个版本的完整快照支持回滚checkpoint_write写入操作的详细日志write_keys_zset按时间排序的日志索引为什么要这么设计快速读取直接读checkpoint_latest不用遍历历史版本回滚出问题时可以恢复到任意版本调试方便可以查看完整的操作日志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. 初始化模型modelinit_chat_model(modelqwen-turbo,model_provideropenai,base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1)# 2. 创建 Agent带内存记忆agentcreate_agent(modelmodel,system_promptSystemMessage(content你是一个幽默大师),checkpointerInMemorySaver()# ⚠️ 仅用于测试)# 3. 第一轮对话config{configurable:{thread_id:1}}responseagent.invoke({messages:[HumanMessage(content我喜欢一个女生她叫小美)]},config)print(response[messages][-1].content)# 4. 第二轮对话AI 会记住小美responseagent.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. 初始化模型modelinit_chat_model(modelqwen-turbo,model_provideropenai,base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1)# 2. 创建数据目录os.makedirs(data,exist_okTrue)# 3. 使用 SQLite 存储持久化到文件withSqliteSaver.from_conn_string(data/checkpoints.sqlite# 数据库文件路径)asmemory:# 4. 创建 Agentagentcreate_agent(modelmodel,system_promptSystemMessage(content你是一个幽默大师),checkpointermemory)# 5. 会话配置config{configurable:{thread_id:1}}# 6. 调用responseagent.invoke({messages:[HumanMessage(content我喜欢一个女生她叫小美)]},config)print(response[messages][-1].content)# 重启服务后记忆依然存在 ✅✅ 优点✅ 数据持久化到文件✅ 无需安装额外数据库✅ 适合小型项目⚠️ 注意事项❌ 不支持高并发写入❌ 不适合分布式部署❌ 大量数据时性能下降方案三RedisSaver生产环境✅✅ 强烈推荐适用场景企业级应用、高并发场景、分布式系统 前置依赖uvaddlanggraph-checkpoint-redis⚠️ 重要提示必须使用 Redis Stack普通 Redis 不行因为需要RediSearch用于搜索和索引RedisJSON用于存储 JSON 数据 Docker Compose 配置version:3.8services: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. 初始化模型modelinit_chat_model(modelqwen-turbo,model_provideropenai,base_urlhttps://dashscope.aliyuncs.com/compatible-mode/v1)# 2. 初始化 Redis Checkpointerredis_cmRedisSaver.from_conn_string(redis://:123456192.168.191.128:6379# 格式redis://:密码主机:端口)# 3. 手动获取连接对象checkpointerredis_cm.__enter__()# 4. 首次执行需要 setup创建必要的索引checkpointer.setup()# 5. 创建 Agentagentcreate_agent(modelmodel,system_prompt你是一个幽默大师能够用幽默的语言回答用户的问题。,checkpointercheckpointer)# 6. 会话配置config{configurable:{thread_id:1# 每个 user 用不同的 thread_id}}# 7. 第一轮对话responseagent.invoke({messages:[HumanMessage(content我喜欢一个女生她叫小美)]},config)print(response[messages][-1].content)# 8. 第二轮对话即使重启服务记忆依然存在responseagent.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()方法来创建必要的索引结构。正确做法checkpointerRedisSaver.from_conn_string(redis://:passwordhost:6379)checkpointercheckpointer.__enter__()checkpointer.setup()# ⚠️ 这一步不能少注意setup()只需要执行一次后续启动不需要再次调用除非清空了 Redis 数据 坑 3thread_id 混用导致串台错误现象用户 A 说我叫张三用户 B 问我叫什么AI 回答你叫张三原因多个用户使用了相同的thread_id。正确做法# ❌ 错误所有用户共用一个 thread_idconfig{configurable:{thread_id:1}}# ✅ 正确每个用户使用唯一的 thread_idimportuuid user_idget_current_user_id()# 从 session/token 获取config{configurable:{thread_id:str(user_id)}}最佳实践defget_user_config(user_id):return{configurable:{thread_id:fuser_{user_id},# 加前缀避免冲突user_id:user_id}} 坑 4消息过长导致上下文溢出错误现象Error: This models maximum context length is XXX tokens原因长时间对话导致消息历史超过模型的上下文窗口限制。解决方案三种方案 A裁剪消息Trim Messagesfromlangchain.agents.middlewareimportbefore_modelfromlangchain.messagesimportRemoveMessagefromlanggraph.graph.messageimportREMOVE_ALL_MESSAGESbefore_modeldeftrim_messages(state,runtime):只保留最近的 N 条消息messagesstate[messages]iflen(messages)10:returnNone# 不需要裁剪# 保留第一条系统消息 最近 9 条first_msgmessages[0]recent_messagesmessages[-9:]return{messages:[RemoveMessage(idREMOVE_ALL_MESSAGES),# 清空所有first_msg,*recent_messages]}agentcreate_agent(modelmodel,middleware[trim_messages],# 注册中间件checkpointercheckpointer)方案 B删除旧消息Delete Messagesafter_modeldefdelete_old_messages(state,runtime):删除最早的消息messagesstate[messages]iflen(messages)20:# 删除最早的 10 条return{messages:[RemoveMessage(idm.id)forminmessages[:10]]}returnNone方案 C消息摘要SummarizationMiddleware✅ 推荐fromlangchain.agents.middlewareimportSummarizationMiddleware agentcreate_agent(modelgpt-4,middleware[SummarizationMiddleware(modelgpt-4-mini,# 用于摘要的小模型trigger(tokens,4000),# 超过 4000 token 时触发keep(messages,20)# 至少保留 20 条消息)],checkpointercheckpointer)工作原理当消息超过阈值时自动调用小模型生成摘要用摘要替换早期的详细消息既保留关键信息又控制 token 数量 坑 5SQLite 并发写入锁冲突错误现象Error: database is locked原因SQLite 同一时间只允许一个写操作高并发时会锁表。解决方案# 方案 1降低并发简单粗暴fromqueueimportQueueimportthreading task_queueQueue()lockthreading.Lock()defsafe_invoke(agent,message,config):withlock:returnagent.invoke(message,config)# 方案 2改用 PostgreSQL 或 Redis推荐fromlanggraph.checkpoint.postgresimportPostgresSaver checkpointerPostgresSaver.from_conn_string(postgresql://user:passlocalhost:5432/langchain) 最后总结 核心原因AI 失忆不是因为模型笨而是因为你没给它配笔记本Checkpointer。✅ 正确方案测试用 InMemorySaver小项目用 SQLite生产环境用 Redis Stack。⚠️ 避坑提醒普通 Redis 存不了 LangChain 记忆必须用 Redis Stack这个坑我替你踩过了。 一句话经验很多时候 Bug 不是代码写错了而是选错了工具。就像你不能用记事本去管理企业级数据库一样你也不能用 InMemorySaver 去支撑生产环境。 附录快速选型指南你的场景推荐方案理由本地学习/DemoInMemorySaver零配置开箱即用个人项目/工具SqliteSaver无需装数据库文件即存储企业应用/SAASRedisSaver (Stack)高性能、支持集群已有 PostgreSQLPostgresSaver复用现有基础设施需要审计日志PostgreSQL RedisPG 存日志Redis 存热数据 参考资源官方文档https://docs.langchain.com/oss/python/langchain/short-term-memoryCheckpoint 列表https://docs.langchain.com/oss/python/integrations/checkpointers/indexRedis Stackhttps://redis.io/docs/stack/如果这篇文章帮到了你欢迎点赞、收藏、转发有问题评论区见我看到都会回复~ 作者介绍写文不易Bug 更不易。如果这篇文章对你有帮助可以搜一搜空门技术栈这里分享✅ Java / Spring AI / 企业级项目实战✅ Docker / RAG知识库 / 微服务踩坑✅ Python、前端、AI应用落地✅ 偶尔分享一些「头发保卫战」经验 一个热爱技术、持续填坑的开发者陪你一起少踩坑少加班多写优雅代码。 推荐阅读看了 3 天官方文档后我决定自己写一篇 LangChain 人话教程GPT-5.5 变强、Spring AI 更新、Ollama 爆漏洞今天值得看的技术热点还在复制粘贴 if-else模板方法模式专治重复代码CSDNLangChain 入门实战指南