Dual-rate 长记忆断裂方向开源中间件—— 给 Agent 装上“快/慢记忆齿轮”

最近针对许多agent项目的长信息断裂问题，我做了一个“双速递归记忆”中间件：让 Agent 不再长对话就失忆（Topical-Chat 全量 10,784 对话 avg_recall +10.0%，d4 +21.7%）

做长对话智能体（Agent）的时候，你一定遇到过一个很真实的痛：

对话一旦变长，Agent 开始“断片”。

不是完全忘光，而是关键点突然断流：

记得最近几轮，却把早期的重要信息丢了。

更难受的是：

你明明已经做了“长期记忆”（向量库、摘要、存档…），它还是会在长距离上掉链子。

我最近围绕这个问题做了一些研究与工程尝试，最后落地成一个双速递归记忆中间件（Dual-rate Agent Memory Middleware），核心是：

✅ Fast Memory（快记忆）：高频更新，紧贴当前语境

✅ Slow Memory（慢记忆）：低频沉淀，只保存稳定关键事实

✅ Consistency Check（一致性校验）：防止“快记忆噪声”污染长期记忆

最终在 Topical-Chat 全量 10,784 条对话上验证：

avg_recall +10.0%

d4（长距离段）+21.7%
同时 线上 token 成本保持可控（基本不显著增加）。

1. 为什么“长期记忆”会断流？

很多人的长期记忆方案看起来是完整的：

存历史 → 向量化 → 检索 → 塞回 prompt

但长对话里会出现两个典型问题：

（1）记忆被噪声淹没

你越记越多，越检索越“杂”。

模型拿到的不是“最关键的旧信息”，而是一堆“看起来相关但不重要”的片段。

（2）记忆发生漂移

早期事实在后续被多轮“改写/重述”，如果你把这些都存进去，慢慢就会出现：

新的说法覆盖旧事实

片段之间互相矛盾

关键设定被稀释
最后表现就是：Agent 明明有记忆库，但关键时刻检索不到该检索的东西——断流。

2. 我的思路：给 Agent 装两套“记忆齿轮”⚙️

我把记忆系统拆成两层，像人一样：

快记忆：像随手记的便签，更新很快，但不保证长期可靠

慢记忆：像“永久笔记”，更新谨慎，但足够稳定

✅ Fast Memory：实时贴着对话走

它的目标不是“存得久”，而是“跟得上”。

任何新信息、临时意图、当下约束，都先进入 Fast Memory，保证短期决策不掉线。

✅ Slow Memory：只收“沉淀后的结论”

Slow Memory 不追求全量，而追求少而准。

只把真正稳定、可复用、对未来有价值的信息写进去，比如：

用户长期偏好

任务关键设定

可复用的经验结论

✅ Consistency Check：防止快记忆把慢记忆带偏

最关键的一步是“一致性校验”：

不是 Fast Memory 有啥就同步到 Slow Memory，

而是先判断：它是否稳定？是否重要？是否与已有慢记忆冲突？

可以把它理解成：

快记忆负责“记录”，慢记忆负责“定稿”，一致性校验就是“审核编辑”。

from typing import List, Dict
from dataclasses import dataclass
import numpy as np

# ========== 基础结构定义 ==========

@dataclass
class MemoryItem:
    content: str
    embedding: np.ndarray
    importance: float
    timestamp: float

