ZerdaZerda

FAISSVectorDatabase

On this page

Table of contents

Overview

A wrapper around FAISS that stores vectors in a local, in-process index plus a parallel Python list of DocumentChunk objects (so it can map FAISS result indices back to chunk text/metadata). All state — index, chunks, and lookup tables — lives in memory unless you call save.

Use it when you want exact or approximate nearest-neighbor search with no external services and full control over the index type.

Class attributes

Python · 5 lines
SUPPORTED_INDEX_TYPES = {'flat', 'ivf', 'hnsw'}
SUPPORTED_METRICS = {'cosine', 'l2', 'ip'}
DEFAULT_IVF_CLUSTERS = 100
DEFAULT_IVF_TRAIN_SIZE = 100
DEFAULT_HNSW_M = 32

Constructor

Python · 8 lines
FAISSVectorDatabase(
    embedder: Optional[Any] = None,
    embedding_dim: Optional[int] = None,
    index_type: str = 'flat',
    distance_metric: str = 'cosine',
    ivf_clusters: Optional[int] = None,
    hnsw_m: Optional[int] = None,
)
ParameterTypeDefaultNotes
embedderOptional[Any]NoneMust implement .encode(texts) -> np.ndarray.
embedding_dimOptional[int]NoneRequired if embedder is None. Must be positive.
index_typestr'flat'One of 'flat' (exact), 'ivf' (inverted file, faster at scale), 'hnsw' (approximate graph search).
distance_metricstr'cosine'One of 'cosine', 'l2', 'ip'.
ivf_clustersOptional[int]None → 100Only relevant for index_type='ivf'.
hnsw_mOptional[int]None → 32Only relevant for index_type='hnsw'; connections per layer.

Raises: ImportError if faiss isn't installed; ValueError for an unsupported metric, unsupported index type, non-positive embedding_dim, or if neither embedder nor embedding_dim is given (or the embedder's .encode() call fails while probing dimension).

