短期记忆 — ConversationMemory#

短期记忆负责维护当前对话的上下文。它的核心需求是：保序、能淘汰、够快。

数据结构选的是 OrderedDict。Python 3.7+ 的普通 dict 也保序了，但 OrderedDict 有一个独有方法：popitem(last=False)，可以在 O(1) 时间内弹出最早的条目，实现 FIFO 淘汰。

淘汰的逻辑：

1
def store(self, entry: MemoryEntry) -> None:
2
    self._entries[entry.id] = entry
3
    self._current_tokens += entry.token_count
4
    while self._current_tokens > self._max_tokens and len(self._entries) > 1:
5
        self._evict_oldest()

三个细节值得注意：

1. while 不是 if — 一条新消息可能很大，需要淘汰多条旧的才能腾出空间
2. len > 1 的保护 — 至少保留 1 条消息，即使它超出预算。否则记忆完全清空，Agent 彻底失忆
3. 被淘汰的不直接丢弃 — 而是追加到 _compressed 列表，后续压缩时可以回溯

为什么选 FIFO 而不是 LRU？对话天然是时间有序的，早期的消息通常不如近期的相关。FIFO 实现简单（一行代码），覆盖 80% 的场景，剩下的 20% 靠压缩来兜底。如果用 LRU，每次访问都要移动位置，代码复杂度上去了，但对话场景下的收益很有限。

长期记忆 — LongTermMemory#

长期记忆负责跨会话持久化关键事实。它的核心需求是：持久化、去重、可检索。

存储后端选的是 JSON 文件（~/.paicli/memory/long_term_memory.json）。为什么不用数据库？零依赖 — 不需要 Redis、SQLite、Milvus。可以直接打开文件看里面存了什么。对于学习项目来说，重点是理解 Memory 设计模式，不是存储引擎。生产环境当然会换数据库。

两个关键设计：

内容去重 — 存入时遍历所有已有条目，如果内容完全相同就跳过。防止同一个事实被存 10 次。但这是精确匹配，“我用 Python”和”语言：Python”是同一个事实却不会被去重。语义级去重需要向量相似度，成本更高。

每次写入都存盘 — 最简单但最低效的策略。如果一秒内调 10 次 store()，就写 10 次文件。改进方案是 dirty flag + 定时刷盘，但当前不需要这个优化。

短期 vs 长期对比#

两者的接口完全相同（都有 store、retrieve、search、clear 等），但内在实现完全不同。这就是 Protocol（协议）的好处 — 统一接口，各自实现。

什么时候触发#

不是每次存消息都压缩，而是达到预算的 80% 时才触发。

为什么不是 100%？因为压缩需要调 LLM，需要时间。如果等到 100% 再压缩，在压缩完成之前新来的消息可能已经导致溢出了。留 20% 的缓冲区，大多数情况够压缩完成。

压缩流程 — 三步走#

假设当前有 50 条消息：

1
Step 1: 分割
2
  旧消息 = msg1~msg47（要压缩的）
3
  近期消息 = msg48~msg50（保留最近 3 轮，不压缩）
4

5
Step 2: Map-Reduce 压缩旧消息
6
  Map: [msg1~5]→摘要1, [msg6~10]→摘要2, ..., [msg41~47]→摘要9
7
  Reduce: [摘要1, 摘要2, ..., 摘要9] → 最终摘要
8

9
Step 3: 重建记忆
10
  清空短期记忆
11
  存入: [最终摘要, msg48, msg49, msg50]

为什么是 Map-Reduce 而不是一次性压缩？

如果把 47 条旧消息一次性丢给 LLM 说「帮我压缩」，有两个问题：第一，47 条消息本身就可能超出 LLM 的输出能力；第二，信息密度不均匀，一次性压缩容易丢重要信息。分而治之 — 先分组压缩（Map），再合并摘要（Reduce），每组只有 5 条消息，LLM 处理起来更精确。

如果旧消息不超过 5 条，就只有一个 chunk，跳过 Reduce 阶段，省一次 LLM 调用。

压缩的副产物：事实提取#

压缩时旧消息即将丢失，但其中可能有值得长期保留的事实。所以在清空短期记忆之前，会先提取事实存入长期记忆：

1
# 先提取事实，再清空 — 顺序不能反
2
facts = memory.extract_and_store_facts()
3
memory.clear_short_term()

事实提取是两阶段过滤：第一阶段让 LLM 从对话中提取候选事实，第二阶段用规则过滤掉临时任务（“帮我修改代码”）和推测性内容（“可能使用了 Redis”）。通过两道关卡的事实才会被存入长期记忆。

分词 — 检索的基础#

中文没有天然的词分隔符，需要特殊处理。这里用的是 n-gram 滑动窗口：2-gram 和 3-gram。

比如 tokenize("数据库配置") 会产生：{数据, 据库, 库配, 配置, 数据库, 据库配, 库配置}。

n-gram 的好处是零依赖、速度快，对于检索场景够用了 — 检索不需要精确的语义分词，只需要”query 中的片段能在 content 中找到”就行。

三级评分#

检索时对每条记忆算一个相关性分数：

1
第 1 级：精确匹配 — 整个 query 是 content 的子串 → 1.0 分（最高）
2
第 2 级：关键词匹配 — query 的 token 有多少命中 content → 0~1 分
3
第 3 级：时间衰减 — 越新的记忆权重越高（24 小时内 1.0 → 0.5）

最终分数 = 关键词匹配分 × 时间衰减系数。

长期记忆还有 1.2x 加权，因为长期记忆是经过筛选的，信息密度比短期记忆更高，同等相关性下更值得关注。

检索的局限#

关键词匹配解决不了「用不同词表达同一个意思」的问题。用户说「数据库」，记忆里存的是「DB」— 匹配不到。解决这个需要向量检索（embedding + 相似度匹配），但需要额外的模型和存储。当前方案是零依赖的关键词匹配，够用但不够好。

两种检索模式#

• retrieve(query, limit) — 搜短期 + 长期，返回 list[MemoryEntry]，Agent 内部使用
• build_context_for_query(query, max_tokens) — 只搜长期，返回格式化文本，注入到 LLM 的 system prompt

为什么第二个只搜长期？因为短期记忆已经在 messages 列表里了，不需要额外注入。只有长期记忆（磁盘上的事实）需要在新会话时主动注入到 prompt 里。