class DualRateMemory:

    def __init__(self, embed_fn, consistency_threshold=0.75):
        self.embed_fn = embed_fn

        # 快记忆（高频更新）
        self.fast_memory: List[MemoryItem] = []

        # 慢记忆（低频沉淀）
        self.slow_memory: List[MemoryItem] = []

        self.consistency_threshold = consistency_threshold

    # ========== Fast Memory 更新 ==========
    def update_fast_memory(self, content: str, timestamp: float):
        embedding = self.embed_fn(content)

        item = MemoryItem(
            content=content,
            embedding=embedding,
            importance=self.compute_importance(content),
            timestamp=timestamp
        )

        self.fast_memory.append(item)

    # ========== 一致性校验 ==========
    def consistency_check(self, new_item: MemoryItem) -> bool:
        """
        判断是否应该写入 Slow Memory：
        1. 重要性是否达标
        2. 是否与已有 slow memory 冲突
        """
        if new_item.importance < 0.6:
            return False

        for old_item in self.slow_memory:
            similarity = self.cosine_similarity(
                new_item.embedding, old_item.embedding
            )
            if similarity > self.consistency_threshold:
                # 已存在相似长期记忆，不重复写入
                return False

        return True

    # ========== Slow Memory 更新 ==========
    def promote_to_slow_memory(self):
        """
        从 fast memory 中筛选稳定内容沉淀到 slow memory
        """
        for item in self.fast_memory:
            if self.consistency_check(item):
                self.slow_memory.append(item)

        # 可选：清理旧的 fast memory
        self.fast_memory = self.fast_memory[-20:]

    # ========== 检索 ==========
    def retrieve(self, query: str, top_k=5):
        query_emb = self.embed_fn(query)

        all_memory = self.slow_memory + self.fast_memory

        scored = [
            (item, self.cosine_similarity(query_emb, item.embedding))
            for item in all_memory
        ]

        scored.sort(key=lambda x: x[1], reverse=True)

        return [item.content for item, _ in scored[:top_k]]

    # ========== 工具函数 ==========
    def cosine_similarity(self, v1, v2):
        return np.dot(v1, v2) / (
            np.linalg.norm(v1) * np.linalg.norm(v2) + 1e-8
        )

    def compute_importance(self, content: str) -> float:
        """
        可替换为：
        - LLM 评分
        - 规则评分
        - 关键词加权
        """
        if len(content) > 80:
            return 0.8
        return 0.4

3. 这个中间件在系统里怎么插？

我把它做成一个“可插拔中间件”，放在 Agent 的执行链路中间：

用户输入

→ Agent 推理 / 工具调用

→ Fast Memory 更新（高频）

→ 必要时触发一致性校验

→ Slow Memory 低频沉淀

→ 检索（优先慢记忆，再补快记忆）

→ 拼回上下文 → 下一步推理

这样做的好处是：

Agent 仍然是 Agent，LLM 仍然是 LLM。

我只是换了一种方式，让“记忆进出”更像一个工程系统，而不是堆向量库。

4. 实验：Topical-Chat 全量 10,784 条对话

为了验证“长对话断流”是否真的改善，我在 Topical-Chat 全量 10,784 条对话上做了评估。

结果（我这里给最关键的两项）：

avg_recall +10.0%

d4（更长距离段）+21.7%

这两个数字对我来说意义非常明确：

短距离提升是正常的，

真正的价值是 d4 这种长距离段显著抬起来了——说明确实在缓解“长期记忆断流”。

同时在工程侧我也很关注成本问题：

这个中间件的设计目标之一就是别把 token 成本搞炸。

实际线上/推理链路里，整体 token 开销 保持可控、基本不显著增加（主要因为慢记忆更精简，减少了无效召回和无效上下文堆叠）。

5. 我认为它解决的不是“能不能记”，而是“怎么记才不乱”

很多人做记忆，第一反应是“加容量”。

但长对话的问题往往不是容量，而是：

记得太多导致噪声

更新太快导致漂移

没筛选导致冲突
最后检索质量下降，等价于失忆。

双速 + 一致性校验的核心价值就是：

把“记忆系统”从存储问题，升级为信息治理问题。

6. 总结

我在研究中引入了双速递归记忆中间件，通过 fast/slow 两级记忆和一致性校验机制，缓解长对话场景下的长期记忆断流问题；在 Topical-Chat 全量 10,784 对话上实现 avg_recall +10.0%、d4 +21.7%，同时线上 token 成本保持可控。

目前这个中间件已开源：开源中间件链接