services/memory_manager.py

import json
import os

MEMORY_FILE = "database/memory.json"
MAX_HISTORY = 10


class MemoryManager:

    def __init__(self):
        self._ensure_memory_file()

    # =========================
    # INIT MEMORY FILE
    # =========================

    def _ensure_memory_file(self):

        if not os.path.exists("database"):
            os.makedirs("database")

        if not os.path.exists(MEMORY_FILE):

            memory_structure = {
                "history": [],
                "preferences": {},
                "projects": {},
                "knowledge": []
            }

            with open(MEMORY_FILE, "w", encoding="utf-8") as f:
                json.dump(memory_structure, f, indent=2, ensure_ascii=False)

    # =========================
    # LOAD MEMORY
    # =========================

    def load_memory(self):

        with open(MEMORY_FILE, "r", encoding="utf-8") as f:
            return json.load(f)

    # =========================
    # SAVE MEMORY
    # =========================

    def save_memory(self, memory):

        with open(MEMORY_FILE, "w", encoding="utf-8") as f:
            json.dump(memory, f, indent=2, ensure_ascii=False)

    # =========================
    # ADD CONVERSATION
    # =========================

    def add_exchange(self, user_message, assistant_response):

        memory = self.load_memory()

        history = memory.get("history", [])

        history.append({
            "user": user_message,
            "assistant": assistant_response
        })

        history = history[-MAX_HISTORY:]

        memory["history"] = history

        self.save_memory(memory)

    # =========================
    # BUILD CONTEXT
    # =========================

    def build_context(self):

        memory = self.load_memory()

        history = memory.get("history", [])

        if not history:
            return ""

        context = "\n\n## Conversation Context\n"

        for exchange in history:

            context += f"User: {exchange['user']}\n"
            context += f"Assistant: {exchange['assistant']}\n"

        return context

    # =========================
    # STORE PREFERENCE
    # =========================

    def store_preference(self, key, value):

        memory = self.load_memory()

        memory["preferences"][key] = value

        self.save_memory(memory)

    # =========================
    # STORE PROJECT
    # =========================

    def store_project(self, name, data):

        memory = self.load_memory()

        memory["projects"][name] = data

        self.save_memory(memory)

    # =========================
    # ADD KNOWLEDGE
    # =========================

    def add_knowledge(self, info):

        memory = self.load_memory()

        memory["knowledge"].append(info)

        self.save_memory(memory)

    # =========================
    # CLEAR MEMORY
    # =========================

    def clear(self):

        memory_structure = {
            "history": [],
            "preferences": {},
            "projects": {},
            "knowledge": []
        }

        self.save_memory(memory_structure)