LangChain 高级用法之长期记忆

Apr 28, 2026 / Apr 29, 2026 --- · 6 min read · LangChain LLM ·

Share on:

关于短期记忆已写过两篇 LangChain - 关于会话记忆和 LangChain 核心组件之短期记忆. 有短期记忆就有长期，记忆的短与长的区分标准是看记忆是否能跨越会话，与选择的存储介质, 时效性，中途模型切换都无关。知期记忆限定在同一个会话当中，只要没跨会话，即使是一年前聊过的天，重新拣起来继续聊也是短期记忆; 而长期记忆是专指跨越会话的，在一个会话中聊过的，重开一个新的会话，Agent 还能知道你在别的会话中聊过的内容, 这就是长期记忆。即便这种记忆用内存保存数据，Agent 重启数据会丢失，但只要能跨会话就是长期记忆。

所谓的会话就是像 ChatGPT, Claude 桌面应用对应的 Chat, New chat 就创建了一个新的会话，短期记忆局限于同一个 Chat, 长期记忆则跨越 Chat。同一个会话中聊天，Agent 的回答一直有当前会话上下文中，是好理解的。长期记忆则是无论你 New chat 重开了一个新的 Chat，Agent 都知道你在其他会话中聊过什么。

现在的 ChatGPT 和 Claude 都具有了长期记忆，这带来一个恐怖的事情，随着你使用它们的时间越来越长，AI 可能比你还更了解你，它有了你的隐私，能描绘你的性格，甚至能预测你的下一步行动。看来不想 AI 介入的太多，难道要经常切换着帐号来使用某个 AI 工具？还得探索能不能要求删除长期记忆。如果它们像广告那样粘住你的设备与 IP 就更可怕了。

长期记忆赋予了 AI 能像人与人那样的对话，无论多久，或如何切换话题，两个人之间的对话总会与所有的对话有所关联。业界有少专门实现 Agent 长期记忆的组件，如 Mem0, ZEP, LangChain Memory, LlamaIndex Memory, Letta, Cognee, SuperMemory 等。

短期记忆的上下文虽说相对较短，但经多轮对话后，很快就会超过 Agent 的上下文长度，短期记忆可以通过 Summarization 来压缩上下文。而对于大量会话的长期记忆来说，通过总结肯定是无法有效压缩上下文的，看到一般的实现是把历史会话向量化，然后在对话过程中用 RAG 取加相关片断。这也是为什么 LangChain 中把文档 Long-term memory 放在 Retrieval 之后，这里假定我们对 RAG 有所了解之后来学习 LangChain 的长期记忆。

若对 RAG 不熟悉的话，可参考本人写过的一篇简单例子用 Python + PostgreSQL 演示 RAG.

回到 LangChain 的长期记忆实现，它是内置于 LangGraph stores 的功能，它用 Namespace 和 Key 来组织的 JSON 文件存储长期记忆。

LangChain 实现长期记忆是使用 create_agent() 的 store 参数，LangChain 的基类 BaseStore 有两个实现，InMemoryStore 和 InMemoryByteStore.

1store = InMemoryStore()
2
3agent = create_agent(
4    model="ollama:gemma4:e4b",
5    store=store
6)

回顾 LangChain 的短期记忆是指定 create_agent() 的参数如 checkpointer=InMemorySaver()，所以 checkpointer: 短期记忆，store: 长期记忆。官方文档在演示长期记忆时使用的是 InMemoryStore 和 PostgresStore, 而本文为了方便使用了 SqliteStore, 需先安装 Python 库

1uv add langgraph-checkpoint-sqlite

下面是一个完整的例子，阅读时提前知道几个要点

对存入 store 的消息进行向量化，嵌入模型是 ollama:embeddinggemma:latest
@before_model 中把用户消息存入长期记忆，并从长期记忆中取出与当前问题相关的消息，一并发给模型。取长期记忆数据时用了 (APP, user_id) 作为 namespace
@after_model 中把 AI 非工具调用的消息存入长期记忆，以备后面唤醒长期记忆

 1from typing import Any
 2
 3from langchain.agents.middleware import before_model, after_model
 4from langchain_core.messages import HumanMessage, AIMessage
 5from langgraph.runtime import Runtime
 6import hashlib
 7
 8from dataclasses import dataclass
 9
