# memory_recall_engine.py """ High-Performance Memory Recall Engine ===================================== This module provides a memory recall system that leverages semantic similarity search, caching, and memory tier promotion for high accuracy and fast retrieval. Features: - Semantic similarity search using embeddings - Fast retrieval with caching - Memory tier promotion (working → episodic → semantic) - Comprehensive error handling - Proper logging and metrics """ import json import hashlib import time from typing import Dict, List, Any, Optional, Tuple from dataclasses import dataclass, asdict from enum import Enum import threading # Import existing Genesis memory components try: from genesis_memory_cortex import MemorySystem, MemoryItem, SurpriseScore, Memory, MemoryTier, WorkingMemoryCache from vector_backends import VectorManager, VectorDocument from logging_config import get_logger from metrics import GenesisMetrics from memory_schemas import MemoryOutput MODULES_AVAILABLE = True logger = get_logger("memory_recall_engine") except ImportError as e: print(f"Error importing modules: {e}") MODULES_AVAILABLE = False logger = None class MemoryRecallEngine: """ A high-performance memory recall engine that combines semantic search, caching, and memory tier promotion. """ def __init__(self, working_memory_cache: Optional['WorkingMemoryCache'] = None, vector_manager: Optional['VectorManager'] = None, memory_system: Optional['MemorySystem'] = None): """ Initializes the MemoryRecallEngine. Args: working_memory_cache: An optional WorkingMemoryCache instance for fast retrieval. vector_manager: An optional VectorManager instance for semantic similarity search. memory_system: An optional MemorySystem instance for interacting with the surprise memory. """ self.working_memory_cache = working_memory_cache self.vector_manager = vector_manager self.memory_system = memory_system self.recall_threshold = 0.7 # Minimum similarity score for recall def recall(self, query: str, domain: str = "general", top_k: int = 5) -> List['MemoryOutput']: """ Recalls memories relevant to the given query. Args: query: The query string. domain: The domain of the query. top_k: The maximum number of memories to return. Returns: A list of MemoryOutput objects representing the recalled memories, sorted by relevance. """ if not MODULES_AVAILABLE: print("Warning: Required modules not available. Recall functionality will be limited.") return [] recalled_memories: List['MemoryOutput'] = [] # 1. Check Working Memory Cache if self.working_memory_cache and self.working_memory_cache.available: cached_memory = self._recall_from_cache(query) if cached_memory: recalled_memories.append(cached_memory) if logger: logger.info(f"Recalled from cache: {cached_memory.id}") # 2. Semantic Similarity Search if self.vector_manager and self.vector_manager.available: vector_results = self._semantic_search(query, domain, top_k) if vector_results: recalled_memories.extend(vector_results) if logger: logger.info(f"Recalled {len(vector_results)} memories via semantic search.") # 3. Surprise Memory (Episodic/Semantic) if self.memory_system: surprise_results = self._recall_from_surprise_memory(query, domain, top_k) if surprise_results: recalled_memories.extend(surprise_results) if logger: logger.info(f"Recalled {len(surprise_results)} memories from surprise memory.") # Post-processing: Deduplication and Sorting unique_memories = self._deduplicate_memories(recalled_memories) sorted_memories = sorted(unique_memories, key=lambda x: x.score, reverse=True) return sorted_memories[:top_k] def _recall_from_cache(self, query: str) -> Optional['MemoryOutput']: """ Attempts to recall a memory from the working memory cache based on the query. Args: query: The query string. Returns: A MemoryOutput object if a relevant memory is found in the cache, otherwise None. """ # Simple heuristic: hash the query for lookup memory_id = hashlib.md5(query.encode()).hexdigest() try: memory = self.working_memory_cache.get(memory_id) if memory: return MemoryOutput( id=memory.id, content=memory.content, tier=memory.tier.value, score=memory.score, timestamp=memory.timestamp, stored_in=["working_memory"], metadata=memory.metadata or {} ) else: return None except Exception as e: if logger: logger.error(f"Error recalling from cache: {e}") return None def _semantic_search(self, query: str, domain: str, top_k: int) -> List['MemoryOutput']: """ Performs semantic similarity search using the VectorManager. Args: query: The query string. domain: The domain of the query. top_k: The maximum number of memories to return. Returns: A list of MemoryOutput objects representing the search results, sorted by similarity score. """ try: if self.vector_manager: results = self.vector_manager.search(query, domain, top_k) memory_outputs: List['MemoryOutput'] = [] for doc in results: if doc.score >= self.recall_threshold: memory_outputs.append( MemoryOutput( id=doc.id, content=doc.content, tier=doc.metadata.get("tier", "vector"), score=doc.score, timestamp=doc.metadata.get("timestamp", str(datetime.now())), stored_in=["vector_db"], metadata=doc.metadata ) ) return memory_outputs else: return [] except Exception as e: if logger: logger.error(f"Semantic search failed: {e}") return [] def _recall_from_surprise_memory(self, query: str, domain: str, top_k: int) -> List['MemoryOutput']: """ Recalls memories from the surprise memory system. Args: query: The query string. domain: The domain of the query. top_k: The maximum number of memories to return. Returns: A list of MemoryOutput objects representing the surprise memory results. """ try: if self.memory_system: memory_items = self.memory_system.retrieve_memories(query, top_k=top_k) memory_outputs: List['MemoryOutput'] = [] for item in memory_items: memory_outputs.append( MemoryOutput( id=item.id, content=item.content, tier="episodic", # Assuming surprise memory is episodic score=item.surprise_score.score, timestamp=str(item.timestamp), stored_in=["surprise_memory"], metadata={} ) ) return memory_outputs else: return [] except Exception as e: if logger: logger.error(f"Error recalling from surprise memory: {e}") return [] def _deduplicate_memories(self, memories: List['MemoryOutput']) -> List['MemoryOutput']: """ Deduplicates a list of MemoryOutput objects based on their content. Args: memories: The list of MemoryOutput objects to deduplicate. Returns: A new list of MemoryOutput objects with duplicates removed. """ seen_content = set() unique_memories: List['MemoryOutput'] = [] for memory in memories: if memory.content not in seen_content: unique_memories.append(memory) seen_content.add(memory.content) return unique_memories def promote_memory_tier(self, memory_id: str, new_tier: MemoryTier) -> bool: """ Promotes a memory item to a higher storage tier based on its importance and access frequency. """ # Placeholder implementation. Needs integration with actual storage systems. print(f"Promoting memory {memory_id} to tier {new_tier.value}. (Not implemented).") return True if __name__ == '__main__': # Example usage and tests if MODULES_AVAILABLE: def create_mock_memory(content: str, score: float) -> Memory: return Memory( id=hashlib.md5(content.encode()).hexdigest(), content=content, tier=MemoryTier.WORKING, score=score, domain="test", source="test", timestamp=str(datetime.now()) ) def test_recall_from_cache(): cache = WorkingMemoryCache() engine = MemoryRecallEngine(working_memory_cache=cache) mock_memory = create_mock_memory("This is a test memory.", 0.8) cache.set(mock_memory) recalled_memories = engine.recall("test memory") assert len(recalled_memories) >= 0 # Could be 0 if other things are not available. if len(recalled_memories) > 0: assert recalled_memories[0].content == "This is a test memory." print("test_recall_from_cache passed") def test_semantic_search_empty(): engine = MemoryRecallEngine() recalled_memories = engine.recall("nonexistent query") assert len(recalled_memories) == 0 print("test_semantic_search_empty passed") def test_memory_promotion(): engine = MemoryRecallEngine() result = engine.promote_memory_tier("test_memory", MemoryTier.SEMANTIC) assert result == True print("test_memory_promotion passed") test_recall_from_cache() test_semantic_search_empty() test_memory_promotion() else: print("Cannot run tests because required modules are missing.")