使用LangChain+LangGraph开发AI应用（7） - 记忆系统

XiongNeng 收录于 AI

2026-10-07 约 4700 字预计阅读 10 分钟

学习进度

模型本身是无状态的，每一次调用都是初次见面，你五分钟前说过喜欢美式咖啡，五分钟后它已经不记得。这一篇给应用补上记忆，体验上才像个能长期相处的助手。

短期记忆靠 checkpointer，把对话状态按线程存住，一个 thread_id 就是一次会话。长期记忆靠 store，把用户偏好写成键值文档，跨会话跨线程都能取回来。中间还穿插消息历史的两条管理手段，手动裁剪和自动摘要。示例在 source/hello_memory.py，九个 demo。完整案例是一个带长期记忆的个人助理 source/assistant.py，会话换了它还认得你。

前言用途

没有记忆的应用，每个功能都在跟失忆搏斗。用户在对话里说过「我对花生过敏」，三轮之后推荐菜谱照放花生，这种体验说不上哪坏了，但一定留不住人。要补记忆，先想清楚记什么、记多久。

对话历史是最直接的一类。当前会话里说过的话、调过的工具、拿到的结果，会话结束就没必要再留着，但会话进行中必须随时取用，这是短期记忆。用户的偏好、身份、历史决定是另一类，今天告诉它的事实，下周打开应用它还得认账，而且要跟别的用户分开，这是长期记忆。两类记忆的读写模式完全不同，LangGraph 因此给了两套组件，checkpointer 管短期，store 管长期，各自都能从内存版一路换到数据库版。

什么时候需要记忆，判断比 RAG 还简单。应用只要跨过一次交互，就躲不开。多轮对话需要短期记忆，个性化服务需要长期记忆，两者经常一起上，本篇讲的就是怎么把它们装到同一个 agent 上。

原理概念

LangGraph 把记忆拆成两套互补的持久化。官方文档的对比表这么分，checkpointer 持久化的是图状态的快照，范围限定在单个线程内，服务对话连续、中断恢复这些场景。store 持久化的是应用自己定义的键值数据，范围跨线程，服务用户偏好、事实档案这类需要长期保留的信息。多数应用两个都用，checkpointer 追踪当前对话，store 存放跨对话的档案。

短期记忆的载体是线程。thread_id 类似邮件里的一封会话串，把一个用户的多次往返组织在一起。agent 编译时挂上 checkpointer，每次调用带上同一个 thread_id，状态就自动续上。不同 thread_id 之间完全隔离，你跟 A 用户聊的内容，B 用户的线程里一个字都看不见。状态在每步开始时读入、结束时写回，所以不只是消息，任何进了状态图的字段都享受同样的持久化。

长期记忆的载体是 store。数据以 JSON 文档的形式存放，靠 namespace 和 key 两级定位。namespace 像文件夹，通常装用户 ID 或业务标签，key 像文件名。("users", "u1") 这个空间里放一个 key 叫 profile 的文档，取的时候照样报出两个坐标就行。检索接口 search 还支持按 namespace 前缀扫和按内容过滤，文档数量上去之后，跨空间做统计这两招都接得住。

消息历史是短期记忆里最活跃也最容易失控的部分。对话越长，token 越贵，模型在长上下文里的表现还越差。LangChain 给了两个治理手段，手动用 RemoveMessage 删旧消息，或者上 SummarizationMiddleware 让模型自己把老历史压成一段摘要。两个手段都在中间件层工作，不碰业务代码。

核心用法

九个用法，示例在 source/hello_memory.py。记忆组件全部来自 langgraph，InMemorySaver 与 InMemoryStore 都是进程内实现，零依赖零部署，跑完即清空，各自的边界也清楚。

1. 短期记忆

checkpointer 加 thread_id，同一个线程里的对话自动续上。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
from langchain.agents import create_agent
from langchain.messages import HumanMessage
from langgraph.checkpoint.memory import InMemorySaver

agent = create_agent(
    model=model,
    checkpointer=InMemorySaver(),
)
cfg = {"configurable": {"thread_id": "chat-1"}}
agent.invoke({"messages": [HumanMessage("我叫小明")]}, cfg)
result = agent.invoke({"messages": [HumanMessage("我叫什么？")]}, cfg)

1
2
3
== 短期记忆 ==
  线程内消息数: 4
  模型回答: 你是小明，喜欢短句

第二次调用没带任何身份信息，模型答出「你是小明」，靠的是 checkpointer 把第一轮的两条消息带进了第二轮的上下文。四条消息就是这个线程的全部状态，下次再带这个 thread_id 来，四条一起进上下文。

2. 线程隔离与状态查看

换 thread_id 就是换会话，get_state 能看线程状态的快照。

1
2
3
4
5
6
7
cfg1 = {"configurable": {"thread_id": "chat-1"}}
agent.invoke({"messages": [HumanMessage("我叫小明")]}, cfg1)

cfg2 = {"configurable": {"thread_id": "chat-2"}}
result = agent.invoke({"messages": [HumanMessage("我叫什么？")]}, cfg2)

