ZerdaZerda

retriever module

On this page

Overview

Implements four classes: a from-scratch BM25 scorer, three individual retrieval strategies (vector, keyword, metadata), and one orchestrator (HybridRetriever) that fuses their results via weighted scoring plus Reciprocal Rank Fusion (RRF). VectorRetriever and MetadataRetriever both natively understand chunks/backends from the rest of zerda (zerda.chunks.DocumentChunk, zerda.vector_database.VectorDatabaseBase), in addition to any plain duck-typed chunk object.

BM25Scorer

Python · 2 lines
class BM25Scorer:
    def __init__(self, k1: float = 1.5, b: float = 0.75): ...

A dependency-free BM25 implementation — no rank_bm25, no elasticsearch, just pure Python + math.log.

Algorithm

Standard BM25 (Best Matching 25):

Text
score(d, q) = Σ idf(t) · (tf(t,d) · (k1 + 1)) / (tf(t,d) + k1 · (1 − b + b · |d| / avgdl))
  • idf(t) = log((N − df(t) + 0.5) / (df(t) + 0.5) + 1) — computed once at index time over the whole corpus
  • tf(t, d) — raw term frequency within document d
  • |d| — document length in tokens; avgdl — average document length across the corpus
  • k1=1.5, b=0.75 — standard defaults, overridable in the constructor

index(documents)

Python
def index(self, documents: List[str]) -> None: ...

Tokenizes every document (re.findall(r"[\w\u0600-\u06FF]{2,}", text.lower()) — words/Arabic sequences of 2+ characters), computes average document length and per-term IDF. Side effect: replaces self._corpus, self._idf, self._avg_dl in place. If documents is empty, returns immediately leaving an empty index (subsequent score() calls return []).

score(query, top_k=20)

Python
def score(self, query: str, top_k: int = 20) -> List[tuple]: ...

Returns [(doc_index, score), ...] sorted descending, truncated to top_k. Returns [] if index() was never called or called with no documents.

Complexity: O(D × Q) where D = corpus size, Q = number of unique query terms — every document is scored against every query term that exists in the global vocabulary.

Thread safety: Not thread-safe for concurrent index() calls (mutates shared state without locks); concurrent score() calls on an already-built index are safe since score() doesn't mutate state.

_cosine(a, b) (module-private helper)

Python
def _cosine(a: np.ndarray, b: np.ndarray) -> float: ...

Standard cosine similarity; returns 0.0 if either vector has zero norm (avoids division by zero). Used by both VectorRetriever and ChunkFilter._dedup_semantic.

VectorRetriever

Python · 10 lines
class VectorRetriever(HybridRetrieverStrategy):
    def __init__(
        self,
        config:               ContextEngineConfig,
        chunks:               Optional[List[Any]]              = None,
        embeddings:           Optional[List[np.ndarray]]       = None,
        embed_fn:             Optional[Any]                    = None,
        vector_store_adapter: Optional[Any]                    = None,
        vector_db:            Optional["VectorDatabaseBase"]   = None,
    ): ...

Operates in one of three mutually-exclusive modes, decided per-call in retrieve(), checked in this order:

  1. Native zerda.vector_database mode — if vector_db is not None, every retrieve() call delegates to vector_db.search(query, top_k=..., score_threshold=..., **backend_kwargs), i.e. the backend's own method (ChromaVectorDatabase, FAISSVectorDatabase, PineconeVectorDatabase, or any other zerda.vector_database.VectorDatabaseBase subclass) — no LangChain-shaped adapter required. This is checked first, ahead of vector_store_adapter.
  2. External vector store mode — if vector_db is None and vector_store_adapter is not None, every retrieve() call delegates to vector_store_adapter.similarity_search_with_score(query, k=top_k, filter=filters). This matches the LangChain VectorStore interface signature.
  3. In-memory mode — otherwise, uses self._chunks / self._embeddings populated via index_chunks(), computing cosine similarity against every stored embedding.

VectorDatabaseBase, DocumentChunk, etc. are only imported under TYPE_CHECKING in retriever.py — so import context (or import zerda.context) never forces zerda.vector_database/zerda.chunks (and their optional heavy deps: chromadb, faiss, pinecone, sentence-transformers) to be importable. They're only actually needed at the moment you pass a real vector_db/native chunk into these methods.

index_vector_db(chunks, vector_db, embeddings=None)

Python · 6 lines
def index_vector_db(
    self,
    chunks:      List["DocumentChunk"],
    vector_db:   "VectorDatabaseBase",
    embeddings:  Optional[np.ndarray] = None,
) -> None: ...

Switches this retriever to native vector_db mode and (if chunks is non-empty) writes them into the backend via vector_db.add(chunks, embeddings=embeddings). Whether embeddings are computed at all — and by what model — is entirely up to vector_db's own configured embedder; embeddings here is only needed if you've pre-computed vectors yourself and want to bypass that.

