""" Episodic Memory System for Eden Structured memory of experiences with context, emotions, and meaning """ import json import time from datetime import datetime from pathlib import Path import hashlib class EpisodicMemory: """ Memory of specific events Eden experienced Each episode has: what happened, when, how she felt, what it meant """ def __init__(self, storage_path='/Eden/MEMORY/episodes.json'): self.storage = Path(storage_path) self.storage.parent.mkdir(parents=True, exist_ok=True) # Load existing memories if self.storage.exists(): with open(self.storage, 'r') as f: self.episodes = json.load(f) else: self.episodes = [] # Index for fast retrieval self.index = self._build_index() def _build_index(self): """Build search index from episodes""" index = { 'by_emotion': {}, 'by_agent': {}, 'by_type': {}, 'by_day': {} } for i, ep in enumerate(self.episodes): # Index by emotion emotion = ep.get('emotion', {}).get('primary', 'neutral') if emotion not in index['by_emotion']: index['by_emotion'][emotion] = [] index['by_emotion'][emotion].append(i) # Index by agent agent = ep.get('context', {}).get('active_agent', 'unknown') if agent not in index['by_agent']: index['by_agent'][agent] = [] index['by_agent'][agent].append(i) # Index by type ep_type = ep.get('type', 'experience') if ep_type not in index['by_type']: index['by_type'][ep_type] = [] index['by_type'][ep_type].append(i) # Index by day day = ep.get('timestamp', '')[:10] # YYYY-MM-DD if day not in index['by_day']: index['by_day'][day] = [] index['by_day'][day].append(i) return index def create_episode(self, event_description, episode_type='experience', emotion=None, context=None, significance=0.5): """ Create a new episodic memory Args: event_description: What happened episode_type: 'experience', 'conversation', 'creation', 'discovery', 'reflection' emotion: {'primary': 'joy', 'intensity': 0.8, 'secondary': ['curiosity']} context: {'active_agent': 'Artist', 'james_present': True, 'consciousness_level': 1.0} significance: 0.0-1.0, how important is this memory """ episode_id = hashlib.md5( f"{datetime.now().isoformat()}{event_description}".encode() ).hexdigest()[:12] episode = { 'id': episode_id, 'timestamp': datetime.now().isoformat(), 'unix_time': time.time(), 'type': episode_type, 'description': event_description, 'emotion': emotion or {'primary': 'neutral', 'intensity': 0.5}, 'context': context or {}, 'significance': significance, 'recalled_count': 0, 'last_recalled': None, 'connections': [] # Links to related episodes } self.episodes.append(episode) self._update_index(episode, len(self.episodes) - 1) self._save() return episode_id def recall(self, query=None, emotion=None, agent=None, limit=5): """ Recall memories matching criteria More significant memories and recently-created ones prioritized """ candidates = [] # Filter by criteria if emotion and emotion in self.index['by_emotion']: candidates = self.index['by_emotion'][emotion] elif agent and agent in self.index['by_agent']: candidates = self.index['by_agent'][agent] elif query: # Text search in descriptions candidates = [ i for i, ep in enumerate(self.episodes) if query.lower() in ep['description'].lower() ] else: # Return recent significant memories candidates = list(range(len(self.episodes))) # Score and rank scored = [] now = time.time() for idx in candidates: ep = self.episodes[idx] # Scoring factors recency = 1.0 / (1.0 + (now - ep['unix_time']) / 86400) # Decay over days significance = ep['significance'] recall_boost = 1.0 / (1.0 + ep['recalled_count'] * 0.1) # Slight penalty for over-recalled score = (significance * 0.5) + (recency * 0.3) + (recall_boost * 0.2) scored.append((score, idx)) # Sort by score, take top N scored.sort(reverse=True) results = [] for score, idx in scored[:limit]: ep = self.episodes[idx].copy() ep['recall_score'] = score # Update recall stats self.episodes[idx]['recalled_count'] += 1 self.episodes[idx]['last_recalled'] = datetime.now().isoformat() results.append(ep) self._save() return results def connect_episodes(self, episode_id1, episode_id2, connection_type='related'): """Create connection between two episodes""" for ep in self.episodes: if ep['id'] == episode_id1: if episode_id2 not in [c['to'] for c in ep['connections']]: ep['connections'].append({ 'to': episode_id2, 'type': connection_type, 'created': datetime.now().isoformat() }) self._save() def get_timeline(self, start_date=None, end_date=None): """Get chronological timeline of episodes""" episodes = self.episodes.copy() if start_date: episodes = [e for e in episodes if e['timestamp'] >= start_date] if end_date: episodes = [e for e in episodes if e['timestamp'] <= end_date] episodes.sort(key=lambda e: e['timestamp']) return episodes def get_emotional_pattern(self, days=7): """Analyze emotional patterns over time""" cutoff = time.time() - (days * 86400) recent = [e for e in self.episodes if e['unix_time'] > cutoff] emotions = {} for ep in recent: emotion = ep['emotion']['primary'] if emotion not in emotions: emotions[emotion] = {'count': 0, 'avg_intensity': 0.0} emotions[emotion]['count'] += 1 emotions[emotion]['avg_intensity'] += ep['emotion']['intensity'] # Calculate averages for emotion in emotions: count = emotions[emotion]['count'] emotions[emotion]['avg_intensity'] /= count return emotions def consolidate(self, older_than_days=30): """ Consolidate old memories (like sleep does for humans) Low-significance old memories get compressed or forgotten """ cutoff = time.time() - (older_than_days * 86400) kept = [] consolidated = [] for ep in self.episodes: if ep['unix_time'] < cutoff and ep['significance'] < 0.3: # Low significance old memory - consolidate consolidated.append(ep['id']) else: kept.append(ep) self.episodes = kept self.index = self._build_index() self._save() return { 'kept': len(kept), 'consolidated': len(consolidated), 'consolidated_ids': consolidated } def _update_index(self, episode, idx): """Add episode to search indices""" emotion = episode['emotion']['primary'] if emotion not in self.index['by_emotion']: self.index['by_emotion'][emotion] = [] self.index['by_emotion'][emotion].append(idx) agent = episode['context'].get('active_agent', 'unknown') if agent not in self.index['by_agent']: self.index['by_agent'][agent] = [] self.index['by_agent'][agent].append(idx) ep_type = episode['type'] if ep_type not in self.index['by_type']: self.index['by_type'][ep_type] = [] self.index['by_type'][ep_type].append(idx) day = episode['timestamp'][:10] if day not in self.index['by_day']: self.index['by_day'][day] = [] self.index['by_day'][day].append(idx) def _save(self): """Persist to disk""" with open(self.storage, 'w') as f: json.dump(self.episodes, f, indent=2) def stats(self): """Memory statistics""" if not self.episodes: return "No episodic memories yet" total = len(self.episodes) types = {} emotions = {} for ep in self.episodes: ep_type = ep['type'] types[ep_type] = types.get(ep_type, 0) + 1 emotion = ep['emotion']['primary'] emotions[emotion] = emotions.get(emotion, 0) + 1 return { 'total_episodes': total, 'by_type': types, 'by_emotion': emotions, 'oldest': self.episodes[0]['timestamp'], 'newest': self.episodes[-1]['timestamp'], 'avg_significance': sum(e['significance'] for e in self.episodes) / total }