LangChain 跨会话存储(Store)实战:让 Agent 记住你,而不是只记住对话
Checkpointer 解决了「单个对话内记忆」——同一线程里的多轮对话它能记住。但换个线程、隔天再来,它就又把你忘了。因为用户的偏好、学习进度、账号信息,这些应该跨对话存在,而不是绑死在一次聊天里。
这就是 Store(跨会话存储)的地盘。这篇文章讲清楚:Store 和 Checkpointer 的区别、Store 的基本操作、怎么在 Agent 里用 Store,以及持久化选型。
一、先厘清:Checkpointer vs Store
这是理解记忆体系最关键的对比。两者分工不同,缺一不可:
| 维度 | Checkpointer | Store |
|---|---|---|
| 作用域 | 单个对话线程(thread_id) | 跨所有对话线程 |
| 数据类型 | Agent 状态快照(自动管理) | 任意键值数据(手动管理) |
| 典型用途 | 多轮对话记忆 | 用户偏好、知识库、配置 |
| 数据组织 | thread_id → checkpoint | (namespace, key) → value |
✅ 一句话总结:Checkpointer 负责「对话到哪了」,Store 负责「用户是谁、会什么、喜欢什么」。两者配合,才能构建有持续记忆的智能 Agent。
二、Store 的基本操作
Store 用命名空间(namespace)+ 键(key)+ 值(value)的层级结构组织数据。namespace 是元组(类似文件夹层级),key 是字符串(类似文件名),value 是字典。
fromlanggraph.store.memoryimportInMemoryStore store=InMemoryStore()# 写入数据:put(namespace, key, value)store.put(("users","user_001"),# 命名空间"preferences",# 键{# 值"theme":"dark","language":"zh-CN","level":"入门",})store.put(("users","user_001"),"progress",{"completed_courses":["HTML 基础","Python 基础"],"total_hours":35,})# 读取数据:get(namespace, key)prefs=store.get(("users","user_001"),"preferences")print(f"偏好设置:{prefs.value}")progress=store.get(("users","user_001"),"progress")print(f"学习进度:{progress.value}")# 搜索数据:search(namespace)all_user_data=store.search(("users","user_001"))print(f"\n用户的所有数据 ({len(all_user_data)}项):")foriteminall_user_data:print(f"{item.key}:{item.value}")# 删除数据:delete(namespace, key)store.delete(("users","user_001"),"preferences")deleted=store.get(("users","user_001"),"preferences")print(f"\n删除后:{deleted}")# None运行结果:
偏好设置: {'theme': 'dark', 'language': 'zh-CN', 'level': '入门'} 学习进度: {'completed_courses': ['HTML 基础', 'Python 基础'], 'total_hours': 35} 用户的所有数据 (2 项): preferences: {'theme': 'dark', 'language': 'zh-CN', 'level': '入门'} progress: {'completed_courses': ['HTML 基础', 'Python 基础'], 'total_hours': 35} 删除后: None四个核心操作
| 操作 | 方法 | 说明 |
|---|---|---|
| 写入 | store.put(namespace, key, value) | 存数据 |
| 读取 | store.get(namespace, key) | 取单条 |
| 搜索 | store.search(namespace) | 列某个 namespace 下所有条目 |
| 删除 | store.delete(namespace, key) | 删单条 |
三、在 Agent 中使用 Store
把 Store 传给create_agent(),Agent 里的所有工具都能通过InjectedStore注解访问它。看一个课程顾问的例子——课程目录和用户会员信息都存在 Store 里,工具按需读。
fromdotenvimportload_dotenv load_dotenv()fromtypingimportAnnotatedfromlanggraph.store.baseimportBaseStorefromlanggraph.store.memoryimportInMemoryStorefromlangchain.toolsimporttool,InjectedStorefromlangchain.agentsimportcreate_agentfromlangchain.chat_modelsimportinit_chat_modelfromlangchain.messagesimportHumanMessage# 创建 Store 并预置数据store=InMemoryStore()store.put(("runoob","courses"),"catalog",{"Python3 基础教程":{"price":"免费","hours":20,"level":"入门"},"Python 数据分析":{"price":"会员","hours":30,"level":"进阶"},"Java 面向对象":{"price":"免费","hours":25,"level":"进阶"},})store.put(("runoob","users"),"user_vip_001",{"name":"小明","membership":"VIP","joined":"2024-01-15",})@tooldefquery_course_info(course_name:str,store:Annotated[BaseStore,InjectedStore()],)->str:"""查询菜鸟教程 RUNOOB 中课程的详细信息。 Args: course_name: 课程名称 """item=store.get(("runoob","courses"),"catalog")catalog=item.valueifitemelse{}ifcourse_nameincatalog:info=catalog[course_name]return(f"《{course_name}》- 价格:{info['price']},"f"时长:{info['hours']}小时,难度:{info['level']}")returnf"未找到课程《{course_name}》"@tooldefget_user_membership(user_id:str,store:Annotated[BaseStore,InjectedStore()],)->str:"""查询用户会员信息。 Args: user_id: 用户 ID """item=store.get(("runoob","users"),user_id)ifitemisNone:returnf"未找到用户{user_id}"user=item.valuereturn(f"用户{user['name']},{user['membership']}会员,"f"注册日期{user['joined']}")model=init_chat_model("deepseek:deepseek-v4-flash",temperature=0)agent=create_agent(model=model,tools=[query_course_info,get_user_membership],store=store,system_prompt="你是菜鸟教程 RUNOOB 的课程顾问。",)# 查询课程信息(数据来自 Store)result=agent.invoke({"messages":[HumanMessage(content="Python3 基础教程多少钱?")]})print(f"查询课程:{result['messages'][-1].content}")# 查询用户信息(数据来自 Store)result=agent.invoke({"messages":[HumanMessage(content="帮我查一下用户 user_vip_001 的信息")]})print(f"查询用户:{result['messages'][-1].content}")运行结果:
查询课程: 《Python3 基础教程》是免费的,学习时长约20小时,难度为入门级别。 查询用户: 用户小明是 VIP 会员,注册日期为 2024年1月15日。🔴 重点:关键在
store: Annotated[BaseStore, InjectedStore()]。它告诉 LangChain,「这个参数是注入进来的 Store」——模型看不到它(不会出现在工具签名里),但工具运行时能直接读写 Store。
四、Store 的持久化
InMemoryStore数据在程序重启后丢失,和InMemorySaver一样只适合开发和测试。生产环境用数据库版。
# 开发阶段fromlanggraph.store.memoryimportInMemoryStore store=InMemoryStore()# 生产环境(需要 PostgreSQL)# from langgraph.store.postgres import PostgresStore# store = PostgresStore.from_conn_string("postgresql://...")⚠️ 注意:生产环境建议用
PostgresStore(数据库持久化),并配合PostgresSaver(Checkpointer)一起用——前者管跨会话长期记忆,后者管单线程对话状态。
五、Store 使用建议
| 场景 | namespace 示例 | key 示例 | 说明 |
|---|---|---|---|
| 用户偏好 | ("users", user_id) | preferences | 主题、语言、通知设置 |
| 学习进度 | ("users", user_id) | progress | 已完成课程、学习时长 |
| 知识库 | ("kb", collection) | doc_id | 文档、FAQ、产品信息 |
| 会话摘要 | ("sessions", thread_id) | summary | 长对话的摘要,供 Checkpointer 之外使用 |
🔴 重点:Store 的数据是跨线程的,这正是它和 Checkpointer 的根本区别。今天在 A 线程存的偏好,明天在 B 线程也能取到。
六、总结:你真正需要记住的这几件事
- Checkpointer = 短期记忆(对话到哪了),Store = 长期记忆(用户是谁、会什么)。
- Store 用 (namespace, key) → value 的层级结构组织数据,四个核心操作:put / get / search / delete。
- 在 Agent 里用
InjectedStore注解把 Store 注入工具,模型看不到、工具能读写。 - 开发用
InMemoryStore,生产用PostgresStore,最好和PostgresSaver一起上。 - Store 跨线程共享——这是跨会话记忆的关键,也是和 Checkpointer 的本质区别。
验证清单
- 能
put/get/search/delete操作 Store - 工具里能通过
InjectedStore读 Store 数据 - 同一 Store 在不同 thread_id(跨会话)都能读到
InMemoryStore重启后数据丢失、PostgresStore持久化- 能说出 Store 和 Checkpointer 的职责边界
参考资源
- LangChain 官方 · Long-term memory: https://docs.langchain.com/oss/python/langchain/long-term-memory
- LangGraph 官方 · Stores / Persistence: https://docs.langchain.com/oss/python/langgraph/stores
- 菜鸟教程 · LangChain 跨会话存储 Store: https://www.runoob.com/langchain/langchain-store.html
说明:文中模型名与代码示例以官方文档和菜鸟教程为准,具体 API 细节请以你安装的 LangChain / LangGraph 版本为准。