记忆压缩高效管理 AI Agent 的记忆前言随着使用时间增长Agent 的记忆会越来越多需要压缩策略来保持系统高效。记忆压缩能节省存储空间并提升检索速度。我在多个项目中实现过记忆压缩今天分享一些策略和实现。重要性评估from typing import List, Dict, Any import time from dataclasses import dataclass dataclass class MemoryImportance: 记忆重要性 score: float reason: str factors: Dict[str, float] class ImportanceEvaluator: 重要性评估器 def __init__(self): pass def evaluate(self, memory: MemoryItem) - MemoryImportance: 评估重要性 factors {} # 访问频率 recency_factor self._recency_factor(memory) factors[recency] recency_factor # 访问次数 access_factor self._access_factor(memory) factors[access] access_factor # 内容重要性基于标签 content_factor self._content_factor(memory) factors[content] content_factor # 综合评分 total_score ( 0.4 * recency_factor 0.3 * access_factor 0.3 * content_factor ) reason f基于时间({recency_factor:.2f})、访问({access_factor:.2f})、内容({content_factor:.2f}) return MemoryImportance(total_score, reason, factors) def _recency_factor(self, memory: MemoryItem) - float: 时间因子 if memory.last_access: time_diff time.time() - memory.last_access else: time_diff time.time() - memory.timestamp # 时间越近分数越高 decay_hours 24 * 7 # 一周衰减 decay_factor max(0.1, 1.0 - time_diff / (decay_hours * 3600)) return decay_factor def _access_factor(self, memory: MemoryItem) - float: 访问因子 # 访问次数越多越重要 return min(1.0, memory.access_count / 10.0) def _content_factor(self, memory: MemoryItem) - float: 内容因子 important_tags {important, critical, key, user} for tag in memory.tags: if tag in important_tags: return 1.0 return 0.5摘要压缩from typing import List, Optional from abc import ABC, abstractmethod class MemorySummarizer(ABC): 记忆摘要器 abstractmethod def summarize(self, memory: MemoryItem) - MemoryItem: 生成摘要 pass class SimpleSummarizer(MemorySummarizer): 简单摘要器 def __init__(self, max_length: int 200): self.max_length max_length def summarize(self, memory: MemoryItem) - MemoryItem: 简单摘要 if memory.type ! MemoryType.TEXT: return memory content str(memory.content) if len(content) self.max_length: return memory # 简单截断 summary content[:self.max_length] ... # 创建摘要记忆 return MemoryItem( idmemory.id _summary, typeMemoryType.TEXT, contentsummary, metadata{ is_summary: True, original_id: memory.id, original_length: len(content) }, importancememory.importance * 0.9, # 稍微降低重要性 tagsmemory.tags [summary] ) class LLMSummarizer(MemorySummarizer): LLM 摘要器 def __init__(self, llm_client): self.llm_client llm_client def summarize(self, memory: MemoryItem) - MemoryItem: 使用 LLM 摘要 if memory.type ! MemoryType.TEXT: return memory content str(memory.content) # 使用 LLM 生成摘要 prompt f请为以下内容生成简洁摘要不超过100字:\n\n{content} summary self.llm_client.generate(prompt) return MemoryItem( idmemory.id _summary, typeMemoryType.TEXT, contentsummary, metadata{ is_summary: True, original_id: memory.id, original_length: len(content) }, importancememory.importance * 0.9, tagsmemory.tags [summary] )记忆压缩策略from typing import List, Tuple, Optional from dataclasses import dataclass dataclass class CompressionAction: 压缩动作 action: str # keep, summarize, delete memory_id: str reason: str class MemoryCompressor: 记忆压缩器 def __init__(self, evaluator: ImportanceEvaluator, summarizer: MemorySummarizer): self.evaluator evaluator self.summarizer summarizer def analyze(self, memories: List[MemoryItem]) - List[CompressionAction]: 分析并决定压缩动作 actions [] for memory in memories: importance self.evaluator.evaluate(memory) if importance.score 0.7: # 高重要性保留 action keep reason f高重要性 ({importance.score:.2f}) elif importance.score 0.3: # 中重要性摘要 action summarize reason f中重要性 ({importance.score:.2f}) else: # 低重要性删除 action delete reason f低重要性 ({importance.score:.2f}) actions.append(CompressionAction( actionaction, memory_idmemory.id, reasonreason )) return actions def compress(self, store: MemoryStore, keep_summary: bool True) - Tuple[int, int]: 执行压缩 memories list(store.memories.values()) actions self.analyze(memories) kept 0 compressed 0 deleted 0 for action in actions: memory store.memories.get(action.memory_id) if not memory: continue if action.action keep: kept 1 elif action.action summarize: if keep_summary: summary self.summarizer.summarize(memory) # 替换原记忆为摘要 store.memories[action.memory_id] summary compressed 1 else: del store.memories[action.memory_id] deleted 1 elif action.action delete: del store.memories[action.memory_id] deleted 1 # 重建索引 store._rebuild_index() return compressed, deleted压缩管理class MemoryManager: 记忆管理器 def __init__(self, store: MemoryStore, compressor: MemoryCompressor): self.store store self.compressor compressor self.max_memory_count 10000 self.compression_threshold 8000 def maybe_compress(self) - Tuple[int, int]: 检查并在需要时压缩 if len(self.store.memories) self.compression_threshold: compressed, deleted self.compressor.compress(self.store) return compressed, deleted return 0, 0 def periodic_cleanup(self): 定期清理 # 删除非常老的记忆 cutoff time.time() - (90 * 24 * 3600) # 90天前 to_delete [] for mid, mem in self.store.memories.items(): if mem.timestamp cutoff and mem.importance 0.2: to_delete.append(mid) for mid in to_delete: del self.store.memories[mid] if to_delete: self.store._rebuild_index() return len(to_delete)总结记忆压缩要点重要性评估多因素综合评估摘要策略保留核心信息压缩动作保留/摘要/删除定期清理管理记忆数量性能优化保持检索效率实践建议从简单策略开始监控压缩效果保留重要记忆定期测试压缩逻辑
记忆压缩：高效管理 AI Agent 的记忆
记忆压缩高效管理 AI Agent 的记忆前言随着使用时间增长Agent 的记忆会越来越多需要压缩策略来保持系统高效。记忆压缩能节省存储空间并提升检索速度。我在多个项目中实现过记忆压缩今天分享一些策略和实现。重要性评估from typing import List, Dict, Any import time from dataclasses import dataclass dataclass class MemoryImportance: 记忆重要性 score: float reason: str factors: Dict[str, float] class ImportanceEvaluator: 重要性评估器 def __init__(self): pass def evaluate(self, memory: MemoryItem) - MemoryImportance: 评估重要性 factors {} # 访问频率 recency_factor self._recency_factor(memory) factors[recency] recency_factor # 访问次数 access_factor self._access_factor(memory) factors[access] access_factor # 内容重要性基于标签 content_factor self._content_factor(memory) factors[content] content_factor # 综合评分 total_score ( 0.4 * recency_factor 0.3 * access_factor 0.3 * content_factor ) reason f基于时间({recency_factor:.2f})、访问({access_factor:.2f})、内容({content_factor:.2f}) return MemoryImportance(total_score, reason, factors) def _recency_factor(self, memory: MemoryItem) - float: 时间因子 if memory.last_access: time_diff time.time() - memory.last_access else: time_diff time.time() - memory.timestamp # 时间越近分数越高 decay_hours 24 * 7 # 一周衰减 decay_factor max(0.1, 1.0 - time_diff / (decay_hours * 3600)) return decay_factor def _access_factor(self, memory: MemoryItem) - float: 访问因子 # 访问次数越多越重要 return min(1.0, memory.access_count / 10.0) def _content_factor(self, memory: MemoryItem) - float: 内容因子 important_tags {important, critical, key, user} for tag in memory.tags: if tag in important_tags: return 1.0 return 0.5摘要压缩from typing import List, Optional from abc import ABC, abstractmethod class MemorySummarizer(ABC): 记忆摘要器 abstractmethod def summarize(self, memory: MemoryItem) - MemoryItem: 生成摘要 pass class SimpleSummarizer(MemorySummarizer): 简单摘要器 def __init__(self, max_length: int 200): self.max_length max_length def summarize(self, memory: MemoryItem) - MemoryItem: 简单摘要 if memory.type ! MemoryType.TEXT: return memory content str(memory.content) if len(content) self.max_length: return memory # 简单截断 summary content[:self.max_length] ... # 创建摘要记忆 return MemoryItem( idmemory.id _summary, typeMemoryType.TEXT, contentsummary, metadata{ is_summary: True, original_id: memory.id, original_length: len(content) }, importancememory.importance * 0.9, # 稍微降低重要性 tagsmemory.tags [summary] ) class LLMSummarizer(MemorySummarizer): LLM 摘要器 def __init__(self, llm_client): self.llm_client llm_client def summarize(self, memory: MemoryItem) - MemoryItem: 使用 LLM 摘要 if memory.type ! MemoryType.TEXT: return memory content str(memory.content) # 使用 LLM 生成摘要 prompt f请为以下内容生成简洁摘要不超过100字:\n\n{content} summary self.llm_client.generate(prompt) return MemoryItem( idmemory.id _summary, typeMemoryType.TEXT, contentsummary, metadata{ is_summary: True, original_id: memory.id, original_length: len(content) }, importancememory.importance * 0.9, tagsmemory.tags [summary] )记忆压缩策略from typing import List, Tuple, Optional from dataclasses import dataclass dataclass class CompressionAction: 压缩动作 action: str # keep, summarize, delete memory_id: str reason: str class MemoryCompressor: 记忆压缩器 def __init__(self, evaluator: ImportanceEvaluator, summarizer: MemorySummarizer): self.evaluator evaluator self.summarizer summarizer def analyze(self, memories: List[MemoryItem]) - List[CompressionAction]: 分析并决定压缩动作 actions [] for memory in memories: importance self.evaluator.evaluate(memory) if importance.score 0.7: # 高重要性保留 action keep reason f高重要性 ({importance.score:.2f}) elif importance.score 0.3: # 中重要性摘要 action summarize reason f中重要性 ({importance.score:.2f}) else: # 低重要性删除 action delete reason f低重要性 ({importance.score:.2f}) actions.append(CompressionAction( actionaction, memory_idmemory.id, reasonreason )) return actions def compress(self, store: MemoryStore, keep_summary: bool True) - Tuple[int, int]: 执行压缩 memories list(store.memories.values()) actions self.analyze(memories) kept 0 compressed 0 deleted 0 for action in actions: memory store.memories.get(action.memory_id) if not memory: continue if action.action keep: kept 1 elif action.action summarize: if keep_summary: summary self.summarizer.summarize(memory) # 替换原记忆为摘要 store.memories[action.memory_id] summary compressed 1 else: del store.memories[action.memory_id] deleted 1 elif action.action delete: del store.memories[action.memory_id] deleted 1 # 重建索引 store._rebuild_index() return compressed, deleted压缩管理class MemoryManager: 记忆管理器 def __init__(self, store: MemoryStore, compressor: MemoryCompressor): self.store store self.compressor compressor self.max_memory_count 10000 self.compression_threshold 8000 def maybe_compress(self) - Tuple[int, int]: 检查并在需要时压缩 if len(self.store.memories) self.compression_threshold: compressed, deleted self.compressor.compress(self.store) return compressed, deleted return 0, 0 def periodic_cleanup(self): 定期清理 # 删除非常老的记忆 cutoff time.time() - (90 * 24 * 3600) # 90天前 to_delete [] for mid, mem in self.store.memories.items(): if mem.timestamp cutoff and mem.importance 0.2: to_delete.append(mid) for mid in to_delete: del self.store.memories[mid] if to_delete: self.store._rebuild_index() return len(to_delete)总结记忆压缩要点重要性评估多因素综合评估摘要策略保留核心信息压缩动作保留/摘要/删除定期清理管理记忆数量性能优化保持检索效率实践建议从简单策略开始监控压缩效果保留重要记忆定期测试压缩逻辑
