Memory 系统详解——短期记忆 + 长期记忆双架构——从零开始学 LangGraph（十）

Memory 系统详解——短期记忆 长期记忆双架构写在前面第 6 期我们学了 Checkpointer——Agent 能在同一个 thread_id 里记住对话历史。但 Checkpointer 有一个边界它只在一个线程会话内有效。# 会话 ACheckpointer 记住了 agent.invoke(input, {configurable: {thread_id: A}}) # 会话 B完全不记得会话 A agent.invoke(input, {configurable: {thread_id: B}})真实世界的 Agent 需要跨会话的记忆“我昨天跟 Agent 聊过 Python 项目今天它应该记得我的技术栈偏好”“用户每次来都问英文问题Agent 应该记住他喜欢英文”“Agent 学会了用户的名字换了一个会话不应该忘记”LangGraph 提供了两层记忆架构层级组件作用域存储用途短期记忆Checkpointer单线程checkpoint 快照对话历史、当前状态长期记忆Store跨线程自定义键值对用户偏好、知识图谱、设置这一期我们重点搞定Store——长期记忆。全部代码不需要 API Key。1. 两层记忆的直观感受先看一个对比 demofrom typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END, add_messages from langgraph.checkpoint.memory import InMemorySaver from langgraph.store.memory import InMemoryStore class State(TypedDict): messages: Annotated[list, add_messages] def chat_node(state: State, config, *, store: InMemoryStore) - State: 带长期记忆的聊天节点 user_id config[configurable].get(user_id, default) # 从 Store 读取长期记忆 user_mem store.get((users, user_id), profile) user_name user_mem.value.get(name, 用户) if user_mem else 用户 return {messages: [(ai, f{user_name}你好这是第几次聊天)]} saver InMemorySaver() store InMemoryStore() # 预先存入用户信息 store.put((users, user-001), profile, {name: 三丰, lang: zh}) builder StateGraph(State) builder.add_node(chat, chat_node) builder.add_edge(START, chat) builder.add_edge(chat, END) graph builder.compile(checkpointersaver, storestore) # ─── 不同会话但记住了用户信息 ─── # 会话 A r1 graph.invoke( {messages: [(user, 你好)]}, {configurable: {thread_id: session-a, user_id: user-001}}, ) print(r1[messages][-1].content) # 三丰你好这是第几次聊天 # 会话 B不同的 thread_id但同一个 user_id r2 graph.invoke( {messages: [(user, hi)]}, {configurable: {thread_id: session-b, user_id: user-001}}, ) print(r2[messages][-1].content) # 三丰你好这是第几次聊天 # ✅ 换了一个会话但记得名字关键区别Checkpointer 记得这个会话的消息历史Store 记得用户三丰的偏好配置2. Store 的核心 APIStore 本质上是一个命名空间的键值存储命名空间 (namespace) → 键 (key) → 值 (value)2.1 写入store.put()store.put( (users, user-123), # namespace两层像文件系统的目录 profile, # key数据项的名称 {name: 三丰, age: 30, pref_lang: zh}, # value任意 dict )namespace是分层结构的元组用来组织数据示例 namespace含义(users, user-123)用户 123 的数据(conversations, thread-abc)会话 abc 的数据(global, config)全局配置(agents, weather-agent, stats)天气 Agent 的统计2.2 读取store.get()result store.get((users, user-123), profile) if result: print(result.value) # {name: 三丰, ...} print(result.key) # profile print(result.namespace) # (users, user-123) else: print(未找到)2.3 搜索store.search()# 搜索某个 namespace 下的所有数据 results store.search((users,)) for item in results: print(f用户 {item.key}: {item.value[name]}) # 用户 user-123: 三丰 # 用户 user-456: 李四store.search()支持通配符过滤# 搜索所有以 user- 开头的 key results store.search((users,), filter{key: {$like: user-%}}) # 带分页 results store.search((users,), limit10, offset0)2.4 删除store.delete()store.delete((users, user-123), profile)3. 在 Node 中读写 StoreNode 函数可以通过store关键字参数访问 Storedef my_node(state, config, *, store): 参数说明 - state: 当前的 State - config: 运行配置含 thread_id 等 - store: 长期记忆存储自动注入 # 读 item store.get((namespace,), key) # 写 store.put((namespace,), key, {data: value}) return {updated: True}完整示例from typing import Annotated from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END, add_messages from langgraph.checkpoint.memory import InMemorySaver from langgraph.store.memory import InMemoryStore class MemState(TypedDict): messages: Annotated[list, add_messages] def profile_node(state: MemState, config, *, store: InMemoryStore) - MemState: 读取和更新用户画像 user_id config[configurable].get(user_id, default) # 读取已有画像 existing store.get((profiles, user_id), prefs) prefs existing.value if existing else {} last_msg state[messages][-1].content if state[messages] else # 检测用户偏好 if 我叫 in last_msg: name last_msg.split(我叫)[-1].split()[0].split(。)[0].strip() prefs[name] name store.put((profiles, user_id), prefs, prefs) return {messages: [(ai, f记住了{name})]} if 英文 in last_msg or English in last_msg: prefs[lang] en store.put((profiles, user_id), prefs, prefs) return {messages: [(ai, Switched to English!)]} # 使用偏好生成回复 name prefs.get(name, 朋友) lang prefs.get(lang, zh) greeting fHello {name}! if lang en else f{name}你好呀 return {messages: [(ai, greeting)]} saver InMemorySaver() store InMemoryStore() builder StateGraph(MemState) builder.add_node(profile, profile_node) builder.add_edge(START, profile) builder.add_edge(profile, END) graph builder.compile(checkpointersaver, storestore) # ─── 测试 ─── config {configurable: {thread_id: mem-1, user_id: u001}} # 第 1 轮自我介绍 r1 graph.invoke({messages: [(user, 你好我叫三丰)]}, config) print(r1[messages][-1].content) # 记住了三丰 # 第 2 轮切换语言 r2 graph.invoke({messages: [(user, Please use English)]}, config) print(r2[messages][-1].content) # Switched to English! # 第 3 轮新会话但记得名字和语言 r3 graph.invoke( {messages: [(user, hi)]}, {configurable: {thread_id: mem-2, user_id: u001}}, ) print(r3[messages][-1].content) # Hello 三丰! ← 跨会话记得4. 实战一个带长期记忆的完整 Agent把 Store 和第 4 期的天气 Agent 结合起来from typing import Annotated, Literal from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END, add_messages from langgraph.checkpoint.memory import InMemorySaver from langgraph.store.memory import InMemoryStore from langchain_core.messages import SystemMessage, ToolMessage from langchain.chat_models import init_chat_model from langchain.tools import tool tool def get_weather(city: str) - str: 查询天气 data {北京: 晴 25°C, 上海: 多云 28°C, 深圳: 雷阵雨 30°C} return data.get(city, f暂无{city}数据) tool def search_nearby(city: str, category: str 餐厅) - str: 搜索附近场所 return f{city}的{category}推荐店铺A、店铺B、店铺C tools [get_weather, search_nearby] tools_by_name {tool.name: tool for tool in tools} model init_chat_model(gpt-4o-mini, temperature0) model_with_tools model.bind_tools(tools) class AgentState(TypedDict): messages: Annotated[list, add_messages] def extract_preferences(user_input: str) - dict: 从用户输入提取偏好 prefs {} if 北京 in user_input: prefs[city] 北京 if 上海 in user_input: prefs[city] 上海 if 餐厅 in user_input or 吃 in user_input: prefs[interest] 餐厅 return prefs def llm_node(state: AgentState, config, *, store: InMemoryStore) - dict: 带长期记忆的 LLM 节点 user_id config[configurable].get(user_id, default) # 读取长期记忆 user_mem store.get((users, user_id), prefs) preferences user_mem.value if user_mem else {} # 检测输入中隐含的偏好 last_msg state[messages][-1].content if state[messages] else new_prefs extract_preferences(last_msg) if new_prefs: merged {**preferences, **new_prefs} store.put((users, user_id), prefs, merged) preferences merged # 构建包含记忆的系统提示 city_hint f用户常查询的城市{preferences.get(city, 未知)} if preferences.get(city) else interest_hint f用户感兴趣的类别{preferences.get(interest, 通用)} if preferences.get(interest) else memory_context f\n关于用户的记忆\n{city_hint}\n{interest_hint}.strip() if (city_hint or interest_hint) else response model_with_tools.invoke([ SystemMessage(contentf你是一个助手可查天气和周边信息。{memory_context}), *state[messages] ]) return {messages: [response]} def tool_node(state: AgentState) - dict: last state[messages][-1] results [] for tc in last.tool_calls: t tools_by_name[tc[name]] results.append(ToolMessage(contentstr(t.invoke(tc[args])), tool_call_idtc[id])) return {messages: results} def should_continue(state: AgentState) - Literal[tools, END]: last state[messages][-1] return tools if last.tool_calls else END saver InMemorySaver() store InMemoryStore() builder StateGraph(AgentState) builder.add_node(llm, llm_node) builder.add_node(tools, tool_node) builder.add_edge(START, llm) builder.add_conditional_edges(llm, should_continue, {tools: tools, END: END}) builder.add_edge(tools, llm) agent builder.compile(checkpointersaver, storestore) # ─── 测试跨会话记忆 ─── def ask(query: str, user_id: str u001, thread_id: str t001): config {configurable: {thread_id: thread_id, user_id: user_id}} result agent.invoke({messages: [(user, query)]}, config) for m in result[messages]: if m.type ai and m.content: print(f {m.content}) # 第 1 天用户查北京天气 print(\n--- 第 1 天 ---) ask(北京今天天气怎么样) # 第 2 天新会话系统记得用户在北京 print(\n--- 第 2 天新会话自动记得城市偏好---) ask(这附近有什么好吃的, thread_idt002) # 自动知道 user 在北京搜索北京的餐厅运行输出--- 第 1 天 --- 北京今天天气晴朗25°C... --- 第 2 天新会话自动记得城市偏好--- 北京附近的好吃推荐店铺A、店铺B、店铺C...Agent 是怎么记住的用户说北京天气 → llm_node 提取到 {city: 北京} → store.put((users, u001), prefs, {city: 北京}) → 当前会话正常查天气 第二天 用户说附近有什么好吃的 → llm_node 读取 store → 拿到 {city: 北京} → 自动查北京的餐厅 → 用户甚至不用再提北京5. 三种 Store 实现跟 Checkpointer 一样Store 也有多种后端5.1 InMemoryStore开发用from langgraph.store.memory import InMemoryStore store InMemoryStore()数据存在内存重启丢失。适合开发测试。5.2 PostgresStore生产用pip install langgraph-checkpoint-postgresfrom langgraph.checkpoint.postgres import PostgresStore store PostgresStore.from_conn_string( postgresql://user:passhost:5432/langgraph ) await store.setup() # 建表数据持久化到 PostgreSQL支持分布式。5.3 自定义 Store如果你有自己的存储系统Redis、MongoDB、文件系统可以实现BaseStore接口from langgraph.store.base import BaseStore from typing import Optional class MyRedisStore(BaseStore): def __init__(self, redis_client): self.redis redis_client def get(self, namespace: tuple, key: str) - Optional[object]: data self.redis.get(f{:.join(namespace)}:{key}) if data: return InMemoryStore._Item(namespace, key, json.loads(data)) return None def put(self, namespace: tuple, key: str, value: dict) - None: self.redis.set(f{:.join(namespace)}:{key}, json.dumps(value)) def search(self, namespace: tuple, **kwargs) - list: # 实现搜索逻辑 pass def delete(self, namespace: tuple, key: str) - None: self.redis.delete(f{:.join(namespace)}:{key})6. Memory 设计的最佳实践6.1 存储什么该存不该存用户显式告诉你的信息名字、城市、偏好语言每次对话的完整 raw 日志提取的总结性信息敏感密码、令牌跨会话有用的上下文片段会话中 5 分钟前刚说的具体内容用户选择主题、格式、语言大量原始文本6.2 数据模型设计推荐的结构# 用户资料 store.put((users, user_id), profile, { name: str, preferred_lang: zh | en, timezone: Asia/Shanghai, created_at: 2026-07-06T00:00:00, updated_at: 2026-07-06T00:00:00, }) # 用户偏好 store.put((users, user_id), preferences, { city: 北京, interests: [技术, 美食], response_style: detailed, # concise / detailed / fun }) # 会话摘要 store.put((users, user_id), fsummary_{session_id}, { topics: [Python, LangGraph], decisions: [选择用 LangGraph 开发], pending_items: [部署到服务器], })6.3 更新策略def update_preference(store, user_id, key, value): 安全更新不覆盖已有数据 existing store.get((users, user_id), prefs) prefs existing.value if existing else {} prefs[key] value prefs[updated_at] __import__(datetime).datetime.now().isoformat() store.put((users, user_id), prefs, prefs)6.4 过期策略生产环境中 Store 会持续增长建议设 TTL# 可以把 TTL 信息存在数据中 store.put((sessions, session_id), cache, { data: ..., expires_at: 2026-07-07T00:00:00, }) # 读取时检查 TTL def get_with_ttl(store, namespace, key, max_age_hours24): item store.get(namespace, key) if not item: return None from datetime import datetime, timedelta expires datetime.fromisoformat(item.value.get(expires_at, 2000-01-01)) if datetime.now() expires: store.delete(namespace, key) return None return item7. 用 Store 实现记住-遗忘机制真正的聊天记忆不是无限增长的。你需要决定什么该忘class MemoryState(TypedDict): messages: Annotated[list, add_messages] memory_summary: str # 压缩后的记忆摘要 def summarize_memory_node(state: MemoryState, config, *, store: InMemoryStore) - MemoryState: 当消息太多时压缩长期记忆 msgs state[messages] user_id config[configurable].get(user_id, default) # 只在消息超过 10 条时才触发摘要 if len(msgs) 10 and not state.get(memory_summary): # 提取关键信息 names, topics set(), set() for m in msgs: content m.content if hasattr(m, content) else str(m) # 简单关键词提取 for name_keyword in [我叫, 我是]: if name_keyword in content: names.add(content.split(name_keyword)[-1].split()[0].split(。)[0].strip()) for topic_word in [喜欢, 研究, 做]: if topic_word in content: topics.add(content) summary f用户: {, .join(names) if names else 未知}。讨论过: {, .join(list(topics)[:3])} # 存入长期记忆 store.put((users, user_id), summary, { summary: summary, created_at: str(__import__(datetime).datetime.now()), }) return {memory_summary: summary} return {}8. 常见坑点坑 1Node 函数签名里少了*, store# ❌ 没加 * (关键字参数分隔符) def my_node(state, config, store): ... # ✅ 正确——store 是关键字参数 def my_node(state, config, *, store): ... # ✅ 或者用 **kwargs def my_node(state, config, **kwargs): store kwargs.get(store)坑 2compile 时忘了传 store# ❌ 忘了传 graph builder.compile(checkpointersaver) # ✅ 必须传 graph builder.compile(checkpointersaver, storestore)坑 3store.get 不检查 None# ❌ 如果用户没有 profilestore.get 返回 None result store.get((users, xxx), profile) name result.value[name] # AttributeError! # ✅ 检查 None result store.get((users, xxx), profile) name result.value[name] if result else 默认用户坑 4大规模使用 InMemoryStore# ❌ 生产环境用 InMemoryStore — 重启丢数据 # ❌ InMemoryStore 不限制大小 — 会撑爆内存 # ✅ 生产环境用 PostgresStore 或自定义持久化 Store坑 5Store 数据不一致Store 不提供事务保证。如果多个 Node 同时写同一个 namespace/key不同线程可能出现数据覆盖。设计上尽量让每个 namespace 只有特定 Node 写入。9. Checkpointer Store 完整架构┌──────────────────────────────────────┐ │ LangGraph │ │ │ │ ┌─────────────────────────────┐ │ │ │ Checkpointer │ │ │ │ (短期记忆自动保存checkpoint) │ │ │ │ scope 一个 thread_id │ │ │ │ 存State 快照、消息历史 │ │ │ └─────────────────────────────┘ │ │ │ │ ┌─────────────────────────────┐ │ │ │ Store │ │ │ │ (长期记忆手动读写) │ │ │ │ scope 整个应用 │ │ │ │ 存用户偏好、知识、设置 │ │ │ └─────────────────────────────┘ │ └──────────────────────────────────────┘ Checkpointer 自动读/写 Store 手动读/写 (开发者不需要额外代码) (开发者显式调用)维度CheckpointerStore数据来源Node 返回的 State 更新开发者手动调用store.put()读取方式自动加载同 thread_id手动调用store.get()作用域单线程全局生命周期与线程绑定由开发者控制使用场景对话历史、状态追踪用户偏好、跨会话记忆10. 本期核心总结概念一句话两层记忆Checkpointer 短期线程内Store 长期跨线程namespace分层组织数据的目录结构如(users, user-123)store.put()写入长期记忆参数namespace, key, valuestore.get()读取长期记忆返回 Item 或 Nonestore.search()搜索 namespace 下的所有记录store.delete()删除指定记录Node 中访问函数签名要加*, storecompile 时传 store 参数下期预告第 11 期Subgraph 与多 Agent 协作模式Subgraph 图中的图。把 Agent 封装成可复用的子图构建路由、委托、交接三种多 Agent 协作模式。参考资料LangGraph Memory 概念Memory overview - Docs by LangChainStores 指南Stores - Docs by LangChainLangGraph MemoryAI 版Memory overview - Docs by LangChain短期记忆 长期记忆的双架构设计是 LangGraph 区别于大多数 Agent 框架的核心能力之一。Checkpointer 让 Agent 在当前对话中连贯Store 让 Agent 在不同对话中越来越了解你。两者配合才是一个真正有记忆的 Agent。