生命不息，折腾不止。上一篇画出了第一张图、接上了工具，还用 checkpoint 记住了「这次聊到哪」——但那只是短期记忆，进程一重启、换个会话就全忘。今天给它装上真正的长期记忆，让 Agent 隔天回来还认得你是谁。

上一篇结尾我留了个尾巴：checkpoint 只能记住「当前这一条会话」，换个 thread_id 就翻脸不认人。这不叫记忆，顶多算「金鱼式记忆」。真正的记忆系统得能跨会话、跨天、跨重启地记住「这个用户到底是谁、之前说过什么」。

这就是今天的主角——LangGraph 的 Store。它和 checkpointer 是两套东西，配合起来才是完整的记忆：一个管「这次聊到哪」，一个管「我都了解这个人什么」。

一、先分清两件事：短期记忆 vs 长期记忆

很多教程把这两件事揉在一起讲，越讲越晕。先甩一张表把它们掰开：

短期记忆（Checkpointer） 长期记忆（Store）
存的是什么 这条会话里每一步的完整状态快照 你主动挑出来、想长期留存的「事实」
作用范围 单个 thread_id，换线程就隔离 全局的，跨线程、跨会话都能读
回答的问题 「这次聊天聊到哪了？」 「这个用户我了解什么？」
典型内容 消息历史、中间推理步骤 用户偏好、身份信息、业务参数
生产落点 InMemorySaver → SQLite/Postgres InMemoryStore → PostgresStore

一句话记死：checkpointer 按 thread_id 隔离，Store 按 user_id（或任意你想用的维度）隔离。同一个用户换十个会话，checkpointer 各存各的，但 Store 里的资料是共享的。这就是「隔天也认得你」的底层逻辑——昨天那条会话早就结清了，但 Store 里存着「他叫吴峰、在承德做网络科技」这些事实，今天一上来就能捞出来用。

二、最小可跑：给图挂一个 Store

Store 用法和 checkpointer 一样，编译时挂上去就行，一个 store= 参数：

1
pip install langgraph langchain-openai
1
2
3
4
5
6
7
8
9
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()

# 编译时 checkpointer 和 store 一起挂
graph = builder.compile(
    checkpointer=InMemorySaver(),
    store=store,
)

挂上去之后，LangGraph 会把 store 自动注入到每一个节点函数里——你不用自己传来传去，在节点签名里声明一个 store 参数就能拿到。这是和 checkpointer 最大的不同：checkpointer 你得靠 config 里的 thread_id 用它，而 store 直接就是一个对象，节点里想读就读、想写就写：

1
2
3
4
5
6
from langgraph.store.base import BaseStore
from langchain_core.runnables import RunnableConfig

def call_model(state, config: RunnableConfig, *, store: BaseStore):
    # store 就在这里，直接用
    ...

（补充一句：旧文档里就是 store: BaseStore 这种注入写法；新一点的文档推荐通过 Runtime 对象取 runtime.store。你手上装的版本用哪个、以官方文档为准，两套调用 put / search 的方法都一模一样，只是「怎么拿到 store 这个对象」的姿势不同。）

三、写：让 Agent 主动记东西

Store 存的是「json 文档」，按 namespace（命名空间）+ key（键） 组织。namespace 是个元组，你可以理解成文件夹路径，最常用的组织方式就是把 user_id 放进去，让不同用户的记忆天然隔离：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import uuid
from langgraph.store.base import BaseStore
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_core.runnables import RunnableConfig

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
store = InMemoryStore()

