5.8 长期记忆检索与遗忘#
在前面的内容中，我们已经完成了长期记忆的持久化存储，但如果一个系统只是不断累积记忆，却无法在需要时准确找回，也无法在信息过时后及时清理，那么这些记忆最终就会从系统资产变成系统负担。本节内容将系统介绍长期记忆的检索方式、过期机制以及自定义遗忘策略的实现思路。
5.8.1 检索与遗忘#
对于短期记忆而言，当前对话上下文通常规模有限，模型可以直接在消息列表中读取信息；但长期记忆不同，它往往跨越多个会话持续积累，数据量会不断增长。如果没有检索机制，即便是拥有记忆也无法在真正需要时把它拿出来使用；而如果没有遗忘机制，记忆库又会不断膨胀，使早已失效的内容长期占据存储空间，甚至影响召回结果的准确性。
从工程实现角度看，长期记忆系统通常会面临如下几类问题。
（1）记忆越多，越难直接遍历
当历史记忆数量较少时，开发者或许还可以通过简单的枚举方式读取所有内容；但随着用户数量和交互轮次增加，遍历式读取很快就会失去效率，必须引入向量化检索。
（2）用户提问与记忆原文的差异
例如，记忆中存储的是“用户最喜欢的编程语言是 Python”，用户提问却可能是“我一般使用哪个编程语言？”。如果只做关键词匹配，系统未必能够稳定命中；而语义检索恰恰就是为了解决“表达不同但含义相近”的问题。
（3）并非所有记忆都值得永久保留
有些记忆是稳定事实，如用户姓名、长期偏好；有些记忆却是阶段性上下文，如最近一次报错、某段临时安排、短期任务状态。后者如果长期留存，不仅价值有限，反而可能干扰后续决策。
因此，一个成熟的长期记忆系统不应只有写入能力，还需要形成写入、检索、衰减、清理的完整闭环。
5.8.2 记忆检索#
下面先来看长期记忆的语义检索实现，对应完整代码参见 Code/Chapter05/C09_long_memory_store_pg.py 文件。
（1）向量索引
与第5.5节中仅把记忆写入 PostgreSQL 不同，在上一节内容中我们安装完成了 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"，详见第3.3.4节内容。
（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)]从上述结果可以看出，搜索返回的内容除了包含记忆本身还有其与用户请求的相似度得分。当获取到与用户请求相关的记忆内容以后便可以进行后续操作。
5.8.3 记忆遗忘的必要性#
在理解了如何找回记忆之后，还要进一步考虑哪些记忆不应永久存在。现实中的长期记忆并不是一个只增不减的仓库，很多信息都具有明显时效性。例如：
- 临时性偏好，例如用户最近阶段的关注重点，也许只在最近几天内有参考价值；
- 某个短期项目安排，例如一段时间内持续跟进的事项，任务结束后就应失效；
- 某些用户状态信息会随时间变化，旧记录继续存在反而会误导系统。
如果系统不提供遗忘机制，那么数据库中的长期记忆会不断堆积，造成至少3方面问题。第一，召回噪声增加，真正重要的信息更难被排到前面；第二，存储与维护成本持续上升；第三，过时记忆可能与最新事实冲突，影响 Agent 的判断质量。因此，遗忘并不是长期记忆系统的附属功能，而是和检索同样重要的核心能力。
当然，像用户姓名、长期职业背景、长期稳定兴趣等高价值事实，通常并不适合遗忘，否则系统会频繁遗忘本应长期保留的信息破坏个性化体验。因此，记忆遗忘的核心价值不在于让所有记忆都会过期，而在于帮助开发者区分永久记忆和可衰减记忆。
5.8.4 记忆遗忘机制#
对于工程系统来说，最常见也最容易落地的遗忘方式，就是为记忆配置生存时间（Time To Live, TTL）。此时，可以借助 PostgresStore 提供的定时扫描机制来完成长期记忆的清除。以下完整示例代码可参见 Code/Chapter05/C10_long_memory_store_ttl.py 文件。
（1）TTL 全局配置
为了实现长期记忆的以外，可以在配置 PostgresStore 上下文环境的时候指定 TTL 全局配置 from_conn_string( ttl=ttl_config)，如下所示：
1 ttl_config = {"refresh_on_read": True,
2 "sweep_interval_minutes": 1,
3 "default_ttl": 1}在上述代码中，refresh_on_read 表示记忆在被读取后是否自动刷新过期时间。这里设置为 True意味着只要某条记忆仍然被访问（ get() 或search() 操作 ），它的有效期就可以向后延长，即长期不再使用的记忆才逐渐遗忘。例如本该1小时后过期，但是第59分的时候被访问，那么其过期时间会再次延后1小时。这种设计体现出并不是所有记忆都需要开发者手动决定删留，系统可以根据是否还在被使用来自动估计其价值。被持续访问的记忆，说明它依然与当前用户、任务或系统行为相关；而长期无人访问的记忆，更适合进入遗忘流程。sweep_interval_minutes 表示后台清理线程多久执行一次过期扫描。这里设为 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分钟后，预期结果这里为空，所有记忆均被删除。5.8.5 自定义定制遗忘策略#
虽然默认 TTL 已经能完成基础过期删除，但在实际项目中，开发者往往还会有更细粒度的控制需求。例如，默认情况下会删除 store 表中所有命名空间的内容，但是系统中可能存在多个业务根命名空间，不同类别的记忆并不希望被同一套清理器一起处理，或者某些租户、某些任务域的记忆需要独立回收，而其他区域则保持不动。这时我们可以自行修改执行删除的 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在上述代码中，我们定义了两个命名空间，实验结果将是 namespace2中的记忆被清除，而 namespace1 中的记忆依旧存在。 各位读者可自行运行测试案例观察结果。
5.8.7 小结#
长期记忆系统真正的价值，不在于数据库里存了多少条记忆，而在于系统能否在恰当的时候把正确的记忆取出来，并在不再需要的时候及时将其遗忘。至此，我们已经把长期记忆从如何存推进到了如何找和如何忘。在接下来的内容中，就可以进一步把短期记忆、长期记忆与完整 Agent 工作流结合起来，构建出真正具备跨会话个性化能力的智能体系统。