5.2 Memory Store - 详细解读

一、概述

1.1 本节简介

本节是 LangChain Academy Module-5 的第二部分，主要内容是深入理解 LangGraph Memory Store（记忆存储系统）。这是构建具有长期记忆能力的 AI 应用的基础组件。

与 5.1 节相比，本节更加基础和聚焦：

5.1 构建了一个复杂的 Memory Agent（task_mAIstro），涉及多种记忆类型、Trustcall、条件路由等
5.2 专注于 Store 的基本概念和使用方法，构建一个简单但完整的记忆聊天机器人

1.2 学习目标

通过学习本节内容，你将掌握：

LangGraph Store 的核心概念：namespace、key、value
Store 的三个基本操作：put（保存）、get（获取）、search（搜索）
短期记忆与长期记忆的协作：Checkpointer + Store
在热路径中保存记忆："热路径"意味着在用户聊天时实时保存，而不是异步后台处理
跨会话记忆：如何让 AI 在不同对话会话中记住用户信息

1.3 实际应用场景

这种记忆系统适用于：

个性化客服机器人：记住用户的偏好和历史问题
个人助理应用：记住用户的习惯、日程安排
教育辅导系统：记住学生的学习进度和弱点
医疗健康应用：记住患者的病史和治疗偏好

二、记忆（Memory）的认知科学基础

2.1 什么是记忆？

根据认知科学研究，记忆是一种认知功能，允许人们存储、检索和使用信息来理解现在和未来。

在 AI 应用中，记忆系统模拟了人类记忆的某些特征：

存储：保存重要信息
检索：在需要时找到相关信息
应用：使用记忆来个性化响应

2.2 长期记忆的类型

在 AI 应用中，长期记忆可以分为几种类型：

语义记忆（Semantic Memory）

关于事实和知识的记忆：

用户的姓名、位置、职业
用户的兴趣和爱好
历史事件和事实

本节重点：我们将专注于构建语义记忆系统，保存关于用户的事实信息。

情节记忆（Episodic Memory）

关于特定事件和经历的记忆：

"上周我们讨论了项目 A"
"三次对话前，你提到了..."

程序性记忆（Procedural Memory）

关于如何做某事的记忆：

如何格式化输出
用户喜欢的交互方式

2.3 本节的记忆架构

┌─────────────────────────────────────────────────┐
│              Chatbot with Memory                 │
│                                                  │
│  ┌────────────────────────────────────────────┐ │
│  │  短期记忆（Within-Thread Memory）          │ │
│  │  使用：MemorySaver (Checkpointer)          │ │
│  │  作用：保存单次会话的对话历史              │ │
│  │  生命周期：会话期间                        │ │
│  └────────────────────────────────────────────┘ │
│                                                  │
│  ┌────────────────────────────────────────────┐ │
│  │  长期记忆（Across-Thread Memory）          │ │
│  │  使用：InMemoryStore (Store)               │ │
│  │  作用：跨会话保存用户信息                  │ │
│  │  生命周期：持久化（跨所有会话）            │ │
│  └────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────┘

三、LangGraph Store 基础

3.1 什么是 LangGraph Store？

LangGraph Memory Store 是一个用于在 LangGraph 中跨线程（会话）存储和检索信息的持久化键值存储系统。

核心特点：

跨线程持久化：数据在不同的对话会话中保持可访问
键值存储：简单而强大的数据组织方式
开源基类：可以实现不同的存储后端（内存、数据库等）
命名空间支持：分层组织数据

3.2 Store 的基本概念

LangGraph Store 使用三个核心概念来组织数据：

3.2.1 Namespace（命名空间）

定义：用元组表示的分层结构，类似于文件系统的目录。

示例：

python

("user_1", "memories")           # 用户 1 的记忆
("user_2", "memories")           # 用户 2 的记忆
("memory", "user_123")           # 另一种组织方式
("project", "alpha", "settings") # 多层命名空间

为什么使用元组？

元组是不可变的，适合作为键
天然支持层级结构
可以灵活组织数据

类比：

Namespace           类似于
("user_1", "memories")  →  /user_1/memories/

3.2.2 Key（键）

定义：在命名空间内标识特定对象的字符串，类似于文件名。

示例：

python

"user_memory"                    # 简单的键
"preference_settings"            # 描述性的键
str(uuid.uuid4())                # UUID 键（唯一标识）
# 例如："a754b8c5-e8b7-40ec-834b-c426a9a7c7cc"

最佳实践：

使用描述性的键名
对于集合类数据，使用 UUID
对于单一记录，使用固定的键名

3.2.3 Value（值）

定义：存储的实际数据，必须是字典类型。

示例：

python

{"food_preference": "I like pizza"}
{"memory": "User's name is Lance"}
{"task": "Buy milk", "status": "pending"}

重要约束：

值必须是 Python 字典（dict）
可以包含嵌套结构
通常包含字符串、数字、列表等可序列化的数据

3.3 Store 的数据模型图解

┌─────────────────────────────────────────────────────┐
│                   InMemoryStore                      │
│                                                      │
│  Namespace: ("user_1", "memories")                  │
│  ┌───────────────────────────────────────────────┐ │
│  │ Key: "a754..."                                │ │
│  │ Value: {"food_preference": "I like pizza"}    │ │
│  │ Created: 2024-11-04T22:48:16                  │ │
│  │ Updated: 2024-11-04T22:48:16                  │ │
│  └───────────────────────────────────────────────┘ │
│  ┌───────────────────────────────────────────────┐ │
│  │ Key: "b892..."                                │ │
│  │ Value: {"hobby": "biking"}                    │ │
│  │ Created: 2024-11-04T22:50:00                  │ │
│  │ Updated: 2024-11-04T22:50:00                  │ │
│  └───────────────────────────────────────────────┘ │
│                                                      │
│  Namespace: ("user_2", "memories")                  │
│  ┌───────────────────────────────────────────────┐ │
│  │ Key: "c123..."                                │ │
│  │ Value: {"name": "Alice"}                      │ │
│  │ Created: 2024-11-04T23:00:00                  │ │
│  │ Updated: 2024-11-04T23:00:00                  │ │
│  └───────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────┘

四、环境准备

4.1 安装依赖包

python

%%capture --no-stderr
%pip install -U langchain_openai langgraph langchain_core

依赖包说明：

langchain_openai：OpenAI 模型的 LangChain 集成
langgraph：构建有状态图的框架
langchain_core：LangChain 核心组件

4.2 配置环境变量

python

import os, getpass

def _set_env(var: str):
    if not os.environ.get(var):
        os.environ[var] = getpass.getpass(f"{var}: ")

# 配置 LangSmith（用于追踪）
_set_env("LANGSMITH_API_KEY")
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_PROJECT"] = "langchain-academy"

LangSmith 的作用：

追踪（Tracing）：记录每次 LLM 调用的详细信息
调试：查看输入输出、token 使用、延迟等
优化：分析性能瓶颈

五、Store 的基本操作

