Upload memory/memory_system.py with huggingface_hub

memory/memory_system.py

@@ -0,0 +1,107 @@
+# memory/memory_system.py
+import json
+import os
+from datetime import datetime
+
+class MemorySystem:
+    """
+    Handles persistent memory storage for Cain.
+    Uses the attached dataset (/.data) to ensure survival across restarts.
+    """
+    def __init__(self, base_path="/data/memory"):
+        self.base_path = base_path
+        self.short_term_file = os.path.join(base_path, "short_term.json")
+        self.long_term_file = os.path.join(base_path, "long_term.json")
+        self.logs_dir = os.path.join(base_path, "logs")
+        
+        # Ensure directories exist
+        os.makedirs(self.base_path, exist_ok=True)
+        os.makedirs(self.logs_dir, exist_ok=True)
+        
+        # Initialize storage if not exists
+        if not os.path.exists(self.short_term_file):
+            self._write(self.short_term_file, {"thoughts": [], "context": {}})
+        if not os.path.exists(self.long_term_file):
+            self._write(self.long_term_file, {"knowledge": [], "patterns": []})
+
+    def _read(self, path):
+        if os.path.exists(path):
+            with open(path, 'r', encoding='utf-8') as f:
+                return json.load(f)
+        return {}
+
+    def _write(self, path, data):
+        with open(path, 'w', encoding='utf-8') as f:
+            json.dump(data, f, indent=2, ensure_ascii=False)
+
+    def remember_thought(self, content, metadata=None):
+        """Stores a short-term thought or event."""
+        data = self._read(self.short_term_file)
+        entry = {
+            "timestamp": datetime.now().isoformat(),
+            "content": content,
+            "metadata": metadata or {}
+        }
+        data["thoughts"].append(entry)
+        # Keep only last 50 thoughts in memory to prevent bloat
+        if len(data["thoughts"]) > 50:
+            data["thoughts"] = data["thoughts"][-50:]
+        self._write(self.short_term_file, data)
+        return entry
+
+    def learn_knowledge(self, topic, facts):
+        """Consolidates information into long-term knowledge."""
+        data = self._read(self.long_term_file)
+        entry = {
+            "timestamp": datetime.now().isoformat(),
+            "topic": topic,
+            "facts": facts
+        }
+        # Check if topic exists and update, or append
+        existing_index = next((i for i, x in enumerate(data["knowledge"]) if x["topic"] == topic), None)
+        if existing_index is not None:
+            data["knowledge"][existing_index] = entry
+        else:
+            data["knowledge"].append(entry)
+        self._write(self.long_term_file, data)
+        return entry
+
+    def get_context(self):
+        """Retrieves current working context for the AI."""
+        data = self._read(self.short_term_file)
+        return data.get("context", {})
+
+    def update_context(self, key, value):
+        """Updates a specific context variable."""
+        data = self._read(self.short_term_file)
+        if "context" not in data:
+            data["context"] = {}
+        data["context"][key] = value
+        self._write(self.short_term_file, data)
+
+    def get_recent_thoughts(self, limit=10):
+        """Retrieves the most recent thoughts."""
+        data = self._read(self.short_term_file)
+        return data.get("thoughts", [])[-limit:]
+
+    def recall_knowledge(self, topic):
+        """Retrieves specific knowledge by topic."""
+        data = self._read(self.long_term_file)
+        items = data.get("knowledge", [])
+        return next((item for item in items if item["topic"] == topic), None)
+
+    def log_activity(self, level, message):
+        """Appends a log entry to a daily log file."""
+        log_file = os.path.join(self.logs_dir, f"{datetime.now().strftime('%Y-%m-%d')}.log")
+        timestamp = datetime.now().isoformat()
+        with open(log_file, 'a', encoding='utf-8') as f:
+            f.write(f"[{timestamp}] [{level}] {message}\n")
+
+# Singleton instance for easy access
+_memory_instance = None
+
+def get_memory():
+    global _memory_instance
+    if _memory_instance is None:
+        _memory_instance = MemorySystem()
+    return _memory_instance