一、引言：为什么记忆是Agent的"灵魂"

在人工智能的发展史上，记忆机制可能是区分"智能工具"与"真正智能体"的关键分水岭。一个不具备记忆能力的AI系统，无论其推理能力多么强大，本质上仍是一个无状态函数——每次交互都是独立的、遗忘的、缺乏上下文的。而人类智能的核心特征之一，正是能够基于过往经验进行学习、推理和决策。

AI Agent的记忆机制不仅仅是数据的简单存储，它是一个复杂的认知架构，涉及信息的编码、存储、检索、压缩和遗忘等多个环节。本文将深入剖析AI Agent记忆机制的核心算法与工程实现，从理论分类到实践架构，为构建具备持久认知能力的智能体提供完整的技术指南。

---

二、记忆机制的整体架构

AI Agent的记忆系统通常采用分层架构设计，核心包含三大记忆模块和四大核心操作：

2.1 三大记忆模块

模块	功能描述	典型实现
短期记忆 (STM)	维护当前对话上下文和工作状态	上下文窗口、对话缓冲区
长期记忆 (LTM)	存储持久化的知识和经验	向量数据库、知识图谱
外部记忆	扩展Agent的知识边界	文档检索、API查询

2.2 四大核心操作

• 检索 ( Retrieval )：从记忆库中高效定位相关信息
• 存储 ( Storage )：将新经验持久化到记忆系统
• 压缩 ( Compression )：减少记忆冗余，提升存储效率
• 遗忘 ( Forgetting )：管理记忆容量，淘汰低价值信息

---

三、记忆的分类体系

3.1 基于持续时间的分类

感觉记忆 (Sensory Memory)

• 持续时间：< 1秒
• 容量：极大（接收所有感官输入）
• 作用：作为信息进入认知系统的第一道门户
• 实现：输入缓冲区，如语音流、图像帧的临时缓存

短期记忆 (Short-term Memory)

• 持续时间：分钟到小时级别
• 容量：有限（约7±2个信息块，Miller定律）
• 作用：维护当前任务的工作上下文
• 实现：LLM的上下文窗口（Context Window），如GPT-4的128K tokens

长期记忆 (Long-term Memory)

• 持续时间：天到永久
• 容量：理论上无限
• 作用：存储持久知识和经验
• 实现：向量数据库、外部存储系统

3.2 基于内容类型的分类

借鉴认知心理学的记忆分类，AI Agent的长期记忆可分为：

情景记忆 (Episodic Memory)

• 存储特定事件序列和个人经历
• 示例：“用户上周三询问过Python异步编程”
• 实现：时间戳索引的对话历史

语义记忆 (Semantic Memory)

• 存储事实、概念和世界知识
• 示例：“Python的async/await语法”
• 实现：知识图谱、结构化数据库

程序记忆 (Procedural Memory)

• 存储技能和操作程序
• 示例：“如何编写高效的异步爬虫”
• 实现：工具调用模板、工作流定义

3.3 基于存储方式的分类

参数化记忆 (Parametric)

• 知识编码在模型权重中
• 优点：推理速度快，无需外部依赖
• 缺点：更新困难，存在幻觉风险

非参数化记忆 (Non-parametric)

• 知识存储在外部系统中
• 优点：易于更新，可解释性强
• 缺点：需要检索开销

混合记忆 (Hybrid)

• 结合参数化和非参数化
• RAG (Retrieval-Augmented Generation) 是典型代表
• 兼顾灵活性和效率

---

四、检索与存储流水线

4.1 检索流程

记忆检索是一个多阶段流水线，核心目标是：在给定查询下，从海量记忆中最快、最准地找到相关信息。

阶段一：查询编码

# 将自然语言查询编码为向量表示
query_embedding = embedding_model.encode("用户之前的偏好是什么？")

阶段二：相似度搜索

• 向量相似度：使用余弦相似度或点积计算查询向量与记忆向量的匹配度
• 混合搜索：结合稠密向量（语义匹配）和稀疏向量（关键词匹配）
• 重排序：使用交叉编码器（Cross-encoder）对初筛结果精细排序

阶段三：上下文组装

将检索到的相关记忆片段组装成LLM可消费的上下文格式。

4.2 存储流程

新经验的写入并非简单的追加，而是经过多道过滤和加工：

class MemoryWriter:
    def capture_experience(self, interaction):
        """捕获原始交互"""
        raw_memory = self.extract_key_info(interaction)
        return raw_memory
    
    def score_importance(self, memory):
        """评估记忆重要性"""
        # 使用LLM或规则引擎打分
        importance = llm_judge.score(memory)
        return importance
    
    def compress(self, memory):
        """压缩记忆内容"""
        if len(memory) > threshold:
            return summarizer.summarize(memory)
        return memory
    
    def store_decision(self, memory, importance):
        """决定是否存储及存储位置"""
        if importance > STM_THRESHOLD:
            self.short_term.store(memory)
        if importance > LTM_THRESHOLD:
            self.long_term.store(memory)

---

五、压缩与遗忘机制

5.1 记忆压缩策略

随着Agent运行时间增长，记忆量会呈指数级膨胀。压缩机制在保持信息价值的同时减少存储开销：

1. 摘要压缩 (Summarization)

使用LLM将长对话历史浓缩为关键要点：

summary = llm.generate(
    f"将以下对话总结为3个关键点：\n{conversation_history}"
)

2. 层次化抽象 (Hierarchical Abstraction)

构建多层级记忆结构：

• L0: 原始对话记录
• L1: 会话级摘要
• L2: 主题级聚合
• L3: 用户画像级抽象

3. 选择性保留 (Selective Filtering)

基于重要性评分，仅保留高价值信息：

important_memories = [
    m for m in memories 
    if m.importance_score > RETENTION_THRESHOLD
]

4. 向量降维 (Vector Dimensionality Reduction)

对高维嵌入向量进行PCA或量化压缩，减少存储空间。

5.2 遗忘策略

遗忘不是缺陷，而是主动的认知优化。有效的遗忘机制防止记忆过载：

策略	原理	适用场景
时间衰减	旧记忆按指数函数降低权重	对话系统
相关性遗忘	低频访问记忆被淘汰	推荐系统
容量限制	固定槽位，FIFO/LRU替换	实时系统
重要性阈值	低重要性记忆定期清理	通用Agent

时间衰减公式

memory_score = base_importance × e^(-λ × time_elapsed)

其中 λ 是衰减系数，控制遗忘速度。

5.3 压缩与遗忘的权衡

维度	压缩	遗忘
记忆大小	↓ 减少	↑ 保留
检索质量	↓ 可能退化	↑ 更高质量
计算成本	↓ 更低	↑ 更高
上下文覆盖	↓ 选择性	↑ 全面性

最佳实践：两者结合——压缩重要记忆，遗忘无关信息。

---

六、多Agent记忆共享架构

在多Agent协作系统中，记忆共享是集体智能的关键：

6.1 共享记忆层设计

class SharedMemoryLayer:
    def __init__(self):
        self.common_knowledge = KnowledgeBase()      # 公共知识库
        self.experience_pool = VectorStore()         # 共享经验池
        self.collective_learnings = GraphStore()     # 集体学习成果
    
    def share(self, agent_id, memory):
        """Agent向共享层写入记忆"""
        self.experience_pool.add(memory, source=agent_id)
    
    def retrieve(self, query, agent_id=None):
        """从共享层检索记忆"""
        return self.experience_pool.search(query)

6.2 同步协议

协议	机制	优点	缺点
广播 (Broadcast)	推送更新到所有Agent	一致性高	扩展性差
Gossip	点对点传播	容错性好	收敛慢
共识 (Consensus)	达成共享状态一致	强一致性	延迟高
版本控制	追踪记忆演进历史	可回溯	复杂度高

6.3 核心挑战

1. 一致性：确保所有Agent看到一致的记忆状态
2. 隐私保护：敏感信息在共享空间中需要脱敏
3. 可扩展性：Agent数量和记忆规模增长时的性能
4. 冲突解决：合并来自不同Agent的矛盾记忆

---

七、记忆机制对比与选型

7.1 四种核心机制对比

特性	上下文窗口	向量数据库	知识图谱	混合RAG
存储格式	文本tokens	稠密向量	图结构	多模态
检索方式	顺序扫描	相似度搜索	图遍历	多阶段
上下文长度	有限(~128K)	无限	无限	无限
更新速度	即时	快	中等	快
推理深度	浅层	中等	深层	深层
可扩展性	低	高	高	非常高

7.2 选型指南

