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)