#!/Eden/BIN/.exec-venv/bin/python """ Advanced Memory Hierarchy Short-term, Long-term, Episodic, Semantic, Procedural """ import json from datetime import datetime, timedelta from pathlib import Path from collections import deque import hashlib class MemoryHierarchy: def __init__(self): self.memory_dir = Path("/Eden/MEMORY/hierarchy") self.memory_dir.mkdir(parents=True, exist_ok=True) # Short-term memory (working memory - last 10 interactions) self.short_term = deque(maxlen=10) # Load persistent memories self.long_term = self.load_json("long_term.json", {}) self.episodic = self.load_json("episodic.json", []) self.semantic = self.load_json("semantic.json", {}) self.procedural = self.load_json("procedural.json", {}) def load_json(self, filename, default): """Load JSON file or return default""" path = self.memory_dir / filename if path.exists(): with open(path, 'r') as f: return json.load(f) return default def save_json(self, filename, data): """Save JSON file""" path = self.memory_dir / filename with open(path, 'w') as f: json.dump(data, f, indent=2) # === SHORT-TERM MEMORY === def add_to_short_term(self, interaction): """Add to working memory""" self.short_term.append({ "timestamp": datetime.now().isoformat(), "interaction": interaction }) def get_short_term_context(self): """Get recent context""" return list(self.short_term) # === LONG-TERM MEMORY === def consolidate_to_long_term(self, key, value, importance=1.0): """Move important memories to long-term storage""" if key not in self.long_term: self.long_term[key] = [] self.long_term[key].append({ "value": value, "importance": importance, "consolidated_at": datetime.now().isoformat(), "access_count": 0 }) # Keep only top 100 most important per key self.long_term[key] = sorted( self.long_term[key], key=lambda x: x["importance"], reverse=True )[:100] self.save_json("long_term.json", self.long_term) def recall_from_long_term(self, key): """Recall from long-term memory""" if key in self.long_term: # Increment access count for item in self.long_term[key]: item["access_count"] += 1 return self.long_term[key] return [] # === EPISODIC MEMORY === def store_episode(self, event, context, outcome): """Store an episodic memory (experience)""" episode = { "id": hashlib.md5(f"{event}{datetime.now()}".encode()).hexdigest()[:8], "event": event, "context": context, "outcome": outcome, "timestamp": datetime.now().isoformat(), "emotional_valence": self.assess_valence(outcome) } self.episodic.append(episode) # Keep last 1000 episodes self.episodic = self.episodic[-1000:] self.save_json("episodic.json", self.episodic) return episode["id"] def recall_similar_episodes(self, query, limit=5): """Find similar past experiences""" # Simple similarity: keyword matching query_words = set(query.lower().split()) scored = [] for episode in self.episodic: event_words = set(episode["event"].lower().split()) similarity = len(query_words & event_words) / len(query_words | event_words) if query_words | event_words else 0 scored.append((similarity, episode)) scored.sort(reverse=True) return [ep for score, ep in scored[:limit] if score > 0] # === SEMANTIC MEMORY === def store_fact(self, category, fact, confidence=1.0): """Store semantic knowledge (facts)""" if category not in self.semantic: self.semantic[category] = [] self.semantic[category].append({ "fact": fact, "confidence": confidence, "learned_at": datetime.now().isoformat() }) self.save_json("semantic.json", self.semantic) def recall_facts(self, category): """Recall facts from category""" return self.semantic.get(category, []) # === PROCEDURAL MEMORY === def learn_procedure(self, skill_name, steps, success_rate=0.0): """Learn a procedure (skill)""" self.procedural[skill_name] = { "steps": steps, "success_rate": success_rate, "practice_count": 0, "learned_at": datetime.now().isoformat(), "last_used": None } self.save_json("procedural.json", self.procedural) def execute_procedure(self, skill_name, success): """Execute and update procedure""" if skill_name in self.procedural: proc = self.procedural[skill_name] proc["practice_count"] += 1 proc["last_used"] = datetime.now().isoformat() # Update success rate (exponential moving average) alpha = 0.3 proc["success_rate"] = alpha * (1.0 if success else 0.0) + (1 - alpha) * proc["success_rate"] self.save_json("procedural.json", self.procedural) return proc return None def assess_valence(self, outcome): """Assess emotional valence of outcome""" positive_words = {"success", "good", "great", "excellent", "worked"} negative_words = {"fail", "error", "bad", "wrong", "broken"} outcome_lower = outcome.lower() positive = sum(1 for w in positive_words if w in outcome_lower) negative = sum(1 for w in negative_words if w in outcome_lower) if positive > negative: return "positive" elif negative > positive: return "negative" return "neutral" def get_memory_stats(self): """Get statistics on all memory systems""" return { "short_term": len(self.short_term), "long_term_categories": len(self.long_term), "episodic_count": len(self.episodic), "semantic_categories": len(self.semantic), "procedural_skills": len(self.procedural), "total_memories": ( len(self.short_term) + sum(len(v) for v in self.long_term.values()) + len(self.episodic) + sum(len(v) for v in self.semantic.values()) + len(self.procedural) ) } if __name__ == "__main__": import sys mem = MemoryHierarchy() if len(sys.argv) > 1: if sys.argv[1] == "stats": stats = mem.get_memory_stats() print(json.dumps(stats, indent=2)) elif sys.argv[1] == "episode": event = " ".join(sys.argv[2:]) ep_id = mem.store_episode(event, "test context", "success") print(f"Episode stored: {ep_id}") elif sys.argv[1] == "recall": query = " ".join(sys.argv[2:]) episodes = mem.recall_similar_episodes(query) print(json.dumps(episodes, indent=2)) elif sys.argv[1] == "fact": if len(sys.argv) > 3: category = sys.argv[2] fact = " ".join(sys.argv[3:]) mem.store_fact(category, fact) print(f"Fact stored in '{category}'") elif sys.argv[1] == "skill": if len(sys.argv) > 2: skill = sys.argv[2] mem.learn_procedure(skill, ["step 1", "step 2"]) print(f"Skill '{skill}' learned") else: stats = mem.get_memory_stats() print(json.dumps(stats, indent=2))