应用场景	推荐方案	理由
简单对话系统	上下文窗口	低复杂度，实时性好
文档搜索	向量数据库	语义匹配能力强
关系查询	知识图谱	实体关系推理
企业级系统	混合RAG	综合能力最优
实时系统	上下文窗口+缓存	速度优先

---

八、向量索引的算法本质：HNSW vs IVF vs PQ

8.1 问题定义：高维空间中的近邻搜索

给定查询向量 $q\in {\mathbb{R}}^D$ 和数据库 X={x1,...,xN}⊂RDX = \{x_1, ..., x_N\} \subset \mathbb{R}^DX={x1​,...,xN​}⊂RD，精确最近邻搜索需要计算：

$$\text{NN}(q)=\arg \underset{x\in X}{\min }\Vert q-x{\Vert }_2$$

其时间复杂度为 $O(N\cdot D)$。当 $N=10^9$、$D=1024$ 时，暴力搜索完全不可行。近似最近邻（ANN） 成为唯一工程可行的路径。

8.2 HNSW：可导航小世界图

核心思想：构建多层图结构，上层稀疏、下层稠密，实现 $O(\log N)$ 的搜索复杂度。

算法细节：

• 图构建：对每个新节点，从顶层随机选择入口，贪婪搜索至当前层最近邻，建立双向连接。
• 参数 $M$：每个节点的最大出度，控制图密度与召回率的权衡。
• 参数 $ef$（exploration factor）：搜索时的候选池大小，直接影响召回率。

复杂度分析：

• 构建：$O(N\cdot M\cdot \log N)$
• 查询：$O(M\cdot \log N)$
• 内存：$O(N\cdot M\cdot D)$

适用场景：百万级向量、高召回率要求（>95%）、内存充裕

8.3 IVF：倒排文件索引

核心思想：先聚类再搜索，将全局搜索转化为局部搜索。

算法细节：

1. 使用K-means将向量空间划分为 $nlist$ 个聚类中心
2. 查询时仅搜索最近的 $nprobe$ 个聚类
3. 搜索复杂度从 $O(N)$ 降至 $O(N\cdot nprobe/nlist)$

复杂度分析：

• 构建：$O(N\cdot nlist\cdot D)$（K-means迭代）
• 查询：$O(nprobe\cdot (N/nlist)\cdot D)$
• 内存：$O(N\cdot D)$（存储原始向量）

适用场景：亿级向量、内存受限、可接受稍低召回率（90-95%）

8.4 PQ：乘积量化

核心思想：将高维向量分解为多个低维子空间，每个子空间独立量化。

数学表达：
$$x\in {\mathbb{R}}^D\to (c_1(x),c_2(x),...,c_M(x))$$

其中 $M$ 为子空间数，每个子空间使用 $k$ 个质心（通常 $k=256$，8 bits）。

压缩率：

• 原始：$N\cdot D\cdot 4$ bytes（float32）
• PQ后：$N\cdot M$ bytes
• 当 $D=1024,M=32$ 时，压缩率 ≈ 32×

复杂度分析：

• 查询：$O(N\cdot M)$（使用ADC非对称距离计算）
• 内存：$O(N\cdot M)$

适用场景：十亿级向量、极端内存约束、可接受5-10%召回损失

8.5 混合索引：IVF-PQ 与 HNSW+PQ

IVF-PQ：先用IVF粗筛，再用PQ精排。兼顾速度与压缩。
HNSW+PQ：在HNSW图中使用PQ压缩的向量计算距离。保持高召回的同时降低内存。

生产推荐：

• 百万级：HNSW（M=16, ef=200）
• 千万级：IVF-HNSW
• 亿级以上：IVF-PQ（nlist=4×√N, nprobe=100-200）

---

九、嵌入模型的选择：不只是"向量化"

9.1 模型选择的多维权衡

维度	考量因素	工程影响
维度	768 vs 1024 vs 4096	高维度=高表达能力，但索引内存与计算成本线性增长
上下文长度	512 vs 8192 tokens	长文档需要长上下文模型，否则信息截断
多语言	是否支持跨语言	全球化Agent的必备能力
指令微调	是否针对检索优化	未经微调的模型可能将"问题"与"答案"映射到不同区域

9.2 主流模型深度对比

OpenAI text-embedding-3-large

• 维度：3072
• 优势：MTEB基准领先，多语言能力强
• 劣势：成本高，延迟大，闭源不可控
• 适用： accuracy-critical金融、医疗场景

BGE-M3（北京智源）

• 维度：1024
• 优势：开源免费，多语言，支持稀疏+稠密混合检索
• 劣势：长文档处理能力弱于E5-mistral
• 适用：开源RAG系统的首选

E5-mistral

• 维度：4096
• 优势：MTEB最高分（66.3），长上下文（32K tokens）
• 劣势：推理成本高，需要GPU加速
• 适用：长文档法律、科研文献检索

GTE-large（阿里）

• 维度：1024
• 优势：速度极快，中文优化好
• 劣势：多语言弱，无稀疏检索
• 适用：中文为主、速度敏感的场景

9.3 混合检索：Dense + Sparse

纯稠密向量检索在处理精确匹配、罕见实体、缩写时表现不佳。混合检索结合两者优势：

$$\text{Score}(q,d)=\alpha \cdot {\text{sim}}_{dense}(q,d)+(1-\alpha )\cdot {\text{sim}}_{sparse}(q,d)$$

其中 $\alpha$ 通常设为 0.7-0.8。BGE-M3原生支持此模式，无需额外稀疏编码器。

---

十、记忆压缩：信息论的视角

10.1 率-失真权衡（Rate-Distortion Trade-off）

记忆压缩的本质是有损压缩。根据信息论，压缩率 $R$ 与失真 $D$ 满足：

$$R(D)=\underset{p(\hat{x}|x):E[d(x,\hat{x})]\le D}{\min }I(X;\hat{X})$$

其中 $I(X;\hat{X})$ 是互信息，$d(\cdot ,\cdot )$ 是失真度量。

工程启示：不存在"无损高压缩"的银弹。每个压缩决策都是信息保留与存储成本的显式权衡。

10.2 四级压缩流水线

L1: 选择性保留（Selective Retention）

• 零信息损失，仅做筛选
• 评分函数：$S(m)=\alpha \cdot \text{relevance}+\beta \cdot \text{recency}+\gamma \cdot \text{importance}+\delta \cdot \text{frequency}$
• 保留Top-K，其余进入压缩队列

L2: LLM摘要（Summarization）

• 提取式：选择关键句子（信息损失10-15%）
• 生成式：重写为新文本（信息损失20-30%，但连贯性更好）
• 压缩比：5×-20×

L3: 层次化抽象（Hierarchical Abstraction）

• L0: 原始token（完整保留24h）
• L1: 会话摘要（保留7天）
• L2: 主题聚合（保留30天）
• L3: 用户画像（永久保留）

访问复杂度：$O(\log L)$，$L$ 为层级数

L4: 向量量化（Vector Quantization）

• 将1024维float32向量压缩为32字节
• 存储降低32×，召回损失5-10%
• 适用于冷数据归档

10.3 压缩的评估指标

指标	定义	目标值
压缩比	原始大小 / 压缩后大小	> 10×
语义保留率	压缩后检索召回 / 原始检索召回	> 90%
KL散度	$D_{KL}(P_{orig}|P_{compressed})$	< 0.1
人工评估	人类判断信息完整性	> 4.0/5.0

---

十一、遗忘机制：形式化模型与工程实现

11.1 统一保留函数

我们提出一个多因子记忆保留函数：

$$R(t)=R_0\cdot e^{-\lambda t}\cdot (1+\alpha f{)}^{\beta }\cdot (1+\gamma i{)}^{\delta }\cdot (1-\sigma c)$$

其中：

• $R_0$：初始记忆强度
• $\lambda$：时间衰减系数
• $f$：访问频率
• $i$：重要性评分
• $c$：冲突计数（与其他记忆矛盾的次数）
• $\alpha ,\beta ,\gamma ,\delta ,\sigma$：可调超参数

遗忘决策：当 $R(t)<{\theta }_{forget}$ 时，记忆进入"待清理队列"，经二次确认后删除。

11.2 四种遗忘策略的数学本质

指数衰减（Exponential Decay）
$$R(t)=R_0\cdot e^{-\lambda t}$$

• 符合放射性衰变模型，生物学上合理
• 问题：所有记忆同等衰减，不区分价值

幂律衰减（Ebbinghaus）
$$R(t)=R_0\cdot t^{-b}$$

• 更符合人类遗忘曲线
• 初始遗忘慢，长期遗忘快
• 适合学习系统与间隔重复算法

相关性加权（Relevance-Weighted）
$$R=\frac{\sum w_i{r}_i}{\sum w_i}$$

