VectorDatabaseBase & sanitize_metadata

Table of contents

• sanitize_metadata
• VectorDatabaseBase
  • Constructor
  • Abstract methods
  • Methods with default implementations
  • Async wrappers
  • Context-manager support
  • Dunder methods

sanitize_metadata

def sanitize_metadata(metadata: Dict) -> Dict

Overview

The single, shared place where a chunk's metadata dictionary is normalized so it's safe to store in any of the three backends. Every backend's add() calls this before persisting metadata, so the rules only need to live in one place.

Parameters

Returns

Dict — a new dictionary with every value coerced into one of: str, int, float, bool, or a string (list/dict/other are stringified). Keys whose value was None are dropped entirely.

Rules (from source)

Raises

Does not raise for the documented value types. If json.dumps fails on an unusual object and str() also fails, an unhandled exception from str() would propagate — this edge case is not guarded further in the source.

Side effects

None — pure function, no I/O, no logging.

Examples

sanitize_metadata({"category": "tech", "score": 0.9, "flag": None})
# {"category": "tech", "score": 0.9}   # 'flag' dropped

sanitize_metadata({"tags": ["ai", "ml", "nlp"]})
# {"tags": "ai, ml, nlp"}

sanitize_metadata({"nested": {"a": 1}})
# {"nested": '{"a": 1}'}

from dataclasses import dataclass
@dataclass
class Meta:
    source: str
sanitize_metadata(Meta(source="web"))
# {"source": "web"}

Best practices

• Call this before handing metadata to a backend if you're bypassing add() and talking to a backend's underlying client/collection/index directly.
• Don't rely on list values surviving round-trip — they become a single joined string. If you need to re-filter on individual list members later (e.g. Pinecone's $in), you'll need to re-split that string yourself; the source comments mention this as the intended pattern but no re-parsing helper is provided in this codebase.

Common mistakes

• Assuming None values are preserved as null — they are silently dropped, which can hide bugs where a field was expected but never set.

VectorDatabaseBase

class VectorDatabaseBase(ABC):

Purpose

Abstract base class defining the contract every backend (FAISSVectorDatabase, ChromaVectorDatabase, PineconeVectorDatabase) must satisfy, so orchestration code can treat all three interchangeably.

Per the class docstring, subclasses must implement add, search, remove_by_doc_id, get_stats, and __len__. They should implement (optional, with a sensible default provided) clear, save/load, get_chunk_by_id, get_chunks_by_doc_id, and the async wrappers asearch/aadd.

Constructor

def __init__(self, embedder: Optional[Any] = None) -> None

Stores embedder on self.embedder. No validation is performed at this level (subclasses do their own validation).

Embedding helpers (protected, used internally by subclasses)

These are not part of the "public API" per the project's own convention (leading underscore), but are documented here because subclasses rely on them and they explain observable error behavior:

• _require_embedder() — raises ValueError if self.embedder is None.
• _encode(texts: List[str]) -> np.ndarray — calls _require_embedder(), then self.embedder.encode(texts), casts to float32. Wraps any exception from the embedder in RuntimeError(f"Embedding generation failed: {exc}").
• _encode_query(query: Union[str, np.ndarray]) -> np.ndarray — if query is a str, encodes it via _encode([query])[0]; otherwise casts it to a float32 array.

Abstract methods (must implement)

add

@abstractmethod
def add(self, chunks: List[DocumentChunk], embeddings: Optional[np.ndarray] = None) -> None

Add or upsert a list of chunks. If embeddings is None, the backend is expected to compute them via self._encode(texts). Per the docstring, every backend must call sanitize_metadata(chunk.metadata) before storing metadata.

search

@abstractmethod
def search(
    self,
    query: Union[str, np.ndarray],
    top_k: int = 5,
    score_threshold: Optional[float] = None,
    **kwargs: Any,
) -> List[Tuple[DocumentChunk, float]]

Search for the most similar chunks to query (text or a pre-computed 1-D vector). Must return results sorted by score descending. **kwargs carries backend-specific parameters (e.g. filter_dict for Pinecone, filters for Chroma, doc_id_filter for FAISS — these names are not unified across backends, only the return type is).

remove_by_doc_id

@abstractmethod
def remove_by_doc_id(self, doc_id: str) -> int

Remove all chunks belonging to doc_id. Returns the number of chunks removed (0 if not found). Note from the docstring: this is the unified name — FAISS uses this name natively, Pinecone's native name is delete_by_doc_id (aliased), Chroma's is delete_by_filter (aliased).

get_stats

@abstractmethod
def get_stats(self) -> Dict[str, Any]

Must return a dict containing at least: total_vectors (int), embedding_dim (int), distance_metric (str), has_embedder (bool). See each backend's page for the actual keys returned — they vary beyond this minimum.

__len__

@abstractmethod
def __len__(self) -> int

Return the total number of stored vectors.

Methods with default implementations

Async wrappers

async def asearch(self, query, top_k=10, score_threshold=None, **kwargs) -> List[Tuple[DocumentChunk, float]]
async def aadd(self, chunks, embeddings=None, **kwargs) -> None
async def aremove_by_doc_id(self, doc_id: str) -> int

Default implementations run the corresponding synchronous method inside a worker thread via asyncio.to_thread(...), so calling them never blocks the event loop even though the underlying backend has no native async support. Backends may override these for native async behavior — as documented on the backend pages, none of the three included backends currently do (they redefine these methods but still delegate to asyncio.to_thread).

[!NOTE] Default top_k differs between search and asearch search()'s default top_k is 5; asearch()'s default top_k is 10. This is exactly as written in the source — not a typo introduced by this documentation.

Context-manager support

def __enter__(self) -> "VectorDatabaseBase": return self
def __exit__(self, *_): return False
async def __aenter__(self) -> "VectorDatabaseBase": return self
async def __aexit__(self, *_): return False

Both sync and async context-manager protocols are supported at the base level (no setup/teardown logic — __exit__/__aexit__ never suppress exceptions, since they return False).

Dunder methods

def __repr__(self) -> str

Returns f"{type(self).__name__}(vectors={len(self)}, embedder={type(self.embedder).__name__ if self.embedder else None})". Each concrete backend overrides __repr__ with more backend-specific fields — see their pages.

Thread safety

Not documented or enforced anywhere in base.py. The async wrappers only guarantee that calling them won't block the event loop by running the sync method in a thread pool — they do not make the underlying backend safe for concurrent access from multiple threads at once. Whether a given backend is safe for concurrent add()/search() calls cannot be determined from this code and isn't asserted anywhere in the docstrings.

Related APIs

• FAISSVectorDatabase
• ChromaVectorDatabase
• PineconeVectorDatabase