第 12 章 BaseStore 与跨会话长期记忆

12.1 Checkpointer 与 Store 的职责分离

LangGraph 的持久化体系由两套机制组成，它们在设计目标、生命周期、读写方式、隔离粒度上完全分离，必须严格区分使用，误用任何一方都会导致架构问题。

维度	Checkpointer（检查点）	Store（长期存储）
核心角色	运行时状态快照（执行轨迹存档）	长期键值与向量记忆（知识库/画像）
生命周期	短期，会话级（单 thread_id 内有效）	长期，用户级/业务级（跨 thread、跨会话、跨部署）
写入方式	自动：每个超步末由运行时 put	手动：节点/工具代码主动调用 put
读取方式	自动：超步开始由运行时加载到 State	手动：节点/工具代码主动调用 get/search
数据内容	完整运行时 State（含 messages、中间字段、断点、pending sends）	跨会话事实（用户画像、偏好、知识库条目、共享黑板）
隔离粒度	thread_id（会话级隔离）	namespace tuple + key（多层级命名空间隔离）
数据进入 checkpoint	是（核心职责）	否（Store 的读写独立于 checkpoint）
典型用途	断点续跑、多轮对话上下文、时间旅行、interrupt 恢复	用户画像、长期事实记忆、业务知识库、多 Agent 共享黑板

职责分离的工程意义：将长期数据错误放入 State 会导致 checkpoint_blobs 体积持续膨胀（每条事实都被每个快照完整保存）、跨会话无法共享（其他 thread_id 读不到）；将短期运行状态放入 Store 则无法享受自动快照与 resume 恢复（Store 写入不会自动与超步同步，中途崩溃可能丢失）。两者是互补关系，不是替代关系。

12.2 BaseStore 接口

BaseStore 位于 langgraph.store.base，是所有 Store 实现的抽象基类，定义了四个同步核心方法与对应的异步版本（方法名加 a 前缀）。

方法	签名（简化）	说明
put	put(namespace: tuple[str,...], key: str, value: dict, index: list[str] | None = None, *, ttl: float | None = None) -> UpsertResponse	存入/更新一条记录；index 指定需要做 embedding 的字段（用于语义检索）；ttl 单条记录过期时间（分钟）
get	get(namespace: tuple[str,...], key: str, *, refresh_ttl: bool = True) -> Item | None	按 namespace+key 精确读取单条记录，未找到返回 None；refresh_ttl 控制读操作是否刷新 TTL 计时
search	search(namespace_prefix: tuple[str,...], *, query: str | None = None, filter: dict | None = None, limit: int = 10, offset: int = 0) -> list[Item]	列出/语义检索某命名空间下的记录；query 为自然语言时启用向量相似度检索，query=None 时按 metadata/时间排序列出；filter 做元数据精确过滤；limit/offset 分页
delete	delete(namespace: tuple[str,...], key: str) -> bool	删除单条记录，返回是否实际删除
batch	batch_ops(ops: list[Op]) -> None	批量原子操作（PutOp/SearchOp/DeleteOp/GetOp），减少 IO 往返

异步版本为 aput / aget / asearch / adelete / abatch。

命名空间（namespace）：一个 tuple[str, ...]，支持多层级隔离，语义类似文件系统路径。例如 ("users", user_id, "facts") 表示某用户下的事实记忆命名空间；("kb", tenant_id, "product_docs") 表示某租户的产品文档库。search 方法使用 namespace_prefix 做前缀匹配，传入 ("users", user_id) 可查询该用户下所有子命名空间的记录，传入完整 tuple ("users", user_id, "facts") 则精确限定到某子命名空间。命名空间第一段通常是业务域/租户 ID，保证多租户隔离。

Item 对象：get/search 返回的记录对象（TypedDict），核心字段：

字段	类型	说明
namespace	tuple[str, ...]	记录所在命名空间
key	str	记录键（同 namespace 内唯一）
value	dict	存储的实际数据（必须是 JSON dict）
created_at	datetime	创建时间（UTC）
updated_at	datetime	最近更新时间（UTC）
score	float / None	语义检索时的余弦相似度分数（0-1，越高越相似）；精确 get/非语义 search 为 None