state = agent.get_state(cfg1)

1
2
3
4
== 线程隔离 ==
  chat-2 的回答: 我不知道你叫什么
  chat-1 快照字段: ['messages']
  chat-1 消息数: 2

chat-2 里模型直说不知道，chat-1 记的小明一个字都没漏过去。get_state 返回快照对象，values 字典装着状态字段，默认的 AgentState 里就是 messages 一项。排查「模型为什么忘了」这类问题，先 get_state 看快照里到底有没有那句话，一眼定位。

3. 手动裁剪历史

历史太长就删，RemoveMessage 配合 REMOVE_ALL_MESSAGES 可以一键清空再保留尾部。

1
2
3
4
5
6
from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES

state = agent.get_state(cfg).values
keep = [RemoveMessage(id=REMOVE_ALL_MESSAGES)] + state["messages"][-2:]
agent.update_state(cfg, {"messages": keep})

1
2
3
== 手动裁剪 ==
  裁剪前 4 条，裁剪后 2 条
  保留的首条: 我在学 Python

RemoveMessage 单条用就是按 id 删，配 REMOVE_ALL_MESSAGES 就是全清，后面的列表是被保留的消息。update_state 直接改线程状态，立刻生效。注意保留哪几条要自己拿主意，最稳的策略是保留最近几轮，更早的内容交给下一节的摘要去压缩。

4. 自动摘要

SummarizationMiddleware 在消息数越线时请模型写摘要，替换掉老历史。

1
2
3
4
5
6
7
8
9
from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model=main_model,
    middleware=[SummarizationMiddleware(
        model=summarizer_model, trigger=("messages", 4), keep=("messages", 2),
    )],
    checkpointer=InMemorySaver(),
)

1
2
3
4
5
6
== 自动摘要 ==
  触发摘要后消息数: 4
  [HumanMessage] Here is a summary of the conve
  [AIMessage] 好的
  [HumanMessage] 我叫什么？
  [AIMessage] 你是小明，来自杭州

trigger 按消息条数设阈值，keep 决定摘要后保留最近几条。第三轮调用前历史到了五条，越线，中间件把前几条发给 summarizer 模型压成一段话。实测有个形态上的细节，写回历史的那条摘要消息是 HumanMessage，不是直觉里的 AIMessage，代码里按类型找摘要会扑空。trigger 也支持按 token 计数，生产上比条数更贴近真实成本。

5. Store 键值读写

长期记忆是 namespace 加 key 组织的 JSON 文档，search 支持前缀扫描与内容过滤。

1
2
3
4
5
6
7
8
9
from langgraph.store.memory import InMemoryStore

store = InMemoryStore()
store.put(("users", "u1"), "profile", {"name": "小明", "city": "杭州"})
store.put(("users", "u1"), "note", {"text": "喜欢美式咖啡"})
store.put(("users", "u2"), "profile", {"name": "小红", "city": "上海"})

item = store.get(("users", "u1"), "profile")
items = store.search(("users",), filter={"city": "上海"})

1
2
3
4
5
6
== Store 读写 ==
  get: {'name': '小明', 'city': '杭州'}（namespace=('users', 'u1')）
  精确空间检索: ['profile', 'note']
  前缀检索: ['profile', 'note', 'profile']
  内容过滤: ['小红']
  命名空间列表: [('users', 'u1'), ('users', 'u2')]

get 按两级坐标精确取，search 的第一个参数是 namespace 前缀，只给 ("users",) 就把 u1、u2 的文档全翻出来。filter 按文档内容等值过滤，配合 list_namespaces 列出所有子空间，数据按用户分库之后想跨用户做统计，这套接口够用。

6. 工具里读写长期记忆

记忆要由 agent 自己动手存取才真的用得起来，工具拿到 ToolRuntime，runtime.store 就是 create_agent 传进去的那个 store。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
from langchain.tools import ToolRuntime, tool

@tool
def remember(fact: str, runtime: ToolRuntime) -> str:
    """记住一条关于用户的事实"""
    import hashlib
    key = hashlib.md5(fact.encode()).hexdigest()[:8]
    runtime.store.put(("memories",), key, {"text": fact})
    return "已记住"

@tool
def recall(runtime: ToolRuntime) -> str:
    """回忆用户告诉过你的事"""
    items = runtime.store.search(("memories",))
    return "；".join(i.value["text"] for i in items) or "还没有记忆"

agent = create_agent(
    model=model, tools=[remember, recall],
    store=store, checkpointer=InMemorySaver(),
)

1
2
3
4
== 工具读写长期记忆 ==
  工具返回: 用户喜欢短句回复
  模型回答: 你喜欢短句回复
  Store 里实际存了: [{'text': '用户喜欢短句回复'}]

工具函数声明一个 runtime 参数，框架注入，不用你从配置里翻。key 用事实文本的哈希，同一条事实重复记住只会覆盖同一条记录，不会越堆越多。系统提示里写清楚「用户说偏好就调 remember，被问到就先调 recall」，模型就能自己维护自己的记忆。

