💾 记忆系统

一句话总结：记忆系统是 Agent 的"大脑硬盘"，让模型在对话和任务中保持上下文连贯性和经验积累。

📋 目录

🧩 记忆分类

记忆层级对比

层级容量保留时间检索方式类比
工作记忆几千 Token当前会话直接访问工作台面
短期记忆几万 Token数天/周顺序/最近记事本
长期记忆无限永久语义检索档案库
程序记忆模型参数永久内隐肌肉记忆

⏱️ 短期记忆

滑动窗口管理

摘要压缩

class ShortTermMemory:
    def __init__(self, max_tokens=8000, llm=None):
        self.messages = []
        self.max_tokens = max_tokens
        self.llm = llm
    
    def add_message(self, message):
        self.messages.append(message)
        self._trim_if_needed()
    
    def _trim_if_needed(self):
        """当超出窗口时，使用摘要压缩"""
        current_tokens = count_tokens(self.messages)
        if current_tokens > self.max_tokens:
            # 保留最近 N 轮 + 摘要旧对话
            recent = self.messages[-6:]  # 最近 6 轮
            old = self.messages[:-6]
            summary = self.llm.generate(
                f"请总结以下对话：{''.join(old)}"
            )
            self.messages = [{"role": "system", "content": f"[对话摘要]: {summary}"}] + recent

关键对话提取

# 不是所有对话都同样重要
importance_score = calculate_importance(message, context)
# 高重要性标记：
# - 用户提供了关键信息
# - 达成了重要决定
# - 触发了异常状态

📚 长期记忆

记忆类型

事件记忆存储

@dataclass
class Episode:
    """事件记忆：一次完整的交互事件"""
    timestamp: str
    user_query: str
    agent_action: str
    outcome: str
    metadata: dict
    embedding: list  # 向量表示
    
class EpisodicMemory:
    def __init__(self, vector_store):
        self.store = vector_store
        self.indexed = set()
    
    def store_episode(self, episode: Episode):
        if episode.timestamp in self.indexed:
            return
        self.store.add(episode.embedding, {
            "episode": episode.__dict__,
            "timestamp": episode.timestamp
        })
        self.indexed.add(episode.timestamp)
    
    def retrieve_relevant(self, query: str, top_k=5):
        results = self.store.search(query, top_k=top_k)
        return [r["episode"] for r in results]

语义记忆构建

# 从事件记忆中提取语义知识
def extract_semantic_memory(episodes: List[Episode], llm):
    """定期从事件记忆中提炼语义知识"""
    prompt = """
    分析以下用户交互事件，提取通用的用户偏好、习惯和知识。
    
    事件列表：
    {episodes}
    
    请提取：
    1. 用户偏好（如编程语言偏好、风格偏好）
    2. 用户习惯（如常用工具、工作流程）
    3. 领域知识（用户关注的技术方向）
    
    以结构化格式输出。
    """
    semantic = llm.generate(prompt.format(episodes=episodes))
    return parse_semantic(semantic)

🗄️ 向量数据库

向量检索流程

常用向量数据库

数据库特点适用场景
Chroma轻量、嵌入式本地开发、小型项目
Milvus高性能、分布式大规模向量检索
Pinecone全托管 SaaS快速上线、生产环境
Weaviate混合搜索向量 + 关键词联合
QdrantRust 实现高性能、日本开发

相似度度量

方法公式特点
Cosine$\cos(\theta) = \frac{A \cdot B}{|A||B|}$最常用，关注方向
L2$d = |A - B|_2$关注绝对距离
Dot Product$A \cdot B$结合大小和方向

Embedding 模型选择

模型维度速度质量推荐场景
text-embedding-3-small1536快⭐⭐⭐通用场景
text-embedding-3-large3072中⭐⭐⭐⭐⭐高质量检索
bge-large1024快⭐⭐⭐⭐开源方案
nomic-embed768快⭐⭐⭐⭐本地部署

🔄 记忆管理

记忆写入策略

记忆遗忘策略

策略描述适用场景
时间衰减按时间衰减重要性通用
重要性衰减基于信息重要性长期记忆
冗余消除合并相似记忆语义记忆
主动遗忘定期清理低频记忆资源受限

检索策略

class MemoryRetriever:
    def retrieve(self, context, agent_state):
        """多策略记忆检索"""
        memories = []
        
        # 1. 语义检索（基于内容相似性）
        semantic = self.vector_store.search(
            query=context, top_k=10
        )
        memories.extend(semantic)
        
        # 2. 时间检索（最近的重要记忆）
        temporal = self.get_recent_memories(hours=24, top_k=5)
        memories.extend(temporal)
        
        # 3. 关联检索（相关记忆图遍历）
        related = self.graph_traversal(
            seed=context, depth=2, top_k=5
        )
        memories.extend(related)
        
        # 4. 排序与去重
        return self.rank_and_dedup(memories, context)

记忆冲突解决

冲突类型解决方案
信息矛盾时间优先（新的覆盖旧的）
信息矛盾置信度优先（高可信度覆盖）
信息矛盾请求用户确认
信息冗余合并相似条目

📚 延伸阅读