Spaces:
Build error
Build error
File size: 2,840 Bytes
f48c478 2248e6e f48c478 2248e6e f48c478 2248e6e f48c478 2248e6e f48c478 2248e6e f48c478 2248e6e f48c478 2248e6e f48c478 2248e6e f48c478 2248e6e f48c478 2248e6e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 | 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) |