4.2 短期记忆与长期记忆

4.2 短期记忆与长期记忆 — Agent智能体开发实战

本节导读：深入理解智能体记忆系统的双模态设计，掌握短期记忆与长期记忆的架构差异、实现策略和优化方法，为构建高效的智能体记忆系统奠定基础。

学习目标

• 理解短期记忆与长期记忆的核心区别和设计原则
• 掌握双模态记忆系统的架构设计和实现方法
• 学习记忆容量管理和性能优化的关键技术
• 了解记忆衰减机制与长期记忆压缩策略

核心概念

短期记忆与长期记忆是智能体记忆系统的双模态设计，就像人类的短期记忆和长期记忆一样，各自承担不同的职责。

记忆系统双模态架构

短期记忆特点：

• 访问速度：微秒级响应
• 容量限制：通常存储 100-1000 个条目
• 保留期限：任务会话内持久
• 访问模式：频繁读写操作

长期记忆特点：

• 访问速度：毫秒级响应
• 容量限制：可扩展至百万级条目
• 保留期限：长期持久化
• 访问模式：读多写少，定期归档

分步实战

步骤 1：短期记忆系统设计与实现

@dataclass
class MemoryItem:
    """记忆条目数据结构"""
    id: str
    content: str
    timestamp: datetime
    access_count: int
    last_access: datetime
    metadata: Dict[str, Any]
    importance_score: float = 0.0

class ShortTermMemory:
    """短期记忆系统"""
    
    def __init__(self, max_items: int = 1000, retention_hours: int = 24):
        self.max_items = max_items
        self.retention_hours = retention_hours
        self.items: Dict[str, MemoryItem] = {}
        self.access_heap = []  # 按访问频率排序的堆
        self.lock = threading.RLock()
        
    def add_item(self, content: str, metadata: Dict[str, Any] = None, 
                 importance_score: float = 0.0) -> str:
        """添加记忆条目"""
        item_id = hashlib.md5(content.encode()).hexdigest()
        timestamp = datetime.now()
        
        with self.lock:
            # 检查是否已存在
            if item_id in self.items:
                self.items[item_id].access_count += 1
                self.items[item_id].last_access = timestamp
                return item_id
            
            # 创建新条目
            item = MemoryItem(
                id=item_id,
                content=content,
                timestamp=timestamp,
                access_count=1,
                last_access=timestamp,
                metadata=metadata or {},
                importance_score=importance_score
            )
            
            self.items[item_id] = item
            heapq.heappush(self.access_heap, (item.access_count, item_id))
            
            # 执行内存管理
            self._manage_memory()
            
        return item_id
    
    def get_item(self, item_id: str) -> Optional[MemoryItem]:
        """获取记忆条目"""
        with self.lock:
            item = self.items.get(item_id)
            if item:
                item.access_count += 1
                item.last_access = datetime.now()
                heapq.heappush(self.access_heap, (item.access_count, item_id))
            return item
    
    def search_items(self, query: str, top_k: int = 10) -> List[MemoryItem]:
        """搜索相似记忆条目"""
        if not self.items:
            return []
        
        # 使用TF-IDF进行相似度计算
        contents = [(item_id, item.content) for item_id, item in self.items.items()]
        if not contents:
            return []
        
        vectorizer = TfidfVectorizer(max_features=1000)
        tfidf_matrix = vectorizer.fit_transform([content for _, content in contents])
        query_vector = vectorizer.transform([query])
        
        # 计算相似度
        similarities = cosine_similarity(query_vector, tfidf_matrix).flatten()
        
        # 获取top-k结果
        results = []
        for idx, sim_score in enumerate(similarities):
            if sim_score > 0.1:  # 相似度阈值
                item_id, _ = contents[idx]
                item = self.items[item_id]
                item.similarity_score = sim_score
                results.append(item)
        
        # 按相似度排序
        results.sort(key=lambda x: getattr(x, 'similarity_score', 0), reverse=True)
        return results[:top_k]
    
    def _manage_memory(self):
        """内存管理：LRU淘汰策略"""
        if len(self.items) > self.max_items:
            # 计算需要淘汰的数量
            items_to_remove = len(self.items) - self.max_items
            
            # 获取最少访问的条目
            removed_count = 0
            temp_heap = []
            
            # 重新构建堆，确保访问频率正确
            for item_id, item in self.items.items():
                temp_heap.append((item.access_count, item_id))
            
            # 排序并选择要移除的条目
            temp_heap.sort()
            
            for count, item_id in temp_heap:
                if removed_count >= items_to_remove:
                    break
                    
                # 保留高重要性的条目
                if self.items[item_id].importance_score < 0.5:
                    del self.items[item_id]
                    removed_count += 1
        
        # 清理过期条目
        current_time = datetime.now()
        expired_items = [
            item_id for item_id, item in self.items.items()
            if (current_time - item.timestamp).total_seconds() > self.retention_hours * 3600
        ]
        
        for item_id in expired_items:
            del self.items[item_id]