def call_model(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    user_id = config["configurable"]["user_id"]          # 从 config 拿到当前用户
    namespace = ("memories", user_id)                    # 这个用户的记忆文件夹

    last = state["messages"][-1].content                 # 用户最后一句话
    if "记住" in last:                                    # 用户让「记住」，就存一条
        store.put(namespace, str(uuid.uuid4()), {"data": last})

    return {"messages": [model.invoke(state["messages"])]}

builder = StateGraph(MessagesState)
builder.add_node("agent", call_model)
builder.add_edge(START, "agent")
builder.add_edge("agent", END)
graph = builder.compile(
    checkpointer=InMemorySaver(),
    store=store,
)

核心就一句 store.put(namespace, key, value)：

  • namespace：元组，比如 ("memories", "wufeng")，第一条是「记忆」，第二条是「谁」。
  • key：这条记忆的唯一标识。上面用 uuid.uuid4() 每次随机——这是方便演示，生产上其实要小心，后面第五节专门讲为什么。
  • value：一个 dict，你想存啥都行，这里是 {"data": "..."}。

user_id 从哪来？来自 config。和上一篇的 thread_id 是邻居，同一个 configurable 字典里多塞一个字段就行：

1
config = {"configurable": {"thread_id": "t-1", "user_id": "wufeng"}}

一个 thread_id 管会话，一个 user_id 管身份，两个钥匙各管各的事。

四、读：下次会话把记忆捞出来喂给模型

光存不读等于白存。把上一节的节点补上「读」这一半——每次对话前，先把这个用户名下的记忆全捞出来，塞进 system 提示词里：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
def call_model(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    user_id = config["configurable"]["user_id"]
    namespace = ("memories", user_id)

    # 读：把这个用户所有的记忆捞出来
    items = store.search(namespace)
    info = "\n".join([m.value["data"] for m in items])

    # 写：用户让「记住」，就存一条
    last = state["messages"][-1].content
    if "记住" in last:
        store.put(namespace, str(uuid.uuid4()), {"data": last})

    system = f"你是一个贴心助手。已知用户信息：\n{info}"
    return {"messages": [model.invoke([{"role": "system", "content": system}] + state["messages"])]}

store.search(namespace) 把该命名空间下的记忆按条返回，每条记起来是 item.value["data"]。跑一遍验证跨会话的记忆：

1
2
3
4
5
6
7
8
# 会话1：同一个用户，第一次聊天，让他记住名字
config1 = {"configurable": {"thread_id": "t-1", "user_id": "wufeng"}}
graph.invoke({"messages": [("user", "我叫吴峰，记住了")]}, config=config1)

# 会话2：换了个全新线程（thread_id 变 t-2），但同一个 user_id
config2 = {"configurable": {"thread_id": "t-2", "user_id": "wufeng"}}
res = graph.invoke({"messages": [("user", "我叫什么名字？")]}, config=config2)
print(res["messages"][-1].content)   # 它能答出「吴峰」

关键就在 thread_id 从 t-1 换成了 t-2——短期记忆（checkpointer）那边已经断了，但长期记忆（store）因为 user_id 还是 wufeng，照样把「吴峰」捞了出来。这就是跨会话记忆的全部秘密：换线程，不换用户，agent 依然认得你。

这里顺手带一句：上面 model 用的是 gpt-4o-mini，如果你想接 Claude/GPT 这些国外模型又不想折腾翻墙和绑卡，把 ChatOpenAI 的 base_url 指到中转站 ai.aklibk.com 就行，国内直连、免绑卡、人民币按量付费，一行 base_url 的事。跟记忆本身没关系，纯属顺手省事。

五、进阶：语义搜索 + 稳定 key，别让记忆变成一锅乱炖

上一节的 store.search(namespace) 是「全捞出来」，记忆少还行，记几十条就成「每句话都往提示词里灌一堆」，又贵又慢。LangGraph 的 Store 支持语义检索——按「含义相近」搜，而不是按精确匹配。只要在创建 store 时配一个 embedding 索引：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from langchain.embeddings import init_embeddings
from langgraph.store.memory import InMemoryStore

store = InMemoryStore(
    index={
        "embed": init_embeddings("openai:text-embedding-3-small"),
        "dims": 1536,   # text-embedding-3-small 是 1536 维
    }
)

store.put(("memories", "wufeng"), "1", {"data": "爱吃火锅"})
store.put(("memories", "wufeng"), "2", {"data": "在承德做网络科技"})

# 按语义搜：问「吃啥」能命中「爱吃火锅」
items = store.search(("memories", "wufeng"), query="今天想吃什么", limit=1)
print(items[0].value["data"])   # -> 爱吃火锅

（init_embeddings 是 LangChain 新包装里的统一入口，需要 pip install langchain langchain-openai 才拿得到；你要是不想引这个包，用现成的 langchain_openai.OpenAIEmbeddings(model="text-embedding-3-small") 这类 embedding 对象塞进 embed 字段也一样能跑，姿势以你手上的版本为准。）

配上索引后，store.search(namespace, query="...", limit=1) 就是向量相似度检索，只捞最相关的几条回来，不用全灌。读的时候把 query= 换成用户当前那句话，模型就能「想起」跟这句话最相关的那点事。

但这里必须泼一盆冷水，也是我踩过的坑：语义搜索按「相似度」排序，不按「真假」排序。 如果你每次都拿随机 key（uuid）去存，同一个事实会被存成几十条互相矛盾的版本，检索时哪个「措辞更像」就捞出哪个，旧信息永远清不掉、纠正也覆盖不了。解法是——用稳定的 key，让 put 变成「覆盖式更新」：

1
2
3
4
5
6
7
# 坏的：每次随机 key，同一个事实存 N 份
store.put(namespace, str(uuid.uuid4()), {"data": "用户住北京"})
store.put(namespace, str(uuid.uuid4()), {"data": "用户搬上海了"})  # 两条都在，搜出哪个全看命

# 好的：稳定 key，写第二次直接覆盖第一次
store.put(namespace, "prefs-location", {"data": "用户住北京"})
store.put(namespace, "prefs-location", {"data": "用户搬上海了"})  # 同 key，新值覆盖旧值

put 本质是 upsert（有则更新，无则插入），key 定了，纠错就是覆盖，而不是堆一条新的。记住这条铁律：身份、偏好这类「一个用户只有一份」的字段，用固定 key；只有「日记、事件流水」这类天然会越来越多的，才用随机 key。

六、生产落地：换 PostgresStore + 踩坑清单

InMemoryStore 存在内存里，进程一重启记忆全没，跟 InMemorySaver 一个德行，只配拿来调试。生产上换成 Postgres 版：

1
2
3
4
5
6
from langgraph.store.postgres import PostgresStore

DB_URI = "postgresql://user:pass@localhost:5432/db"
with PostgresStore.from_conn_string(DB_URI) as store:
    store.setup()   # 第一次用必须先建表
    graph = builder.compile(checkpointer=checkpointer, store=store)

套路和上一篇的 PostgresSaver 一模一样，换个 Store 而已；要语义检索，创建时同样传 index={"embed": ..., "dims": ...}。（装哪个包、类名落在哪个模块，版本之间略有变动，以官方文档为准。）

踩坑清单，替我趟过的这几个：

  1. checkpointer 和 store 傻傻分不清：想知道「聊到哪」用 thread_id，想知道「这人是谁」用 user_id + store。别拿 thread_id 去当存储的 namespace，那等于把长期记忆降级成了短期记忆。
  2. 忘了在 config 里传 user_id：namespace 拿不到 user_id 会报错，或者干脆所有用户共用一套记忆串台。多用户场景这是头号事故。
  3. 随机 key 存「唯一事实」：纠正不覆盖、旧数据淤积，踩过的人了。身份/偏好用固定 key。
  4. 语义搜索当「事实核查」用：向量只比相似度不比真假，旧事实照样高置信地被捞出来。该清就清、该覆盖就覆盖。
  5. InMemoryStore 直接上生产：重启即失忆，务必换 PostgresStore，并且第一次用先 store.setup()。

到这里，你的 Agent 就摆脱了「金鱼记忆」：checkpointer 管住一次会话的上下文，store 管住跨越时间的人格与知识，一个 user_id 把它们串成一个「认得你的长期搭档」。配合上一篇的断点续跑和人审门禁，你手里这套东西已经在往「能上线服务」那一步走了。

生命不息，折腾不止。下一篇咱们把这套东西从本地脚本变成真正的服务——LangGraph 实战第三课：把 Agent 打包成 API，langgraph serve 一键起服。今天先把 store 挂上、亲手跑一遍「记住→换线程→问名字」这三步，确认它真认得你了，咱们下一篇见。