| """ |
| Nova Persistent Chat Memory Engine |
| Manages long-term conversational history, saves memory to disk, and restores context across sessions. |
| """ |
|
|
| import os |
| import json |
| from typing import List, Dict |
|
|
|
|
| class NovaChatMemory: |
| def __init__(self, storage_path: str = None): |
| if storage_path is None: |
| root_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")) |
| storage_path = os.path.join(root_dir, "checkpoints", "chat_memory.json") |
|
|
| self.storage_path = storage_path |
| self.history: List[Dict[str, str]] = [] |
| self.load_memory() |
|
|
| def add_user_message(self, content: str): |
| self.history.append({"role": "user", "content": content}) |
| self.save_memory() |
|
|
| def add_assistant_message(self, content: str): |
| self.history.append({"role": "assistant", "content": content}) |
| self.save_memory() |
|
|
| def get_messages(self) -> List[Dict[str, str]]: |
| return self.history |
|
|
| def clear(self): |
| self.history = [] |
| self.save_memory() |
|
|
| def save_memory(self): |
| try: |
| os.makedirs(os.path.dirname(self.storage_path), exist_ok=True) |
| with open(self.storage_path, "w", encoding="utf-8") as f: |
| json.dump(self.history, f, indent=2) |
| except Exception as e: |
| print(f"Error saving chat memory: {e}") |
|
|
| def load_memory(self): |
| if os.path.exists(self.storage_path): |
| try: |
| with open(self.storage_path, "r", encoding="utf-8") as f: |
| self.history = json.load(f) |
| except Exception as e: |
| print(f"Error loading chat memory: {e}") |
| self.history = [] |
| else: |
| self.history = [] |
|
|