5.1 创建 Store 实例

python

import uuid
from langgraph.store.memory import InMemoryStore

# 创建内存存储实例
in_memory_store = InMemoryStore()

InMemoryStore：

在内存中保存数据
适合开发和测试
生产环境通常使用持久化存储（如数据库）

Python 知识点：

uuid：通用唯一识别码（Universally Unique Identifier）
用于生成全局唯一的 ID

5.2 保存数据：put() 方法

5.2.1 基本用法

python

# 定义命名空间
user_id = "1"
namespace_for_memory = (user_id, "memories")

# 生成唯一的键
key = str(uuid.uuid4())
# 例如：'a754b8c5-e8b7-40ec-834b-c426a9a7c7cc'

# 值必须是字典
value = {"food_preference": "I like pizza"}

# 保存到 store
in_memory_store.put(namespace_for_memory, key, value)

方法签名：

python

def put(
    namespace: tuple,      # 命名空间（元组）
    key: str,              # 键（字符串）
    value: dict            # 值（字典）
) -> None:

参数详解：

namespace：(user_id, "memories")
- 第一个元素：用户 ID（字符串）
- 第二个元素：数据类型（"memories"）
- 可以有更多层级
key：str(uuid.uuid4())
- 使用 UUID 确保唯一性
- 也可以使用自定义字符串
value：{"food_preference": "I like pizza"}
- 必须是字典
- 可以包含任意可序列化的数据

5.2.2 UUID 详解

python

import uuid

# 生成 UUID
unique_id = uuid.uuid4()
print(unique_id)
# 输出：UUID('a754b8c5-e8b7-40ec-834b-c426a9a7c7cc')

# 转换为字符串
key = str(unique_id)
print(key)
# 输出：'a754b8c5-e8b7-40ec-834b-c426a9a7c7cc'

为什么使用 UUID？

全局唯一：极低的碰撞概率（2^122）
无需中央协调：可以在分布式系统中独立生成
标准化：符合 RFC 4122 标准

UUID 的版本：

uuid.uuid1()：基于时间戳和 MAC 地址
uuid.uuid4()：随机生成（最常用）
uuid.uuid5()：基于命名空间和名称的哈希

5.3 搜索数据：search() 方法

5.3.1 基本用法

python

# 搜索命名空间中的所有数据
memories = in_memory_store.search(namespace_for_memory)

# 返回值是列表
print(type(memories))
# 输出：<class 'list'>

方法签名：

python

def search(
    namespace: tuple     # 命名空间
) -> list:               # 返回项目列表

5.3.2 查看搜索结果

python

# 查看第一个结果的所有信息
print(memories[0].dict())

输出：

python

{
    'value': {'food_preference': 'I like pizza'},
    'key': 'a754b8c5-e8b7-40ec-834b-c426a9a7c7cc',
    'namespace': ['1', 'memories'],
    'created_at': '2024-11-04T22:48:16.727572+00:00',
    'updated_at': '2024-11-04T22:48:16.727574+00:00'
}

返回对象的属性：

value：存储的数据（字典）
key：对象的键
namespace：命名空间（列表形式）
created_at：创建时间（ISO 8601 格式）
updated_at：最后更新时间

Python 知识点：

.dict() 方法：

Pydantic 模型的方法
将对象转换为字典
便于查看和序列化

5.3.3 访问键和值

python

# 获取键和值
print(memories[0].key, memories[0].value)

输出：

a754b8c5-e8b7-40ec-834b-c426a9a7c7cc {'food_preference': 'I like pizza'}

访问嵌套数据：

python

# 获取具体的偏好
food_pref = memories[0].value['food_preference']
print(food_pref)
# 输出：I like pizza

5.4 获取数据：get() 方法

5.4.1 基本用法

python

# 通过命名空间和键获取特定对象
memory = in_memory_store.get(namespace_for_memory, key)

# 查看结果
print(memory.dict())

输出：

python

{
    'value': {'food_preference': 'I like pizza'},
    'key': 'a754b8c5-e8b7-40ec-834b-c426a9a7c7cc',
    'namespace': ['1', 'memories'],
    'created_at': '2024-11-04T22:48:16.727572+00:00',
    'updated_at': '2024-11-04T22:48:16.727574+00:00'
}

方法签名：

python

def get(
    namespace: tuple,    # 命名空间
    key: str            # 键
) -> Item:              # 返回单个项目

5.4.2 search() vs get()

特性	search()	get()
参数	仅需要 namespace	需要 namespace 和 key
返回类型	列表（可能为空）	单个对象或 None
使用场景	获取某类所有数据	获取特定的一条数据
性能	可能较慢（扫描）	更快（直接查找）

示例对比：

python

# search - 获取用户的所有记忆
all_memories = store.search(("user_1", "memories"))
# 返回：[memory1, memory2, memory3, ...]

# get - 获取特定的一条记忆
specific_memory = store.get(("user_1", "memories"), "key_123")
# 返回：memory1 或 None

5.5 Store 的三个操作总结

python

# 1. put - 保存数据
store.put(
    namespace=("user_1", "memories"),
    key="pref_001",
    value={"food": "pizza"}
)

# 2. search - 搜索命名空间
memories = store.search(("user_1", "memories"))
# 返回该命名空间下的所有项目

# 3. get - 获取特定项目
memory = store.get(("user_1", "memories"), "pref_001")
# 返回特定的项目

记忆口诀：

put：保存新数据或覆盖现有数据
search：搜索命名空间，返回列表
get：精确获取，返回单个对象

六、构建记忆聊天机器人

现在让我们使用 Store 构建一个完整的聊天机器人，它具有：

短期记忆：记住当前会话的对话历史
长期记忆：记住用户的个人信息，跨会话使用

6.1 初始化模型

python

from langchain_openai import ChatOpenAI

# 配置 OpenAI API Key
_set_env("OPENAI_API_KEY")

# 初始化模型
model = ChatOpenAI(model="gpt-4o", temperature=0)

参数说明：

model="gpt-4o"：使用 GPT-4o 模型
temperature=0：确定性输出（每次输入相同，输出也相同）

6.2 定义系统提示词

6.2.1 聊天机器人提示词

python

MODEL_SYSTEM_MESSAGE = """You are a helpful assistant with memory that provides information about the user.
If you have memory for this user, use it to personalize your responses.
Here is the memory (it may be empty): {memory}"""

提示词设计要点：

角色定义："You are a helpful assistant with memory"
- 明确告诉模型它有记忆能力
使用指令："use it to personalize your responses"
- 指示如何使用记忆
上下文占位符：{memory}
- 将被实际的记忆内容替换

6.2.2 创建记忆的提示词

python

CREATE_MEMORY_INSTRUCTION = """You are collecting information about the user to personalize your responses.

CURRENT USER INFORMATION:
{memory}

INSTRUCTIONS:
1. Review the chat history below carefully
2. Identify new information about the user, such as:
   - Personal details (name, location)
   - Preferences (likes, dislikes)
   - Interests and hobbies
   - Past experiences
   - Goals or future plans
3. Merge any new information with existing memory
4. Format the memory as a clear, bulleted list
5. If new information conflicts with existing memory, keep the most recent version

Remember: Only include factual information directly stated by the user. Do not make assumptions or inferences.

Based on the chat history below, please update the user information:"""

