optimized_store.py

import pickle
from pathlib import Path
from typing import Optional, Dict, Any
import logging
import mmap
import os

logger = logging.getLogger(__name__)

class OptimizedVectorStore:
    _instance = None
    
    def __init__(self, file_path: Path):
        self.file_path = file_path
        self._store = None
        self._docstore = None
        self._index_to_docstore_id = None
        self._embeddings = None
        
    @property
    def store(self):
        """Lazy Loading mit Memory Mapping"""
        if self._store is None:
            self._load_store()
        return self._store
    
    def _load_store(self):
        """Lädt den Vector Store mit Memory Mapping"""
        if not self.file_path.exists():
            raise FileNotFoundError(f"Vector Store nicht gefunden: {self.file_path}")
            
        logger.info(f"Lade Vector Store von {self.file_path}")
        try:
            # Memory Mapping für große Dateien
            with open(self.file_path, 'rb') as f:
                # Memory-Map the file
                mm = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
                self._store = pickle.load(mm)
                
                # Extrahiere wichtige Komponenten
                if hasattr(self._store, 'docstore'):
                    self._docstore = self._store.docstore
                if hasattr(self._store, 'index_to_docstore_id'):
                    self._index_to_docstore_id = self._store.index_to_docstore_id
                if hasattr(self._store, 'embeddings'):
                    self._embeddings = self._store.embeddings
                
                mm.close()
                logger.info("Vector Store erfolgreich geladen")
                
        except Exception as e:
            logger.error(f"Fehler beim Laden des Vector Stores: {str(e)}")
            raise
    
    @property
    def docstore(self):
        """Lazy Loading des Docstores"""
        if self._store is None:
            self._load_store()
        return self._docstore
    
    @property
    def index_to_docstore_id(self):
        """Lazy Loading der Index-Mapping"""
        if self._store is None:
            self._load_store()
        return self._index_to_docstore_id

    @property
    def embedding_function(self):
        """Lazy Loading der Embedding-Funktion"""
        if self._embeddings is None and self._store is not None:
            self._embeddings = self._store.embedding_function
        return self._embeddings

    def similarity_search_with_relevance_scores(self, *args, **kwargs):
        """Delegiert Suche an den Store"""
        return self.store.similarity_search_with_relevance_scores(*args, **kwargs)
    
    def similarity_search(self, *args, **kwargs):
        """Delegiert Suche an den Store"""
        return self.store.similarity_search(*args, **kwargs)