Side effects: sets self._vector_db = vector_db (unconditionally, even if chunks is empty — useful for attaching to an already-populated backend); logs at INFO level before and after, including the resulting len(vector_db).

Raises: propagates whatever vector_db.add() raises (missing/misconfigured embedder, dimension mismatch, backend connection error) — not swallowed, same posture as index_chunks() below.

index_chunks(chunks, embed_fn)

Python
def index_chunks(self, chunks: List[Any], embed_fn: Any) -> None: ...

Eagerly computes and stores an embedding for every chunk by calling embed_fn(text) once per chunk — not batched. For large corpora with a slow/rate-limited embed_fn, this will be the dominant cost of ContextEngine.index(). (This is the in-memory-mode indexer; for native backends, use index_vector_db() above instead, which lets the backend batch/embed on its own terms.)

Side effects: Replaces self._chunks, self._embed_fn, self._embeddings. Logs at INFO level before and after.

Raises: Propagates any exception raised by embed_fn directly (no try/except around the list comprehension) — unlike query-time embedding (see below), indexing failures are not swallowed.

retrieve(query, top_k, filters=None)

Python · 6 lines
def retrieve(
    self,
    query:   str,
    top_k:   int,
    filters: Optional[Dict[str, Any]] = None,
) -> List[ScoredChunk]: ...
ModeBehavior
Native vector_dbResolves filters to whatever kwarg name the concrete backend's search() expects (see below), then calls vector_db.search(query, top_k=top_k, score_threshold=cfg.min_vector_score or None, **backend_kwargs). On any exception, logs an error and returns [] (does not raise) — same fail-safe posture as the adapter mode.
Vector store adapterCalls the adapter; on any exception, logs an error and returns [] (does not raise). filters is passed through as-is to the adapter, unmodified.
In-memoryReturns [] immediately if chunks/embeddings/embed_fn are missing. Otherwise embeds the query (cached — see below), computes cosine similarity per stored embedding, drops anything below RetrieverConfig.min_vector_score, sorts descending, returns top top_k. filters is accepted as a parameter but never applied in in-memory mode — only the native vector_db and vector-store-adapter branches actually filter.

vector_db filter-kwarg resolution

VectorDatabaseBase subclasses each accept metadata filters under a different keyword name — this is a real inconsistency across the three built-in zerda.vector_database backends, not a bug introduced here:

Backendsearch() filter kwarg
ChromaVectorDatabasefilters
FAISSVectorDatabasedoc_id_filter (doc_id only — FAISS has no general metadata filter)
PineconeVectorDatabasefilter_dict

_resolve_vector_db_filter_kwargs(vector_db, filters) (module-private) bridges this: it looks up the backend's class name in a small static map; for an unrecognized/custom backend, it falls back to inspecting vector_db.search's signature for one of filters/filter_dict/doc_id_filter/where. If filters is falsy, or no matching kwarg can be found at all, it returns {} (i.e. an unfiltered search) — a custom backend with no filter-shaped kwarg logs a DEBUG line and silently ignores filters rather than raising.

[!NOTE] FAISS filtering is doc_id-only Because FAISSVectorDatabase.search() only supports doc_id_filter, passing e.g. filters={"category": "legal"} against a FAISS-backed VectorRetriever will resolve to {} (no doc_id key present) and silently perform an unfiltered search — it does not raise, and does not fall back to client-side filtering. If you need FAISS + arbitrary metadata filters, pass filters={"doc_id": "..."} specifically, or switch to ChromaVectorDatabase/PineconeVectorDatabase.

Query embedding cache

Python
def _get_query_embedding(self, query: str) -> Optional[np.ndarray]: ...

