4.4 记忆系统的实现与优化

C --> F[缓存层]
C --> G[索引层]
D --> H[连接池]
E --> I[持久化存储]
​

</div>

## 分步实战

### 步骤 1：记忆系统核心实现

```python
import time
import threading
from concurrent.futures import ThreadPoolExecutor
from dataclasses import dataclass
from typing import Dict, List, Any, Optional, Tuple
import numpy as np
import sqlite3
import json
import pickle
import logging
from abc import ABC, abstractmethod

@dataclass
class SystemConfig:
    """系统配置"""
    max_concurrent_requests: int = 100
    default_timeout: int = 30
    cache_size: int = 10000
    cache_ttl: int = 3600
    db_path: str = "memory_system.db"
    db_pool_size: int = 10
    embedding_dim: int = 768
    index_type: str = "hnsw"

class MemorySystemBase(ABC):
    """记忆系统基类"""
    
    def __init__(self, config: SystemConfig):
        self.config = config
        self.logger = self._setup_logger()
        self._initialize_system()
    
    @abstractmethod
    def _initialize_system(self):
        """初始化系统"""
        pass
    
    @abstractmethod
    def add_memory(self, memory_id: str, content: str, 
                  metadata: Dict[str, Any] = None) -> bool:
        """添加记忆"""
        pass
    
    @abstractmethod
    def get_memory(self, memory_id: str) -> Optional[Dict[str, Any]]:
        """获取记忆"""
        pass
    
    def _setup_logger(self) -> logging.Logger:
        """设置日志系统"""
        logger = logging.getLogger('MemorySystem')
        logger.setLevel(logging.INFO)
        
        # 控制台处理器
        console_handler = logging.StreamHandler()
        console_handler.setLevel(logging.INFO)
        
        # 文件处理器
        file_handler = logging.FileHandler('memory_system.log')
        file_handler.setLevel(logging.DEBUG)
        
        # 格式化器
        formatter = logging.Formatter(
            '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
        )
        console_handler.setFormatter(formatter)
        file_handler.setFormatter(formatter)
        
        logger.addHandler(console_handler)
        logger.addHandler(file_handler)
        
        return logger

class ConnectionPool:
    """数据库连接池"""
    
    def __init__(self, db_path: str, pool_size: int = 10):
        self.db_path = db_path
        self.pool_size = pool_size
        self.pool = []
        self.lock = threading.RLock()
        self._initialize_pool()
    
    def _initialize_pool(self):
        """初始化连接池"""
        for _ in range(self.pool_size):
            conn = self._create_connection()
            self.pool.append(conn)
    
    def _create_connection(self) -> sqlite3.Connection:
        """创建数据库连接"""
        conn = sqlite3.connect(self.db_path, check_same_thread=False)
        conn.execute('PRAGMA journal_mode=WAL')
        conn.execute('PRAGMA synchronous=NORMAL')
        return conn
    
    def get_connection(self) -> sqlite3.Connection:
        """获取数据库连接"""
        with self.lock:
            if self.pool:
                return self.pool.pop()
            else:
                return self._create_connection()
    
    def return_connection(self, conn: sqlite3.Connection):
        """归还数据库连接"""
        with self.lock:
            if len(self.pool) < self.pool_size:
                self.pool.append(conn)
            else:
                conn.close()

class MemorySystem(MemorySystemBase):
    """记忆系统实现"""
    
    def __init__(self, config: SystemConfig):
        super().__init__(config)
        self.connection_pool = ConnectionPool(config.db_path, config.db_pool_size)
        self.executor = ThreadPoolExecutor(max_workers=config.max_concurrent_requests)
        self.cache = LRUCache(config.cache_size, config.cache_ttl)
        self.metrics = SystemMetrics()
        self._initialize_database()
    
    def _initialize_system(self):
        """初始化系统"""
        self.logger.info("Initializing memory system...")
        self._start_background_workers()
        self.logger.info("Memory system initialized successfully")
    
    def _initialize_database(self):
        """初始化数据库"""
        conn = self.connection_pool.get_connection()
        try:
            # 创建记忆表
            conn.execute('''
                CREATE TABLE IF NOT EXISTS memories (
                    id TEXT PRIMARY KEY,
                    content TEXT NOT NULL,
                    embedding BLOB,
                    metadata TEXT,
                    timestamp DATETIME DEFAULT CURRENT_TIMESTAMP,
                    access_count INTEGER DEFAULT 0,
                    importance_score REAL DEFAULT 0.0,
                    version INTEGER DEFAULT 1,
                    is_active BOOLEAN DEFAULT 1
                )
            ''')
            
            # 创建索引
            conn.execute('CREATE INDEX IF NOT EXISTS idx_timestamp ON memories(timestamp)')
            conn.execute('CREATE INDEX IF NOT EXISTS idx_importance ON memories(importance_score)')
            conn.execute('CREATE INDEX IF NOT EXISTS idx_access ON memories(access_count)')
            
            conn.commit()
        finally:
            self.connection_pool.return_connection(conn)
    
    def add_memory(self, memory_id: str, content: str, 
                  metadata: Dict[str, Any] = None) -> bool:
        """添加记忆"""
        start_time = time.time()
        
        try:
            # 验证输入
            if not memory_id or not content:
                self.logger.error("Invalid memory_id or content")
                return False
            
            # 生成embedding
            embedding = self._generate_embedding(content)
            
            # 序列化数据
            metadata_json = json.dumps(metadata or {})
            embedding_blob = pickle.dumps(embedding)
            
            # 存储到数据库
            conn = self.connection_pool.get_connection()
            try:
                conn.execute('''
                    INSERT INTO memories (id, content, embedding, metadata, timestamp, access_count, importance_score, version)
                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
                ''', (memory_id, content, embedding_blob, metadata_json, 
                      time.time(), 0, metadata.get('importance_score', 0.5) if metadata else 0.5, 1))
                conn.commit()
                
                # 添加到缓存
                memory_data = {
                    'content': content,
                    'embedding': embedding,
                    'metadata': metadata or {},
                    'timestamp': time.time(),
                    'access_count': 0,
                    'importance_score': metadata.get('importance_score', 0.5) if metadata else 0.5
                }
                self.cache.set(memory_id, memory_data)
                
                # 更新指标
                self.metrics.increment('memories_added', 1)
                self.metrics.record_operation_time('add_memory', time.time() - start_time)
                
                self.logger.info(f"Memory {memory_id} added successfully")
                return True
                
            except Exception as e:
                conn.rollback()
                self.logger.error(f"Failed to add memory {memory_id}: {e}")
                return False
            finally:
                self.connection_pool.return_connection(conn)
                
        except Exception as e:
            self.logger.error(f"Error in add_memory: {e}")
            self.metrics.increment('errors', 1)
            return False
    
    def get_memory(self, memory_id: str) -> Optional[Dict[str, Any]]:
        """获取记忆"""
        start_time = time.time()
        
        try:
            # 先从缓存获取
            cached_data = self.cache.get(memory_id)
            if cached_data:
                cached_data['access_count'] = cached_data.get('access_count', 0) + 1
                self.cache.set(memory_id, cached_data)
                self.metrics.increment('cache_hits', 1)
                self.metrics.record_operation_time('get_memory', time.time() - start_time)
                return cached_data
            
            # 从数据库获取
            conn = self.connection_pool.get_connection()
            try:
                cursor = conn.execute('''
                    SELECT content, embedding, metadata, timestamp, access_count, importance_score, version
                    FROM memories WHERE id = ? AND is_active = 1
                ''', (memory_id,))
                
                row = cursor.fetchone()
                if not row:
                    self.metrics.increment('cache_misses', 1)
                    return None
                
                content, embedding_blob, metadata_json, timestamp, access_count, importance_score, version = row
                
                # 反序列化数据
                embedding = pickle.loads(embedding_blob)
                metadata = json.loads(metadata_json)
                
                memory_data = {
                    'content': content,
                    'embedding': embedding,
                    'metadata': metadata,
                    'timestamp': timestamp,
                    'access_count': access_count + 1,
                    'importance_score': importance_score,
                    'version': version
                }
                
                # 更新缓存
                self.cache.set(memory_id, memory_data)
                
                # 更新数据库访问计数
                conn.execute('''
                    UPDATE memories SET access_count = access_count + 1 
                    WHERE id = ?
                ''', (memory_id,))
                conn.commit()
                
                # 更新指标
                self.metrics.increment('cache_misses', 1)
                self.metrics.record_operation_time('get_memory', time.time() - start_time)
                self.metrics.increment('memories_accessed', 1)
                
                return memory_data
                
            finally:
                self.connection_pool.return_connection(conn)
                
        except Exception as e:
            self.logger.error(f"Error in get_memory: {e}")
            self.metrics.increment('errors', 1)
            return None
    
    def _generate_embedding(self, text: str) -> np.ndarray:
        """生成文本embedding"""
        # 使用TF-IDF生成特征向量（简化实现）
        from sklearn.feature_extraction.text import TfidfVectorizer
        vectorizer = TfidfVectorizer(max_features=self.config.embedding_dim)
        features = vectorizer.fit_transform([text])
        return features.toarray()[0]
    
    def _start_background_workers(self):
        """启动后台工作线程"""
        # 定期清理过期缓存
        threading.Thread(target=self._cache_cleaner, daemon=True).start()
        
        # 定期健康检查
        threading.Thread(target=self._health_check, daemon=True).start()
    
    def _cache_cleaner(self):
        """缓存清理线程"""
        while True:
            time.sleep(300)  # 每5分钟清理一次
            self.cache.cleanup_expired()
    
    def _health_check(self):
        """健康检查线程"""
        while True:
            time.sleep(300)  # 每5分钟检查一次
            self._perform_health_check()
    
    def _perform_health_check(self):
        """执行健康检查"""
        try:
            # 检查数据库连接
            conn = self.connection_pool.get_connection()
            conn.execute('SELECT 1').fetchone()
            self.connection_pool.return_connection(conn)
            
            # 检查系统资源
            import psutil
            cpu_percent = psutil.cpu_percent()
            memory_percent = psutil.virtual_memory().percent
            disk_percent = psutil.disk_usage('/').percent
            
            if cpu_percent > 90 or memory_percent > 90 or disk_percent > 90:
                self.logger.warning(f"High resource usage: CPU={cpu_percent}%, Memory={memory_percent}%, Disk={disk_percent}%")
            
        except Exception as e:
            self.logger.error(f"Health check failed: {e}")

class LRUCache:
    """LRU缓存实现"""
    
    def __init__(self, capacity: int, ttl: int):
        self.capacity = capacity
        self.ttl = ttl
        self.cache = {}
        self.access_times = {}
        self.expiration_times = {}
        self.lock = threading.RLock()
    
    def get(self, key: str) -> Optional[Any]:
        """获取缓存值"""
        with self.lock:
            if key not in self.cache:
                return None
            
            # 检查是否过期
            if key in self.expiration_times and time.time() > self.expiration_times[key]:
                del self.cache[key]
                del self.access_times[key]
                if key in self.expiration_times:
                    del self.expiration_times[key]
                return None
            
            # 更新访问时间
            self.access_times[key] = time.time()
            return self.cache[key]
    
    def set(self, key: str, value: Any, ttl: int = None):
        """设置缓存值"""
        with self.lock:
            if key in self.cache:
                # 更新现有值
                self.cache[key] = value
                self.access_times[key] = time.time()
                if ttl:
                    self.expiration_times[key] = time.time() + ttl
            else:
                # 添加新值
                if len(self.cache) >= self.capacity:
                    # 移除最久未使用的项
                    oldest_key = min(self.access_times.keys(), key=self.access_times.get)
                    del self.cache[oldest_key]
                    del self.access_times[oldest_key]
                    if oldest_key in self.expiration_times:
                        del self.expiration_times[oldest_key]
                
                self.cache[key] = value
                self.access_times[key] = time.time()
                if ttl:
                    self.expiration_times[key] = time.time() + ttl
    
    def cleanup_expired(self):
        """清理过期项"""
        with self.lock:
            current_time = time.time()
            expired_keys = [
                key for key, exp_time in self.expiration_times.items()
                if current_time > exp_time
            ]
            
            for key in expired_keys:
                if key in self.cache:
                    del self.cache[key]
                if key in self.access_times:
                    del self.access_times[key]
                if key in self.expiration_times:
                    del self.expiration_times[key]

class SystemMetrics:
    """系统指标收集器"""
    
    def __init__(self):
        self.metrics = {
            'memories_added': 0,
            'memories_accessed': 0,
            'searches_performed': 0,
            'cache_hits': 0,
            'cache_misses': 0,
            'errors': 0,
            'operation_times': {}
        }
        self.lock = threading.RLock()
    
    def increment(self, metric: str, value: int = 1):
        """增加指标值"""
        with self.lock:
            if metric in self.metrics:
                self.metrics[metric] += value
            else:
                self.metrics[metric] = value
    
    def record_operation_time(self, operation: str, time_taken: float):
        """记录操作时间"""
        with self.lock:
            if operation not in self.metrics['operation_times']:
                self.metrics['operation_times'][operation] = []
            self.metrics['operation_times'][operation].append(time_taken)
    
    def get_statistics(self) -> Dict[str, Any]:
        """获取统计信息"""
        with self.lock:
            stats = self.metrics.copy()
            
            # 计算操作时间统计
            for operation, times in stats['operation_times'].items():
                stats['operation_times'][operation] = {
                    'count': len(times),
                    'avg_time': sum(times) / len(times) if times else 0,
                    'min_time': min(times) if times else 0,
                    'max_time': max(times) if times else 0
                }
            
            return stats
​

常见问题 FAQ

Q1：如何处理大规模记忆系统的性能瓶颈？

A：处理大规模记忆系统性能瓶颈的策略：

1. 分片处理：将数据分布到多个节点，水平扩展存储和计算能力
2. 缓存优化：使用多层缓存（本地缓存+分布式缓存），减少数据库访问
3. 索引优化：使用高效的向量索引，如FAISS、HNSW等
4. 批量操作：使用批量读写操作，减少网络开销
5. 异步处理：使用异步队列处理非关键操作，提高响应速度

Q2：分布式记忆系统如何保证数据一致性？

A：保证数据一致性的策略：

1. 复制策略：使用主从复制或多副本复制，确保数据冗余
2. 一致性协议：使用Paxos、Raft等一致性协议
3. 版本控制：使用版本号和时间戳，支持冲突解决
4. 最终一致性：允许短期不一致，最终达到一致状态
5. 故障检测：使用心跳检测和故障转移机制

Q3：如何实现记忆系统的高可用性？

A：实现高可用性的策略：

1. 冗余设计：使用多副本、多节点冗余
2. 故障转移：自动检测故障并切换到备用节点
3. 负载均衡：使用负载均衡器分散请求
4. 数据备份：定期数据备份和恢复测试
5. 监控告警：实时监控系统状态，及时发现和处理问题

本节小结

本节深入探讨了记忆系统的实现与优化，从系统架构设计到工程实现，全面介绍了记忆系统的核心技术。通过学习本节内容，读者应该能够：

1. 掌握记忆系统的整体架构设计和工程实现
2. 学习性能优化和资源管理的关键技术
3. 理解分布式记忆系统的实现方案

至此，第4章"记忆系统架构"全部内容已完成，为智能体的记忆管理提供了完整的技术解决方案。

关键词：Agent智能体开发实战, 记忆系统, 系统架构, 性能优化, 分布式实现
难度：高级
预计阅读：25 分钟