提示词结构分析：

任务说明：
- "You are collecting information..."
- 明确任务目标
当前状态：
- "CURRENT USER INFORMATION: {memory}"
- 显示现有记忆
详细指令：
- 5 条明确的步骤
- 告诉模型要提取什么信息
约束条件：
- "Only include factual information"
- 避免模型推测或臆断
输出格式：
- "Format as a clear, bulleted list"
- 确保输出结构化

提示词工程技巧：

✅ 使用大写标题（CURRENT USER INFORMATION）突出重要部分
✅ 使用编号列表使指令清晰
✅ 提供具体示例（"name, location"）
✅ 设置约束条件（"Only include factual information"）
✅ 明确输出格式（"bulleted list"）

6.3 定义节点函数

6.3.1 call_model 节点

这是主节点，负责与用户交互：

python

from langgraph.graph import MessagesState
from langchain_core.runnables.config import RunnableConfig
from langgraph.store.base import BaseStore
from langchain_core.messages import SystemMessage

def call_model(state: MessagesState, config: RunnableConfig, store: BaseStore):
    """从 store 加载记忆并用它来个性化聊天机器人的响应"""

    # 从 config 获取用户 ID
    user_id = config["configurable"]["user_id"]

    # 从 store 检索记忆
    namespace = ("memory", user_id)
    key = "user_memory"
    existing_memory = store.get(namespace, key)

    # 提取实际的记忆内容
    if existing_memory:
        # Value 是一个包含 memory 键的字典
        existing_memory_content = existing_memory.value.get('memory')
    else:
        existing_memory_content = "No existing memory found."

    # 在系统提示中格式化记忆
    system_msg = MODEL_SYSTEM_MESSAGE.format(memory=existing_memory_content)

    # 使用记忆和聊天历史生成响应
    response = model.invoke([SystemMessage(content=system_msg)] + state["messages"])

    return {"messages": response}

代码逐行解析：

第 1-2 行：函数签名

python

def call_model(state: MessagesState, config: RunnableConfig, store: BaseStore):

state：包含消息列表的状态对象
config：运行时配置（包含 user_id、thread_id）
store：长期记忆存储

第 5 行：获取用户 ID

python

user_id = config["configurable"]["user_id"]

从配置中提取用户标识
用于查找该用户的记忆

第 7-10 行：检索记忆

python

namespace = ("memory", user_id)
key = "user_memory"
existing_memory = store.get(namespace, key)

构建命名空间：("memory", "1")
使用固定的键："user_memory"
调用 get() 方法获取记忆

第 12-16 行：提取记忆内容

python

if existing_memory:
    existing_memory_content = existing_memory.value.get('memory')
else:
    existing_memory_content = "No existing memory found."

如果记忆存在，提取 value['memory']
如果不存在，使用默认消息

为什么使用 .get('memory')？

python

# store 中保存的格式：
{
    "memory": "- User's name is Lance\n- Likes biking"
}

# 所以需要提取 'memory' 键的值
existing_memory.value.get('memory')

第 18-19 行：格式化系统消息

python

system_msg = MODEL_SYSTEM_MESSAGE.format(memory=existing_memory_content)

将记忆内容插入提示词模板
例如："Here is the memory: - User's name is Lance..."

第 21-22 行：生成响应

python

response = model.invoke([SystemMessage(content=system_msg)] + state["messages"])

构建消息列表：[SystemMessage, HumanMessage, AIMessage, ...]
调用模型生成响应

第 24 行：返回状态更新

python

return {"messages": response}

返回包含新消息的字典
LangGraph 会将其合并到状态中

6.3.2 write_memory 节点

这个节点负责反思对话并保存记忆：

python

from langchain_core.messages import SystemMessage

def write_memory(state: MessagesState, config: RunnableConfig, store: BaseStore):
    """反思聊天历史并保存记忆到 store"""

    # 获取用户 ID
    user_id = config["configurable"]["user_id"]

    # 从 store 检索现有记忆
    namespace = ("memory", user_id)
    existing_memory = store.get(namespace, "user_memory")

    # 提取记忆内容
    if existing_memory:
        existing_memory_content = existing_memory.value.get('memory')
    else:
        existing_memory_content = "No existing memory found."

    # 在系统提示中格式化记忆
    system_msg = CREATE_MEMORY_INSTRUCTION.format(memory=existing_memory_content)

    # 让模型生成更新后的记忆
    new_memory = model.invoke([SystemMessage(content=system_msg)] + state['messages'])

    # 覆盖 store 中的现有记忆
    key = "user_memory"

    # 将值写为包含 memory 键的字典
    store.put(namespace, key, {"memory": new_memory.content})

代码逐行解析：

第 7-11 行：获取现有记忆

python

user_id = config["configurable"]["user_id"]
namespace = ("memory", user_id)
existing_memory = store.get(namespace, "user_memory")

与 call_model 类似，获取用户的现有记忆

第 13-17 行：提取记忆内容

python

if existing_memory:
    existing_memory_content = existing_memory.value.get('memory')
else:
    existing_memory_content = "No existing memory found."

提供现有记忆给模型作为上下文

第 19-23 行：生成新记忆

python

system_msg = CREATE_MEMORY_INSTRUCTION.format(memory=existing_memory_content)
new_memory = model.invoke([SystemMessage(content=system_msg)] + state['messages'])

使用 CREATE_MEMORY_INSTRUCTION 提示词
让模型分析对话并生成更新的记忆
new_memory.content 是模型生成的文本

第 25-29 行：保存到 store

python

key = "user_memory"
store.put(namespace, key, {"memory": new_memory.content})

使用固定的键："user_memory"
保存格式：{"memory": "实际的记忆内容"}
如果已存在，会被覆盖

重要细节：

为什么覆盖而不是追加？

因为模型已经在 CREATE_MEMORY_INSTRUCTION 中被要求合并新旧信息
模型生成的 new_memory 已经包含了完整的更新记忆
简化了数据管理逻辑

6.4 构建状态图

python

from IPython.display import Image, display
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, START, END

# 定义图
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_node("write_memory", write_memory)

# 定义边
builder.add_edge(START, "call_model")
builder.add_edge("call_model", "write_memory")
builder.add_edge("write_memory", END)

# 长期记忆（跨线程）的 Store
across_thread_memory = InMemoryStore()

# 短期记忆（线程内）的 Checkpointer
within_thread_memory = MemorySaver()

# 编译图
graph = builder.compile(
    checkpointer=within_thread_memory,
    store=across_thread_memory
)

# 可视化
display(Image(graph.get_graph(xray=1).draw_mermaid_png()))

图结构：

START → call_model → write_memory → END

流程说明：

