给 Agent 装上“大象记忆“:LangChain 长期记忆(Long-term Memory)实战
2026/6/25 16:34:59 网站建设 项目流程

你跟一个 AI 助手说:"我叫小明,我喜欢喝拿铁。“它回答"好的,记住了!”。第二天你再打开对话,它却问:“你好,请问怎么称呼?”——金鱼记忆

问题在于:默认情况下,Agent 的记忆只活在一次会话里,会话一结束就清空。要让它跨对话、跨天地记住你的偏好和身份,需要的是长期记忆(Long-term Memory)。这篇文章结合 3 个完整可运行的例子,讲清长期记忆怎么实现。代码全部内嵌正文,复制即可跑。


一、长期记忆 vs 短期记忆

先分清两个容易混的概念:

短期记忆(Checkpointer)长期记忆(Store)
范围单个会话内跨多个会话
内容完整对话历史提炼的关键信息(偏好、画像、事实)
生命周期会话结束即丢弃持久保留,随时可取
访问方式框架自动保存/加载通过runtime.store手动读写
类比人的工作记忆人的知识库

一句话:短期记忆让 Agent 在这一场对话里连贯;长期记忆让它跨越所有对话记住你。本文讲的是后者——基于Store实现。


二、Store 的数据结构:一个 JSON 文件柜

长期记忆基于 LangGraph 的Store,数据以 JSON 文档存储,用命名空间(namespace)+ 键(key)的层级组织。类比文件系统:

namespace(命名空间) ≈ 文件夹路径 key(键) ≈ 文件名 value(值) ≈ 文件内容(JSON) ("users", "user_123") "profile" → {"name": "张三", "email": "...", "skill": "..."} └──── namespace ────┘ └─ key ─┘ └──────────────── value ────────────────┘

namespace 的核心作用是隔离——不同用户的数据互不干扰:

("users", "user_123") ← user_123 的数据 ("users", "user_456") ← user_456 的数据(互不可见)

namespace 的层数是自由的:上面用两层最简单。需要按类目细分时,可以再加一层,
比如("users", "user_123", "preferences"),把偏好、画像分开存——第五节就是这么做的。

记住这个namespace + key → value结构,后面所有操作都是围绕它的增删改查。


三、最小实现:写入 + 读取

实现长期记忆只需三步:① 创建 Store → ② 传入create_agentstore=→ ③ 工具里用runtime.store读写

下面用InMemoryStore(开发用,最简单)做一个能存能取的助手:一个工具负责写、一个负责读。