TTL 配置：通过构造 Store 时传入 ttl_config=TTLConfig(...) 配置过期策略，字段包括：default_ttl（分钟，None 表示永不过期）、refresh_on_read（读取/写入时是否刷新 TTL 计时，默认 True）、sweep_interval_minutes（定期清理过期条目的频率）。单条 put 时也可通过 ttl 参数覆盖默认值。

12.3 InMemoryStore 与 PostgresStore

InMemoryStore（langgraph.store.memory.InMemoryStore）是基于嵌套字典的内存实现，支持语义检索（需要在构造时传入 embedding 模型配置），用于开发期、Notebook、单元测试。进程退出数据丢失。

from langgraph.store.memory import InMemoryStore
from langchain_openai import OpenAIEmbeddings

# 开发期：内存 Store + OpenAI Embedding
store = InMemoryStore(
    index={
        "embed": OpenAIEmbeddings(model="text-embedding-3-small"),
        "dims": 1536,
    }
)

PostgresStore（langgraph-checkpoint-postgres 包）是基于 PostgreSQL 的生产级实现，与 PostgresSaver 可共享同一个数据库实例甚至同一个连接池，但使用独立的数据表（默认 store 表）。PostgresStore 依赖 pgvector 扩展存储向量字段，setup() 方法首次执行时会自动 CREATE EXTENSION IF NOT EXISTS vector 并建立 HNSW 或 IVFFlat 向量索引。

from langgraph.store.postgres import PostgresStore
from psycopg_pool import ConnectionPool

pool = ConnectionPool(
    conninfo=DB_URI, min_size=5, max_size=20, timeout=5,
    kwargs={"autocommit": True},
)
pool.open()
store = PostgresStore(
    pool,
    index={
        "embed": "openai:text-embedding-3-small",  # embedding 模型
        "dims": 1536,                              # 向量维度
        "distance_type": "cosine",                 # 距离度量：cosine/euclidean/inner_product
        "index_type": "hnsw",                      # 索引类型：hnsw（高精度）/ivfflat（省内存）
        "m": 16,                                   # HNSW 参数
    },
)
store.setup()  # 建表、pgvector 扩展、向量索引
graph = builder.compile(checkpointer=checkpointer, store=store)

异步版本为 langgraph.store.postgres.aio.AsyncPostgresStore，配合 AsyncConnectionPool 在 FastAPI 异步栈中使用。

12.4 在节点与工具中使用 Store

Store 实例必须通过 langgraph.config.get_store() 在节点函数、工具函数、条件路由函数中获取，前提是 compile 时传入了 store 参数（否则 get_store() 返回 None）。Store 不应通过函数参数从调用方手动传递——这样做会破坏 Pregel 的序列化假设（Store 包含连接池等不可序列化资源）。工具函数中还可以使用 InjectedStore 类型注解自动注入（见第 8 章）。

from langgraph.config import get_store

def remember_user_name(state) -> dict:
    """记忆节点：对话结束后提取用户名存入 Store。"""
    store = get_store()
    if store is None:
        return {}  # 未配置 store 时静默跳过
    user_id = state["user_id"]
    last_user_msg = next(
        (m.content for m in reversed(state["messages"]) if m.type == "human"),
        "",
    )
    if "我叫" in last_user_msg:
        name = last_user_msg.split("我叫")[-1].strip().split()[0]
        store.put(
            namespace=("users", user_id, "profile"),
            key="name",
            value={"name": name, "source": "self_report"},
        )
    return {}

12.5 典型记忆模式

基于 BaseStore 的抽象能力，实践中形成三类高频记忆模式。

用户画像/偏好（单条记录模式）：namespace 为 ("users", user_id, "profile")，key 为固定 "data" 或具体字段名（如 "language"、"theme"），value 为结构化字典（姓名、语言、回答风格、产品偏好等）。单条记录适合固定字段集，节点在对话开始时通过 get 读取一次并注入 SystemMessage，更新时 put 覆盖。