START：开始执行
call_model：
- 加载用户的长期记忆
- 加载对话历史（短期记忆）
- 生成个性化响应
write_memory：
- 反思整个对话
- 提取新信息
- 更新长期记忆
END：结束执行

为什么是线性流程？

与 5.1 节的复杂路由不同，这里是简单的线性流程：

5.1：Agent 决定是否保存记忆（条件路由）
5.2：每次对话后都保存记忆（固定流程）

优缺点对比：

特性	5.2 线性流程	5.1 条件路由
简单性	✅ 更简单	❌ 更复杂
灵活性	❌ 固定流程	✅ 智能决策
资源使用	❌ 每次都调用 LLM	✅ 按需调用
适用场景	简单应用	复杂 Agent

6.5 编译配置

python

graph = builder.compile(
    checkpointer=within_thread_memory,
    store=across_thread_memory
)

参数说明：

checkpointer：within_thread_memory（MemorySaver）
- 保存短期记忆（对话历史）
- 在单个 thread 内持久化
- 支持中断和恢复
store：across_thread_memory（InMemoryStore）
- 保存长期记忆（用户信息）
- 跨所有 thread 共享
- 持久化用户数据

两种记忆的配合：

用户第一次对话（Thread 1）：
┌─────────────────────────────────────┐
│ Checkpointer (Thread 1)             │
│ - "Hi, my name is Lance"            │
│ - "Hello, Lance!"                   │
└─────────────────────────────────────┘
         │
         ↓ write_memory 提取信息
┌─────────────────────────────────────┐
│ Store (User 1)                      │
│ - Name: Lance                       │
└─────────────────────────────────────┘

用户第二次对话（Thread 2）：
┌─────────────────────────────────────┐
│ Checkpointer (Thread 2)             │
│ - "Where should I go biking?"       │
└─────────────────────────────────────┘
         ↓ call_model 加载记忆
┌─────────────────────────────────────┐
│ Store (User 1)                      │
│ - Name: Lance                       │
│ - Likes biking in San Francisco     │
└─────────────────────────────────────┘
         ↓
响应："Hi Lance! For biking in SF..."

七、实战演示

7.1 配置和初始化

python

# 提供 thread ID（短期记忆）和 user ID（长期记忆）
config = {
    "configurable": {
        "thread_id": "1",      # 会话标识
        "user_id": "1"         # 用户标识
    }
}

配置说明：

thread_id：标识一次对话会话
user_id：标识用户，用于命名空间

7.2 第一次交互：介绍自己

python

# 用户输入
input_messages = [HumanMessage(content="Hi, my name is Lance")]

# 运行图
for chunk in graph.stream(
    {"messages": input_messages},
    config,
    stream_mode="values"
):
    chunk["messages"][-1].pretty_print()

输出：

================================ Human Message =================================
Hi, my name is Lance

================================== Ai Message ==================================
Hello, Lance! It's nice to meet you. How can I assist you today?

执行流程分析：

call_model 节点：
- 检查 store 中的记忆：没有找到
- 使用默认消息："No existing memory found."
- 生成响应："Hello, Lance!..."
write_memory 节点：
- 分析对话历史
- 识别新信息："User's name is Lance"
- 保存到 store

7.3 第二次交互：分享兴趣

python

# 用户输入
input_messages = [HumanMessage(content="I like to bike around San Francisco")]

# 运行图
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

输出：

================================ Human Message =================================
I like to bike around San Francisco

================================== Ai Message ==================================
That sounds like a great way to explore the city, Lance! San Francisco has some
beautiful routes and views. Do you have a favorite trail or area you like to bike in?

注意：

AI 使用了用户的名字："Lance"
说明它已经从记忆中读取了用户信息

7.4 查看短期记忆（对话历史）

python

# 获取线程状态
thread = {"configurable": {"thread_id": "1"}}
state = graph.get_state(thread).values

# 打印所有消息
for m in state["messages"]:
    m.pretty_print()

输出：

================================ Human Message =================================
Hi, my name is Lance

================================== Ai Message ==================================
Hello, Lance! It's nice to meet you. How can I assist you today?

================================ Human Message =================================
I like to bike around San Francisco

================================== Ai Message ==================================
That sounds like a great way to explore the city, Lance! San Francisco has some
beautiful routes and views. Do you have a favorite trail or area you like to bike in?

分析：

所有的对话历史都保存在 Checkpointer 中
通过 thread_id 可以访问
这是短期记忆，仅在当前会话中

7.5 查看长期记忆（用户信息）

python

# 定义命名空间
user_id = "1"
namespace = ("memory", user_id)

# 获取记忆
existing_memory = across_thread_memory.get(namespace, "user_memory")

# 查看完整信息
print(existing_memory.dict())

输出：

python

{
    'value': {
        'memory': "**Updated User Information:**\n- User's name is Lance.\n- Likes to bike around San Francisco."
    },
    'key': 'user_memory',
    'namespace': ['memory', '1'],
    'created_at': '2024-11-05T00:12:17.383918+00:00',
    'updated_at': '2024-11-05T00:12:25.469528+00:00'
}

记忆内容解析：

**Updated User Information:**
- User's name is Lance.
- Likes to bike around San Francisco.

关键观察：

模型提取了两条信息
格式化为清晰的项目符号列表
包含时间戳（创建和更新时间）

7.6 跨会话测试：新的对话

现在让我们创建一个新的会话（不同的 thread_id），但使用相同的用户（相同的 user_id）：

python

# 新的 thread_id，但相同的 user_id
config = {
    "configurable": {
        "thread_id": "2",      # 新会话
        "user_id": "1"         # 相同用户
    }
}

# 用户输入
input_messages = [HumanMessage(
    content="Hi! Where would you recommend that I go biking?"
)]

# 运行图
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

输出：

================================ Human Message =================================
Hi! Where would you recommend that I go biking?

================================== Ai Message ==================================
Hi Lance! Since you enjoy biking around San Francisco, there are some fantastic
routes you might love. Here are a few recommendations:

1. **Golden Gate Park**: This is a classic choice with plenty of trails and
   beautiful scenery. You can explore the park's many attractions, like the
   Conservatory of Flowers and the Japanese Tea Garden.

2. **The Embarcadero**: A ride along the Embarcadero offers stunning views of
   the Bay Bridge and the waterfront. It's a great way to experience the city's
   vibrant atmosphere.

3. **Marin Headlands**: If you're up for a bit of a challenge, biking across
   the Golden Gate Bridge to the Marin Headlands offers breathtaking views of
   the city and the Pacific Ocean.

4. **Presidio**: This area has a network of trails with varying difficulty
   levels, and you can enjoy views of the Golden Gate Bridge and the bay.

5. **Twin Peaks**: For a more challenging ride, head up to Twin Peaks. The
   climb is worth it for the panoramic views of the city.

Let me know if you want more details on any of these routes!

长期记忆的威力：