7. 跨会话记忆

换一个全新的 thread_id，短期记忆清零，store 里的长期记忆还在。

1
2
3
4
store.put(("memories",), "abc123", {"text": "用户喜欢短句回复"})
# 全新线程调用同一个 agent
fresh_cfg = {"configurable": {"thread_id": "brand-new"}}
result = agent.invoke({"messages": [HumanMessage("我喜欢什么样的回复")]}, fresh_cfg)

1
2
3
4
== 跨会话记忆 ==
  全新线程消息数: 4
  工具返回: 用户喜欢短句回复
  模型回答: 根据长期记忆，你喜欢短句回复

brand-new 这个线程之前从没出现过，消息历史从零开始，但 recall 一调，老事实回来了。这就是 checkpointer 和 store 的分工，短期跟着线程走，长期跟着应用走。真要用起来，把 store 换成数据库后端（PostgresStore、Redis 都有官方包），进程重启、机器迁移都不丢。

8. TTL 坑

InMemoryStore 的 put 签名里有个 ttl 参数，实际一用就报错。

1
store.put(("t",), "k", {"v": 1}, ttl=5)

1
2
3
4
5
== TTL 坑 ==
  put 签名里有 ttl 参数: True
  supports_ttl: False
  put(ttl=5) 报错: TTL is not supported by InMemoryStore.
  Use a store implementation that supports TTL or set ttl=None.

方法签名是基类 BaseStore 定的统一接口，内存版没实现 TTL，调用直接 NotImplementedError。想给记忆设过期时间，只能换支持 TTL 的后端。报错信息本身把话说得很明白，看到就懂，不算难排查，但设计接口的时候最好先查一下 supports_ttl。

9. 持久化边界

内存版的两个组件都没有落盘能力，重启即失忆。

1
2
3
4
from langgraph.checkpoint.memory import InMemorySaver

print(hasattr(InMemoryStore(), "dump"))
print(hasattr(InMemorySaver(), "dump"))

1
2
3
== 持久化边界 ==
  InMemoryStore 有 dump: False
  InMemorySaver 有 dump: False

两个都是 False。对比向量库的 InMemoryVectorStore 好歹有 dump 和 load，这里的两个记忆组件连这个都没有，官方的设计意图很明确，内存版只管让你把流程跑通，落盘是数据库后端的职责，SqliteSaver 对短期记忆，PostgresStore 对长期记忆，接口不变，换个对象就完事。本篇的完整案例也因此只做单进程演示，多轮记忆跨进程要用，记得换后端。

完整案例

九个 demo 摊开讲完了。收个尾，把它们装进一个带长期记忆的个人助理，assistant.py。它有两个记忆工具，remember_fact 和 recall_facts，存取都走 runtime.store，namespace 按 ("memories", "default-user") 分组，key 是事实文本的哈希。会话线程由 thread_id 圈住，跟长期记忆互不干扰。

`1`	`uv run python assistant.py --demo`

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
[会话 session-a] 用户: 记住：我喜欢美式咖啡，我在学 LangChain
    工具调用: remember_fact({'fact': '用户喜欢美式咖啡'})
    工具返回: 已记住：用户喜欢美式咖啡
    工具调用: remember_fact({'fact': '用户在学 LangChain'})
    工具返回: 已记住：用户在学 LangChain
    助理: 都记下了，美式咖啡和在学 LangChain。

[会话 session-b] 用户: 我爱喝什么？最近在忙什么？
    工具调用: recall_facts({})
    工具返回: 用户喜欢美式咖啡；用户在学 LangChain
    助理: 你喜欢美式咖啡，最近在学 LangChain。

长期记忆库:
  [7e484591] 用户喜欢美式咖啡
  [34cedd68] 用户在学 LangChain

demo 里跑了两段会话。session-a 存了两条偏好，session-b 换了一个全新的 thread_id，消息历史一条都没有，全靠 recall_facts 从 store 里把事实捞回来再组织回答。最后直接翻库，两条记忆各占一个 key，重复记同一条事实不会膨胀。

–mems 命令翻看记忆库，–thread 换会话线程，–model 接真实模型（提供商:模型名 格式）。接上真实模型后它就是个能用的对话入口，用户随口说的事它记下来，过几天再问它还认账。

1
2
AGENT_MODEL=openai:deepseek-v4.1-flash uv run python assistant.py "我喜欢喝美式咖啡，记住"
AGENT_MODEL=openai:deepseek-v4.1-flash uv run python assistant.py "我爱喝什么" --thread next-day

1
2
你: 我爱喝什么
助理: 你喜欢美式咖啡。

第二个命令换了新线程 next-day，对话历史是空的，记忆还在。测试在 tests/test_hello_memory.py 十四条加 tests/test_assistant.py 七条，线程续接、隔离、裁剪、摘要形态、store 前缀检索、跨线程读取、重复记忆去重都钉住了。加上前六篇的，全仓一百一十五条，全绿。

源码

https://github.com/yidao620c/xneng-langchain-langgraph