"""Memory system using JSON for simple key-value storage""" import os import json import logging import sys from datetime import datetime # Add parent directory to path for imports sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from config import MODEL_CONFIG class MemorySystem: """Memory system using JSON for simple key-value storage""" def __init__(self, json_db_path=None, config=None): self.config = config or MODEL_CONFIG or {} # Get paths from config or use defaults memory_config = self.config.get('memory', {}) if self.config else {} self.json_db_path = json_db_path or memory_config.get('json_path', './memory.json') self.json_memory = {} # Initialize JSON database self.load_json_memory() def is_ready(self): """Check if memory system is fully initialized""" return self.json_memory is not None def load_json_memory(self): """Load JSON memory database""" try: if os.path.exists(self.json_db_path): with open(self.json_db_path, 'r', encoding='utf-8') as f: self.json_memory = json.load(f) logging.info(f"Loaded JSON memory with {len(self.json_memory)} entries") else: self.json_memory = {} logging.info("Created new JSON memory database") except Exception as e: logging.error(f"Error loading JSON memory: {e}") self.json_memory = {} def save_json_memory(self): """Save JSON memory database""" try: with open(self.json_db_path, 'w', encoding='utf-8') as f: json.dump(self.json_memory, f, indent=2, ensure_ascii=False) except Exception as e: logging.error(f"Error saving JSON memory: {e}") def store_memory(self, text, metadata=None, memory_type="conversation"): """Store a memory in JSON""" timestamp = datetime.now().isoformat() # Store in JSON memory_id = f"{memory_type}_{timestamp}" self.json_memory[memory_id] = { "text": text, "metadata": metadata or {}, "type": memory_type, "timestamp": timestamp } self.save_json_memory() logging.info(f"Stored memory in JSON: {memory_id[:20]}...") def retrieve_relevant_memories(self, query, n_results=5): """Retrieve relevant memories using keyword search in JSON""" relevant_memories = [] # Simple keyword search in JSON if self.json_memory: query_lower = query.lower() query_words = set(query_lower.split()) for memory_id, memory_data in self.json_memory.items(): text_lower = memory_data.get("text", "").lower() text_words = set(text_lower.split()) # Simple overlap check overlap = len(query_words & text_words) if overlap > 0: relevant_memories.append({ "text": memory_data["text"], "metadata": memory_data.get("metadata", {}), "distance": 1.0 - (overlap / max(len(query_words), len(text_words))) }) # Sort by relevance (lower distance = more relevant) relevant_memories.sort(key=lambda x: x.get("distance", 1.0)) relevant_memories = relevant_memories[:n_results] logging.info(f"Retrieved {len(relevant_memories)} relevant memories from JSON DB") return relevant_memories def get_json_memory(self, key): """Get a specific memory by key from JSON database""" return self.json_memory.get(key) def set_json_memory(self, key, value, metadata=None): """Set a key-value memory in JSON database""" self.json_memory[key] = { "value": value, "metadata": metadata or {}, "timestamp": datetime.now().isoformat() } self.save_json_memory() def get_all_json_memories(self): """Get all JSON memories""" return self.json_memory.copy()