事实三元组（多条记录模式）：namespace 为 ("users", user_id, "facts")，每条事实一条独立记录（key 可使用 UUID 或事实 hash），value 结构为 {"fact": "...", "category": "preference|biography|project|...", "confidence": 0.9, "created_at": "..."}。多条模式支持语义检索：每轮对话开始时用当前问题做 query 检索最相关的 N 条事实注入 prompt。该模式是开放式记忆的主流方案。

业务知识库：namespace 为 ("kb", tenant_id, doc_type)，按租户与文档类型分层；离线管线（文档解析→分块→embedding）批量 put 条目，Agent 通过检索节点 search 召回相关文档片段用于 RAG。该模式与向量数据库（Pinecone/Milvus/Weaviate）定位类似，但 PostgresStore 优势在于与 checkpoint/业务数据同库、事务一致、运维成本低。

12.6 语义检索（Semantic Search）

PostgresStore 配置 embedding 后，search 方法支持自然语言语义检索。底层机制：put 时若指定 index=["field1","field2"]，Store 会对这些字段的文本调用 embedding 模型生成向量，写入 pgvector 字段；search 传入 query 时，对 query 也调用 embedding 生成向量，SQL 中执行向量相似度排序（ORDER BY embedding <=> query_vector 即余弦距离），返回 Top-K 结果。filter 参数的元数据过滤下推到 SQL 的 JSONB where 条件（如 WHERE value @> '{"category": "preference"}'），与向量检索在同一查询中完成。

# 语义检索：召回与当前问题最相关的 5 条用户事实
results = store.search(
    namespace_prefix=("users", user_id, "facts"),
    query="用户喜欢什么编程语言？最近在做什么项目？",
    filter={"category": "preference"},   # 元数据过滤
    limit=5,
)
for item in results:
    print(f"[score={item.score:.3f}] {item.key}: {item.value['fact']}")

语义检索的关键参数：limit 控制召回条数（通常 3–10 条，过多会注入无关事实导致模型困惑）；filter 做硬性分类过滤提升准确率；namespace_prefix 限定检索范围避免跨用户/跨租户泄露。

12.7 与 Checkpointer 协同、多 Agent 黑板、Memory Service

与 Checkpointer 协同：Checkpointer 在每个超步末自动保存执行轨迹与当前 State，Store 保存跨会话长期知识；节点按需读写 Store，Store 的读写不会自动进入 checkpoint，这是设计上的刻意分离——长期记忆不应当随每个超步快照冗余保存。节点可以在一个超步内多次 put 新记忆，这些写入立即对后续 get/search 可见（同线程内），其他 thread 调用 get/search 也能读到最新值（数据库隔离级别为 READ COMMITTED）。

共享黑板模式（Blackboard）：多智能体协作场景下，多个 Agent（研究员、程序员、测试员、审稿人）共享同一个 Store，工作过程中将中间产物（调研笔记、代码草稿、测试结果、审稿意见）写入约定的 namespace（如 ("blackboard", task_id)），其他 Agent 通过 search 读取相关条目后推进各自分工。相比让所有消息都过 messages 字段（会造成 checkpoint 体积膨胀、所有 Agent 被迫读取全部历史），黑板模式只持久化需要共享的关键产物，读取时通过命名空间+语义检索精准召回。

Memory Service 模式：在 ToolNode 或主 Agent 循环的前后挂载独立的"记忆节点"，自动完成记忆读写：对话开始节点（load_memory）通过 search 检索与当前问题相关的记忆，拼接到 SystemMessage 注入；对话结束节点（extract_memory）通过 LLM 结构化提取本轮对话中的新事实/偏好，去重后 put 到 Store。该模式将记忆的读写逻辑从业务节点中解耦，业务 Agent 不需要显式感知记忆存在。官方 LangMem 库即基于此模式提供更高层的记忆管理能力。

12.8 完整示例：带记忆的对话机器人

from typing import Annotated, TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.memory import InMemoryStore
from langgraph.config import get_store
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
from langchain_openai import ChatOpenAI, OpenAIEmbeddings

class State(TypedDict):
    messages: Annotated[list, add_messages]
    user_id: str