记住了名字："Hi Lance!"
- 虽然这是新会话，但 AI 记得用户的名字
记住了兴趣："Since you enjoy biking around San Francisco"
- AI 知道用户喜欢在旧金山骑车
个性化建议：
- 提供了具体的旧金山骑行路线
- 针对用户的兴趣定制响应

7.7 继续对话：进一步个性化

python

# 用户输入
input_messages = [HumanMessage(
    content="Great, are there any bakeries nearby that I can check out? I like a croissant after biking."
)]

# 运行图
for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

输出：

================================ Human Message =================================
Great, are there any bakeries nearby that I can check out? I like a croissant after biking.

================================== Ai Message ==================================
Absolutely, Lance! Here are a few bakeries in San Francisco where you can enjoy
a delicious croissant after your ride:

1. **Tartine Bakery**: Located in the Mission District, Tartine is famous for
   its pastries, and their croissants are a must-try.

2. **Arsicault Bakery**: This bakery in the Richmond District has been praised
   for its buttery, flaky croissants. It's a bit of a detour, but worth it!

3. **b. Patisserie**: Situated in Lower Pacific Heights, b. Patisserie offers
   a variety of pastries, and their croissants are particularly popular.

4. **Le Marais Bakery**: With locations in the Marina and Castro, Le Marais
   offers a charming French bakery experience with excellent croissants.

5. **Neighbor Bakehouse**: Located in the Dogpatch, this bakery is known for
   its creative pastries, including some fantastic croissants.

These spots should provide a delightful treat after your biking adventures.
Enjoy your ride and your croissant!

新信息会被保存：

用户喜欢骑车后吃可颂
这条信息会被添加到长期记忆中
在未来的对话中可以使用

记忆更新流程：

对话前的记忆：
- User's name is Lance
- Likes to bike around San Francisco

对话后的记忆：
- User's name is Lance
- Likes to bike around San Francisco
- Enjoys eating croissants after biking  ← 新增

八、在热路径中保存记忆

8.1 什么是"热路径"？

热路径（Hot Path）：在用户等待响应的过程中执行的操作。

对比：

特性	热路径（Hot Path）	冷路径（Cold Path）
执行时机	用户等待响应时	后台异步处理
用户体验	可能增加延迟	不影响响应速度
实时性	立即可用	有延迟
实现复杂度	简单	需要队列/任务调度

本节的实现：

用户发送消息
    ↓
call_model（生成响应）  ← 用户等待
    ↓
write_memory（保存记忆） ← 用户等待
    ↓
返回响应给用户

8.2 热路径的优缺点

优点：

✅ 实时更新：记忆立即保存，下次对话可用
✅ 实现简单：不需要额外的后台任务系统
✅ 数据一致性：避免并发问题

缺点：

❌ 增加延迟：用户需要等待记忆保存
❌ 资源消耗：每次对话都调用 LLM 提取记忆
❌ 可能失败：如果保存失败，用户会感知到

8.3 冷路径的替代方案

如果对响应速度要求高，可以考虑冷路径：

python

from langgraph.graph import StateGraph, START, END

# 方案 1：分离响应和记忆保存
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_node("write_memory", write_memory)

# 响应后立即结束
builder.add_edge(START, "call_model")
builder.add_edge("call_model", END)

# 记忆保存作为后台任务（需要额外实现）
# background_task_queue.add(write_memory_task)

实现方式：

使用消息队列（如 RabbitMQ、Celery）
使用 LangGraph 的 Background Execution
定期批处理记忆更新

8.4 选择策略

场景	推荐方案
对话频率低	热路径
响应速度关键	冷路径
记忆必须实时	热路径
大规模用户	冷路径 + 缓存

九、调试和追踪

9.1 使用 LangSmith 追踪

LangSmith 提供了强大的追踪功能，可以查看每次执行的详细信息。

查看追踪：

每次运行后，LangSmith 会自动记录
可以在 LangSmith UI 中查看
示例追踪：https://smith.langchain.com/public/10268d64-82ff-434e-ac02-4afa5cc15432/r

追踪内容：

输入输出：
- 用户消息
- 系统提示（包含记忆）
- AI 响应
中间步骤：
- call_model 节点的执行
- write_memory 节点的执行
性能指标：
- Token 使用量
- 延迟时间
- 成本估算

9.2 查看记忆加载过程

从追踪中，我们可以看到记忆是如何被加载和使用的：

System Message 示例：

You are a helpful assistant with memory that provides information about the user.
If you have memory for this user, use it to personalize your responses.
Here is the memory (it may be empty):

**Updated User Information:**
- User's name is Lance.
- Likes to bike around San Francisco.

分析：

记忆被成功加载并插入到系统提示中
模型可以看到完整的用户信息
这解释了为什么响应能够个性化

9.3 调试常见问题

问题 1：记忆没有被使用

症状：AI 的响应没有体现用户的个人信息

检查清单：

✅ 确认 user_id 配置正确

✅ 检查 store 中是否有记忆

python

memory = store.get(("memory", user_id), "user_memory")
print(memory)

✅ 查看 LangSmith 追踪，确认系统提示中包含记忆

问题 2：记忆没有被保存

症状：下次对话时，AI 不记得之前的信息

检查清单：

✅ 确认 write_memory 节点被执行

✅ 检查 store 的内容

python

memories = store.search(("memory", user_id))
for m in memories:
    print(m.value)

✅ 查看是否有异常抛出

问题 3：记忆格式错误

症状：记忆存在但无法正确解析

检查清单：

✅ 确认 value 是字典格式

✅ 确认包含 'memory' 键

python

# 正确格式
{"memory": "- User's name is Lance"}

# 错误格式
"- User's name is Lance"  # 不是字典
{"info": "..."}           # 缺少 'memory' 键

十、Python 知识点总结

10.1 UUID（通用唯一识别码）

python

import uuid

# 生成 UUID
id1 = uuid.uuid4()
print(id1)
# 输出：UUID('a754b8c5-e8b7-40ec-834b-c426a9a7c7cc')

# 转换为字符串
id_str = str(id1)
print(id_str)
# 输出：'a754b8c5-e8b7-40ec-834b-c426a9a7c7cc'

# 比较
id2 = uuid.uuid4()
print(id1 == id2)  # False（几乎不可能相同）

UUID 的特点：

全局唯一（概率意义上）
无需中央协调
128 位（32 个十六进制字符）
标准格式：8-4-4-4-12

10.2 字典的 get() 方法

python

# 基本用法
data = {"name": "Alice", "age": 30}

# 获取存在的键
print(data.get("name"))  # 'Alice'

# 获取不存在的键
print(data.get("email"))  # None

# 提供默认值
print(data.get("email", "no-email@example.com"))
# 输出：'no-email@example.com'

为什么使用 get() 而不是 []？

python

# 使用 []
data = {"name": "Alice"}
email = data["email"]  # KeyError!

# 使用 get()
email = data.get("email", "unknown")  # 安全，返回 'unknown'

10.3 f-string 格式化

python

name = "Alice"
age = 30