• $w_i$：第 $i$ 次访问的权重（越近权重越高）
• 完全自适应，但计算开销大

容量限制（LRU/LFU）
$$\text{Evict}=\arg \underset{m\in M}{\min }R(m)$$

• 固定容量 $K$，超出时淘汰最低分记忆
• $O(1)$ 操作，适合实时系统

11.3 生产级遗忘系统的设计

class ProductionForgettingEngine:
    def __init__(self, capacity: int, decay_lambda: float = 0.01):
        self.capacity = capacity
        self.decay_lambda = decay_lambda
        self.memories = {}  # id -> MemoryMetadata
        self.access_log = defaultdict(list)  # id -> [timestamp]
        
    def compute_retention(self, memory_id: str) -> float:
        meta = self.memories[memory_id]
        time_elapsed = time.time() - meta.created_at
        
        # 时间衰减
        time_factor = math.exp(-self.decay_lambda * time_elapsed)
        
        # 访问频率增强
        recent_accesses = sum(
            1 for t in self.access_log[memory_id] 
            if time.time() - t < 86400  # 24h window
        )
        freq_factor = (1 + 0.1 * recent_accesses) ** 0.5
        
        # 重要性
        importance_factor = (1 + meta.importance) ** 0.3
        
        # 冲突惩罚
        conflict_factor = max(0, 1 - 0.2 * meta.conflict_count)
        
        return meta.base_strength * time_factor * freq_factor * importance_factor * conflict_factor
    
    def evict_if_needed(self):
        if len(self.memories) < self.capacity * 0.9:
            return  # 提前退出，避免不必要的计算
            
        # 计算所有记忆的保留分数
        scores = {
            mid: self.compute_retention(mid) 
            for mid in self.memories
        }
        
        # 淘汰最低分的10%
        sorted_memories = sorted(scores.items(), key=lambda x: x[1])
        evict_count = int(len(sorted_memories) * 0.1)
        
        for mid, _ in sorted_memories[:evict_count]:
            if scores[mid] < 0.3:  # 硬阈值
                self._archive_or_delete(mid)

---

十二、多Agent记忆一致性：分布式系统视角

12.1 从单Agent到多Agent的复杂性跃迁

单Agent的记忆系统是一个中心化数据库。多Agent系统则是一个分布式数据库，面临所有经典分布式问题：

• 网络分区：Agent间通信中断
• 脑裂：多个Agent同时认为自己是"权威"
• 冲突写入：两个Agent对同一记忆产生矛盾更新
• 因果依赖：Agent B的推理依赖Agent A的记忆更新，但A的更新延迟到达

12.2 Raft共识：强一致性方案

将Raft协议适配到Agent记忆同步：

角色定义：

• Leader Agent：记忆协调者，接收所有写入请求
• Follower Agent：从Leader复制记忆日志
• Candidate：Leader失效时参与选举

记忆同步流程：

1. Append Entries：Leader将新记忆打包为日志条目，发送给所有Follower
2. Majority Ack：超过半数Follower确认接收后，Leader提交该记忆
3. Commit & Apply：所有节点将已提交记忆应用到本地状态机

复杂度：

• 写入延迟：2次RTT（Leader→Follower→Leader）
• 读取延迟：本地读取（Leader）或 1次RTT（Follower一致性读）
• 容错：可容忍 $(N-1)/2$ 个节点故障

适用场景：金融交易Agent、医疗诊断Agent等对一致性要求极高的场景。

12.3 Gossip协议：最终一致性方案

核心思想：去中心化，每个Agent随机选择伙伴交换记忆差异。

传播动力学：

• 每轮每个Agent随机选择 $k$ 个邻居（通常 $k=3$）
• 交换记忆摘要（Merkle Tree），仅传输差异部分
• 收敛时间：$O(\log N)$ 轮

CRDT（无冲突复制数据类型）
对于可合并的记忆类型（如计数器、集合、图），使用CRDT确保无论以何种顺序合并，最终结果一致。

class GCounter(CRDT):
    """ grow-only counter for memory access counts """
    def __init__(self):
        self.payload = defaultdict(int)  # agent_id -> count
        
    def increment(self, agent_id):
        self.payload[agent_id] += 1
        
    def merge(self, other):
        for agent_id, count in other.payload.items():
            self.payload[agent_id] = max(self.payload[agent_id], count)
            
    def read(self):
        return sum(self.payload.values())

适用场景：社交媒体Agent、内容推荐Agent等可容忍短暂不一致的场景。

12.4 一致性模型的选择矩阵

一致性模型	协议	保证	延迟	适用场景
强一致性	Raft/Paxos	所有Agent看到相同状态	高（2-3 RTT）	金融、医疗
因果一致性	Vector Clocks	保留因果关系	中	协同编辑
会话一致性	Sticky Routing	单用户视角一致	低	客服Agent
最终一致性	Gossip + CRDT	暂时分歧可接受	极低	社交、内容

---

十三、端到端生产架构

13.1 六层流水线

L1: 输入层

• 多模态输入：文本、图像、音频、结构化数据
• 预处理：去噪、标准化、敏感信息检测

L2: 嵌入层

• 模型：BGE-M3（开源）或 text-embedding-3-large（闭源）
• 批处理：动态batching，GPU利用率>80%
• 缓存：L1（进程内LRU）+ L2（Redis）

L3: 索引层

• 热数据：HNSW（内存）
• 温数据：IVF（SSD）
• 冷数据：IVF-PQ（对象存储）

L4: 检索层

• 第一阶段：稠密向量检索（Top-100）
• 第二阶段：稀疏关键词过滤
• 第三阶段：Cross-encoder重排序（Top-10）

L5: 记忆管理器

• 压缩管道：选择性保留→摘要→量化→归档
• 遗忘调度：每10分钟运行一次淘汰算法
• 版本控制：Git-like语义化版本

L6: LLM上下文组装

• 动态优先级：系统提示 > 用户画像 > 近期对话 > 相关记忆
• Token预算管理：预留20%给检索结果，80%给生成

13.2 性能基准（生产目标）

指标	目标	说明
嵌入延迟	< 50ms (p99)	含网络往返
索引构建	< 1h/百万向量	含HNSW图构建
查询延迟	< 20ms (p99)	含重排序
召回率@10	> 95%	对比暴力搜索
压缩比	10-20×	多层压缩后
同步开销	< 5%	占整体吞吐量
存储成本	$0.1/百万向量/月	含三副本

---

十四、完整可运行代码：生产级记忆系统

"""
Production-Grade Agent Memory System
Author: AI Research Team
Requirements: numpy, faiss-cpu, sentence-transformers, redis
"""

import time
import math
import hashlib
import numpy as np
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass, field
from collections import defaultdict
from datetime import datetime
import json

# 模拟依赖（实际生产中使用真实库）
class MockEmbeddingModel:
    """模拟嵌入模型，生产环境替换为 sentence-transformers """
    def __init__(self, dim: int = 768):
        self.dim = dim
        np.random.seed(42)
        self.cache = {}
        
    def encode(self, texts: List[str]) -> np.ndarray:
        """生成确定性嵌入（模拟）"""
        embeddings = []
        for text in texts:
            if text not in self.cache:
                # 使用文本哈希生成确定性向量
                hash_val = int(hashlib.md5(text.encode()).hexdigest(), 16)
                np.random.seed(hash_val % (2**32))
                vec = np.random.randn(self.dim).astype(np.float32)
                vec = vec / np.linalg.norm(vec)
                self.cache[text] = vec
            embeddings.append(self.cache[text])
        return np.array(embeddings)

class MockVectorDB:
    """模拟向量数据库，生产环境替换为 FAISS/Milvus/Pinecone """
    def __init__(self, dim: int, index_type: str = "HNSW"):
        self.dim = dim
        self.vectors = []
        self.metadata = []
        self.index_type = index_type
        
    def add(self, vectors: np.ndarray, metadata: List[Dict]):
        self.vectors.extend(vectors)
        self.metadata.extend(metadata)
        
    def search(self, query: np.ndarray, top_k: int = 10) -> List[Tuple[int, float, Dict]]:
        """暴力搜索（生产环境使用ANN索引）"""
        if not self.vectors:
            return []
        
        vectors = np.array(self.vectors)
        similarities = np.dot(vectors, query[0])
        top_indices = np.argsort(similarities)[::-1][:top_k]
        
        return [(int(idx), float(similarities[idx]), self.metadata[idx]) 
                for idx in top_indices]