# 1) 初始化 Store（生产换成 PostgresStore）
store = InMemoryStore(
    index={
        "embed": OpenAIEmbeddings(model="text-embedding-3-small"),
        "dims": 1536,
    }
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 2) 记忆加载节点：对话前检索相关记忆
def load_memory(state: State) -> dict:
    s = get_store()
    if s is None:
        return {}
    user_id = state["user_id"]
    last_user_msg = state["messages"][-1].content if state["messages"] else ""
    # 语义检索相关记忆
    memories = s.search(
        namespace_prefix=("memories", user_id),
        query=last_user_msg,
        limit=3,
    )
    if not memories:
        return {}
    memory_lines = "\n".join(
        f"- [{m.value.get('category', 'fact')}] {m.value['fact']}"
        for m in memories if m.score and m.score > 0.5
    )
    sys_msg = SystemMessage(
        content=f"以下是你了解到的用户信息（请自然地融入回答，不要生硬提及）：\n{memory_lines}"
    )
    return {"messages": [sys_msg]}

# 3) 对话节点
def chatbot(state: State) -> dict:
    # filter 掉系统临时注入消息，或者直接全量传入（LLM 能处理）
    return {"messages": [llm.invoke(state["messages"])]}

# 4) 记忆提取节点：对话后提取新事实（演示版，生产用 LLM 结构化提取）
def extract_memory(state: State) -> dict:
    s = get_store()
    if s is None:
        return {}
    user_id = state["user_id"]
    # 遍历本轮 human 消息提取简单模式（真实场景用 LLM）
    for m in state["messages"]:
        if m.type == "human" and "我叫" in m.content:
            name = m.content.split("我叫")[-1].strip().split()[0].rstrip("，。！？")
            s.put(
                namespace=("memories", user_id),
                key=f"name_{name}",
                value={"fact": f"用户名字是 {name}", "category": "profile"},
                index=["fact"],
            )
        if m.type == "human" and "我喜欢" in m.content:
            pref = m.content.split("我喜欢")[-1].strip().split("。")[0]
            import uuid
            s.put(
                namespace=("memories", user_id),
                key=f"pref_{uuid.uuid4().hex[:8]}",
                value={"fact": f"用户喜欢 {pref}", "category": "preference"},
                index=["fact"],
            )
    return {}

builder = StateGraph(State)
builder.add_node("load_memory", load_memory)
builder.add_node("chatbot", chatbot)
builder.add_node("extract_memory", extract_memory)
builder.add_edge(START, "load_memory")
builder.add_edge("load_memory", "chatbot")
builder.add_edge("chatbot", "extract_memory")
builder.add_edge("extract_memory", END)

checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer, store=store)

# 第一次对话：用户告知名字
graph.invoke(
    {"messages": [HumanMessage(content="你好，我叫小明，我喜欢Python编程")], "user_id": "u-1"},
    config={"configurable": {"thread_id": "u-1:conv-1"}},
)

# 第二次对话：新 thread_id（即新会话），但同一 user_id，模型应能通过 Store 记住名字与偏好
graph.invoke(
    {"messages": [HumanMessage(content="你还记得我吗？能结合我的喜好推荐学习方向吗？")], "user_id": "u-1"},
    config={"configurable": {"thread_id": "u-1:conv-2"}},
)
state = graph.get_state({"configurable": {"thread_id": "u-1:conv-2"}})
print("AI 回复：", state.values["messages"][-1].content)
# 预期：回复中包含"小明"与"Python 相关"

本章小结

Checkpointer 与 Store 在 LangGraph 持久化体系中职责严格分离：Checkpointer 负责会话级短期运行快照（自动写入、thread_id 隔离、服务于断点续跑与 resume），Store 负责跨会话长期记忆（手动读写、namespace 隔离、服务于用户画像/知识库/共享黑板）。BaseStore 提供 put/get/search/delete/batch 五个核心方法，namespace tuple 做多层级隔离，value 必须是 JSON dict。InMemoryStore 用于开发测试，PostgresStore 基于 PostgreSQL+pgvector 实现生产级持久化与语义检索，支持 TTL 过期。节点与工具通过 get_store() 或 InjectedStore 获取 Store 实例。典型记忆模式包括用户画像（单条记录）、事实三元组（多条记录+语义检索）、业务知识库（离线灌入+在线检索）。语义检索通过 embedding 向量+余弦相似度实现，filter 做元数据下推过滤。多 Agent 场景下 Store 充当共享黑板，Memory Service 模式通过独立记忆节点自动完成记忆读写，解耦业务逻辑。