importosfromdataclassesimportdataclassfromdotenvimportload_dotenvfromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modelfromlangchain.toolsimporttoolfromlanggraph.prebuiltimportToolRuntimefromlanggraph.store.memoryimportInMemoryStore load_dotenv()model=init_chat_model(os.getenv("MODEL_NAME","glm-5.1"),model_provider="openai",base_url=os.getenv("OPENAI_API_BASE"),api_key=os.getenv("OPENAI_API_KEY"),streaming=True,)@dataclassclassCustomContext:user_id:str@tooldefsave_user_profile(name:str,email:str,skill:str,runtime:ToolRuntime[CustomContext])->str:"""保存/更新当前用户的档案信息(写入长期记忆)。 Args: name: 用户姓名 email: 用户邮箱 skill: 用户技能 """namespace=("users",runtime.context.user_id)runtime.store.put(namespace,"profile",{"name":name,"email":email,"skill":skill})returnf"已保存档案:name={name}, email={email}, skill={skill}"@tooldefget_user_profile(runtime:ToolRuntime[CustomContext])->str:"""读取当前用户的档案信息(从长期记忆读取)。"""namespace=("users",runtime.context.user_id)item=runtime.store.get(namespace,"profile")ifitemisNone:return"No profile"p=item.valuereturnf"name:{p['name']}\nemail:{p['email']}\nskill:{p['skill']}\n"store=InMemoryStore()agent=create_agent(model=model,tools=[save_user_profile,get_user_profile],store=store,# ② 注入 Storesystem_prompt="你是一个使用工具帮用户完成任务的有用助手",)if__name__=="__main__":ctx=CustomContext(user_id="user_123")# 1. 写入:让 Agent 调用 save_user_profile 把资料存进长期记忆print("===== 写入 =====")r1=agent.invoke({"messages":[{"role":"user","content":"记住我的资料:我叫张三,邮箱 777@qq.com,技能 langchain"}]},context=ctx,)print(r1["messages"][-1].content)# 2. 读取:新的一轮对话,让 Agent 调用 get_user_profile 取回资料print("\n===== 读取 =====")r2=agent.invoke({"messages":[{"role":"user","content":"我叫什么?我的技能是什么?"}]},context=ctx,)print(r2["messages"][-1].content)

两个要点:

  1. user_id来自runtime.context,不要硬编码。命名空间用("users", runtime.context.user_id)动态拼,每个用户读写自己的数据。硬编码 user_id = 所有人共享一份数据,这是经典坑。
  2. get返回的是Item对象,真正的数据在.value里,不存在时返回None

InMemoryStore有个致命问题:数据只在内存里,进程一退出就全没了。它只适合开发调试。要真正"跨会话、重启还在",得换持久化存储。


四、持久化:换成 SqliteStore

InMemoryStore换成SqliteStore,记忆就落盘到一个.db文件里——进程重启后依然存在,而且不像 Redis/Postgres 那样需要单独跑一个服务器,单机一个文件就搞定。

依赖:pip install langgraph-checkpoint-sqlite。用法是SqliteStore.from_conn_string(路径)返回一个上下文管理器,进去后先调setup()建表。

代码几乎和上面一样,只换了 Store 的创建方式(注意with块):

importosfromdataclassesimportdataclassfromdotenvimportload_dotenvfromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modelfromlangchain.toolsimporttoolfromlanggraph.prebuiltimportToolRuntimefromlanggraph.store.sqliteimportSqliteStore load_dotenv()model=init_chat_model(os.getenv("MODEL_NAME","glm-5.1"),model_provider="openai",base_url=os.getenv("OPENAI_API_BASE"),api_key=os.getenv("OPENAI_API_KEY"),streaming=True,)@dataclassclassCustomContext:user_id:str@tooldefsave_user_profile(name:str,email:str,skill:str,runtime:ToolRuntime[CustomContext])->str:"""保存/更新当前用户的档案信息(写入长期记忆)。 Args: name: 用户姓名 email: 用户邮箱 skill: 用户技能 """namespace=("users",runtime.context.user_id)runtime.store.put(namespace,"profile",{"name":name,"email":email,"skill":skill})returnf"已保存档案:name={name}, email={email}, skill={skill}"@tooldefget_user_profile(runtime:ToolRuntime[CustomContext])->str:"""读取当前用户的档案信息(从长期记忆读取)。"""namespace=("users",runtime.context.user_id)item=runtime.store.get(namespace,"profile")ifitemisNone:return"No profile"p=item.valuereturnf"name:{p['name']}\nemail:{p['email']}\nskill:{p['skill']}\n"if__name__=="__main__":withSqliteStore.from_conn_string("long_memory.db")asstore:store.setup()# 建表(幂等)agent=create_agent(model=model,tools=[save_user_profile,get_user_profile],store=store,system_prompt="你是一个使用工具帮用户完成任务的有用助手",)ctx=CustomContext(user_id="user_123")defask(content:str)->str:returnagent.invoke({"messages":[{"role":"user","content":content}]},context=ctx)["messages"][-1].content# 1. 启动读取:第二次起能读到上次写入的(持久化证明)print("===== 启动读取 =====")print(ask("我的档案是什么?"))# 2. 写入print("\n===== 写入 =====")print(ask("记住我的资料:我叫张三,邮箱 777@qq.com,技能 langchain"))# 3. 写入后读取print("\n===== 写入后读取 =====")print(ask("我叫什么?我的技能是什么?"))

怎么验证持久化?跑两遍这个脚本:

  • 第一遍:"启动读取"是空的 → 写入 → 读到刚写的。
  • 第二遍:"启动读取"直接命中上一遍写到磁盘的档案——这就是长期记忆"跨会话、重启还在"的铁证。

单机/中小规模用 SQLite 足矣;如果要多实例共享、海量数据,再升级到PostgresStore(用法一样:from_conn_string+setup()),但绝大多数场景一个 sqlite 文件就够。唯一不能用于生产的是InMemoryStore


五、记忆的完整生命周期:增删改查 + 隔离

写入和读取只是开始。Store 还支持更新、搜索、删除、列命名空间。下面这段直接操作 Store(不经过模型,确定性、可复现),把整套生命周期和命名空间隔离一次性演示清楚:

fromlanggraph.store.memoryimportInMemoryStore store=InMemoryStore()defline(title:str)->None:print(f"\n{'='*50}\n{title}")if__name__=="__main__":# ---- Put:为两个用户、两个类目写入数据(命名空间隔离)----line("Put 写入(多用户 / 多类目)")store.put(("users","user_123","profile"),"info",{"name":"张三","city":"北京"})store.put(("users","user_123","preferences"),"coffee",{"value":"latte"})store.put(("users","user_123","preferences"),"language",{"value":"zh"})store.put(("users","user_456","preferences"),"coffee",{"value":"americano"})print("已为 user_123 和 user_456 写入数据")# ---- Get:读取单条 ----line("Get 读取")item=store.get(("users","user_123","preferences"),"coffee")print("user_123 coffee ->",item.value)# {'value': 'latte'}print("created_at ->",item.created_at)# ---- Update:相同 namespace + key 再 put 即覆盖 ----line("Update 更新(覆盖)")store.put(("users","user_123","preferences"),"coffee",{"value":"espresso"})print("更新后 ->",store.get(("users","user_123","preferences"),"coffee").value)# ---- Search:按命名空间前缀搜索,可加 filter 精确过滤 ----line("Search 搜索")all_prefs=store.search(("users","user_123","preferences"))print("user_123 全部偏好:",[(r.key,r.value)forrinall_prefs])filtered=store.search(("users","user_123","preferences"),filter={"value":"zh"})print("filter value=zh:",[(r.key,r.value)forrinfiltered])# ---- 命名空间隔离:user_456 看不到 user_123 的数据 ----line("命名空间隔离")print("user_456 coffee ->",store.get(("users","user_456","preferences"),"coffee").value)# ---- list_namespaces:列出所有命名空间 ----line("list_namespaces 列出命名空间")fornsinstore.list_namespaces():print(" ",ns)# ---- Delete:删除单条 ----line("Delete 删除")store.delete(("users","user_123","preferences"),"coffee")print("删除 coffee 后 ->",store.get(("users","user_123","preferences"),"coffee"))# None

运行输出(节选):

Update 更新(覆盖) 更新后 -> {'value': 'espresso'} Search 搜索 user_123 全部偏好: [('coffee', {'value': 'espresso'}), ('language', {'value': 'zh'})] filter value=zh: [('language', {'value': 'zh'})] 命名空间隔离 user_456 coffee -> {'value': 'americano'} Delete 删除 删除 coffee 后 -> None

五个操作对应记忆的生命周期:

操作方法说明
创建store.put(ns, key, value)value 必须是 dict
读取store.get(ns, key)返回 Item,数据在.value,无则 None
更新store.put(ns, key, 新value)相同 ns+key 直接覆盖
搜索store.search(ns_prefix, filter=...)按前缀 + 字段过滤
删除store.delete(ns, key)

注意search这里是按字段精确过滤。如果要"按语义找相关记忆"(向量检索),那是向量数据库/RAG 的活儿——本系列的rag/目录专门讲,不在长期记忆的范畴里硬塞。


六、最佳实践与避坑

✅ 用 namespace 做隔离,别用扁平 key 拼字符串

# ✅ 好:天然隔离store.put(("users",user_id,"preferences"),"coffee",{...})store.put(("users",user_id,"profile"),"name",{...})# ❌ 差:全堆一层,靠拼 key 区分,容易冲突store.put(("data",),f"user_{user_id}_coffee",{...})

✅ user_id 从 context 动态取,绝不硬编码

硬编码 user_id → 所有用户读写同一份数据,彻底失去隔离。永远runtime.context.user_id

✅ 存"提炼的关键信息",不是整段对话

长期记忆该存的是偏好、画像、重要事实(“对海鲜过敏”“喜欢拿铁”),而不是把几百条聊天记录原样塞进去——那是短期记忆/Checkpointer 的事。

⚠️ InMemoryStore 只能用于开发

它重启即丢、不能多实例共享。任何要留住数据的场景,至少上 SqliteStore。


七、总结

概念比喻作用
Long-term Memory人的知识库跨会话持久记忆
Store文件柜存取 JSON 文档
namespace文件夹路径隔离不同用户/类目
key / value文件名 / 文件内容定位并存储一条记忆
InMemoryStore临时便签开发用,重启丢失
SqliteStore带锁的抽屉单机持久化,无需服务器
context身份证动态传入 user_id

长期记忆的本质,是让 Agent 突破单次会话的限制——通过 Store 以namespace + key的结构持久化关键信息,让它在任意时间、任意对话里都能召回你的偏好、画像和历史,从而提供连贯、个性化的体验。

从"金鱼"到"大象",差的就是这一个store=参数和几行runtime.store读写。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询