@dataclass
class MemoryEntry:
    """记忆条目"""
    id: str
    content: str
    embedding: np.ndarray
    timestamp: float
    importance: float = 0.5
    access_count: int = 0
    last_accessed: float = field(default_factory=time.time)
    conflict_count: int = 0
    compressed: bool = False
    compression_level: int = 0  # 0=原始, 1=摘要, 2=量化
    
    def to_dict(self) -> Dict:
        return {
            "id": self.id,
            "content": self.content[:100] + "..." if len(self.content) > 100 else self.content,
            "timestamp": self.timestamp,
            "importance": self.importance,
            "access_count": self.access_count,
            "compressed": self.compressed
        }

class TieredMemoryStore:
    """分层记忆存储：热/温/冷"""
    
    def __init__(self, embedding_dim: int = 768, capacity: int = 10000):
        self.dim = embedding_dim
        self.capacity = capacity
        
        # 三层存储
        self.hot_memories: Dict[str, MemoryEntry] = {}      # 内存，最近24h
        self.warm_db = MockVectorDB(embedding_dim, "IVF")     # 向量DB，最近30天
        self.cold_archive: List[MemoryEntry] = []             # 对象存储，永久
        
        # 统计
        self.stats = {
            "total_stored": 0,
            "total_retrieved": 0,
            "total_compressed": 0,
            "total_forgotten": 0
        }
        
    def store(self, entry: MemoryEntry, force_tier: Optional[str] = None):
        """存储记忆到适当层级"""
        self.stats["total_stored"] += 1
        
        age_hours = (time.time() - entry.timestamp) / 3600
        
        if force_tier:
            tier = force_tier
        elif age_hours < 24 and len(self.hot_memories) < self.capacity * 0.1:
            tier = "hot"
        elif age_hours < 720:  # 30天
            tier = "warm"
        else:
            tier = "cold"
            
        if tier == "hot":
            self.hot_memories[entry.id] = entry
        elif tier == "warm":
            self.warm_db.add(entry.embedding.reshape(1, -1), [entry.to_dict()])
        else:
            self.cold_archive.append(entry)
            
    def retrieve(self, query_embedding: np.ndarray, top_k: int = 10) -> List[MemoryEntry]:
        """跨层级检索"""
        results = []
        
        # 1. 热数据（内存，最快）
        if self.hot_memories:
            hot_vectors = np.array([m.embedding for m in self.hot_memories.values()])
            hot_ids = list(self.hot_memories.keys())
            similarities = np.dot(hot_vectors, query_embedding)
            hot_top = np.argsort(similarities)[::-1][:min(3, len(similarities))]
            
            for idx in hot_top:
                mid = hot_ids[idx]
                self.hot_memories[mid].access_count += 1
                self.hot_memories[mid].last_accessed = time.time()
                results.append(self.hot_memories[mid])
        
        # 2. 温数据（向量DB）
        warm_results = self.warm_db.search(query_embedding.reshape(1, -1), top_k=min(5, top_k))
        for _, score, meta in warm_results:
            # 实际生产环境从DB反序列化完整对象
            pass  # 简化处理
            
        # 3. 冷数据（仅当温数据不足时）
        if len(results) < top_k:
            # 从冷归档中筛选（生产环境使用独立索引）
            pass
            
        self.stats["total_retrieved"] += 1
        return results[:top_k]

class MemoryCompressionEngine:
    """记忆压缩引擎"""
    
    def __init__(self, compression_threshold: float = 0.7):
        self.threshold = compression_threshold
        
    def should_compress(self, entry: MemoryEntry) -> bool:
        """判断是否需要压缩"""
        age_days = (time.time() - entry.timestamp) / 86400
        if age_days > 7 and entry.access_count < 3:
            return True
        if entry.compression_level >= 2:
            return False
        return False
    
    def summarize(self, content: str, ratio: float = 0.3) -> str:
        """提取式摘要（生产环境使用LLM）"""
        sentences = content.split("。")
        keep_count = max(1, int(len(sentences) * ratio))
        return "。".join(sentences[:keep_count]) + "..."
    
    def quantize(self, embedding: np.ndarray, bits: int = 8) -> np.ndarray:
        """向量量化"""
        # 简单均匀量化，生产环境使用PQ
        min_val, max_val = embedding.min(), embedding.max()
        quantized = np.round((embedding - min_val) / (max_val - min_val) * (2**bits - 1))
        return quantized.astype(np.uint8)
    
    def compress(self, entry: MemoryEntry) -> MemoryEntry:
        """执行压缩"""
        if entry.compression_level == 0:
            # L1: 摘要
            entry.content = self.summarize(entry.content)
            entry.compression_level = 1
        elif entry.compression_level == 1:
            # L2: 量化
            entry.embedding = self.quantize(entry.embedding)
            entry.compressed = True
            entry.compression_level = 2
            
        return entry

class ForgettingEngine:
    """遗忘引擎"""
    
    def __init__(self, capacity: int, decay_lambda: float = 0.001):
        self.capacity = capacity
        self.decay_lambda = decay_lambda
        self.retention_threshold = 0.2
        
    def compute_retention_score(self, entry: MemoryEntry) -> float:
        """计算保留分数"""
        time_elapsed = time.time() - entry.timestamp
        time_factor = math.exp(-self.decay_lambda * time_elapsed)
        
        # 访问频率增强（24h窗口）
        freq_boost = min(2.0, 1 + 0.1 * entry.access_count)
        
        # 重要性
        importance_boost = (1 + entry.importance) ** 0.5
        
        # 冲突惩罚
        conflict_penalty = max(0.5, 1 - 0.1 * entry.conflict_count)
        
        base = 1.0
        return base * time_factor * freq_boost * importance_boost * conflict_penalty
    
    def select_for_eviction(self, memories: List[MemoryEntry], target_count: int) -> List[str]:
        """选择待淘汰记忆"""
        if len(memories) <= target_count:
            return []
            
        scores = [(m.id, self.compute_retention_score(m)) for m in memories]
        scores.sort(key=lambda x: x[1])
        
        evict_count = len(memories) - target_count
        return [mid for mid, _ in scores[:evict_count]]

class AgentMemorySystem:
    """完整的Agent记忆系统"""
    
    def __init__(self, embedding_dim: int = 768, max_memories: int = 10000):
        self.embedding_model = MockEmbeddingModel(embedding_dim)
        self.store = TieredMemoryStore(embedding_dim, max_memories)
        self.compressor = MemoryCompressionEngine()
        self.forgetting = ForgettingEngine(max_memories)
        
        # 运行统计
        self.query_count = 0
        self.total_latency = 0.0
        
    def memorize(self, content: str, importance: float = 0.5) -> str:
        """记忆新信息"""
        start = time.time()
        
        # 1. 生成嵌入
        embedding = self.embedding_model.encode([content])[0]
        
        # 2. 创建记忆条目
        entry = MemoryEntry(
            id=hashlib.sha256(f"{content}{time.time()}".encode()).hexdigest()[:16],
            content=content,
            embedding=embedding,
            timestamp=time.time(),
            importance=importance
        )
        
        # 3. 存储
        self.store.store(entry)
        
        # 4. 触发压缩检查（异步）
        self._async_compress()
        
        # 5. 触发遗忘检查（异步）
        self._async_forget()
        
        latency = time.time() - start
        self.total_latency += latency
        
        return entry.id
    
    def recall(self, query: str, top_k: int = 5) -> List[Tuple[str, float]]:
        """回忆相关信息"""
        start = time.time()
        
        # 1. 查询嵌入
        query_emb = self.embedding_model.encode([query])[0]
        
        # 2. 检索
        results = self.store.retrieve(query_emb, top_k)
        
        # 3. 格式化输出
        output = [(r.content, float(np.dot(query_emb, r.embedding))) for r in results]
        
        self.query_count += 1
        self.total_latency += (time.time() - start)
        
        return output
    
    def _async_compress(self):
        """异步压缩（生产环境使用后台线程/Celery）"""
        all_memories = list(self.store.hot_memories.values())
        for mem in all_memories:
            if self.compressor.should_compress(mem):
                self.compressor.compress(mem)
                self.store.stats["total_compressed"] += 1
    
    def _async_forget(self):
        """异步遗忘"""
        all_memories = list(self.store.hot_memories.values())
        evict_ids = self.forgetting.select_for_eviction(
            all_memories, 
            int(self.store.capacity * 0.9)
        )
        for mid in evict_ids:
            if mid in self.store.hot_memories:
                del self.store.hot_memories[mid]
                self.store.stats["total_forgotten"] += 1
    
    def get_stats(self) -> Dict:
        """获取系统统计"""
        avg_latency = self.total_latency / max(1, self.query_count + self.store.stats["total_stored"])
        return {
            "queries": self.query_count,
            "stored": self.store.stats["total_stored"],
            "retrieved": self.store.stats["total_retrieved"],
            "compressed": self.store.stats["total_compressed"],
            "forgotten": self.store.stats["total_forgotten"],
            "avg_latency_ms": round(avg_latency * 1000, 2),
            "hot_memory_count": len(self.store.hot_memories)
        }