# 基本用法
print(f"Name: {name}, Age: {age}")
# 输出：Name: Alice, Age: 30

# 表达式
print(f"Next year: {age + 1}")
# 输出：Next year: 31

# 格式化
pi = 3.14159
print(f"Pi: {pi:.2f}")
# 输出：Pi: 3.14

# 对齐
print(f"{name:>10}")  # 右对齐，宽度 10
# 输出：'     Alice'

10.4 字符串的 format() 方法

python

# 基本用法
template = "Hello, {name}!"
result = template.format(name="Alice")
print(result)
# 输出：Hello, Alice!

# 多个占位符
template = "{greeting}, {name}! You are {age} years old."
result = template.format(greeting="Hi", name="Bob", age=25)
print(result)
# 输出：Hi, Bob! You are 25 years old.

# 位置参数
template = "{0} {1} {0}"
result = template.format("Hello", "World")
print(result)
# 输出：Hello World Hello

在本节中的应用：

python

MODEL_SYSTEM_MESSAGE = """Here is the memory: {memory}"""
system_msg = MODEL_SYSTEM_MESSAGE.format(memory="User likes pizza")
# 结果：Here is the memory: User likes pizza

十一、LangGraph 知识点总结

11.1 Checkpointer vs Store

特性	Checkpointer	Store
作用域	单个 thread（会话）	跨 thread（用户）
生命周期	会话期间	持久化
存储内容	状态快照（消息列表等）	结构化数据（键值对）
访问方式	自动加载到状态	需要显式调用 get/search
主要用途	对话历史、中断恢复	用户信息、长期记忆
标识符	thread_id	namespace + key

11.2 Config 的使用

python

# 定义配置
config = {
    "configurable": {
        "thread_id": "session_123",
        "user_id": "user_456"
    }
}

# 在节点中访问
def my_node(state, config, store):
    thread_id = config["configurable"]["thread_id"]
    user_id = config["configurable"]["user_id"]
    # 使用这些 ID...

Config 的作用：

传递运行时参数
标识会话和用户
可以包含其他自定义配置

11.3 状态更新机制

python

def my_node(state: MessagesState):
    # 状态包含 messages 列表
    current_messages = state["messages"]

    # 生成新消息
    new_message = AIMessage(content="Hello!")

    # 返回状态更新
    return {"messages": [new_message]}
    # 或者返回多条消息
    return {"messages": [message1, message2]}

LangGraph 会自动合并：

python

# 原始状态
state = {"messages": [msg1, msg2]}

# 节点返回
return {"messages": [msg3]}

# 合并后的状态
state = {"messages": [msg1, msg2, msg3]}

11.4 流式执行

python

# stream() - 流式执行
for chunk in graph.stream(input_data, config, stream_mode="values"):
    # 每次状态更新时产生一个 chunk
    latest_message = chunk["messages"][-1]
    latest_message.pretty_print()

# invoke() - 一次性执行
result = graph.invoke(input_data, config)
# 返回最终状态

stream_mode 选项：

"values"：每次产生完整状态
"updates"：只产生更新部分
"messages"：只产生新消息

十二、架构设计分析

12.1 为什么每次都保存记忆？

本节的设计是每次对话后都调用 write_memory：

call_model → write_memory → END

优点：

✅ 简单直接
✅ 不会遗漏信息
✅ 记忆始终最新

缺点：

❌ 浪费资源（有时对话没有新信息）
❌ 增加延迟
❌ 增加成本（每次都调用 LLM）

改进方案：参考 5.1 节，使用条件路由：

python

def should_save_memory(state):
    # 检查是否有新信息
    if has_new_information(state):
        return "write_memory"
    else:
        return END

builder.add_conditional_edges("call_model", should_save_memory)

12.2 为什么使用固定的 key？

python

key = "user_memory"
store.put(namespace, key, {"memory": new_memory.content})

使用固定 key 的原因：

每个用户只有一个记忆文档
每次更新都覆盖整个文档
简化了检索逻辑（不需要搜索）

对比 UUID key：

特性	固定 key	UUID key
数量	每个用户一个	每个用户多个
更新方式	覆盖	追加
检索	get()	search()
适用场景	单一记忆文档	记忆集合

在 5.1 节中：

使用 UUID key 保存多个 ToDo 项
每个 ToDo 是独立的文档
可以单独更新每个 ToDo

12.3 记忆合并策略

本节使用 LLM 来合并新旧记忆：

python

CREATE_MEMORY_INSTRUCTION = """
...
3. Merge any new information with existing memory
4. Format the memory as a clear, bulleted list
5. If new information conflicts with existing memory, keep the most recent version
...
"""

优点：

LLM 可以理解语义
自动解决冲突
格式化输出

缺点：

依赖 LLM 能力
可能出错或遗漏
成本较高

替代方案：使用 Trustcall（5.1 节）：

结构化更新
JSON Patch 精确控制
更可靠

十三、实践建议

13.1 命名空间设计原则

推荐模式：

python

# 模式 1：类型 + 用户 ID
("memory", "user_123")
("preferences", "user_123")
("history", "user_123")

# 模式 2：用户 ID + 类型
("user_123", "memory")
("user_123", "preferences")
("user_123", "history")

# 模式 3：多层结构
("organization", "team_A", "user_123", "memory")

选择建议：

第一层通常是最常用的查询维度
考虑未来的扩展性
保持一致性

13.2 记忆内容格式

推荐格式：

python

# 项目符号列表（易读）
{
    "memory": """
    - User's name is Lance
    - Lives in San Francisco
    - Likes biking and croissants
    """
}

# JSON 结构（易解析）
{
    "profile": {
        "name": "Lance",
        "location": "San Francisco",
        "interests": ["biking", "croissants"]
    }
}

# 混合格式
{
    "summary": "Lance from SF, likes biking",
    "details": {
        "name": "Lance",
        "location": "San Francisco"
    }
}

选择建议：

简单应用：项目符号列表
复杂应用：JSON 结构
需要搜索：考虑添加标签和元数据

13.3 提示词优化

创建记忆的提示词关键要素：

明确任务：

You are collecting information about the user...

显示现有记忆：
```
CURRENT USER INFORMATION:
{memory}
```
1
2

列出要提取的信息类型：

- Personal details (name, location)
- Preferences (likes, dislikes)
...

处理冲突：

If new information conflicts with existing memory, keep the most recent version

设置约束：

Only include factual information directly stated by the user.

13.4 错误处理

python

def call_model(state, config, store):
    try:
        user_id = config["configurable"]["user_id"]
        namespace = ("memory", user_id)

        # 安全地获取记忆
        existing_memory = store.get(namespace, "user_memory")
        if existing_memory:
            memory_content = existing_memory.value.get('memory', '')
        else:
            memory_content = "No existing memory found."

        # 生成响应
        system_msg = MODEL_SYSTEM_MESSAGE.format(memory=memory_content)
        response = model.invoke([SystemMessage(content=system_msg)] + state["messages"])

        return {"messages": response}

    except KeyError as e:
        # 配置错误
        print(f"Config error: {e}")
        return {"messages": [AIMessage(content="Sorry, configuration error.")]}

    except Exception as e:
        # 其他错误
        print(f"Unexpected error: {e}")
        return {"messages": [AIMessage(content="Sorry, an error occurred.")]}

