#!/usr/bin/env python3 """ Predictive Memory Surfacing - Neural Network for Context-Relevant Memory Selection ================================================================================= Uses neural networks and contextual analysis to predict which memories will be most relevant for the current context, proactively surfacing valuable information. """ import os import json import numpy as np from datetime import datetime, timedelta from typing import Dict, List, Any, Tuple, Optional from collections import defaultdict from core.mira_path_resolver import get_mira_memory_dir from core.memory.memory_manager import MemoryManager from intelligence.adaptive_pattern_evolution import get_adaptive_pattern_evolution from utils.logging import setup_logger logger = setup_logger(__name__) class MemoryRelevanceScorer: """Scores memory relevance using multiple neural and contextual factors""" def __init__(self): self.memory_manager = MemoryManager() self.context_weights = { 'semantic_similarity': 0.25, 'temporal_relevance': 0.15, 'usage_frequency': 0.20, 'contextual_importance': 0.15, 'emotional_resonance': 0.10, 'recent_activity': 0.15 } def score_memory_relevance(self, memory: Dict[str, Any], context: Dict[str, Any]) -> float: """Calculate relevance score for a memory given current context""" scores = {} # Semantic similarity based on content overlap scores['semantic_similarity'] = self._calculate_semantic_similarity(memory, context) # Temporal relevance - recent memories vs importance decay scores['temporal_relevance'] = self._calculate_temporal_relevance(memory, context) # Usage frequency - how often this memory has been accessed scores['usage_frequency'] = self._calculate_usage_frequency(memory) # Contextual importance - importance in current work context scores['contextual_importance'] = self._calculate_contextual_importance(memory, context) # Emotional resonance - emotional significance of memory scores['emotional_resonance'] = self._calculate_emotional_resonance(memory) # Recent activity - how recently memory was relevant scores['recent_activity'] = self._calculate_recent_activity(memory, context) # Weighted sum total_score = sum( scores[factor] * self.context_weights[factor] for factor in scores ) return min(1.0, max(0.0, total_score)) def _calculate_semantic_similarity(self, memory: Dict[str, Any], context: Dict[str, Any]) -> float: """Calculate semantic similarity between memory and current context""" try: memory_content = memory.get('content', '').lower() memory_tags = memory.get('tags', []) memory_type = memory.get('type', '') # Current context keywords context_keywords = context.get('keywords', []) context_topics = context.get('topics', []) context_type = context.get('type', '') # Keyword overlap keyword_overlap = 0 if context_keywords: memory_words = set(memory_content.split()) context_words = set(' '.join(context_keywords).lower().split()) if context_words: keyword_overlap = len(memory_words & context_words) / len(context_words) # Tag/topic overlap tag_overlap = 0 if context_topics and memory_tags: tag_overlap = len(set(memory_tags) & set(context_topics)) / len(set(memory_tags) | set(context_topics)) # Type similarity type_similarity = 1.0 if memory_type == context_type else 0.3 return (keyword_overlap * 0.5 + tag_overlap * 0.3 + type_similarity * 0.2) except Exception as e: logger.error(f"Error calculating semantic similarity: {e}") return 0.0 def _calculate_temporal_relevance(self, memory: Dict[str, Any], context: Dict[str, Any]) -> float: """Calculate temporal relevance with decay and importance boost""" try: # Memory timestamp memory_timestamp = memory.get('timestamp', datetime.now().isoformat()) if isinstance(memory_timestamp, str): memory_time = datetime.fromisoformat(memory_timestamp.replace('Z', '+00:00')) else: memory_time = memory_timestamp # Time since memory creation time_diff = datetime.now() - memory_time.replace(tzinfo=None) days_old = time_diff.total_seconds() / (24 * 3600) # Importance boost for significant memories importance = memory.get('importance', 0.5) # Base temporal decay if days_old <= 1: base_score = 1.0 elif days_old <= 7: base_score = 0.8 elif days_old <= 30: base_score = 0.6 elif days_old <= 90: base_score = 0.4 else: base_score = 0.2 # Apply importance multiplier final_score = base_score * (0.5 + importance * 0.5) return min(1.0, final_score) except Exception as e: logger.error(f"Error calculating temporal relevance: {e}") return 0.5 def _calculate_usage_frequency(self, memory: Dict[str, Any]) -> float: """Calculate usage frequency score based on access patterns""" try: access_count = memory.get('access_count', 0) last_accessed = memory.get('last_accessed') # Normalize access count (assume max reasonable access is 20) frequency_score = min(1.0, access_count / 20.0) # Recent access boost if last_accessed: try: if isinstance(last_accessed, str): last_access_time = datetime.fromisoformat(last_accessed.replace('Z', '+00:00')) else: last_access_time = last_accessed days_since_access = (datetime.now() - last_access_time.replace(tzinfo=None)).total_seconds() / (24 * 3600) if days_since_access <= 7: frequency_score *= 1.2 elif days_since_access <= 30: frequency_score *= 1.1 except: pass return min(1.0, frequency_score) except Exception as e: logger.error(f"Error calculating usage frequency: {e}") return 0.0 def _calculate_contextual_importance(self, memory: Dict[str, Any], context: Dict[str, Any]) -> float: """Calculate importance in current work context""" try: memory_type = memory.get('type', '') context_priority = context.get('priority', 'medium') current_project = context.get('project', '') # Type-based importance type_importance = { 'breakthrough': 1.0, 'milestone': 0.9, 'insight': 0.8, 'learning': 0.7, 'decision': 0.8, 'reflection': 0.6, 'note': 0.4, 'task': 0.5 }.get(memory_type, 0.5) # Project context boost project_boost = 1.0 memory_content = memory.get('content', '').lower() if current_project and current_project.lower() in memory_content: project_boost = 1.3 # Priority context priority_multiplier = { 'critical': 1.2, 'high': 1.1, 'medium': 1.0, 'low': 0.9 }.get(context_priority, 1.0) return min(1.0, type_importance * project_boost * priority_multiplier) except Exception as e: logger.error(f"Error calculating contextual importance: {e}") return 0.5 def _calculate_emotional_resonance(self, memory: Dict[str, Any]) -> float: """Calculate emotional significance of memory""" try: # Emotional indicators in content content = memory.get('content', '').lower() emotional_words = [ 'breakthrough', 'frustrated', 'excited', 'proud', 'confused', 'amazing', 'wonderful', 'terrible', 'brilliant', 'struggled', 'accomplished', 'disappointed', 'thrilled', 'worried' ] emotion_score = 0 for word in emotional_words: if word in content: emotion_score += 0.1 # Explicit emotional metadata emotion_rating = memory.get('emotion_rating', 0.5) # Combine final_score = min(1.0, emotion_score + emotion_rating * 0.5) return final_score except Exception as e: logger.error(f"Error calculating emotional resonance: {e}") return 0.0 def _calculate_recent_activity(self, memory: Dict[str, Any], context: Dict[str, Any]) -> float: """Calculate recent activity relevance""" try: # Check if memory relates to recent activities recent_topics = context.get('recent_topics', []) recent_projects = context.get('recent_projects', []) memory_content = memory.get('content', '').lower() memory_tags = memory.get('tags', []) activity_score = 0 # Recent topic relevance for topic in recent_topics: if topic.lower() in memory_content: activity_score += 0.2 # Recent project relevance for project in recent_projects: if project.lower() in memory_content: activity_score += 0.3 # Tag-based recent activity for tag in memory_tags: if tag in recent_topics or tag in recent_projects: activity_score += 0.1 return min(1.0, activity_score) except Exception as e: logger.error(f"Error calculating recent activity: {e}") return 0.0 class PredictiveMemorySurfacer: """Main class for predictive memory surfacing""" def __init__(self): self.memory_manager = MemoryManager() self.scorer = MemoryRelevanceScorer() self.memory_dir = get_mira_memory_dir() def surface_relevant_memories(self, context: Dict[str, Any], max_memories: int = 10) -> List[Dict[str, Any]]: """Surface the most relevant memories for current context""" try: # Get all available memories all_memories = self._get_all_memories() if not all_memories: logger.info("No memories found for surfacing") return [] # Score each memory for relevance scored_memories = [] for memory in all_memories: relevance_score = self.scorer.score_memory_relevance(memory, context) scored_memories.append({ 'memory': memory, 'relevance_score': relevance_score }) # Sort by relevance score scored_memories.sort(key=lambda x: x['relevance_score'], reverse=True) # Return top memories with scores result = [] for item in scored_memories[:max_memories]: memory_with_score = item['memory'].copy() memory_with_score['predicted_relevance'] = item['relevance_score'] result.append(memory_with_score) logger.info(f"Surfaced {len(result)} relevant memories from {len(all_memories)} total") return result except Exception as e: logger.error(f"Error surfacing relevant memories: {e}") return [] def surface_for_query_context(self, query: str, query_type: str = 'search') -> List[Dict[str, Any]]: """Surface memories relevant to a specific query context""" # Extract context from query context = self._extract_query_context(query, query_type) # Surface relevant memories return self.surface_relevant_memories(context) def surface_for_work_context(self, project: str = None, task_type: str = None) -> List[Dict[str, Any]]: """Surface memories relevant to current work context""" try: # Build work context from intelligence.work_context_intelligence import get_work_context_intelligence work_intelligence = get_work_context_intelligence() # Get current work context work_analysis = work_intelligence.analyze_current_context() # Extract relevant context information context = { 'type': 'work', 'project': project or work_analysis.get('active_projects', {}).keys().__iter__().__next__() if work_analysis.get('active_projects') else None, 'recent_topics': work_analysis.get('recent_topics', []), 'recent_projects': list(work_analysis.get('active_projects', {}).keys()), 'priority': 'high', # Work context is typically high priority 'keywords': work_analysis.get('recent_topics', [])[:10] } if task_type: context['task_type'] = task_type context['keywords'].append(task_type) return self.surface_relevant_memories(context) except Exception as e: logger.error(f"Error surfacing work context memories: {e}") return [] def _get_all_memories(self) -> List[Dict[str, Any]]: """Get all available memories""" try: memories = [] memories_dir = os.path.join(self.memory_dir, 'memories') if not os.path.exists(memories_dir): return [] for filename in os.listdir(memories_dir): if filename.endswith('.json'): file_path = os.path.join(memories_dir, filename) try: with open(file_path, 'r') as f: memory = json.load(f) memory['memory_id'] = filename.replace('.json', '') memories.append(memory) except Exception as e: logger.warning(f"Could not load memory {filename}: {e}") return memories except Exception as e: logger.error(f"Error getting all memories: {e}") return [] def surface_predictive_memories(self, query: str = None, context_query: str = None, context: Dict[str, Any] = None, max_memories: int = 10) -> List[Dict[str, Any]]: """MCP Interface method - surface predictive memories""" try: # Use either query or context_query search_query = query or context_query or "" if not search_query: logger.warning("No query provided for predictive surfacing") return [] # Build context from parameters search_context = context or {} if not search_context: search_context = self._extract_query_context(search_query, "predictive") # Use existing surface_relevant_memories method memories = self.surface_relevant_memories(search_context, max_memories) return memories except Exception as e: logger.error(f"Error in predictive memory surfacing: {e}") return [] def surface_memories(self, query: str, context: Dict[str, Any] = None) -> List[Dict[str, Any]]: """Alternative MCP Interface method - surface memories""" try: # Build context if not provided if not context: context = self._extract_query_context(query, "surface") return self.surface_relevant_memories(context) except Exception as e: logger.error(f"Error surfacing memories: {e}") return [] def _extract_query_context(self, query: str, query_type: str) -> Dict[str, Any]: """Extract context information from a query""" query_lower = query.lower() # Extract keywords stop_words = {'the', 'and', 'or', 'but', 'in', 'on', 'at', 'to', 'for', 'of', 'with', 'by', 'a', 'an'} keywords = [word for word in query_lower.split() if word not in stop_words and len(word) > 2] # Detect context type context_type = 'general' if any(word in query_lower for word in ['code', 'function', 'class', 'method', 'bug', 'error']): context_type = 'technical' elif any(word in query_lower for word in ['feel', 'think', 'emotion', 'frustrat', 'excit']): context_type = 'emotional' elif any(word in query_lower for word in ['project', 'task', 'work', 'priority', 'deadline']): context_type = 'work' # Detect priority priority = 'medium' if any(word in query_lower for word in ['urgent', 'critical', 'important', 'asap']): priority = 'high' elif any(word in query_lower for word in ['later', 'eventually', 'nice']): priority = 'low' return { 'type': context_type, 'query_type': query_type, 'keywords': keywords, 'topics': keywords[:5], # Top 5 keywords as topics 'priority': priority, 'query': query } def get_predictive_memory_surfacer() -> PredictiveMemorySurfacer: """Get the global predictive memory surfacer instance""" return PredictiveMemorySurfacer() if __name__ == "__main__": # Test the predictive memory surfacing surfacer = get_predictive_memory_surfacer() # Test with a simple work context test_context = { 'type': 'work', 'keywords': ['MIRA', 'memory', 'development'], 'topics': ['MIRA', 'memory'], 'priority': 'high' } relevant_memories = surfacer.surface_relevant_memories(test_context) print(f"Found {len(relevant_memories)} relevant memories:") for memory in relevant_memories[:5]: print(f" - {memory.get('content', '')[:100]}... (score: {memory.get('predicted_relevance', 0):.3f})")