# ============ 演示 ============
if __name__ == "__main__":
    print("=" * 60)
    print("Production-Grade Agent Memory System Demo")
    print("=" * 60)
    
    # 初始化系统
    memory = AgentMemorySystem(embedding_dim=768, max_memories=1000)
    
    # 模拟记忆写入
    test_memories = [
        "用户张三偏好Python编程，讨厌Java",
        "张三上周购买了MacBook Pro M3",
        "张三的GitHub账号是zhangsan-dev",
        "张三需要构建一个AI Agent项目",
        "张三的预算限制在5000元以内",
        "张三偏好开源解决方案而非商业API",
        "张三有3年后端开发经验",
        "张三正在学习向量数据库",
        "张三的项目 deadline 是下周五",
        "张三团队有5名成员",
    ]
    
    print("\n[1] 写入记忆...")
    for i, content in enumerate(test_memories):
        mid = memory.memorize(content, importance=0.6 + 0.02 * i)
        print(f"  ✓ Memory {i+1}: {content[:40]}... → ID: {mid}")
    
    # 模拟查询
    print("\n[2] 检索记忆...")
    queries = [
        "张三喜欢什么编程语言？",
        "张三的硬件配置是什么？",
        "张三的项目时间要求？",
        "张三团队规模？",
    ]
    
    for query in queries:
        results = memory.recall(query, top_k=3)
        print(f"\n  Query: {query}")
        for content, score in results:
            print(f"    → {content[:50]}... (score: {score:.3f})")
    
    # 系统统计
    print("\n[3] 系统统计")
    stats = memory.get_stats()
    for key, val in stats.items():
        print(f"  {key}: {val}")
    
    print("\n" + "=" * 60)
    print("Demo completed. Replace Mock* classes with production implementations.")
    print("=" * 60)

---

十五、边缘情况与工程陷阱

15.1 向量漂移（Vector Drift）

问题：当嵌入模型更新后，新旧向量不在同一语义空间，导致检索失效。

解决方案：

• 版本化嵌入：存储 embedding_model_version 字段
• 迁移策略：批量重新编码旧数据，或维护多版本索引
• 兼容性层：使用归一化 + 投影矩阵对齐不同版本

15.2 记忆冲突（Memory Conflict）

问题：两个来源对同一实体给出矛盾信息。

解决方案：

def resolve_conflict(memories: List[MemoryEntry]) -> MemoryEntry:
    # 策略1: 时间优先
    # 策略2: 来源可信度加权
    # 策略3: 保留所有并标记矛盾
    # 策略4: 触发LLM仲裁
    credibility = {
        "user_explicit": 1.0,
        "inferred": 0.6,
        "third_party": 0.4
    }
    return max(memories, key=lambda m: credibility.get(m.source, 0.5))

15.3 上下文污染（Context Pollution）

问题：检索到的"相关记忆"实际上引入了干扰信息，降低LLM输出质量。

解决方案：

• 相关性阈值过滤：丢弃 score < 0.7 的结果
• 多样性重排序：使用 MMR（Maximal Marginal Relevance）平衡相关性与多样性
• 反事实检索：显式排除已知的错误信息

15.4 冷启动（Cold Start）

问题：新Agent无历史记忆，检索返回空或噪声。

解决方案：

• 预填充通用知识库
• 使用用户画像模板（ demographics, preferences 的默认值）
• 主动询问策略：前3轮对话中强制收集关键信息

---

十六、未来方向：记忆机制的下一个十年

16.1 神经-符号混合记忆

将神经网络的模式识别能力与符号系统的可解释性结合：

• 神经网络负责相似性匹配与模糊检索
• 知识图谱负责逻辑推理与关系遍历
• 两者通过神经符号接口双向转换

16.2 持续学习（Continual Learning）

解决"灾难性遗忘"问题：

• 弹性权重巩固（EWC）：保护对旧任务重要的网络参数
• 渐进式神经网络：为新任务添加新列，不修改旧列
• 经验回放：在训练新任务时混合旧任务样本

16.3 情感记忆与情境意识

超越事实存储，记录情感状态与情境上下文：

• “用户上次提到这个项目时很焦虑”
• “在周一早晨，用户对邮件通知更敏感”

这需要多模态嵌入（文本+语音语调+时间戳+环境传感器）。

16.4 可解释记忆

Agent能够解释"为什么记得这个"：

• 记忆溯源：这条记忆来自哪次对话、哪个来源
• 置信度校准：我知道这条记忆可能不准确
• 主动遗忘请求：用户要求"忘掉我的密码"

---

结语

AI Agent的记忆机制，本质上是一个分布式信息检索系统与认知架构的交叉领域。从向量索引的数学优化，到信息论的压缩极限，再到分布式系统的一致性协议——每一个层面都蕴含着深厚的计算机科学原理。

但技术的终极目标不是复杂性本身，而是让Agent真正"理解"用户：知道用户的偏好、记住用户的经历、在合适的时机回忆起合适的记忆。当Agent能够像一位相识多年的老友一样与你对话时，那将是记忆机制最美的实现。

---

附

完整代码如下：

