【langgraph 从入门到精通graphApi 篇】Memory 与长期记忆

第 8 章：Memory 与长期记忆

8.1 本章目标

学完本章你将能够：

1. 理解短期记忆（Checkpoint）与长期记忆（Store）的区别
2. 掌握 InMemoryStore / PostgresStore 的配置和使用
3. 学会在节点中通过 Runtime 访问 Store
4. 实现语义记忆搜索功能

8.2 核心概念

Checkpoint vs Store

比喻：Checkpoint 就像"草稿自动保存"——每次编辑后自动保存当前状态。Store 就像"客户档案"——记录客户的长期偏好，所有对话都能访问。

短期记忆 vs 长期记忆架构

Store 命名空间设计

# 命名空间使用 tuple 结构，支持层级隔离("user_123","memories")# 用户 123 的记忆("user_123","preferences")# 用户 123 的偏好("user_456","memories")# 用户 456 的记忆("global","knowledge")# 全局知识库

8.3 实战

实战 1：用户偏好记忆

fromtypingimportTypedDict,Annotatedfromdataclassesimportdataclassfromlanggraph.graphimportStateGraph,START,END,add_messagesfromlanggraph.checkpoint.memoryimportMemorySaverfromlanggraph.store.memoryimportInMemoryStorefromlanggraph.runtimeimportRuntimefromlangchain_core.messagesimportBaseMessage,HumanMessage,AIMessage# ============================================# 定义 Context（运行时上下文）# ============================================@dataclassclassContext:user_id:str# ============================================# 定义 State# ============================================classState(TypedDict):messages:Annotated[list[BaseMessage],add_messages]# ============================================# 定义节点（使用 Runtime 访问 Store）# ============================================asyncdefmemory_node(state:State,runtime:Runtime[Context])->dict:""" 使用 Runtime 访问 Store 实现长期记忆。 关键点： 1. 在函数签名中声明 runtime: Runtime[Context] 参数 2. LangGraph 自动注入 Runtime 实例 3. 通过 runtime.context 访问上下文 4. 通过 runtime.store 访问 Store """user_id=runtime.context.user_id namespace=(user_id,"memories")# 从 Store 读取用户记忆memories=awaitruntime.store.asearch(namespace,limit=10)# 构建包含记忆的回复ifmemories:memory_text="我记得以下信息：\n"+"\n".join(f" -{m.value.get('content','')}"forminmemories)else:memory_text="我还没有关于你的记忆。"# 如果有新信息，存入 Storelast_msg=state["messages"][-1]if"记住"inlast_msg.contentor"我叫"inlast_msg.content:awaitruntime.store.aput(namespace,f"memory_{len(memories)+1}",{"content":last_msg.content},)return{"messages":[AIMessage(content=f"收到！{memory_text}")]}# ============================================# 构建图# ============================================store=InMemoryStore()checkpointer=MemorySaver()builder=StateGraph(State)builder.add_node("memory",memory_node)builder.add_edge(START,"memory")builder.add_edge("memory",END)# 编译时传入 store 和 context_schemagraph=builder.compile(checkpointer=checkpointer,store=store,context_schema=Context,)# ============================================# 异步运行# ============================================importasyncioasyncdefmain():config={"configurable":{"thread_id":"mem-001"}}context=Context(user_id="user_123")# 第一次对话result=awaitgraph.ainvoke({"messages":[HumanMessage(content="我叫小明，记住这个")]},config=config,context=context,)print(f"AI:{result['messages'][-1].content}")# 第二次对话（新 thread，但同一 user_id）config2={"configurable":{"thread_id":"mem-002"}}result=awaitgraph.ainvoke({"messages":[HumanMessage(content="你还记得我叫什么吗？")]},config=config2,context=context,)print(f"AI:{result['messages'][-1].content}")asyncio.run(main())

实战 2：语义记忆搜索

fromlanggraph.store.memoryimportInMemoryStore# 创建带语义搜索的 Storestore=InMemoryStore(index={"embed":"openai:text-embedding-3-small",# 嵌入模型"dims":1536,# 嵌入维度"fields":["content","$"],# 要索引的字段})# 存储带语义信息的记忆asyncdefstore_with_semantics():awaitstore.aput(("user_123","memories"),"mem_1",{"content":"用户喜欢喝咖啡，尤其是拿铁"},)awaitstore.aput(("user_123","memories"),"mem_2",{"content":"用户是 Python 程序员"},)awaitstore.aput(("user_123","memories"),"mem_3",{"content":"用户住在北京朝阳区"},)# 语义搜索：自然语言查询results=awaitstore.asearch(("user_123","memories"),query="用户喜欢喝什么饮料？",# 自然语言查询limit=3,)foriteminresults:print(f" [{item.key}]{item.value['content']}")# 输出: [mem_1] 用户喜欢喝咖啡，尤其是拿铁asyncio.run(store_with_semantics())

8.4 API 速查

8.5 错误与避坑指南

坑 1：混淆 Store 和 Checkpoint

# ❌ 错误：把用户偏好存在 Checkpoint 中# Checkpoint 是 thread 级别的，换 thread 就丢失了！classState(TypedDict):user_preferences:dict# 不应该放在 State 中# ✅ 正确：用户偏好存在 Store 中# Store 跨 thread 共享，通过 user_id 隔离awaitstore.aput(("user_123","preferences"),"key",value)

坑 2：命名空间设计不当

# ❌ 错误：所有用户共享命名空间awaitstore.aput(("memories"),"key",value)# 所有用户混在一起# ✅ 正确：按用户 ID 隔离awaitstore.aput(("user_123","memories"),"key",value)awaitstore.aput(("user_456","memories"),"key",value)

坑 3：语义搜索未配置 embedding

# ❌ 错误：没有配置 indexstore=InMemoryStore()# 没有 indexawaitstore.asearch(ns,query="自然语言查询")# 无法语义搜索，只能精确匹配# ✅ 正确：配置 indexstore=InMemoryStore(index={"embed":"openai:text-embedding-3-small","dims":1536,"fields":["content"],})

坑 4：忘记在 compile 时传入 store

# ❌ 错误graph=builder.compile(checkpointer=...)# 节点中 runtime.store 不可用！# ✅ 正确graph=builder.compile(checkpointer=...,store=store,# 必须传入)

8.6 最佳实践总结

1. 用户级数据用 Store，会话级数据用 Checkpoint：按数据生命周期选择存储
2. 命名空间使用(user_id, "category")结构：层级隔离，清晰明了
3. 生产环境使用 PostgresStore 持久化：数据不丢失，支持高并发
4. 语义搜索时选择合适的 embedding 模型：平衡成本和效果
5. Context Schema 用于注入运行时信息：如 user_id、tenant_id 等，不需要放在 State 中