A simple unbounded dict cache (self._query_cache) keyed by the raw query string. Grows forever for the lifetime of the VectorRetriever instance — there is no eviction. Only used by the in-memory mode; native vector_db/adapter modes re-embed the query on every call (embedding is the backend's responsibility). If embed_fn(query) raises, the exception is caught, logged as a warning, and None is returned (so retrieve() returns [] rather than propagating the error) — this is the opposite behavior from index_chunks(), which does not catch embedding errors.

Complexity

In-memory mode: O(C) similarity computations per query, where C = number of indexed chunks (no approximate nearest-neighbor structure — every retrieve() is a full linear scan). Native vector_db/adapter modes: whatever the backend implements (e.g. Chroma's HNSW index, FAISS's configured index type, Pinecone's managed ANN service) — not a linear scan.

Thread safety

Not thread-safe: self._query_cache is a plain dict mutated without locking; concurrent retrieve() calls for new (uncached) queries from multiple threads could race on the same key, though the worst outcome is redundant embedding calls, not data corruption. This only applies to in-memory mode — native vector_db/adapter modes have no client-side cache to race on (thread safety there is whatever the backend itself guarantees).

KeywordRetriever

Python · 2 lines
class KeywordRetriever(HybridRetrieverStrategy):
    def __init__(self, config: ContextEngineConfig, chunks: Optional[List[Any]] = None): ...

Thin wrapper around BM25Scorer. If chunks is provided at construction, the BM25 index is built immediately (_build_index() is called from __init__).

retrieve(query, top_k, filters=None)

Requests top_k * 2 raw BM25 scores, normalizes them to [0, 1] by dividing by the maximum score in the batch, and returns the top top_k as ScoredChunks with keyword_score and composite_score both set to the normalized score. filters is accepted but unused — BM25 has no concept of metadata filtering in this implementation.

Note on the top_k * 2 request: this over-fetch only affects which raw BM25 results are normalized against (the max of the fetched batch) before truncating to top_k — it does not change which top-top_k chunks ultimately get returned, since results are already sorted descending before truncation.

MetadataRetriever

Python · 2 lines
class MetadataRetriever(HybridRetrieverStrategy):
    def __init__(self, config: ContextEngineConfig, chunks: Optional[List[Any]] = None): ...

The simplest of the three: does no scoring of the query text at all. Every chunk whose metadata satisfies all key/value pairs in filters gets metadata_score = composite_score = 1.0; everything else is excluded.

Each chunk's metadata is normalized through models.chunk_metadata_dict() before filtering — this is what lets MetadataRetriever work transparently whether chunk.metadata is a plain dict (e.g. from hand-built chunk objects, or from ChromaVectorDatabase.search(), which already unwraps to a dict) or a ChunkMetadata dataclass (the shape produced by zerda.chunks chunkers directly, and by FAISSVectorDatabase/PineconeVectorDatabase).

_matches_filters(metadata, filters)

For each (key, value) in filters:

  • If metadata.get(key) is None → no match.
  • If value is a list → match if metadata[key] is in that list (supports {"category": ["a", "b"]} style "IN" filters).
  • Otherwise → exact equality required.

Returns [] immediately if chunks is empty or filters is falsy — there is no "match everything" mode.

HybridRetriever

Python · 10 lines
class HybridRetriever:
    RRF_K = 60

    def __init__(
        self,
        config:             ContextEngineConfig,
        vector_retriever:   Optional[VectorRetriever]   = None,
        keyword_retriever:  Optional[KeywordRetriever]  = None,
        metadata_retriever: Optional[MetadataRetriever] = None,
    ): ...

The orchestrator. Note this class does not subclass HybridRetrieverStrategy (see strategy for why this matters).

retrieve(query, top_k=None, filters=None)

Python · 6 lines
def retrieve(
    self,
    query:   str,
    top_k:   Optional[int]            = None,
    filters: Optional[Dict[str, Any]] = None,
) -> List[ScoredChunk]: ...

Algorithm, step by step

  1. k = top_k or self.cfg.top_k.
  2. Vector branch (if use_vector and a vector retriever is configured): fetch k*2 results, merge into all_results keyed by content_hash.
  3. Keyword branch (if use_keyword and configured): fetch k*2 results; if a chunk's hash already exists from step 2, only its keyword_score is merged in (the existing ScoredChunk object is reused, not replaced) — otherwise it's added as new.
  4. Metadata branch (only if use_metadata, a metadata retriever is configured, and filters is truthy): same merge pattern.
  5. If nothing was found across all three branches, logs a warning and returns [].
  6. RRF fusion (_apply_rrf): for each of the three original, unmerged ranked lists, every chunk gets 1 / (RRF_K + rank + 1) added to a running RRF score (so a chunk ranked #1 by two different methods accumulates two bonuses). The accumulated RRF scores are normalized by the maximum observed RRF score, then added to each chunk's composite_score scaled by 0.20 (a flat 20%-of-the-max bonus weight, hardcoded, not configurable).
  7. Weighted composite score: every merged chunk's composite_score is recomputed from scratch (overwriting whatever the individual retriever set) as vector_score*vector_weight + keyword_score*keyword_weight + metadata_score*metadata_weight, then the RRF bonus from step 6 is layered on top (step 6 runs before step 7 in the source, so the RRF addition from step 6 is actually overwritten by the reassignment in step 7 — see the warning below).
  8. Sort descending by composite_score, return the top k.

[!WARNING] RRF bonus ordering — read the source carefully Looking at the exact source order: _apply_rrf(chunks_list, ...) runs before the loop that sets sc.composite_score = vector_score*w + keyword_score*w + metadata_score*w. Since that loop does a plain = assignment (not +=), it overwrites the RRF bonus that _apply_rrf just added. In the current implementation, the RRF bonus added in step 6 has no effect on the final composite_score used for sorting and truncation — it's computed and discarded. If you rely on RRF fusion boosting multi-method hits, verify this against your actual installed version, since this looks like an ordering bug rather than intended behavior.

Parameters

NameTypeRequiredNotes
querystrYesPassed to all three sub-retrievers (after any history-enrichment done upstream by ContextEngine)
top_kOptional[int]NoFalls back to RetrieverConfig.top_k (default 20)
filtersOptional[Dict[str, Any]]NoPassed to vector and metadata retrievers; ignored by keyword retrieval (BM25 operates on text only)

Returns

List[ScoredChunk], sorted descending by composite_score, length ≤ top_k. Can be [].

Complexity

O(k log k) for the final sort, plus the cost of each underlying retriever (dominated by VectorRetriever's O(C) linear scan in in-memory mode).

Thread safety

Not explicitly thread-safe; inherits the thread-safety characteristics of whichever sub-retrievers are configured.

Examples

1. Basic — keyword-only retrieval:

Python · 5 lines
from zerda.context import ContextEngineConfig, KeywordRetriever

cfg = ContextEngineConfig()
kw = KeywordRetriever(cfg, chunks=my_chunks)
results = kw.retrieve("machine learning algorithms", top_k=5)

2. Intermediate — vector + keyword hybrid:

Python · 9 lines
from zerda.context import ContextEngineConfig, VectorRetriever, KeywordRetriever, HybridRetriever

cfg = ContextEngineConfig()
vec = VectorRetriever(cfg)
vec.index_chunks(my_chunks, embed_fn=my_embed_fn)
kw  = KeywordRetriever(cfg, chunks=my_chunks)

hybrid = HybridRetriever(cfg, vector_retriever=vec, keyword_retriever=kw)
results = hybrid.retrieve("ما هو الانحدار الخطي؟", top_k=10)

3. Advanced — external vector store adapter (LangChain-compatible):

Python · 2 lines
vec = VectorRetriever(cfg, vector_store_adapter=my_chroma_vectorstore)
results = vec.retrieve("query text", top_k=5, filters={"category": "docs"})

4. Error handling — vector store adapter failure:

Python · 6 lines
# If my_chroma_vectorstore.similarity_search_with_score raises (e.g. connection error),
# VectorRetriever.retrieve() catches it, logs an error, and returns [] rather than propagating:
results = vec.retrieve("query text", top_k=5)  # [] on backend failure, no exception

# Same fail-safe behavior applies to native vector_db mode: if vector_db.search() raises
# (e.g. Chroma collection missing, Pinecone connection error), retrieve() also returns [].

5. Production usage — metadata-filtered retrieval combined with vector search:

Python · 7 lines
cfg.retriever.use_metadata = True
meta = MetadataRetriever(cfg, chunks=my_chunks)
hybrid = HybridRetriever(cfg, vector_retriever=vec, keyword_retriever=kw, metadata_retriever=meta)
results = hybrid.retrieve(
    "renewal policy",
    filters={"category": ["legal", "compliance"], "year": 2026},
)

Best practices

  • Pre-build KeywordRetriever's BM25 index once (pass chunks to the constructor or call index_chunks()), not per-query — _build_index() re-tokenizes the entire corpus.
  • If using in-memory VectorRetriever with a large corpus, expect linear-scan latency; for production scale, supply a vector_db (native zerda.vector_database backend) or vector_store_adapter instead — either lets the backend's own index (HNSW, IVF, managed ANN, ...) do the heavy lifting.
  • Prefer vector_db over vector_store_adapter whenever your store is one of zerda.vector_database's own backends — it avoids the LangChain-shaped duck-typing entirely and gets correct filter-kwarg resolution for free (see above).
  • If you rely on metadata filters against a FAISSVectorDatabase-backed VectorRetriever, remember only doc_id filtering is actually supported — anything else silently becomes an unfiltered search.

Common mistakes

  • Expecting filters to limit BM25/keyword results — it's silently ignored there.
  • Expecting RRF fusion to actually change final ranking — per the warning above, verify this in your version; as read, it does not.
  • Setting use_metadata=True without passing filters to retrieve() — the metadata branch is skipped entirely (use_metadata and self._meta and filters all must be truthy).
  • Assuming filters passed to a vector_db-backed VectorRetriever behaves identically across backends — it's translated to a different kwarg per backend, and FAISS only supports doc_id filtering (see "vector_db filter-kwarg resolution" above).
  • Treating a chunk's .metadata as always a dict when writing custom code against zerda.chunks chunks directly — it's a ChunkMetadata dataclass there. Use chunk_metadata_dict() (as MetadataRetriever/ScoredChunk do internally) rather than calling .get() on it directly.

← Back to API overview

Related pages