​

步骤 2：长期记忆系统设计与实现

class LongTermMemory:
    """长期记忆系统"""
    
    def __init__(self, storage_path: str = "memory.db", index_type: str = "hnsw"):
        self.storage_path = storage_path
        self.index_type = index_type
        self.connection = self._init_database()
        self.vector_index = None
        self.cache = {}
        self.cache_lock = threading.RLock()
        self._init_vector_index()
        
    def _init_database(self) -> sqlite3.Connection:
        """初始化数据库"""
        conn = sqlite3.connect(self.storage_path)
        conn.execute('''
            CREATE TABLE IF NOT EXISTS long_term_memory (
                id TEXT PRIMARY KEY,
                content TEXT NOT NULL,
                embedding BLOB,
                timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                access_count INTEGER DEFAULT 0,
                metadata TEXT,
                importance_score REAL DEFAULT 0.0,
                category TEXT,
                compression_level INTEGER DEFAULT 0
            )
        ''')
        conn.commit()
        return conn
    
    def add_item(self, content: str, embedding: Optional[np.ndarray] = None,
                 metadata: Dict[str, Any] = None, importance_score: float = 0.0,
                 category: str = "general") -> str:
        """添加长期记忆条目"""
        item_id = hashlib.md5(content.encode()).hexdigest()
        timestamp = datetime.now()
        
        # 如果没有提供embedding，需要生成
        if embedding is None:
            embedding = self._generate_embedding(content)
        
        # 序列化embedding
        serialized_embedding = pickle.dumps(embedding)
        
        # 存储到数据库
        self.connection.execute('''
            INSERT OR REPLACE INTO long_term_memory 
            (id, content, embedding, timestamp, access_count, metadata, 
             importance_score, category, compression_level)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
        ''', (item_id, content, serialized_embedding, timestamp, 0, 
              json.dumps(metadata or {}), importance_score, category, 0))
        
        # 添加到向量索引
        if self.vector_index is not None:
            self.vector_index.add(embedding.reshape(1, -1))
        
        # 更新缓存
        with self.cache_lock:
            self.cache[item_id] = {
                'content': content,
                'embedding': embedding,
                'metadata': metadata,
                'importance_score': importance_score,
                'category': category
            }
        
        self.connection.commit()
        return item_id
    
    def get_item(self, item_id: str) -> Optional[Dict[str, Any]]:
        """获取记忆条目"""
        # 首先检查缓存
        with self.cache_lock:
            cached = self.cache.get(item_id)
            if cached:
                cached['access_count'] = cached.get('access_count', 0) + 1
                self.cache[item_id] = cached
                return cached
        
        # 从数据库获取
        cursor = self.connection.execute('''
            SELECT content, embedding, timestamp, access_count, metadata, 
                   importance_score, category, compression_level
            FROM long_term_memory WHERE id = ?
        ''', (item_id,))
        
        row = cursor.fetchone()
        if not row:
            return None
        
        content, embedding_data, timestamp, access_count, metadata_json, \
        importance_score, category, compression_level = row
        
        # 反序列化embedding
        embedding = pickle.loads(embedding_data)
        metadata = json.loads(metadata_json)
        
        item_data = {
            'content': content,
            'embedding': embedding,
            'timestamp': timestamp,
            'access_count': access_count + 1,
            'metadata': metadata,
            'importance_score': importance_score,
            'category': category,
            'compression_level': compression_level
        }
        
        # 更新缓存
        with self.cache_lock:
            self.cache[item_id] = item_data
        
        # 更新数据库中的访问计数
        self.connection.execute('''
            UPDATE long_term_memory SET access_count = access_count + 1 
            WHERE id = ?
        ''', (item_id,))
        self.connection.commit()
        
        return item_data
    
    def search_similar(self, query: str, top_k: int = 10, 
                       threshold: float = 0.7) -> List[Dict[str, Any]]:
        """搜索相似记忆"""
        # 生成查询的embedding
        query_embedding = self._generate_embedding(query)
        
        # 在向量索引中搜索
        if self.vector_index is not None:
            distances, indices = self.vector_index.search(
                query_embedding.reshape(1, -1), top_k
            )
            
            results = []
            for i, (distance, idx) in enumerate(zip(distances[0], indices[0])):
                if idx < 0 or distance > (1 - threshold):  # 距离转换为相似度
                    continue
                
                # 获取对应的条目
                item_id = self._get_id_from_index(idx)
                if item_id:
                    item_data = self.get_item(item_id)
                    if item_data:
                        item_data['similarity_score'] = 1 - distance
                        results.append(item_data)
            
            return results
        
        # 如果没有向量索引，使用文本搜索
        return self._text_search(query, top_k)
    
    def _text_search(self, query: str, top_k: int = 10) -> List[Dict[str, Any]]:
        """文本搜索（备选方案）"""
        cursor = self.connection.execute('''
            SELECT id, content, importance_score, category
            FROM long_term_memory 
            WHERE content LIKE ? OR category = ?
            ORDER BY importance_score DESC, access_count DESC
            LIMIT ?
        ''', (f'%{query}%', query, top_k))
        
        results = []
        for row in cursor.fetchall():
            item_id, content, importance_score, category = row
            item_data = self.get_item(item_id)
            if item_data:
                results.append(item_data)
        
        return results