十四、扩展思考

14.1 与 5.1 节的对比

特性	5.2 Memory Store	5.1 Memory Agent
复杂度	简单	复杂
记忆类型	单一（用户信息）	多种（Profile、ToDo、Instructions）
更新策略	每次都更新	条件更新
数据结构	自由文本	结构化（Pydantic）
工具	无	Trustcall、UpdateMemory
路由	线性	条件路由
适用场景	简单聊天机器人	复杂 Agent 系统

学习路径建议：

先掌握 5.2 的基础概念（Store 的使用）
再学习 5.1 的高级技巧（Trustcall、条件路由）

14.2 生产环境考虑

14.2.1 持久化存储

开发环境：

python

store = InMemoryStore()  # 数据在内存中，重启丢失

生产环境（示例）：

python

# 使用 PostgreSQL
from langgraph.store.postgres import PostgresStore

store = PostgresStore(
    connection_string="postgresql://user:pass@localhost/dbname"
)

# 或使用 Redis
from langgraph.store.redis import RedisStore

store = RedisStore(
    redis_url="redis://localhost:6379"
)

14.2.2 性能优化

缓存：

python

from functools import lru_cache

@lru_cache(maxsize=100)
def get_user_memory(user_id):
    namespace = ("memory", user_id)
    memory = store.get(namespace, "user_memory")
    return memory.value if memory else None

批量操作：

python

# 批量获取多个用户的记忆
user_ids = ["user_1", "user_2", "user_3"]
namespaces = [("memory", uid) for uid in user_ids]

# 使用批量 API（如果支持）
memories = store.batch_get(namespaces, "user_memory")

14.2.3 安全性

数据加密：

python

import json
from cryptography.fernet import Fernet

# 生成密钥
key = Fernet.generate_key()
cipher = Fernet(key)

# 加密保存
def save_encrypted_memory(store, namespace, key, data):
    encrypted_data = cipher.encrypt(json.dumps(data).encode())
    store.put(namespace, key, {"encrypted": encrypted_data.decode()})

# 解密读取
def load_encrypted_memory(store, namespace, key):
    item = store.get(namespace, key)
    if item:
        encrypted_data = item.value["encrypted"].encode()
        decrypted_data = cipher.decrypt(encrypted_data)
        return json.loads(decrypted_data)
    return None

访问控制：

python

def call_model(state, config, store):
    user_id = config["configurable"]["user_id"]

    # 验证用户权限
    if not has_permission(user_id, "read_memory"):
        return {"messages": [AIMessage(content="Access denied.")]}

    # 继续处理...

十五、总结

15.1 核心要点

LangGraph Store 是跨线程（会话）持久化存储的基础
- 使用命名空间、键、值组织数据
- 提供 put、get、search 三个基本操作
短期记忆 + 长期记忆 的架构
- Checkpointer：保存对话历史（短期）
- Store：保存用户信息（长期）
- 两者配合实现完整的记忆系统
在热路径中保存记忆
- 实时更新，立即可用
- 简单但可能影响性能
提示词工程 是记忆质量的关键
- 明确指令
- 显示现有记忆
- 处理冲突

15.2 学到的技能

Python 技能：

UUID 的生成和使用
字典的 get() 方法
f-string 和 format() 格式化
错误处理

LangGraph 技能：

Store 的基本操作
Checkpointer 的使用
配置（Config）的传递
状态更新机制
流式执行

系统设计：

记忆系统架构
命名空间设计
热路径 vs 冷路径
性能优化策略

15.3 下一步

继续学习 Module-5 的其他部分：

5.3 Memory Schema - Profile：
- 更结构化的用户资料管理
- 使用 Pydantic 定义模式
5.4 Memory Schema - Collection：
- 管理记忆集合（如多个兴趣、多个偏好）
- 使用 Trustcall 更新
高级主题：
- 记忆的搜索和检索
- 记忆的优先级和重要性
- 记忆的遗忘机制

十六、附录

A.1 完整代码清单

python

# 导入依赖
from IPython.display import Image, display
from langgraph.checkpoint.memory import MemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.store.base import BaseStore
from langgraph.store.memory import InMemoryStore
from langchain_core.messages import HumanMessage, SystemMessage
from langchain_core.runnables.config import RunnableConfig
from langchain_openai import ChatOpenAI

# 初始化模型
model = ChatOpenAI(model="gpt-4o", temperature=0)

# 系统提示词
MODEL_SYSTEM_MESSAGE = """You are a helpful assistant with memory that provides information about the user.
If you have memory for this user, use it to personalize your responses.
Here is the memory (it may be empty): {memory}"""

CREATE_MEMORY_INSTRUCTION = """You are collecting information about the user to personalize your responses.

CURRENT USER INFORMATION:
{memory}

INSTRUCTIONS:
1. Review the chat history below carefully
2. Identify new information about the user, such as:
   - Personal details (name, location)
   - Preferences (likes, dislikes)
   - Interests and hobbies
   - Past experiences
   - Goals or future plans
3. Merge any new information with existing memory
4. Format the memory as a clear, bulleted list
5. If new information conflicts with existing memory, keep the most recent version

Remember: Only include factual information directly stated by the user. Do not make assumptions or inferences.

Based on the chat history below, please update the user information:"""

# 定义节点
def call_model(state: MessagesState, config: RunnableConfig, store: BaseStore):
    user_id = config["configurable"]["user_id"]
    namespace = ("memory", user_id)
    key = "user_memory"
    existing_memory = store.get(namespace, key)

    if existing_memory:
        existing_memory_content = existing_memory.value.get('memory')
    else:
        existing_memory_content = "No existing memory found."

    system_msg = MODEL_SYSTEM_MESSAGE.format(memory=existing_memory_content)
    response = model.invoke([SystemMessage(content=system_msg)] + state["messages"])
    return {"messages": response}

def write_memory(state: MessagesState, config: RunnableConfig, store: BaseStore):
    user_id = config["configurable"]["user_id"]
    namespace = ("memory", user_id)
    existing_memory = store.get(namespace, "user_memory")

    if existing_memory:
        existing_memory_content = existing_memory.value.get('memory')
    else:
        existing_memory_content = "No existing memory found."

    system_msg = CREATE_MEMORY_INSTRUCTION.format(memory=existing_memory_content)
    new_memory = model.invoke([SystemMessage(content=system_msg)] + state['messages'])
    key = "user_memory"
    store.put(namespace, key, {"memory": new_memory.content})

# 构建图
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)
builder.add_node("write_memory", write_memory)
builder.add_edge(START, "call_model")
builder.add_edge("call_model", "write_memory")
builder.add_edge("write_memory", END)