常见错误

1. 混淆 Checkpointer 与 Store，把长期数据放进 State 导致 checkpoint 膨胀。现象：运行一段时间后 checkpoint_blobs 表体积异常增长，单条 checkpoint 数百 KB 甚至数 MB，延迟升高、数据库 IO 飙升。原因：用户画像、历史事实、知识库条目等跨会话数据全部塞在 messages 或 State 字段中，每个超步快照都完整冗余保存。修复：跨会话数据放入 Store，State 只保留本轮对话需要的上下文；已有的大 State 通过迁移脚本把长期事实搬移到 Store。
2. namespace 设计不当导致数据泄露。现象：A 用户搜索记忆时返回 B 用户的条目；多租户场景下租户 X 看到租户 Y 的知识库。原因：namespace 未包含 user_id/tenant_id 等隔离字段，或 search 时 namespace_prefix 范围过大（如只传 ("memories",) 而不包含 user_id）。修复：namespace 第一段放租户 ID，第二段放用户/业务对象 ID，search 始终传包含隔离字段的完整前缀；多租户场景下在 filter 中强制加 tenant_id 作为二次校验。
3. 误以为 PostgresStore 需要手动 commit。现象：put 后立即在另一个进程/thread 查询，怀疑数据没提交（实际数据已提交可见）。原因：受关系型数据库习惯影响，以为需要手动 commit。修复：PostgresStore 使用 autocommit=True 的连接池，put/delete 立即提交持久化，无需手动 commit；若希望批量操作原子性，使用 batch() 方法。
4. embedding 模型未配置时调用语义检索报错。现象：search(query="xxx") 抛 ValueError（"embed not configured"）或 NoneType 错误。原因：构造 InMemoryStore/PostgresStore 时未传入 index 参数（embed/dims），或配置错误导致 embed 函数为 None。修复：构造 Store 时必须传入 index={"embed": embed_model, "dims": dims}；仅做精确 get/metadata 过滤 list 时不传 query 即可，不需要 embedding。
5. put 时 value 不是 dict。现象：put 抛 TypeError 或 JSON 序列化错误。原因：BaseStore 约定 value 必须是 dict（JSON 对象），直接传入字符串、列表、整数都会失败。修复：用 {"text": "...", ...} 包装非 dict 值，或重构 schema 让 value 始终是有键字典。

生产建议

• 设计 Store namespace 前先画出数据分层图（租户/用户/类别/条目），形成团队规范文档，统一命名风格（snake_case），避免后续跨业务数据冲突；
• 语义检索召回条数 limit 控制在 3–10 条，过多会导致 prompt 膨胀、模型注意力分散、延迟上升；通过阈值（如 score > 0.5）过滤低相似度结果再注入；
• Memory Service 节点（记忆提取/写入）的 LLM 调用使用小模型（如 gpt-4o-mini、Qwen2.5-7B）以降低成本，对提取结果做去重（put 前先 get/search 判断是否已有相同事实）；
• PostgresStore 的表与 Checkpointer 表可放同一数据库实例以降低运维复杂度，但建议使用独立连接池实例，避免 checkpoint 高写入与 Store 检索相互争用连接；
• 为 Store put 操作加容量保护：单用户记忆条数上限（如 500–1000 条），超出触发旧记忆降权/摘要/清理，避免向量检索性能随记忆量线性下降；
• 生产环境使用 Pydantic BaseModel 约束 value 的结构，对记忆条目的 category、created_at、source、confidence 等字段做统一规范，便于检索过滤与质量评估；
• embedding 模型选择与业务语言匹配的方案（中文场景考虑 text-embedding-3-small、bge-large-zh-v1.5 等），向量维度（dims）与模型一致，避免维度不匹配报错。

---