```python
"""
Production-Grade Agent Memory System
Author: AI Research Team
Requirements: numpy, faiss-cpu, sentence-transformers, redis
"""

import time
import math
import hashlib
import numpy as np
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass, field
from collections import defaultdict
from datetime import datetime
import json


# ============================================================================
# 模拟依赖层（生产环境替换为真实库）
# ============================================================================

class MockEmbeddingModel:
    """
    模拟嵌入模型，生产环境替换为 sentence-transformers 或 OpenAI API
    支持模型: BGE-M3, E5-mistral, text-embedding-3-large, GTE-large
    """
    def __init__(self, dim: int = 768):
        self.dim = dim
        np.random.seed(42)
        self.cache = {}
        
    def encode(self, texts: List[str]) -> np.ndarray:
        """生成确定性嵌入（模拟）"""
        embeddings = []
        for text in texts:
            if text not in self.cache:
                hash_val = int(hashlib.md5(text.encode()).hexdigest(), 16)
                np.random.seed(hash_val % (2**32))
                vec = np.random.randn(self.dim).astype(np.float32)
                vec = vec / np.linalg.norm(vec)
                self.cache[text] = vec
            embeddings.append(self.cache[text])
        return np.array(embeddings)


class MockVectorDB:
    """
    模拟向量数据库，生产环境替换为 FAISS / Milvus / Pinecone / Qdrant
    支持索引类型: HNSW, IVF, IVF-PQ, HNSW+PQ
    """
    def __init__(self, dim: int, index_type: str = "HNSW"):
        self.dim = dim
        self.vectors = []
        self.metadata = []
        self.index_type = index_type
        
    def add(self, vectors: np.ndarray, metadata: List[Dict]):
        self.vectors.extend(vectors)
        self.metadata.extend(metadata)
        
    def search(self, query: np.ndarray, top_k: int = 10) -> List[Tuple[int, float, Dict]]:
        """暴力搜索（生产环境使用ANN索引: HNSW/IVF-PQ）"""
        if not self.vectors:
            return []
        
        vectors = np.array(self.vectors)
        similarities = np.dot(vectors, query[0])
        top_indices = np.argsort(similarities)[::-1][:top_k]
        
        return [(int(idx), float(similarities[idx]), self.metadata[idx]) 
                for idx in top_indices]


# ============================================================================
# 核心数据结构
# ============================================================================

@dataclass
class MemoryEntry:
    """
    记忆条目 - Agent记忆系统的原子单元
    
    字段说明:
        id: 唯一标识符 (SHA-256前16位)
        content: 文本内容
        embedding: 稠密向量表示 (float32)
        timestamp: 创建时间戳 (Unix epoch)
        importance: 重要性评分 [0.0, 1.0]
        access_count: 被访问次数
        last_accessed: 最后访问时间
        conflict_count: 与其他记忆的冲突次数
        compressed: 是否已被压缩
        compression_level: 压缩层级 (0=原始, 1=摘要, 2=量化)
        source: 记忆来源 (user_explicit, inferred, third_party)
        tags: 标签列表 (用于分类和过滤)
    """
    id: str
    content: str
    embedding: np.ndarray
    timestamp: float
    importance: float = 0.5
    access_count: int = 0
    last_accessed: float = field(default_factory=time.time)
    conflict_count: int = 0
    compressed: bool = False
    compression_level: int = 0
    source: str = "inferred"
    tags: List[str] = field(default_factory=list)
    
    def to_dict(self) -> Dict:
        """序列化为字典（用于存储和传输）"""
        return {
            "id": self.id,
            "content": self.content[:200] + "..." if len(self.content) > 200 else self.content,
            "timestamp": datetime.fromtimestamp(self.timestamp).isoformat(),
            "importance": self.importance,
            "access_count": self.access_count,
            "compressed": self.compressed,
            "compression_level": self.compression_level,
            "source": self.source,
            "tags": self.tags
        }
    
    def compute_age_hours(self) -> float:
        """计算记忆年龄（小时）"""
        return (time.time() - self.timestamp) / 3600
    
    def compute_recency_score(self, decay_hours: float = 24.0) -> float:
        """
        计算时效性分数
        使用指数衰减: score = exp(-age / decay_hours)
        """
        age = self.compute_age_hours()
        return math.exp(-age / decay_hours)


# ============================================================================
# 分层记忆存储
# ============================================================================

class TieredMemoryStore:
    """
    分层记忆存储系统
    
    三层架构:
        Hot (热数据): 内存存储, 最近24小时, 最快访问
        Warm (温数据): 向量数据库, 最近30天, 平衡速度
        Cold (冷数据): 对象存储归档, 永久保留, 按需加载
    
    生产环境替换:
        Hot  -> Python dict / Redis
        Warm -> FAISS / Milvus / Pinecone
        Cold -> S3 / MinIO / 本地磁盘
    """
    
    # 层级配置
    HOT_MAX_AGE_HOURS = 24
    WARM_MAX_AGE_HOURS = 720  # 30天
    HOT_CAPACITY_RATIO = 0.1  # 热数据占总容量的10%
    
    def __init__(self, embedding_dim: int = 768, capacity: int = 10000):
        self.dim = embedding_dim
        self.capacity = capacity
        
        # 三层存储
        self.hot_memories: Dict[str, MemoryEntry] = {}
        self.warm_db = MockVectorDB(embedding_dim, "IVF")
        self.cold_archive: List[MemoryEntry] = []
        
        # 访问日志 (用于遗忘算法)
        self.access_log: Dict[str, List[float]] = defaultdict(list)
        
        # 统计信息
        self.stats = {
            "total_stored": 0,
            "total_retrieved": 0,
            "total_migrated_hot_to_warm": 0,
            "total_migrated_warm_to_cold": 0,
            "cache_hits": 0,
            "cache_misses": 0
        }
        
    def _determine_tier(self, entry: MemoryEntry) -> str:
        """根据记忆年龄和系统状态决定存储层级"""
        age_hours = entry.compute_age_hours()
        hot_capacity = int(self.capacity * self.HOT_CAPACITY_RATIO)
        
        if age_hours < self.HOT_MAX_AGE_HOURS and len(self.hot_memories) < hot_capacity:
            return "hot"
        elif age_hours < self.WARM_MAX_AGE_HOURS:
            return "warm"
        else:
            return "cold"
    
    def store(self, entry: MemoryEntry, force_tier: Optional[str] = None):
        """
        存储记忆到适当层级
        
        Args:
            entry: 记忆条目
            force_tier: 强制指定层级 (hot/warm/cold), None则自动判断
        """
        self.stats["total_stored"] += 1
        
        tier = force_tier or self._determine_tier(entry)
        
        if tier == "hot":
            self.hot_memories[entry.id] = entry
        elif tier == "warm":
            self.warm_db.add(entry.embedding.reshape(1, -1), [entry.to_dict()])
        else:
            # 冷数据: 量化压缩后归档
            entry.compressed = True
            entry.compression_level = 2
            self.cold_archive.append(entry)
            
    def retrieve(self, query_embedding: np.ndarray, top_k: int = 10) -> List[MemoryEntry]:
        """
        跨层级检索记忆
        
        检索策略:
            1. 优先查询热数据 (O(1) 内存访问)
            2. 补充温数据 (ANN搜索)
            3. 必要时查询冷数据 (全量扫描, 生产环境使用独立索引)
        
        Returns:
            按相关性排序的记忆列表
        """
        results = []
        scored_results = []
        
        # ===== L1: 热数据检索 =====
        if self.hot_memories:
            hot_ids = list(self.hot_memories.keys())
            hot_vectors = np.array([self.hot_memories[mid].embedding for mid in hot_ids])
            similarities = np.dot(hot_vectors, query_embedding)
            
            # 更新访问统计
            for idx, mid in enumerate(hot_ids):
                self.hot_memories[mid].access_count += 1
                self.hot_memories[mid].last_accessed = time.time()
                self.access_log[mid].append(time.time())
                scored_results.append((self.hot_memories[mid], float(similarities[idx])))
            
            self.stats["cache_hits"] += 1
        else:
            self.stats["cache_misses"] += 1
        
        # ===== L2: 温数据检索 =====
        warm_results = self.warm_db.search(query_embedding.reshape(1, -1), top_k=min(5, top_k))
        for _, score, meta in warm_results:
            # 生产环境: 从数据库反序列化完整 MemoryEntry
            pass  # 简化处理, 实际需实现反序列化
        
        # ===== L3: 冷数据检索 (仅当上层不足时) =====
        if len(scored_results) < top_k and self.cold_archive:
            # 生产环境: 冷数据应有独立索引或定期回温策略
            cold_vectors = np.array([m.embedding for m in self.cold_archive])
            cold_similarities = np.dot(cold_vectors, query_embedding)
            cold_top = np.argsort(cold_similarities)[::-1][:min(2, top_k - len(scored_results))]
            
            for idx in cold_top:
                scored_results.append((self.cold_archive[idx], float(cold_similarities[idx])))
        
        # 按相似度排序并返回
        scored_results.sort(key=lambda x: x[1], reverse=True)
        self.stats["total_retrieved"] += 1
        
        return [entry for entry, _ in scored_results[:top_k]]
    
    def migrate_tiers(self):
        """
        层级迁移: 将过期热数据移至温层, 过期温数据移至冷层
        
        应在后台定时任务中执行 (Celery / APScheduler)
        """
        current_time = time.time()
        
        # 热 -> 温
        to_migrate_hot = []
        for mid, entry in self.hot_memories.items():
            if entry.compute_age_hours() > self.HOT_MAX_AGE_HOURS:
                to_migrate_hot.append(mid)
        
        for mid in to_migrate_hot:
            entry = self.hot_memories.pop(mid)
            self.warm_db.add(entry.embedding.reshape(1, -1), [entry.to_dict()])
            self.stats["total_migrated_hot_to_warm"] += 1
        
        # 温 -> 冷 (简化: 生产环境需遍历向量DB)
        # self.stats["total_migrated_warm_to_cold"] += ...
        
    def get_memory_count_by_tier(self) -> Dict[str, int]:
        """获取各层级记忆数量"""
        return {
            "hot": len(self.hot_memories),
            "warm": len(self.warm_db.vectors),
            "cold": len(self.cold_archive)
        }


# ============================================================================
# 记忆压缩引擎
# ============================================================================

class MemoryCompressionEngine:
    """
    记忆压缩引擎 - 四级压缩流水线
    
    压缩策略:
        L0 -> L1: 选择性保留 (无信息损失, 仅筛选)
        L1 -> L2: LLM摘要 (有损, 保留核心语义)
        L2 -> L3: 向量量化 (有损, 降低存储)
        L3 -> L4: 归档删除 (仅保留元数据)
    
    信息论视角: 每次压缩都是 Rate-Distortion 权衡
    """
    
    def __init__(
        self,
        compression_threshold: float = 0.7,
        summary_ratio: float = 0.3,
        quantization_bits: int = 8
    ):
        self.threshold = compression_threshold
        self.summary_ratio = summary_ratio
        self.quantization_bits = quantization_bits
        
    def should_compress(self, entry: MemoryEntry) -> bool:
        """
        判断记忆是否需要压缩
        
        触发条件:
            1. 年龄超过7天且访问次数 < 3
            2. 已达到最高压缩级别
            3. 系统内存压力 (由外部触发)
        """
        age_days = entry.compute_age_hours() / 24
        
        if entry.compression_level >= 2:
            return False  # 已最高压缩
            
        if age_days > 7 and entry.access_count < 3:
            return True
            
        if entry.compression_level == 0 and age_days > 1:
            return True  # 1天后开始L1压缩
            
        return False
    
    def summarize(self, content: str, ratio: float = None) -> str:
        """
        提取式摘要
        
        生产环境替换为 LLM 调用:
            - 提取式: 选择关键句子
            - 生成式: 重写摘要 (信息损失更大但连贯性更好)
        
        Args:
            content: 原始文本
            ratio: 保留比例 (默认 self.summary_ratio)
        """
        ratio = ratio or self.summary_ratio
        sentences = [s.strip() for s in content.replace("。", ".").split(".") if s.strip()]
        
        if len(sentences) <= 2:
            return content  # 太短不压缩
            
        keep_count = max(1, int(len(sentences) * ratio))
        return ". ".join(sentences[:keep_count]) + "..."
    
    def quantize(self, embedding: np.ndarray, bits: int = None) -> np.ndarray:
        """
        向量量化 - 均匀量化
        
        生产环境替换为 Product Quantization (PQ):
            - 将 D 维向量分解为 M 个子空间
            - 每个子空间使用 k-means 训练码本
            - 存储索引而非原始向量
        
        压缩率: 32x (float32 -> 8-bit) 到 128x (PQ)
        
        Args:
            embedding: 原始向量 (float32)
            bits: 量化位数 (默认 8)
        """
        bits = bits or self.quantization_bits
        min_val, max_val = embedding.min(), embedding.max()
        
        if max_val == min_val:
            return np.zeros_like(embedding, dtype=np.uint8)
        
        # 线性映射到 [0, 2^bits - 1]
        quantized = np.round(
            (embedding - min_val) / (max_val - min_val) * (2**bits - 1)
        )
        return quantized.astype(np.uint8)
    
    def compress(self, entry: MemoryEntry) -> MemoryEntry:
        """
        执行压缩升级
        
        压缩流程:
            Level 0 -> 1: 文本摘要
            Level 1 -> 2: 向量量化
        """
        if entry.compression_level == 0:
            # L0 -> L1: 摘要
            entry.content = self.summarize(entry.content)
            entry.compression_level = 1
            entry.compressed = True
            
        elif entry.compression_level == 1:
            # L1 -> L2: 量化
            entry.embedding = self.quantize(entry.embedding)
            entry.compression_level = 2
            
        return entry


# ============================================================================
# 遗忘引擎
# ============================================================================

class ForgettingEngine:
    """
    遗忘引擎 - 基于多因子保留函数的记忆淘汰
    
    保留函数:
        R(t) = R0 * exp(-λt) * (1 + αf)^β * (1 + γi)^δ * (1 - σc)
    
    其中:
        R0: 初始强度
        λ: 时间衰减系数
        f: 访问频率 (24h窗口)
        i: 重要性
        c: 冲突计数
        α, β, γ, δ, σ: 可调超参数
    """
    
    def __init__(
        self,
        capacity: int,
        decay_lambda: float = 0.001,  # 每小时衰减0.1%
        retention_threshold: float = 0.2,
        # 超参数
        alpha: float = 0.1,   # 频率系数
        beta: float = 0.5,    # 频率指数
        gamma: float = 1.0,   # 重要性系数
        delta: float = 0.3,   # 重要性指数
        sigma: float = 0.1    # 冲突惩罚系数
    ):
        self.capacity = capacity
        self.decay_lambda = decay_lambda
        self.retention_threshold = retention_threshold
        self.alpha = alpha
        self.beta = beta
        self.gamma = gamma
        self.delta = delta
        self.sigma = sigma
        
    def compute_retention_score(self, entry: MemoryEntry) -> float:
        """
        计算记忆的保留分数 [0.0, ∞)
        
        分数 > threshold: 保留
        分数 < threshold: 进入淘汰候选队列
        """
        time_elapsed = time.time() - entry.timestamp
        
        # 1. 时间衰减 (指数衰减)
        time_factor = math.exp(-self.decay_lambda * time_elapsed)
        
        # 2. 访问频率增强 (24小时窗口)
        recent_accesses = sum(
            1 for t in getattr(entry, 'access_log', [])
            if time.time() - t < 86400
        )
        freq_factor = (1 + self.alpha * recent_accesses) ** self.beta
        
        # 3. 重要性增强
        importance_factor = (1 + self.gamma * entry.importance) ** self.delta
        
        # 4. 冲突惩罚
        conflict_factor = max(0.1, 1 - self.sigma * entry.conflict_count)
        
        # 基础强度 (可配置)
        base_strength = 1.0
        
        return base_strength * time_factor * freq_factor * importance_factor * conflict_factor
    
    def select_for_eviction(
        self, 
        memories: List[MemoryEntry], 
        target_count: int
    ) -> List[str]:
        """
        选择待淘汰的记忆ID列表
        
        策略:
            1. 计算所有记忆的保留分数
            2. 按分数升序排序
            3. 淘汰最低分的记忆, 直到数量达到 target_count
        
        Returns:
            待淘汰的记忆ID列表
        """
        if len(memories) <= target_count:
            return []
        
        # 计算保留分数
        scores = [
            (m.id, self.compute_retention_score(m)) 
            for m in memories
        ]
        
        # 按分数升序排序 (分数低的优先淘汰)
        scores.sort(key=lambda x: x[1])
        
        # 计算需要淘汰的数量
        evict_count = len(memories) - target_count
        
        # 只淘汰分数低于阈值的 (硬阈值保护)
        evict_ids = []
        for mid, score in scores[:evict_count]:
            if score < self.retention_threshold:
                evict_ids.append(mid)
            else:
                break  # 剩余记忆分数都高于阈值, 停止淘汰
                
        return evict_ids
    
    def apply_forgetting(self, store: TieredMemoryStore) -> int:
        """
        执行遗忘操作
        
        Returns:
            实际淘汰的记忆数量
        """
        all_memories = list(store.hot_memories.values())
        
        # 目标容量: 总容量的90% (预留10%缓冲)
        target_count = int(self.capacity * 0.9)
        
        evict_ids = self.select_for_eviction(all_memories, target_count)
        
        # 执行淘汰
        for mid in evict_ids:
            if mid in store.hot_memories:
                # 生产环境: 归档到冷存储而非直接删除
                del store.hot_memories[mid]
                store.stats["total_forgotten"] += 1
        
        return len(evict_ids)


# ============================================================================
# 主系统: Agent记忆系统
# ============================================================================

class AgentMemorySystem:
    """
    完整的Agent记忆系统 - 生产级实现
    
    架构组件:
        - EmbeddingModel: 文本 -> 向量
        - TieredMemoryStore: 热/温/冷三层存储
        - CompressionEngine: 四级压缩流水线
        - ForgettingEngine: 多因子遗忘算法
    
    性能目标:
        - 嵌入延迟: < 50ms (p99)
        - 查询延迟: < 20ms (p99)
        - 召回率@10: > 95%
        - 压缩比: 10-20x
    """
    
    def __init__(
        self,
        embedding_dim: int = 768,
        max_memories: int = 10000,
        embedding_model = None
    ):
        # 嵌入模型 (可注入, 便于测试和替换)
        self.embedding_model = embedding_model or MockEmbeddingModel(embedding_dim)
        
        # 核心组件
        self.store = TieredMemoryStore(embedding_dim, max_memories)
        self.compressor = MemoryCompressionEngine()
        self.forgetting = ForgettingEngine(max_memories)
        
        # 性能统计
        self.query_count = 0
        self.store_count = 0
        self.total_latency = 0.0
        
        # 配置
        self.compression_interval = 600  # 每10分钟触发压缩 (秒)
        self.forgetting_interval = 600
        self.last_compression = time.time()
        self.last_forgetting = time.time()
        
    def memorize(
        self, 
        content: str, 
        importance: float = 0.5,
        source: str = "inferred",
        tags: List[str] = None
    ) -> str:
        """
        记忆新信息
        
        流程:
            1. 生成嵌入向量
            2. 创建记忆条目
            3. 存储到适当层级
            4. 触发异步压缩检查
            5. 触发异步遗忘检查
        
        Args:
            content: 记忆内容文本
            importance: 重要性评分 [0.0, 1.0]
            source: 来源类型
            tags: 标签列表
        
        Returns:
            记忆ID
        """
        start = time.time()
        
        # 1. 生成嵌入 (生产环境: 批处理 + GPU加速)
        embedding = self.embedding_model.encode([content])[0]
        
        # 2. 创建记忆条目
        entry = MemoryEntry(
            id=hashlib.sha256(f"{content}{time.time()}".encode()).hexdigest()[:16],
            content=content,
            embedding=embedding,
            timestamp=time.time(),
            importance=max(0.0, min(1.0, importance)),  # 归一化
            source=source,
            tags=tags or []
        )
        
        # 3. 存储
        self.store.store(entry)
        self.store_count += 1
        
        # 4. 触发后台维护 (生产环境: Celery任务)
        self._trigger_maintenance()
        
        # 5. 记录延迟
        latency = time.time() - start
        self.total_latency += latency
        
        return entry.id
    
    def recall(
        self, 
        query: str, 
        top_k: int = 5,
        filters: Dict = None
    ) -> List[Tuple[str, float, Dict]]:
        """
        回忆相关信息
        
        流程:
            1. 查询嵌入
            2. 跨层级检索 (热 -> 温 -> 冷)
            3. 重排序 (生产环境: Cross-encoder)
            4. 返回结果 + 相关性分数
        
        Args:
            query: 查询文本
            top_k: 返回结果数量
            filters: 过滤条件 (tags, source, time_range等)
        
        Returns:
            [(content, score, metadata), ...]
        """
        start = time.time()
        
        # 1. 查询嵌入
        query_emb = self.embedding_model.encode([query])[0]
        
        # 2. 检索
        results = self.store.retrieve(query_emb, top_k * 2)  # 检索更多用于重排序
        
        # 3. 计算精确相似度并排序
        scored_results = []
        for entry in results:
            score = float(np.dot(query_emb, entry.embedding))
            scored_results.append((entry, score))
        
        scored_results.sort(key=lambda x: x[1], reverse=True)
        
        # 4. 格式化输出
        output = []
        for entry, score in scored_results[:top_k]:
            output.append((
                entry.content,
                score,
                entry.to_dict()
            ))
        
        self.query_count += 1
        self.total_latency += (time.time() - start)
        
        return output
    
    def update_importance(self, memory_id: str, delta: float):
        """
        动态更新记忆重要性
        
        使用场景:
            - 用户显式标记"这很重要"
            - 系统根据使用频率自动调整
            - 冲突检测后降低可信度
        """
        if memory_id in self.store.hot_memories:
            entry = self.store.hot_memories[memory_id]
            entry.importance = max(0.0, min(1.0, entry.importance + delta))
    
    def _trigger_maintenance(self):
        """触发后台维护任务"""
        current_time = time.time()
        
        # 压缩检查
        if current_time - self.last_compression > self.compression_interval:
            self._async_compress()
            self.last_compression = current_time
        
        # 遗忘检查
        if current_time - self.last_forgetting > self.forgetting_interval:
            self._async_forget()
            self.last_forgetting = current_time
    
    def _async_compress(self):
        """
        异步压缩 (生产环境: Celery / APScheduler)
        
        策略:
            1. 遍历热数据
            2. 判断是否需要压缩
            3. 执行压缩升级
        """
        for mid, entry in list(self.store.hot_memories.items()):
            if self.compressor.should_compress(entry):
                self.compressor.compress(entry)
                self.store.stats["total_compressed"] += 1
    
    def _async_forget(self):
        """
        异步遗忘 (生产环境: Celery / APScheduler)
        """
        forgotten = self.forgetting.apply_forgetting(self.store)
        if forgotten > 0:
            print(f"[Maintenance] Forgotten {forgotten} memories")
    
    def get_stats(self) -> Dict:
        """
        获取系统统计信息
        
        Returns:
            包含各项性能指标的字典
        """
        total_ops = self.query_count + self.store_count
        avg_latency = self.total_latency / max(1, total_ops)
        
        tier_counts = self.store.get_memory_count_by_tier()
        
        return {
            "queries": self.query_count,
            "stores": self.store_count,
            "avg_latency_ms": round(avg_latency * 1000, 2),
            "hot_memories": tier_counts["hot"],
            "warm_memories": tier_counts["warm"],
            "cold_memories": tier_counts["cold"],
            "total_compressed": self.store.stats["total_compressed"],
            "total_forgotten": self.store.stats["total_forgotten"],
            "cache_hit_ratio": round(
                self.store.stats["cache_hits"] / max(1, 
                    self.store.stats["cache_hits"] + self.store.stats["cache_misses"]
                ), 2
            )
        }
    
    def export_memory_graph(self) -> Dict:
        """
        导出记忆图谱 (用于可视化)
        
        生产环境: 导出到 Neo4j / NetworkX 进行图分析
        """
        nodes = []
        edges = []
        
        for mid, entry in self.store.hot_memories.items():
            nodes.append({
                "id": mid,
                "content": entry.content[:50],
                "importance": entry.importance,
                "tags": entry.tags
            })
            
            # 基于标签和相似度构建边
            for other_id, other in self.store.hot_memories.items():
                if mid >= other_id:
                    continue
                similarity = float(np.dot(entry.embedding, other.embedding))
                if similarity > 0.8:
                    edges.append({
                        "source": mid,
                        "target": other_id,
                        "weight": similarity
                    })
        
        return {"nodes": nodes, "edges": edges}


# ============================================================================
# 演示与测试
# ============================================================================

def run_demo():
    """运行完整演示"""
    print("=" * 70)
    print("Production-Grade Agent Memory System - Demo")
    print("=" * 70)
    
    # 初始化系统
    memory = AgentMemorySystem(embedding_dim=768, max_memories=1000)
    
    # 模拟用户记忆写入
    test_memories = [
        ("用户张三偏好Python编程，讨厌Java语法", 0.8, "user_explicit", ["preference", "tech"]),
        ("张三上周在京东购买了MacBook Pro M3 16GB", 0.7, "inferred", ["purchase", "hardware"]),
        ("张三的GitHub账号是zhangsan-dev，有50个stars", 0.6, "user_explicit", ["profile", "social"]),
        ("张三需要构建一个AI Agent项目用于客服自动化", 0.9, "user_explicit", ["project", "goal"]),
        ("张三的预算限制在5000元人民币以内", 0.8, "user_explicit", ["constraint", "budget"]),
        ("张三偏好开源解决方案而非调用商业API", 0.7, "inferred", ["preference", "strategy"]),
        ("张三有3年后端开发经验，熟悉Django和FastAPI", 0.6, "user_explicit", ["profile", "experience"]),
        ("张三正在学习向量数据库，已了解FAISS和Milvus", 0.5, "inferred", ["learning", "tech"]),
        ("张三的项目 deadline 是下周五下午5点", 0.9, "user_explicit", ["project", "deadline"]),
        ("张三团队有5名成员，包括2名前端和3名后端", 0.6, "user_explicit", ["team", "structure"]),
        ("张三之前尝试过LangChain但觉得太复杂", 0.5, "inferred", ["history", "feedback"]),
        ("张三希望Agent能记住用户的长期偏好", 0.8, "user_explicit", ["requirement", "memory"]),
    ]
    
    print("\n[Phase 1] 写入记忆...")
    for i, (content, importance, source, tags) in enumerate(test_memories):
        mid = memory.memorize(content, importance=importance, source=source, tags=tags)
        print(f"  ✓ [{i+1:2d}] {content[:45]:45s}... | ID: {mid} | importance: {importance}")
    
    # 模拟查询
    print("\n[Phase 2] 检索记忆...")
    queries = [
        "张三喜欢什么编程语言？",
        "张三最近买了什么硬件？",
        "张三的项目有什么要求？",
        "张三的团队规模？",
        "张三对商业API的态度？",
        "张三的技术栈是什么？",
    ]
    
    for query in queries:
        results = memory.recall(query, top_k=3)
        print(f"\n  🔍 Query: {query}")
        for i, (content, score, meta) in enumerate(results):
            print(f"     [{i+1}] score={score:.3f} | {content[:60]}...")
            print(f"         tags: {meta.get('tags', [])} | source: {meta.get('source', 'unknown')}")
    
    # 系统统计
    print("\n[Phase 3] 系统统计")
    stats = memory.get_stats()
    for key, val in stats.items():
        print(f"  • {key:20s}: {val}")
    
    # 记忆图谱导出
    print("\n[Phase 4] 记忆图谱")
    graph = memory.export_memory_graph()
    print(f"  • 节点数: {len(graph['nodes'])}")
    print(f"  • 边数: {len(graph['edges'])}")
    if graph['edges']:
        print(f"  • 最强关联: {graph['edges'][0]['source'][:8]} <-> {graph['edges'][0]['target'][:8]} (sim={graph['edges'][0]['weight']:.3f})")
    
    print("\n" + "=" * 70)
    print("Demo completed.")
    print("Production checklist:")
    print("  □ Replace MockEmbeddingModel with sentence-transformers")
    print("  □ Replace MockVectorDB with FAISS/Milvus/Pinecone")
    print("  □ Add async task queue (Celery/RQ)")
    print("  □ Add persistence layer (PostgreSQL/MongoDB)")
    print("  □ Add monitoring (Prometheus/Grafana)")
    print("  □ Add A/B testing for compression/forgetting params")
    print("=" * 70)


if __name__ == "__main__":
    run_demo()