# 编译图
across_thread_memory = InMemoryStore()
within_thread_memory = MemorySaver()
graph = builder.compile(
    checkpointer=within_thread_memory,
    store=across_thread_memory
)

# 使用示例
config = {"configurable": {"thread_id": "1", "user_id": "1"}}
input_messages = [HumanMessage(content="Hi, my name is Lance")]

for chunk in graph.stream({"messages": input_messages}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

A.2 常用命令速查

python

# 创建 Store
from langgraph.store.memory import InMemoryStore
store = InMemoryStore()

# 保存数据
store.put(
    namespace=("memory", "user_123"),
    key="user_memory",
    value={"memory": "User info here"}
)

# 获取数据
item = store.get(("memory", "user_123"), "user_memory")
if item:
    print(item.value)

# 搜索数据
items = store.search(("memory", "user_123"))
for item in items:
    print(item.key, item.value)

# 查看状态
state = graph.get_state({"configurable": {"thread_id": "1"}})
print(state.values)

文档版本：1.0 最后更新：2024-11-05 作者：AI Assistant 基于：LangChain Academy Module-5 Lesson 5.2

希望这份详细解读能帮助你深入理解 LangGraph Store 的使用和记忆系统的设计！

5.2 Memory Store - 详细解读 ​

一、概述 ​

1.1 本节简介 ​

1.2 学习目标 ​

1.3 实际应用场景 ​

二、记忆（Memory）的认知科学基础 ​

2.1 什么是记忆？ ​

2.2 长期记忆的类型 ​

语义记忆（Semantic Memory） ​

情节记忆（Episodic Memory） ​

程序性记忆（Procedural Memory） ​

2.3 本节的记忆架构 ​

三、LangGraph Store 基础 ​

3.1 什么是 LangGraph Store？ ​

3.2 Store 的基本概念 ​

3.2.1 Namespace（命名空间） ​

3.2.2 Key（键） ​

3.2.3 Value（值） ​

3.3 Store 的数据模型图解 ​

四、环境准备 ​

4.1 安装依赖包 ​

4.2 配置环境变量 ​

五、Store 的基本操作 ​

5.1 创建 Store 实例 ​

5.2 保存数据：put() 方法 ​

5.2.1 基本用法 ​

5.2.2 UUID 详解 ​

5.3 搜索数据：search() 方法 ​

5.3.1 基本用法 ​

5.3.2 查看搜索结果 ​

5.3.3 访问键和值 ​

5.4 获取数据：get() 方法 ​

5.4.1 基本用法 ​

5.4.2 search() vs get() ​

5.5 Store 的三个操作总结 ​

六、构建记忆聊天机器人 ​

6.1 初始化模型 ​

6.2 定义系统提示词 ​

6.2.1 聊天机器人提示词 ​

6.2.2 创建记忆的提示词 ​

6.3 定义节点函数 ​

6.3.1 call_model 节点 ​

6.3.2 write_memory 节点 ​

6.4 构建状态图 ​

6.5 编译配置 ​

七、实战演示 ​

7.1 配置和初始化 ​

7.2 第一次交互：介绍自己 ​

7.3 第二次交互：分享兴趣 ​

7.4 查看短期记忆（对话历史） ​

7.5 查看长期记忆（用户信息） ​

7.6 跨会话测试：新的对话 ​

7.7 继续对话：进一步个性化 ​

八、在热路径中保存记忆 ​

8.1 什么是"热路径"？ ​

8.2 热路径的优缺点 ​

8.3 冷路径的替代方案 ​

8.4 选择策略 ​

九、调试和追踪 ​

9.1 使用 LangSmith 追踪 ​

9.2 查看记忆加载过程 ​

9.3 调试常见问题 ​

问题 1：记忆没有被使用 ​

问题 2：记忆没有被保存 ​

问题 3：记忆格式错误 ​

十、Python 知识点总结 ​

10.1 UUID（通用唯一识别码） ​

10.2 字典的 get() 方法 ​

10.3 f-string 格式化 ​

10.4 字符串的 format() 方法 ​

十一、LangGraph 知识点总结 ​

11.1 Checkpointer vs Store ​

11.2 Config 的使用 ​

11.3 状态更新机制 ​

11.4 流式执行 ​

十二、架构设计分析 ​

12.1 为什么每次都保存记忆？ ​

12.2 为什么使用固定的 key？ ​

12.3 记忆合并策略 ​

十三、实践建议 ​

5.2 Memory Store - 详细解读

一、概述

1.1 本节简介

1.2 学习目标

1.3 实际应用场景

二、记忆（Memory）的认知科学基础

2.1 什么是记忆？

2.2 长期记忆的类型

语义记忆（Semantic Memory）

情节记忆（Episodic Memory）

程序性记忆（Procedural Memory）

2.3 本节的记忆架构

三、LangGraph Store 基础

3.1 什么是 LangGraph Store？

3.2 Store 的基本概念

3.2.1 Namespace（命名空间）

3.2.2 Key（键）

3.2.3 Value（值）

3.3 Store 的数据模型图解

四、环境准备

4.1 安装依赖包

4.2 配置环境变量

五、Store 的基本操作

5.1 创建 Store 实例

5.2 保存数据：put() 方法

5.2.1 基本用法

5.2.2 UUID 详解

5.3 搜索数据：search() 方法

5.3.1 基本用法

5.3.2 查看搜索结果

5.3.3 访问键和值

5.4 获取数据：get() 方法

5.4.1 基本用法

5.4.2 search() vs get()

5.5 Store 的三个操作总结

六、构建记忆聊天机器人

6.1 初始化模型

6.2 定义系统提示词

6.2.1 聊天机器人提示词

6.2.2 创建记忆的提示词

6.3 定义节点函数

6.3.1 call_model 节点

6.3.2 write_memory 节点

6.4 构建状态图

6.5 编译配置

七、实战演示

7.1 配置和初始化

7.2 第一次交互：介绍自己

7.3 第二次交互：分享兴趣

7.4 查看短期记忆（对话历史）

7.5 查看长期记忆（用户信息）

7.6 跨会话测试：新的对话

7.7 继续对话：进一步个性化

八、在热路径中保存记忆

8.1 什么是"热路径"？

8.2 热路径的优缺点

8.3 冷路径的替代方案

8.4 选择策略

九、调试和追踪

9.1 使用 LangSmith 追踪

9.2 查看记忆加载过程

9.3 调试常见问题

问题 1：记忆没有被使用

问题 2：记忆没有被保存

问题 3：记忆格式错误

十、Python 知识点总结

10.1 UUID（通用唯一识别码）

10.2 字典的 get() 方法

10.3 f-string 格式化

10.4 字符串的 format() 方法

十一、LangGraph 知识点总结

11.1 Checkpointer vs Store

11.2 Config 的使用

11.3 状态更新机制

11.4 流式执行

十二、架构设计分析

12.1 为什么每次都保存记忆？

12.2 为什么使用固定的 key？

12.3 记忆合并策略

十三、实践建议