第6节 长期记忆的检索与遗忘过程#
在前面的内容中，我们已经完成了长期记忆的持久化存储，但如果一个系统只是不断累积记忆，却无法在需要时准确找回，也无法在信息过时后及时清理，那么这些记忆最终就会从系统资产变成系统负担。本节内容将系统介绍长期记忆的检索方式、过期机制以及自定义遗忘策略的实现思路。
6.1 检索与遗忘#
对于短期记忆而言，当前对话上下文通常规模有限，模型可以直接在消息列表中读取信息；但长期记忆不同，它往往跨越多个会话持续积累，数据量会不断增长。如果没有检索机制，即便是拥有记忆也无法在真正需要时把它拿出来使用；而如果没有遗忘机制，记忆库又会不断膨胀，使早已失效的内容长期占据存储空间，甚至影响召回结果的准确性。
从工程角度看，长期记忆系统通常会面临下面 3 类问题。
（1）记忆越多，遍历就越不靠谱
当历史记忆数量较少时，我们还可以通过简单的枚举方式读取所有内容；但随着用户数量和交互轮次增加，遍历式读取很快就会失去效率，所以必须引入向量化检索。
（2）用户提问 ≠ 记忆原文的字面表达
比如记忆里存的是“用户最喜欢的编程语言是 Python”，用户却问“我一般使用哪个编程语言？”。如果只做关键词匹配，系统未必能够稳定命中；而语义检索恰恰就是为了解决“表达不同但含义相近”的问题。
（3）不是所有记忆都值得永久保留
有些记忆是稳定事实，如用户姓名、长期偏好；有些记忆却是阶段性上下文，如最近一次报错、某段临时安排、短期任务状态。后者如果长期留存，不仅价值有限，反而可能干扰后续决策。
因此，一个成熟的长期记忆系统不应只有写入能力，还需要形成写入、检索、衰减、清理的完整闭环。
6.2 用 pgvector 做长期记忆召回#
先看长期记忆的语义检索怎么写，完整代码见 Code/Chapter05/C09_long_memory_store_pg.py。
（1）构建向量索引
在上一节内容中我们安装完成了 pgvector 插件的安装，因此在创建 PostgresStore 时可以额外传入了 index 参数来对记忆进行向量化并构建索引。
1 embeddings = DashScopeEmbeddings(model="text-embedding-v3")
2 memory_index = {"embed": embeddings,
3 "dims": 1024,
4 "fields": ["content", "context"],
5 "distance_type": "cosine"}在上述代码中，第2~5行便是构建向量索引，这段配置表明后续长期记忆在写入时不仅会把原始字段保存到数据库中，还会对指定字段进行向量化。第4行是指定写入记忆时的内容中需要进行向量化的字段名称，换句话说一条记忆中可以包含多个描述字段均可进行向量化，详见后续示例。第5行是指定相似度的计算方式，常见的有 “cosine” 、“l2” 和 “inner_product”，默认为"cosine"。
（2）写入示例记忆
接下来构造 2 条长期记忆写入数据库做演示：
1 def save_normal_memory(store: PostgresStore, namespace: tuple[str, ...]):
2 store.put(namespace, key=str(uuid.uuid4()),
3 value={"memory_type": "语义记忆",
4 "created_at": datetime.now().isoformat(timespec="seconds"),
5 "content": "用户最喜欢的编程语言是 Python。",
6 "context": "记录用户的编程语言喜好，后续在编程方面可以用到"})
7 store.put(namespace, key=str(uuid.uuid4()),
8 value={"memory_type": "情景记忆",
9 "created_at": datetime.now().isoformat(timespec="seconds"),
10 "content": "2026-01-31，用户完成了一次 RAG 检索结果为空的排查。",
11 "context": "记录用户曾经做过的事情，在谈论到 RAG 话题到时候能用到"})（3）执行检索
再写一个检索方法：
1 def search_result(store: PostgresStore, namespace: tuple[str, ...], query=None):
2 result = store.search(namespace, query=query, limit=2)
3 print(result)在上述代码中，第2行便是根据用户请求 query 去数据库中检索与之匹配的记忆，其中 limit=2 表示返回最相似的前2个。这里需要注意的是，如果上面没有构建 memory_index 索引或者 query=None，那么此时返回的便是按照 store 表中 updated_at 字段降序后的结果。
（4）示例运行
最后用一段代码把上面这些串起来跑：
1 if __name__ == "__main__":
2 namespace = get_namespace()
3 with PostgresStore.from_conn_string(DB_URI, index=memory_index) as store:
4 store.setup()
5 save_normal_memory(store, namespace)
6 query = '我一般使用哪个编程语言？'
7 search_result(store, namespace, query=query)在上述代码中，因为第3行加入了 index=memory_index ，所以第4行代码运行结束后数据库中除了创建 store 和 store_migrations 这两张表以外，还会创建 store_vector 和 vector_migrations 这两张表，其中 store_vector 便是用来存储向量的。当第5行代码执行结束以后，除了表 store 中会记录每条记忆的文本内容，表 store_vector 中还会记录每条记录每个索引字段对应的向量，类似如下结果：
prefix | key |field_name| embedding |created_at|updated_at
-----------+-------+----------+-----------------+----------+-----------
memory.user|318e...| content |[-0.081,0.024...]|2026-06...|2026-06...|
memory.user|318e...| context |[-0.095,0.014...]|2026-06...|2026-06...|
memory.user|2900...| content |[-0.082,0.002...]|2026-06...|2026-06...|
memory.user|2900...| context |[-0.052,0.034...]|2026-06...|2026-06...|在上述结果中，每一条记忆对应其中两行记录，即分别是 “content” 和 “context”，它们对应的记忆ID（key）是相同的。当然，如果写入记忆时有多个文本字段需要向量化，且在 memory_index 中也配置了相应的字段名，那一条记忆便会对应这里的多行记录。
进一步，上述代码在执行第6~7行时，会先将 query 转换成向量然后去 store_vector 表中检索得到相似向量并得到 key，最后通过 key 再去 store 表中取对应的记忆内容返回。
运行结束后输出类似：
[Item(namespace=['memory', 'user'], key='318e706d-bce0-4799-b52a-a67901fc4adc', value={'content': '用户最喜欢的编程语言是 Python。', 'context': '记录用户的编程语言喜好，后续在编程方面可以用到', 'created_at': '2026-06-13T10:47:54', 'memory_type': '语义记忆'}, created_at='2026-06-13T10:47:55', updated_at='2026-06-13T10:47:55', score=0.7497), Item(namespace=['memory', 'user'], key='2900ca37-90ac-428b-a0a1-d4d538b52b4b', value={'content': '2026-01-31，用户完成了一次 RAG 检索结果为空的排查。', 'context': '记录用户曾经做过的事情，在谈论到 RAG 话题到时候能用到', 'created_at': '2026-06-13T10:47:54', 'memory_type': '情景记忆'}, created_at='2026-06-13T10:47:55', updated_at='2026-06-13T10:47:55', score=0.473)]返回的每条 Item 里既有记忆原文，也带一个 score 字段表示跟 query 的相似度。拿到这些内容之后，就可以接着喂给大模型做下一步推理了。
6.3 记忆遗忘的必要性#
在理解了如何找回记忆之后，还要进一步考虑哪些记忆不应永久存在。现实中的长期记忆并不是一个只增不减的仓库，很多信息都具有明显时效性。例如：
-
临时性偏好，例如用户最近阶段的关注重点，也许只在最近几天内有参考价值；
-
某个短期项目安排，例如一段时间内持续跟进的事项，任务结束后就应失效；
-
某些用户状态信息会随时间变化，旧记录继续存在反而会误导系统。
如果没有"忘"的机制，库里的长期记忆会无限堆积，至少带来 3 个问题：
-
召回噪声增加——重要信息更难排到前面；
-
存储与维护成本持续上升；
-
过时记忆与最新事实冲突，影响 Agent 的判断。
所以遗忘不是长期记忆的"附属功能"——它和检索同等重要。
当然，像用户姓名、长期职业背景、稳定兴趣这种高价值事实，不适合也走遗忘流程——不然系统会频繁丢本该长期留的信息，破坏个性化体验。所以遗忘的核心价值不是"所有记忆都过期"，而是帮我们区分永久记忆 vs 可衰减记忆。
6.4 TTL 过期机制遗忘方案#
工程上最容易落地的遗忘方式就是给记忆配一个生存时间（Time To Live, TTL）——LangGraph 的 PostgresStore 自带定时扫描机制可以直接拿来用。完整示例见 Code/Chapter05/C10_long_memory_store_ttl.py。
（1）TTL 全局配置
配置 PostgresStore 时把 ttl=ttl_config 传进 from_conn_string()，就可以给所有记忆开 TTL：
1 ttl_config = {"refresh_on_read": True,
2 "sweep_interval_minutes": 1,
3 "default_ttl": 1}在上述代码中，这 3 个参数分别管不同的东西：
-
refresh_on_read=True：记忆被get()/search()访问时自动续期。比如本该 1 小时过期，但第 59 分钟被读到一次，过期时间就再往后挪 1 小时——“还在被用"的记忆自然活下来，“长期没人读"的记忆才进清理队列。 -
sweep_interval_minutes=1：后台清理线程多久扫一次过期数据。1 分钟就意味着差不多 1 分钟一次。 -
default_ttl=1：每条记忆默认 1 分钟后过期（单位看实现，这里是分钟）。
当然，除了全局配置 TTL 以外，对于特定的记忆内容也可以在持久化存储的时候单独指定 TTL，即
1 store.put(namespace, key, value, ttl=60) #一小时后过期此时，两者的优先级是 store.put(..., ttl=...) 高于 from_conn_string(..., ttl=ttl_config) 里的 default_ttl，底层逻辑是如果 put() 显式传了 ttl 就直接用这次传入的值；如果显式传的是 ttl=None，那就是明确表示不过期，也不会再回退到全局默认值；只有当 put() 不指定 ttl 参数时才回退到 ttl_config["default_ttl"]。
（2）清理过程
具体地，在完成上述 TTL 配置以后还需要开启清理守护进程，示例代码如下：
1 with PostgresStore.from_conn_string(DB_URI, ttl=ttl_config) as store:
2 keys = save_normal_memory(store, namespace)
3 count = 1
4 while True:
5 time.sleep(1)
6 future = store.start_ttl_sweeper(sweep_interval_minutes=1)
7 if count == 30:
8 print(f"当前为30秒后，此时访问一次记忆 {keys[0]}, 对应失效时间增加1分钟，"
9 f"{(datetime.now() + timedelta(minutes=1)).isoformat(timespec="seconds")} 到期，"
10 f"记忆 {keys[1]} 将在30秒后到期被删除")
11 r = store.get(namespace, keys[0])
12 if count == 61:
13 r = store.get(namespace, keys[1])
14 print(f"当前为1分钟后，此时预期结果里只有记忆 {keys[0]} 未被删除，{keys[1]} = {r} 已被删除。")
15 if count > 120:
16 search_result(store, namespace)
17 print(f"当前为2分钟后，预期结果这里为空，所有记忆均被删除。")
18 break
19 count += 1在上述代码中，第6行是触发 TTL 清理器，对已经过期的记忆进行扫描和删除，频率为1分钟。同时在有需要的地方还可以通过 store.stop_ttl_sweeper() 来停止清理器。
对于上述模拟示例来说，首先会保存两条记忆并设置过期时间为1分钟，在第30秒的时候访问其中一条记忆 keys[0]，在第60秒的输出结果中未被访问的记录 keys[1] 已经被删除，在第90秒时记忆 keys[0]过期，但是清理器执行间隔为1分钟，因此第120秒后记忆 keys[0]被删除。
运行完能看到类似这样的输出：
测试周期2分钟，会自动停止
当前为30秒后，此时访问一次记忆 3b3f4e81-8c8f-44de-b580-ce894c1ae0f9, 对应失效时间增加1分钟，2026-06-13T13:53:32 到期，记忆 7eabf9d0-2b0c-4764-a798-1eee63ca5ba1 将在30秒后到期被删除
当前为1分钟后，此时预期结果里只有记忆 3b3f4e81-8c8f-44de-b580-ce894c1ae0f9 未被删除，7eabf9d0-2b0c-4764-a798-1eee63ca5ba1 = None 已被删除。
暂无记忆
当前为2分钟后，预期结果这里为空，所有记忆均被删除。6.5 自定义定制遗忘策略#
虽然默认 TTL 看似够用，但实际项目里几乎一定会撞上"想更细粒度控制"的需求——比如默认会把 store 表里所有 namespace 的过期记录一起清，但系统里往往有多个业务根 namespace，不同业务的记忆不能被同一把扫帚扫掉。
这时候最简单的办法是重写 sweep_ttl() 自己写 SQL。完整代码见 Code/Chapter05/C11_long_memory_custom_store_ttl.py。
1 class myPostgresStore(PostgresStore):
2 def sweep_ttl(self) -> int:
3 with self._cursor() as cur:
4 prefix = self.namespace[0]
5 cur.execute(f"""
6 DELETE FROM store
7 WHERE prefix like '{prefix}%' AND
8 expires_at IS NOT NULL AND expires_at < NOW()
9 """)
10 deleted_count = cur.rowcount
11 return deleted_count在上述代码中，我们重写了 sweep_ttl() 方法，不再按照默认方式（无筛选条件prefix like ‘{prefix}%’）扫描所有过期数据，而是根据当前 namespace 的第一个字段，也就是根前缀定向删除特定范围内已经过期的记忆。这里需要注意的是， store_vector 表里的数据也会被级联删除。当然，其它更多自定义方式可以自行修改SQL。
用一段模拟代码验证一下效果：
1 print(f"\n测试周期1分钟，会自动停止")
2 namespace1 = get_namespace('root1', 'demo_user')
3 with myPostgresStore.from_conn_string(DB_URI, ttl=ttl_config, index=memory_index) as store:
4 save_normal_memory(store, namespace1)
5 namespace2 = ('root2', 'demo_user')
6 with myPostgresStore.from_conn_string(DB_URI, ttl=ttl_config, index=memory_index) as store:
7 save_normal_memory(store, namespace2)
8 count = 1
9 while True:
10 time.sleep(1)
11 future = store.start_ttl_sweeper()
12 count += 1
13 if count > 62:
14 search_result(store, namespace2)
15 search_result(store, namespace1)
16 break上面这段同时建了 namespace1（root1）和 namespace2（root2）两组。跑下来 namespace2 的记忆会被清掉，namespace1 的安然无恙，这就是"按 namespace 隔离清理"的效果。
6.6 小结#
长期记忆系统真正的价值，不在于数据库里存了多少条记忆，而在于系统能否在恰当的时候把正确的记忆取出来，并在不再需要的时候及时将其遗忘。至此，我们已经把长期记忆从如何存推进到了如何找和如何忘。在接下来的内容中，就可以进一步把短期记忆、长期记忆与完整 Agent 工作流结合起来，构建出真正具备跨会话个性化能力的智能体系统。