​

步骤 3：双模态记忆系统整合

class DualMemorySystem:
    """双模态记忆系统：整合短期记忆和长期记忆"""
    
    def __init__(self, config: Dict[str, Any] = None):
        self.config = config or MEMORY_CONFIG
        
        # 初始化短期记忆
        self.short_term = ShortTermMemory(
            max_items=self.config['short_term']['max_items'],
            retention_hours=self.config['short_term']['retention_hours']
        )
        
        # 初始化长期记忆
        self.long_term = LongTermMemory(
            storage_path="long_term_memory.db",
            index_type=self.config['long_term']['index_type']
        )
        
        # 迁移管理
        self.migration_threshold = 0.8  # 迁移阈值
        
    def add_experience(self, experience: str, metadata: Dict[str, Any] = None,
                      importance_score: float = 0.0, category: str = "general"):
        """添加经验到记忆系统"""
        # 首先存储到短期记忆
        item_id = self.short_term.add_item(
            experience, metadata or {}, importance_score
        )
        
        # 高重要性经验直接存储到长期记忆
        if importance_score > 0.8:
            long_term_id = self.long_term.add_item(
                experience, None, metadata or {}, importance_score, category
            )
            return long_term_id
        
        return item_id
    
    def get_memory(self, memory_id: str, memory_type: str = "short") -> Optional[Dict[str, Any]]:
        """获取记忆条目"""
        if memory_type == "short":
            item = self.short_term.get_item(memory_id)
            return {'short_term': item} if item else None
        elif memory_type == "long":
            item = self.long_term.get_item(memory_id)
            return {'long_term': item} if item else None
        else:
            # 优先从短期记忆获取，再查长期记忆
            short_item = self.short_term.get_item(memory_id)
            if short_item:
                return {'short_term': short_item}
            
            long_item = self.long_term.get_item(memory_id)
            if long_item:
                return {'long_term': long_item}
            
            return None
    
    def search_memories(self, query: str, top_k: int = 10, 
                      include_short: bool = True, include_long: bool = True) -> List[Dict[str, Any]]:
        """搜索记忆（双模态）"""
        all_results = []
        
        # 搜索短期记忆
        if include_short:
            short_results = self.short_term.search_items(query, top_k)
            for result in short_results:
                all_results.append({
                    'type': 'short_term',
                    'memory': result,
                    'relevance_score': result.importance_score + result.access_count * 0.1
                })
        
        # 搜索长期记忆
        if include_long:
            long_results = self.long_term.search_similar(query, top_k)
            for result in long_results:
                all_results.append({
                    'type': 'long_term',
                    'memory': result,
                    'relevance_score': result['importance_score'] + result['access_count'] * 0.1
                })
        
        # 按相关性排序
        all_results.sort(key=lambda x: x['relevance_score'], reverse=True)
        return all_results[:top_k]
    
    def _should_migrate(self, item_id: str, importance_score: float) -> bool:
        """判断是否需要迁移到长期记忆"""
        item = self.short_term.get_item(item_id)
        if not item:
            return False
        
        # 条件1：高重要性
        if importance_score > self.migration_threshold:
            return True
        
        # 条件2：访问频率高但内容重要
        if item.access_count > 50 and item.importance_score > 0.6:
            return True
        
        # 条件3：内容长度大于阈值
        if len(item.content) > 1000:  # 超过1000字符的长内容
            return True
        
        return False
