【langgraph 从入门到精通graphApi 篇】Memory 与长期记忆
文章目录
- 第 8 章：Memory 与长期记忆
- 8.1 本章目标
- 8.2 核心概念
- Checkpoint vs Store
- 短期记忆 vs 长期记忆架构
- Store 命名空间设计
- 8.3 实战
- 实战 1：用户偏好记忆
- 实战 2：语义记忆搜索
- 8.4 API 速查
- 8.5 错误与避坑指南
- 坑 1：混淆 Store 和 Checkpoint
- 坑 2：命名空间设计不当
- 坑 3：语义搜索未配置 embedding
- 坑 4：忘记在 compile 时传入 store
- 8.6 最佳实践总结
第 8 章：Memory 与长期记忆
8.1 本章目标
学完本章你将能够：
- 理解短期记忆（Checkpoint）与长期记忆（Store）的区别
- 掌握 InMemoryStore / PostgresStore 的配置和使用
- 学会在节点中通过 Runtime 访问 Store
- 实现语义记忆搜索功能
8.2 核心概念
Checkpoint vs Store
| 特性 | Checkpoint（短期记忆） | Store（长期记忆） |
|---|---|---|
| 存储内容 | 图状态快照（State 的完整副本） | 应用定义的键值数据 |
| 作用范围 | 单个 Thread（对话线程） | 跨 Thread（跨会话） |
| 生命周期 | 对话结束后可清理 | 持久保留 |
| 典型用途 | 对话连续性、HITL、容错 | 用户偏好、知识库、共享信息 |
| 访问方式 | 自动（每个 super-step 保存） | 手动（节点中显式读写） |
比喻：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 速查
| API | 完整签名 | 入参说明 | 返回值 | 说明 |
|---|---|---|---|---|
InMemoryStore() | InMemoryStore(index=...) | index: 语义搜索配置（可选） | Store 对象 | 内存存储 |
PostgresStore(conn) | PostgresStore(conn) | conn: 数据库连接 | Store 对象 | Postgres 持久化 |
store.put(namespace, key, value) | put(ns: tuple, key: str, value: dict) | ns: 命名空间;key: 键;value: 值 | None | 存储数据 |
store.search(namespace, query, limit) | search(ns: tuple, query: str, limit: int) | ns: 命名空间;query: 查询;limit: 数量 | 结果列表 | 语义搜索（需 index） |
store.aget/get/adelete | 异步版本 | 同上 | 同上 | 节点中异步调用 |
Runtime[Context] | Runtime[Context] | Context: 上下文类型 | 运行时对象 | 节点中注入上下文 |
context_schema | compile(context_schema=Type) | context_schema: Context 类型 | CompiledGraph | 编译时声明 Context |
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 最佳实践总结
- 用户级数据用 Store，会话级数据用 Checkpoint：按数据生命周期选择存储
- 命名空间使用
(user_id, "category")结构：层级隔离，清晰明了 - 生产环境使用 PostgresStore 持久化：数据不丢失，支持高并发
- 语义搜索时选择合适的 embedding 模型：平衡成本和效果
- Context Schema 用于注入运行时信息：如 user_id、tenant_id 等，不需要放在 State 中
指南)
