27.4 記憶系統設計

記憶系統概述

記憶系統是Agentic AI系統的重要組成部分，它使AI能夠儲存、檢索和利用過去的資訊，從而實現長期學習和持續改進。

记忆系统的基本概念

1. 什么是记忆系统

記憶系統是指AI系統用於儲存、組織和檢索資訊的機制，類似於人類的記憶功能。

记忆系统的特点 :

• 持久化儲存 : 長期儲存重要資訊
• 快速檢索 : 高效地檢索相關資訊
• 智慧組織 : 自動組織和分類資訊
• 上下文感知 : 根據上下文檢索資訊

2. 記憶型別

記憶型別	特點	儲存時間	容量

感知記憶| 短暫儲存原始輸入| 毫秒到秒| 有限 短期記憶| 儲存當前上下文| 秒到分鐘| 有限 長期記憶| 儲存重要知識和經驗| 永久| 無限 工作記憶| 處理當前任務| 任務期間| 有限

记忆存储机制

1. 向量存储

示例：向量存储

使用者請求： "實現一個基於向量的記憶儲存系統"

Claude Code 生成的程式碼：

    python


    ````python

    > - 使用向量表示记忆内容
    > - 基于相似度检索
    > - 支持多种索引方式
    > - 记录访问统计
    `python

    from typing import Dict, List, Any, Optional
    from datetime import datetime, timedelta
    import json
    import logging

    logger = logging.getLogger(__name__)

    class MemoryLevel(Enum):
    """記憶層級"""
    WORKING = "working"
    SHORT_TERM = "short_term"
    LONG_TERM = "long_term"

    class HierarchicalMemoryStore:
    """層次化記憶儲存"""

    def __init__(self):
    self.working_memory: Dict[str, Memory] = {}
    self.short_term_memory: Dict[str, Memory] = {}
    self.long_term_memory: Dict[str, Memory] = {}

     # 配置

    self.working_memory_limit = 10
    self.short_term_memory_limit = 100
    self.short_term_memory_ttl = timedelta(hours=24)

    def add_memory(self, content: str, embedding: np.ndarray, level: MemoryLevel = MemoryLevel.SHORT_TERM, metadata: Optional[Dict[str, Any]] = None) -> Memory:
    """新增記憶"""
    memory_id = f"mem_{datetime.utcnow().timestamp()}"
    memory = Memory(memory_id, content, embedding, metadata)

     # 根據層級儲存

    if level == MemoryLevel.WORKING:
    self._add_to_working_memory(memory)
    elif level == MemoryLevel.SHORT_TERM:
    self._add_to_short_term_memory(memory)
    elif level == MemoryLevel.LONG_TERM:
    self._add_to_long_term_memory(memory)

    logger.info(f"Memory added to {level.value}: {memory_id}")
    return memory

    def _add_to_working_memory(self, memory: Memory):
    """新增到工作記憶"""

     # 檢查容量

    if len(self.working_memory) >= self.working_memory_limit:

     # 移除最舊的

    oldest_id = min(self.working_memory.keys(), key=lambda k: self.working_memory[k].created_at)
    self._evict_from_working_memory(oldest_id)

    self.working_memory[memory.id] = memory

    def _add_to_short_term_memory(self, memory: Memory):
    """新增到短期記憶"""

     # 檢查容量

    if len(self.short_term_memory) >= self.short_term_memory_limit:

     # 移除最舊的

    oldest_id = min(self.short_term_memory.keys(), key=lambda k: self.short_term_memory[k].created_at)
    self._evict_from_short_term_memory(oldest_id)

    self.short_term_memory[memory.id] = memory

    def _add_to_long_term_memory(self, memory: Memory):
    """新增到長期記憶"""
    self.long_term_memory[memory.id] = memory

    def _evict_from_working_memory(self, memory_id: str):
    """從工作記憶中驅逐"""
    memory = self.working_memory.pop(memory_id)

     # 移動到短期記憶

    self._add_to_short_term_memory(memory)
    logger.info(f"Memory evicted from working memory: {memory_id}")

    def _evict_from_short_term_memory(self, memory_id: str):
    """從短期記憶中驅逐"""
    memory = self.short_term_memory.pop(memory_id)

     # 移動到長期記憶

    self._add_to_long_term_memory(memory)
    logger.info(f"Memory evicted from short-term memory: {memory_id}")

    def retrieve(self, query_embedding: np.ndarray, top_k: int = 5) -> List[Memory]:
    """檢索記憶"""

     # 從所有層級檢索

    all_memories = {
     **self.working_memory,
     **self.short_term_memory,
     **self.long_term_memory
    }

     # 計算相似度

    similarities = []
    for memory_id, memory in all_memories.items():
    similarity = self._cosine_similarity(query_embedding, memory.embedding)
    similarities.append((memory, similarity))

     # 按相似度排序

    similarities.sort(key=lambda x: x[1], reverse=True)

     # 獲取top_k個記憶

    top_memories = [memory for memory, similarity in similarities[:top_k]]

    return top_memories

    def _cosine_similarity(self, embedding1: np.ndarray, embedding2: np.ndarray) -> float:
    """計算餘弦相似度"""
    dot_product = np.dot(embedding1, embedding2)
    norm1 = np.linalg.norm(embedding1)
    norm2 = np.linalg.norm(embedding2)

    if norm1 == 0 or norm2 == 0:
    return 0.0

    return dot_product / (norm1 * norm2)

    def promote_memory(self, memory_id: str, target_level: MemoryLevel):
    """提升記憶層級"""
    memory = None
    source_level = None

     # 查詢記憶

    if memory_id in self.working_memory:
    memory = self.working_memory.pop(memory_id)
    source_level = MemoryLevel.WORKING
    elif memory_id in self.short_term_memory:
    memory = self.short_term_memory.pop(memory_id)
    source_level = MemoryLevel.SHORT_TERM
    elif memory_id in self.long_term_memory:
    memory = self.long_term_memory.pop(memory_id)
    source_level = MemoryLevel.LONG_TERM

    if not memory:
    raise ValueError(f"Memory not found: {memory_id}")

     # 移動到目標層級

    if target_level == MemoryLevel.WORKING:
    self._add_to_working_memory(memory)
    elif target_level == MemoryLevel.SHORT_TERM:
    self._add_to_short_term_memory(memory)
    elif target_level == MemoryLevel.LONG_TERM:
    self._add_to_long_term_memory(memory)

    logger.info(f"Memory promoted from {source_level.value} to {target_level.value}: {memory_id}")

    def cleanup_expired(self):
    """清理過期記憶"""
    current_time = datetime.utcnow()
    expired_ids = []

     # 檢查短期記憶

    for memory_id, memory in self.short_term_memory.items():
    if current_time - memory.created_at > self.short_term_memory_ttl:
    expired_ids.append(memory_id)

     # 移除過期記憶

    for memory_id in expired_ids:
    memory = self.short_term_memory.pop(memory_id)

     # 移動到長期記憶

    self._add_to_long_term_memory(memory)
    logger.info(f"Expired memory moved to long-term: {memory_id}")

    def get_statistics(self) -> Dict[str, Any]:
    """獲取統計資訊"""
    return {
    'working_memory_size': len(self.working_memory),
    'short_term_memory_size': len(self.short_term_memory),
    'long_term_memory_size': len(self.long_term_memory),
    'total_memories': len(self.working_memory) + len(self.short_term_memory) + len(self.long_term_memory)
    }

    # 使用示例

    store = HierarchicalMemoryStore()

    # 新增記憶到不同層級

    store.add_memory("Current task context", np.random.rand(768), MemoryLevel.WORKING, {'type': 'context'})
    store.add_memory("Recent conversation", np.random.rand(768), MemoryLevel.SHORT_TERM, {'type': 'conversation'})
    store.add_memory("Important knowledge", np.random.rand(768), MemoryLevel.LONG_TERM, {'type': 'knowledge'})

    # 獲取統計資訊

    stats = store.get_statistics()
    print(f"Statistics: {stats}")

    # 檢索記憶

    query_embedding = np.random.rand(768)
    memories = store.retrieve(query_embedding, top_k=3)

    print("\nRetrieved memories:")
    for memory in memories:
    print(f"  - {memory.content}")

    # 提升記憶層級

    if store.short_term_memory:
    memory_id = list(store.short_term_memory.keys())[0]
    store.promote_memory(memory_id, MemoryLevel.LONG_TERM)

    ```> **层次化存储特点**:
    > - 分层存储不同重要性的记忆
    > - 自动驱逐和提升记忆
    > - 限制各层容量
    > - 定期清理过期记忆

    ```

    ## 記憶檢索最佳化

    ### 1. 上下文感知檢索

    # 示例：上下文感知檢索

    使用者請求：
    "實現上下文感知的記憶檢索"
    Claude Code 生成的程式碼：

    ````python
    `python

    from typing import Dict, List, Any, Optional
    from datetime import datetime
    import logging

    logger = logging.getLogger(__name__)

    class ContextAwareMemoryStore(VectorMemoryStore):
    """上下文感知记忆存储"""

    def __init__(self, embedding_dim: int = 768):
    super().__init__(embedding_dim)
    self.context_history: List[Dict[str, Any]] = []
    self.context_window_size = 10

    def add_context(self, context: str, embedding: np.ndarray, metadata: Optional[Dict[str, Any]] = None):
    """添加上下文"""
    context_entry = {
    'content': context,
    'embedding': embedding,
    'metadata': metadata or {},
    'timestamp': datetime.utcnow()
    }

    self.context_history.append(context_entry)

     # 限制上下文历史大小
    if len(self.context_history) > self.context_window_size:
    self.context_history.pop(0)

    logger.info(f"Context added: {context[:50]}...")

    def retrieve_with_context(self, query_embedding: np.ndarray, top_k: int = 5) -> List[Dict[str, Any]]:
    """带上下文的检索"""
     # 获取当前上下文
    current_context = self._get_current_context()

     # 结合上下文和查询
    context_enhanced_embedding = self._combine_context_and_query(current_context, query_embedding)

     # 检索记忆
    memories = self.retrieve_similar(context_enhanced_embedding, top_k)

     # 添加上下文信息
    results = []
    for memory in memories:
    results.append({
    'memory': memory,
    'context_relevance': self._calculate_context_relevance(memory, current_context),
    'query_relevance': self._cosine_similarity(query_embedding, memory.embedding)
    })

     # 按综合相关性排序
    results.sort(key=lambda x: 0.7 * x['query_relevance'] + 0.3 * x['context_relevance'], reverse=True)

    return results

    def _get_current_context(self) -> List[Dict[str, Any]]:
    """获取当前上下文"""
    return self.context_history[-self.context_window_size:]

    def _combine_context_and_query(self, context: List[Dict[str, Any]], query_embedding: np.ndarray) -> np.ndarray:
    """结合上下文和查询"""
    if not context:
    return query_embedding

     # 计算上下文的平均嵌入
    context_embeddings = [c['embedding'] for c in context]
    avg_context_embedding = np.mean(context_embeddings, axis=0)

     # 加权组合
    combined = 0.7 * query_embedding + 0.3 * avg_context_embedding

    return combined

    def _calculate_context_relevance(self, memory: Memory, context: List[Dict[str, Any]]) -> float:
    """计算上下文相关性"""
    if not context:
    return 0.0

     # 计算记忆与上下文的平均相似度
    similarities = []
    for ctx in context:
    similarity = self._cosine_similarity(memory.embedding, ctx['embedding'])
    similarities.append(similarity)

    return np.mean(similarities)

    def retrieve_temporal(self, time_range: timedelta, top_k: int = 10) -> List[Memory]:
    """按时间范围检索"""
    current_time = datetime.utcnow()
    cutoff_time = current_time - time_range

     # 筛选时间范围内的记忆
    recent_memories = [
    memory for memory in self.memories.values()
    if memory.created_at >= cutoff_time
    ]

     # 按创建时间排序
    recent_memories.sort(key=lambda m: m.created_at, reverse=True)

    return recent_memories[:top_k]

    # 使用示例
    store = ContextAwareMemoryStore()

    # 添加上下文
    contexts = [
    ("User is asking about Python programming", np.random.rand(768)),
    ("User wants to learn about data structures", np.random.rand(768)),
    ("User is interested in algorithms", np.random.rand(768)),
    ]

    for content, embedding in contexts:
    store.add_context(content, embedding)

    # 添加记忆
    memories = [
    ("Python lists are mutable sequences", np.random.rand(768), {'type': 'knowledge', 'topic': 'python'}),
    ("Dictionaries are key-value pairs", np.random.rand(768), {'type': 'knowledge', 'topic': 'python'}),
    ("Binary search has O(log n) complexity", np.random.rand(768), {'type': 'knowledge', 'topic': 'algorithms'}),
    ]

    for content, embedding, metadata in memories:
    store.add_memory(content, embedding, metadata)

    # 带上下文的检索
    query_embedding = np.random.rand(768)
    results = store.retrieve_with_context(query_embedding, top_k=3)

    print("Retrieved with context:")
    for result in results:
    print(f"  - {result['memory'].content}")
    print(f"    Query relevance: {result['query_relevance']:.3f}")
    print(f"    Context relevance: {result['context_relevance']:.3f}")

    ```> **上下文感知檢索特點**:

    > - 維護上下文歷史
    > - 結合上下文和查詢
    > - 計算上下文相關性
    > - 按時間範圍檢索

    ## 总结

    记忆系统设计包括：

    1. **记忆系统的基本概念**: 什么是记忆系统、记忆类型
    2. **记忆存储机制**: 向量存储、层次化存储
    3. **记忆检索优化**: 上下文感知检索

    通过记忆系统，Claude Code能够存储和检索重要信息，实现长期学习和持续改进。

    在下一节中，我们将探讨多智能体协作。