​

常见问题 FAQ

Q1：短期记忆和长期记忆如何划分标准？

A：短期记忆和长期记忆的划分基于以下几个关键因素：

1. 访问频率：高频访问的记忆优先保留在短期记忆中
2. 内容重要性：重要性评分 > 0.8 的内容直接存入长期记忆
3. 内容长度：超过1000字符的长内容考虑迁移到长期记忆
4. 时间因素：短期记忆保留24小时，之后自动迁移或清理
5. 业务需求：根据具体业务场景自定义迁移策略

Q2：记忆系统的容量如何管理？

A：记忆系统采用多层次的容量管理策略：

• 短期记忆：采用LRU（最近最少使用）淘汰算法，达到最大容量时自动清理最少使用的条目
• 长期记忆：基于重要性评分进行压缩，低重要性记忆会被摘要压缩，释放存储空间
• 缓存管理：长期记忆采用LRU缓存机制，避免频繁的磁盘IO
• 分片存储：大型记忆系统可以按类别或时间进行分片存储

Q3：如何保证记忆检索的效率？

A：记忆检索效率通过以下机制保证：

1. 向量索引：使用FAISS等高效的向量索引库实现快速相似度搜索
2. 缓存机制：热门记忆存储在内存缓存中，实现微秒级访问
3. 分层搜索：先搜索短期记忆，再搜索长期记忆，按相关性排序
4. 预计算特征：对重要记忆预计算embedding和特征向量

本节小结

本节深入探讨了短期记忆与长期记忆的双模态设计，从架构设计到具体实现，全面介绍了记忆系统的核心组件和优化策略。通过学习本节内容，读者应该能够：

1. 理解短期记忆和长期记忆的设计原理和区别
2. 掌握双模态记忆系统的实现方法和优化技巧
3. 学会管理记忆容量和性能的关键策略

下一节我们将探讨记忆检索与更新机制，深入记忆系统的核心算法和优化技术。

关键词：Agent智能体开发实战, 记忆系统, 短期记忆, 长期记忆, 双模态, 算法优化
难度：进阶
预计阅读：30 分钟