Initialization workflow

  1. super().__init__(embedder)
  2. Validate distance_metric and index_type against the supported sets.
  3. Resolve embedding_dim — from the explicit argument, or by calling embedder.encode(["test"]) and reading .shape[1] (or len(...) for 1-D output).
  4. Store ivf_clusters/hnsw_m (falling back to defaults if not given).
  5. Initialize empty storage: self.chunks = [], self._chunk_id_to_idx = {}, self._doc_id_to_indices = {}, self._total_vectors = 0.
  6. Build the FAISS index via _create_index():
    • metric cosine or ip → base index faiss.IndexFlatIP
    • metric l2 → base index faiss.IndexFlatL2
    • index_type='flat' → the base index directly
    • index_type='ivf' → faiss.IndexIVFFlat(quantizer, dim, ivf_clusters) — must be trained before it accepts additions (see add)
    • index_type='hnsw' → faiss.IndexHNSWFlat(dim, hnsw_m) (source note: FAISS's HNSW is natively L2-based; cosine is approximated by normalizing vectors before insertion, not by a different index type)

Thread safety

Not documented or synchronized in source. Concurrent calls to add/search from multiple threads are not guarded against races on self.chunks, the lookup dicts, or the FAISS index itself.

Initialization & validation (internal)

These are private (_-prefixed) but explain observable errors:

  • _validate_metric / _validate_index_type — raise ValueError listing supported values.
  • _determine_embedding_dim — see constructor above.
  • _create_index — see above.
  • _normalize_if_needed(embeddings) — L2-normalizes (faiss.normalize_L2) only when distance_metric == 'cosine'; otherwise returns the array unchanged (cast to float32).
  • _compute_embeddings(texts) — requires self.embedder; raises ValueError if missing, wraps embedder failures in RuntimeError.
  • _validate_embedding_dimension(embeddings) — raises ValueError if not 2-D or if the second dimension doesn't match self.embedding_dim.

Add operations

add_chunk

Python
def add_chunk(self, chunk: DocumentChunk) -> None

Adds a single chunk. If chunk.embedding is None, computes it via _compute_embeddings, then delegates to add(chunks=[chunk], embeddings=...).

Raises: ValueError if chunk isn't a DocumentChunk.

add

Python
def add(self, chunks: List[DocumentChunk], embeddings: Optional[np.ndarray] = None) -> None
ParameterTypeRequiredNotes
chunksList[DocumentChunk]yesEmpty list → logs a warning and returns without error.
embeddingsOptional[np.ndarray]noComputed via _compute_embeddings if omitted.

Internal workflow:

  1. If chunks is empty, warn and return.
  2. Validate every item is a DocumentChunk (else ValueError).
  3. Compute embeddings if not provided.
  4. Validate len(chunks) == len(embeddings) and dimension consistency.
  5. Normalize embeddings if using cosine.
  6. If using an untrained IVF index and total vectors (existing + new) reaches DEFAULT_IVF_TRAIN_SIZE, train it first (_train_ivf_index).
  7. Append each chunk (with sanitize_metadata() applied to its metadata) to self.chunks, updating _chunk_id_to_idx and _doc_id_to_indices.
  8. self.index.add(normalized_embeddings), increment _total_vectors.

Raises: ValueError for invalid/mismatched inputs; RuntimeError wrapping any lower-level failure during the try block.

[!WARNING] IVF index below the training threshold An ivf index only trains once cumulative added vectors reach DEFAULT_IVF_TRAIN_SIZE = 100. Adding fewer than 100 vectors total means the index is never trained, and faiss.IndexIVFFlat.add() will raise if you attempt to add to an untrained index — this is FAISS's own behavior, not something this wrapper suppresses.

Complexity

  • flat: add is O(n) to append n vectors; search is O(N·d) per query (N = stored vectors, d = dimension) — exact but linear.
  • ivf: faster search after training (approximately O(clusters searched × points per cluster)), at the cost of a one-time training pass.
  • hnsw: logarithmic-ish approximate search per FAISS's HNSW implementation; insertion cost is higher than flat per vector.

(These complexity characterizations reflect FAISS's own documented algorithm behavior, not custom logic in this wrapper — the wrapper itself does no extra algorithmic work in the hot path beyond normalization and bookkeeping.)

Search operations

Python · 7 lines
def search(
    self,
    query: Union[str, np.ndarray],
    top_k: int = 5,
    score_threshold: Optional[float] = None,
    doc_id_filter: Optional[Union[str, List[str]]] = None,
) -> List[Tuple[DocumentChunk, float]]
ParameterTypeDefaultNotes
querystr | np.ndarray—Text is encoded via _compute_embeddings; array is cast to float32.
top_kint5Must be > 0.
score_thresholdOptional[float]NoneResults with score < score_threshold are dropped.
doc_id_filterOptional[str | List[str]]NoneKeep only chunks whose doc_id matches (str) or is in the list.

Returns: List[Tuple[DocumentChunk, float]], sorted by FAISS result order (which is score order for the underlying index type).

Internal workflow:

  1. If the database is empty, warn and return [] immediately.
  2. Validate top_k > 0.
  3. Encode/cast the query, validate its dimension matches embedding_dim.
  4. Normalize the query vector the same way stored vectors were normalized.
  5. actual_k = min(top_k, total_vectors); if doc_id_filter is set, search for actual_k * 2 candidates first (to leave room for post-filtering), else actual_k.
  6. Call self.index.search(query_vec, search_k).
  7. For each returned (distance, idx): skip idx == -1 (FAISS's "no result" sentinel) and any out-of-range index (logged as a warning — defensive check, shouldn't normally trigger); apply score_threshold; apply doc_id_filter; stop once top_k results are collected.

Raises: ValueError for empty/invalid top_k or dimension mismatch; RuntimeError wrapping lower-level failures.

Side effects: none beyond logging.

[!NOTE] doc_id_filter isn't guaranteed to return top_k results Because filtering happens after FAISS returns search_k = actual_k * 2 candidates, a heavily-filtered query can return fewer than top_k results even if more matches exist further down FAISS's ranking. This is a direct consequence of the fixed * 2 multiplier in the source and is not adaptive.

search_by_doc_id

Python
def search_by_doc_id(self, doc_id: str, top_k: int = 5, exclude_same_doc: bool = True) -> List[Tuple[DocumentChunk, float]]

Finds chunks similar to every chunk belonging to doc_id, merges and deduplicates the results by chunk_id, sorts by score descending, and truncates to top_k.

Returns [] with a warning if doc_id isn't found. Uses each source chunk's own chunk.embedding as the query for search() (requesting extra results when exclude_same_doc=True to compensate for filtered-out same-document hits).

Delete operations

remove_by_chunk_id

Python
def remove_by_chunk_id(self, chunk_id: str) -> bool

Removes a single chunk. Returns False (with a warning) if not found; otherwise removes it from lookup tables and calls _rebuild_index() (a full index rebuild — see below), returning True.

remove_by_doc_id

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

Removes every chunk for doc_id. Returns 0 if the document isn't found. Otherwise removes the matching entries from self.chunks (highest index first, to keep indices valid during the loop), deletes the doc from _doc_id_to_indices, calls _rebuild_index(), and returns the count removed.

clear

Python
def clear(self) -> int

Empties chunks, both lookup dicts, resets _total_vectors to 0, and recreates a fresh empty index via _create_index(). Returns the number of chunks that were present beforehand.

_rebuild_index (internal, but load-bearing)

Because FAISS indices don't support in-place deletion, every delete operation triggers a full rebuild: recreate the index, re-add every remaining chunk's stored embedding (re-training an IVF index if there are enough vectors), and rebuild both lookup dicts from scratch.

[!WARNING] Deletes are O(n) in the current vector count Every remove_by_chunk_id/remove_by_doc_id call rebuilds the entire index from the remaining chunks. For workloads with frequent deletes against a large index, this can be a meaningful performance cost — there is no incremental/soft-delete path in this backend.

Persistence

save

Python
def save(self, path: Union[str, Path]) -> None

Writes three files under path (created if missing):

  • index.faiss — via faiss.write_index
  • chunks.pkl — a pickled list of dicts (chunk_id, doc_id, text, metadata, and embedding as a plain list or None)
  • metadata.pkl — a pickled dict of embedding_dim, index_type, distance_metric, total_vectors, ivf_clusters, hnsw_m

Raises: RuntimeError wrapping any failure.

load

Python · 2 lines
@classmethod
def load(cls, path: Union[str, Path], embedder: Optional[Any] = None) -> 'FAISSVectorDatabase'

Raises: FileNotFoundError if path doesn't exist or any of the three required files is missing; RuntimeError wrapping any other failure.

Workflow: loads metadata.pkl first, constructs a new instance with that configuration (optionally with a fresh embedder), then overwrites db.index with faiss.read_index(...) and db._total_vectors from the saved metadata, then reconstructs each DocumentChunk from chunks.pkl and rebuilds both lookup tables.

[!NOTE] Falsy embeddings are silently skipped on load The load loop checks if chunk_dict.get('embedding'): before restoring chunk.embedding. An embedding stored as an all-zero vector (or any array whose truthiness check on the underlying list is falsy) would not be restored via this exact condition — this is what the source does, not an inferred edge case.

Statistics & utilities

get_stats

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

Returns:

Python · 11 lines
{
    'total_chunks': int,
    'unique_docs': int,
    'embedding_dim': int,
    'index_type': str,
    'distance_metric': str,
    'ivf_clusters': int | None,   # None unless index_type == 'ivf'
    'hnsw_m': int | None,         # None unless index_type == 'hnsw'
    'is_trained': bool,           # index.is_trained for ivf, else True
    'has_embedder': bool,
}

get_chunk_by_id / get_chunks_by_doc_id / list_document_ids

Straightforward local-cache lookups via the two internal dicts; no I/O.

Dunder methods

Python · 3 lines
def __len__(self) -> int          # self._total_vectors
def __repr__(self) -> str         # vectors, docs, dim, type, metric
def __contains__(self, chunk_id) -> bool   # chunk_id in self._chunk_id_to_idx

Async API

Python · 5 lines
async def asearch(self, query: str, top_k: int = 10, score_threshold: float = 0.0, **kwargs)
async def aadd(self, chunks, **kwargs)
async def aremove_by_doc_id(self, doc_id: str) -> int
async def asave(self, path) -> None
async def __aenter__(self) / async def __aexit__(self, ...)

Every method is asyncio.to_thread(self.<sync_method>, ...) — a thin thread-pool wrapper, not a native async implementation. Note asearch's score_threshold default here is 0.0, not None as in the base class or the synchronous search() — calling asearch() without arguments therefore filters out any result scoring below 0.0, which is not the same as "no filtering."

Common mistakes

  • Passing index_type='ivf' and adding fewer than 100 total vectors, then being surprised when the index either never trains or raises on add/search — see the note under Add operations.
  • Assuming remove_by_doc_id/remove_by_chunk_id are cheap — they trigger a full rebuild.
  • Relying on asearch()'s default threshold matching search()'s default — they differ (0.0 vs None).

Related pages