from typing import List from .vectordb import VectorDB from .embedding import EmbeddingModel class Namespace: def __init__( self, db: VectorDB, name: str, embedding_key: str, index_key: str, embedding_model: EmbeddingModel, ): self._db = db self._name = name self._embedding_key = embedding_key self._index_keys = [k for k in index_key.split(',') if k] self._embedding_model = embedding_model def name(self) -> str: return self._name def text(self, metadata: dict) -> str: return metadata[self._embedding_key] def index(self, metadata: dict) -> List[str]: return [str(metadata[k]) for k in self._index_keys] def retrieve(self, text: str, limit: int = 5) -> List[dict]: vec = self._embedding_model.embedding(text) return self._db.retrieve(self.name(), vec, limit) def upsert(self, metadata: dict): idx = self.index(metadata) vec_id = self._db.vec_id(idx) text = self.text(metadata) vec = self._embedding_model.embedding(text) self._db.upsert(self.name(), vec_id, vec, metadata) def delete(self, vec_id: str): self._db.delete(self.name(), vec_id)