10from langchain.agents import create_agent, AgentState
11from langchain.embeddings import init_embeddings
12from langgraph.store.sqlite import SqliteStore
13from langgraph.store.sqlite.base import SqliteIndexConfig
14
15
16def gen_key(text: Any) -> str:
17  return hashlib.md5(str(text).encode()).hexdigest()
18
19
20@dataclass
21class Context:
22  user_id: str
23
24
25APP = "chat_app"
26
27with (SqliteStore.from_conn_string(
28        "langchain_memory.db",
29        index=SqliteIndexConfig(embed=init_embeddings(model="ollama:embeddinggemma:latest")))
30as store):
31  @before_model
32  def retrieve_long_term_memory(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
33    user_id = runtime.context.user_id
34    additional_messages = []
35
36    last_message = state["messages"][-1]
37    if isinstance(last_message, HumanMessage):
38      for related in store.search((APP, user_id), query=str(last_message.content), limit=2):
39        additional_messages.append(related.value)
40      store.put((APP, user_id), gen_key(last_message.content),
41                {"role": "user", "content": last_message.content})
42
43    new_messages = {"messages": state["messages"][0:-1] + additional_messages + [last_message]}
44    return new_messages
45
46
47  @after_model
48  def save_long_term_memory(state: AgentState, runtime: Runtime) -> None:
49    user_id = runtime.context.user_id
50    last_message = state["messages"][-1]
51    if isinstance(last_message, AIMessage):
52      if len(last_message.content) > 0:
53        store.put((APP, user_id), gen_key(last_message.content),
54                  {"role": "assistant", "content": last_message.content})
55
56
57  agent = create_agent(
58    model="ollama:gemma4:e4b",
59    store=store,
60    middleware=[save_long_term_memory, retrieve_long_term_memory],
61    context_schema=Context,
62  )
63
64  user_id = "user_123"
65
66  agent.invoke({
67    "messages": [{"role": "user", "content": "My name is Yanbin"}]},
68    context=Context(user_id="user_123"),
69  )
70
71  result = agent.invoke({
72    "messages": [{"role": "user", "content": "Please just state my name?"}]},
73    context=Context(user_id="user_123"),
74  )
75
76  for message in result["messages"]:
77    message.pretty_print()

当第一次执行时，会生成一个新的 sqlite 数据库文件 langchain_memory.db, 最后执行的输出结果是

 1================================ Human Message =================================
 2
 3Please just state my name
 4================================ Human Message =================================
 5
 6My name is Yanbin
 7================================== Ai Message ==================================
 8
 9Hello Yanbin! It's nice to meet you.
10
11How can I help you today? 😊
12================================== Ai Message ==================================
13
14Yanbin

我们学习过短期记忆的话，如果没有短期记忆的话，以上第二次问 Please state my name? 的时候，AI 是没有上下文的，无法回答出这个问题来。所以这是长期记忆在起作用。上面的代码注释掉 store=store 后，第二次问 Please state my name? 的时候，AI 就答不上来了。

分析交互过程

第一次 agent.invoke() 时

在 @before_model 中的 store.search((APP, user_id), query=str(last_message.content), limit=2) 时也会请求一次嵌入模型，然后从长期记忆中搜索是否有与问题 My name is Yanbin 相近的结果。此时，数据库为空, 所以没有相关联的数据

发送给嵌入模型的请求数据是

1{"model":"embeddinggemma:latest","input":["My name is Yanbin"],"options":{"mirostat":null,"mirostat_eta":null,"mirostat_tau":null,"num_ctx":null,"num_gpu":null,"num_thread":null,"repeat_last_n":null,"repeat_penalty":null,"temperature":null,"stop":null,"tfs_z":null,"top_k":null,"top_p":null}}

还是在 @before_model 函数中，store.search() 完后，会把把消息 My name is Yanbin, 存入到长期记忆. 在存入之前会对它进行向量化，再存入向量数据库。所以发现向嵌入模型 embeddinggemma:latest, 又发送了与上面完全一样的请求, 从嵌入模型收到的是一堆向量化后的数值，然后存入 sqlite 数据库 langchain_memory.db 中, 后面我们会知道是存入到 store_vectors 这个表中的。

经过 @before_model 后只是把第一条消息存入到长期记忆中，最终发送了主模型的请求只有 My name is Yanbin.

模型返回后，在 @after_model 中把回复的 Hello Yanbin! It's nice to meet you. 直接消息存入到长期记忆中

第二次 agent.invoke() 问 Please just state my name? 这个问题时，在 @before_model 中也会对该问题向量化后从长期记忆中查询相关消息，搜索到两条相关数据，添加到 messages 中作为问题补充。同样会把最后的问话向量化后存入长期记忆中。

所以这时候发给 gemma4:e4b 模型的请求包含了三条消息

1{"model":"gemma4:e4b","stream":true,"options":{},"messages":[{"role":"user","content":"Please just state my name"},{"role":"user","content":"My name is Yanbin"},{"role":"assistant","content":"Hello Yanbin! It's nice to meet you.\n\nHow can I help you today? 😊"}],"tools":[]}

因此模型有这个上下文才能回答出 please state my name? 这个问题

有了长期记忆，以后只执行第二个 agent.invoke()

1    result = agent.invoke({
2        "messages": [{"role": "user", "content": "Please just state my name"}]},
3        context=Context(user_id="user_123"),
4    )

模型也能作出正确的回答，因为在 @before_model 方法中会从长期记忆中查询相关消息，并把相关消息发给模型。我们这里使用长期记忆时使用的 Namespace 是 ("chat_app", "user_123"), 如果第二次 agent_invoke() 时换一个 Namespace, 比如不同的 APP, 或 user_id, 那么模型就没有相关的上下文，只能说

1================================ Human Message =================================
2
3Please just state my name
4================================== Ai Message ==================================
5
6I do not know your name. You haven't told me what it is.

本例中，在使用长期记忆时，如果知道某个其他用户的 user_id，还能访问别的用户的长期记忆，这会造成严重的用户信息泄漏。

BaseStore 相关的 API

前面的例子用到了 store 的两个 API, 分别是 put() 和 search(), 除此之外还有 get(), delete(), batch() 操作，以及函数对应的 async 版本，用前缀 a，如 aget(), asearch() 等。

put(), get() 和 search() 的方法原型分别是

 1    def put(
 2        self,
 3        namespace: tuple[str, ...],
 4        key: str,
 5        value: dict[str, Any],
 6        index: Literal[False] | list[str] | None = None,
 7        *,
 8        ttl: float | None | NotProvided = NOT_PROVIDED,
 9    ) -> None:
10        ...
11
12    def get(
13        self,
14        namespace: tuple[str, ...],
15        key: str,
16        *,
17        refresh_ttl: bool | None = None,
18    ) -> Item | None:
19
20    def search(
21        self,
22        namespace_prefix: tuple[str, ...],
23        /,
24        *,
25        query: str | None = None,
26        filter: dict[str, Any] | None = None,
27        limit: int = 10,
28        offset: int = 0,
29        refresh_ttl: bool | None = None,
30    ) -> list[SearchItem]:
31        ...

记住，LangChain 的长期记忆 value 是一个 dict[str, Any], 可认为是一个 JSON 对象。应自己规划好 namespace 和 key 如何隔离数据。如前面的例子，namespace('<app_name>', '<user_id>') 来限定应用程序及用户 ID 分类，key 可用来进一步划分的小类，比如用 project 为 key.

以上方法有参数 ttl 或 refresh_ttl，用于控制记忆保存多久时间，或查询出来的结果缓存多久。search(query="<str>") 时查询是用向量化后再以相似度进行匹配, 所以我们前面的 @before_model 进行了两个 embedding, 这是可以优化的地方。embedding 是为了能快速进行相似度的查询，如果只需用 namespace 和 key 或 filter 查询的话，可不使用嵌入模型对内容进行向量化。

这里是只用 filter 来查询的例子，比如 put() 的值为

 1    store.put(
 2        namespace,
 3        "a-memory",
 4        {
 5            "rules": [
 6                "User likes short, direct language",
 7                "User only speaks English & python",
 8            ],
 9            "my-key": "my-value",
10        },
11    )

查询时用

1items = store.search(
2        namespace, filter={"my-key": "my-value"}, query="language preferences"
3    )

如果 put() 时用的 namespace 和 key 都是一样的，则会覆盖之前的值.

查看长期记忆中的数据

用 sqlite 查看其中的数据

 1sqlite langchain_memory.db
 2sqlite> .table
 3store              store_migrations   store_vectors      vector_migrations
 4sqlite> .schema store
 5CREATE TABLE store (
 6    -- 'prefix' represents the doc's 'namespace'
 7    prefix text NOT NULL,
 8    key text NOT NULL,
 9    value text NOT NULL,
10    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
11    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, expires_at TIMESTAMP, ttl_minutes REAL,
12    PRIMARY KEY (prefix, key)
13);
14CREATE INDEX store_prefix_idx ON store (prefix);
15CREATE INDEX idx_store_expires_at ON store (expires_at)
16WHERE expires_at IS NOT NULL;
17sqlite> .schema store_vectors
18CREATE TABLE store_vectors (
19    prefix text NOT NULL,
20    key text NOT NULL,
21    field_name text NOT NULL,
22    embedding BLOB,
23    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
24    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
25    PRIMARY KEY (prefix, key, field_name),
26    FOREIGN KEY (prefix, key) REFERENCES store(prefix, key) ON DELETE CASCADE
27);
28sqlite> .schema store_migrations
29CREATE TABLE store_migrations (
30                    v INTEGER PRIMARY KEY
31                );
32sqlite> .schema vector_migrations
33CREATE TABLE vector_migrations (
34                        v INTEGER PRIMARY KEY
35                    );
36sqlite> .headers on
37sqlite> select * from store limit 3;
38prefix|key|value|created_at|updated_at|expires_at|ttl_minutes
39chat_app.user_123|635067f4f6ee04146299a6e4b39fdd94|{"role":"user","content":"My name is Yanbin"}|2026-04-28 22:21:59|2026-04-28 22:21:59||
40chat_app.user_123|91f32494d462a8e3c4c70f859af0c34c|{"role":"assistant","content":"Hello Yanbin! It's nice to meet you.\n\nHow can I help you today? 😊"}|2026-04-28 22:22:00|2026-04-28 22:22:00||
41chat_app.user_123|c5fcad1b11486cd15f73f31aa4b684b9|{"role":"user","content":"Please just state my name"}|2026-04-28 22:34:28|2026-04-28 22:34:28||

总结

本文的例子只是演示了长期记忆能实现什么，实际项目应该不会以这种方式来存储长期记忆，而且在这个会话中只需用短期记忆即可。具体实现长期记忆可以考虑把会话定期，或结束时刷入到长期记忆中，至于对话当中如何从长期记忆中获取内容传给模型，可以附加到系统提示词，后面可以研究一下 Agent 是如何动态使用 Agent Skills 的。

还需进一步借鉴其他的长期记忆实现框架, 比如尝试在 LangChain 实现的 Agent 中使用 Mem0 作为长期记忆。

永久链接 https://yanbin.blog/langchain-advanced-usage-long-term-memory/, 来自隔叶黄莺 Yanbin's Blog
[版权声明]

本文采用署名-非商业性使用-相同方式共享 4.0 国际 (CC BY-NC-SA 4.0) 进行许可。