Vector DB Hybrid Search Indexer

Vector DB Hybrid Search Indexer

1. System Architecture & Prerequisites

Runtime requirements

• Python 3.8+. Only stdlib required: math, collections, re, itertools, functools, dataclasses, statistics.
• Optional: numpy — if importable, dense similarity uses vectorized matrix ops; otherwise the pure-Python _cosine fallback is used. No hard dependency.
• No Qdrant/ChromaDB client required at runtime; the APIs mirror their terminology so you can later swap the in-memory stores for a real server without changing the retrieval logic.

Architecture

Embedding model contract

The module ships a deterministic hashing embedder so the whole pipeline runs with zero installs. To use real embeddings, replace embed() with a call such as:

from sentence_transformers import SentenceTransformer
_model = SentenceTransformer("all-MiniLM-L6-v2")
def embed(text):
    return _model.encode(text).tolist()

The rest of the pipeline is agnostic to the vector source.

2. Input/Output Data Contracts

Document (dataclass)

@dataclass
class Document:
    id: str
    text: str
    metadata: Optional[Dict[str, Any]] = None

SearchResult (dataclass)

@dataclass
class SearchResult:
    doc_id: str
    score: float      # final re-ranked score
    sparse_score: float
    dense_score: float
    rrf_score: float
    text: str

Facade signatures

def index_documents(docs: Sequence[Document]) -> "HybridIndex"
def search(self, query: str, top_k: int = 5) -> List[SearchResult]
def rrf(rankings: Sequence[Sequence[str]], k: int = 60) -> Dict[str, float]

Underlying indices

• HybridIndex.bm25 — BM25Index with search(query, n) returning rank-ordered doc-ids.
• HybridIndex.dense — DenseIndex with search(vec, n) returning (doc_id, score) pairs sorted desc.
• HybridIndex.docs — doc-id → Document.

3. Production Reference Implementation

Save as hybrid_index.py. Runnable standalone.

#!/usr/bin/env python3
"""hybrid_index.py - dependency-light hybrid vector search index.

Pipeline:
  index_documents(docs) -> HybridIndex
      bm25.search(query)      sparse Okapi BM25
      dense.search(query)     cosine similarity over hashing embeddings
  rrf(sparse_rank, dense_rank, k=60)  Reciprocal Rank Fusion
  cross_scorer(a, b)          stand-in Cross-Encoder (dot product)
  search(query, top_k)        fused + re-ranked results

Optional import of numpy for fast dense math; pure-math fallback included.
"""
from __future__ import annotations

import math
import re
from collections import Counter
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional, Sequence, Tuple

try:
    import numpy as _np  # type: ignore

    HAS_NUMPY = True
except ImportError:
    HAS_NUMPY = False


# ---------------------------------------------------------------------------
# Tokenization & hashing embeddings
# ---------------------------------------------------------------------------

_TOKEN_RE = re.compile(r"[a-z0-9_]+")


def tokenize(text: str) -> List[str]:
    """Lowercase word tokens (no stemming, deterministic)."""
    return _TOKEN_RE.findall(text.lower())


_VEC_DIM = 256
_STOPS = frozenset({
    "a", "an", "the", "and", "or", "of", "to", "in", "on", "for", "with",
    "is", "are", "was", "were", "as", "at", "by", "this", "that", "it",
})


def _hash_vec(word: str) -> List[int]:
    """Deterministic hashing-bag vector for a single token."""
    h = 14695981039346656037
    for ch in word:
        h ^= ord(ch)
        h = (h * 1099511628211) & 0xFFFFFFFFFFFFFFFF
    pos = h % _VEC_DIM
    sign = 1 if (h >> 8) & 1 else -1
    vec = [0] * _VEC_DIM
    vec[pos] = sign
    return vec


def embed(text: str) -> Tuple[float, ...]:
    """Hashing-based bag-of-words embedding, tf-weighted, L2-normalized.

    Replace this function body with a real model (e.g. sentence-transformers)
    while keeping the pipeline identical.
    """
    vec = [0.0] * _VEC_DIM
    for tok in tokenize(text):
        if tok in _STOPS:
            continue
        hv = _hash_vec(tok)
        for i, v in enumerate(hv):
            vec[i] += v
    norm = math.sqrt(sum(v * v for v in vec))
    if norm == 0.0:
        return tuple(vec)
    return tuple(v / norm for v in vec)


def _cosine(a: Sequence[float], b: Sequence[float]) -> float:
    """Pure-python cosine similarity with zero-division guard."""
    if len(a) != len(b):
        raise ValueError("embedding dimension mismatch")
    n = len(a)
    dot = 0.0
    na = 0.0
    nb = 0.0
    for i in range(n):
        dot += a[i] * b[i]
        na += a[i] * a[i]
        nb += b[i] * b[i]
    if na == 0.0 or nb == 0.0:
        return 0.0
    return dot / (math.sqrt(na) * math.sqrt(nb))


def _dot(a: Sequence[float], b: Sequence[float]) -> float:
    if len(a) != len(b):
        raise ValueError("embedding dimension mismatch")
    return sum(x * y for x, y in zip(a, b))


# ---------------------------------------------------------------------------
# Sparse BM25 (self-contained Okapi BM25)
# ---------------------------------------------------------------------------

@dataclass
class _Posting:
    doc_id: str
    tf: int


class BM25Index:
    """In-memory Okapi BM25 over token postings."""

    def __init__(self, k1: float = 1.5, b: float = 0.75):
        self.k1 = k1
        self.b = b
        self.postings: Dict[str, List[_Posting]] = {}
        self.doc_len: Dict[str, int] = {}
        self.doc_count = 0
        self.avg_len = 0.0
        self._total_len = 0

    def index_texts(self, doc_ids: Sequence[str], texts: Sequence[str]) -> None:
        if len(doc_ids) != len(texts):
            raise ValueError("doc_ids/texts length mismatch")
        self.postings.clear()
        self.doc_len.clear()
        self.doc_count = len(doc_ids)
        for doc_id, text in zip(doc_ids, texts):
            toks = tokenize(text)
            self.doc_len[doc_id] = len(toks)
            self._total_len += len(toks)
            for tok, tf in Counter(toks).items():
                self.postings.setdefault(tok, []).append(_Posting(doc_id, tf))
        self.avg_len = self._total_len / self.doc_count if self.doc_count else 0.0

    def idf(self, tok: str) -> float:
        n = len(self.postings.get(tok, []))
        if n == 0:
            return 0.0
        return math.log(1 + (self.doc_count - n + 0.5) / (n + 0.5))

    def score(self, query: str, doc_id: str) -> float:
        doc_len = self.doc_len.get(doc_id, 0)
        dl_norm = 1 - self.b + self.b * (doc_len / self.avg_len) if self.avg_len else 1.0
        total = 0.0
        for tok in tokenize(query):
            if tok in _STOPS:
                continue
            idf = self.idf(tok)
            tf = 0
            for p in self.postings.get(tok, []):
                if p.doc_id == doc_id:
                    tf = p.tf
                    break
            if tf == 0:
                continue
            total += idf * (tf * (self.k1 + 1)) / (tf + self.k1 * dl_norm)
        return total

    def search(self, query: str, n: int = 5) -> List[Tuple[str, float]]:
        scored = [(d, self.score(query, d)) for d in self.doc_len]
        scored.sort(key=lambda x: x[1], reverse=True)
        return scored[:n]


# ---------------------------------------------------------------------------
# Dense index (optional numpy, else pure math)
# ---------------------------------------------------------------------------

class DenseIndex:
    """In-memory dense vector index with cosine similarity."""

    def __init__(self, dim: int = _VEC_DIM):
        self.dim = dim
        self.vectors: Dict[str, Tuple[float, ...]] = {}

    def add(self, doc_id: str, vec: Sequence[float]) -> None:
        if len(vec) != self.dim:
            raise ValueError(
                f"dim mismatch: got {len(vec)}, expected {self.dim}")
        self.vectors[doc_id] = tuple(vec)

    def _similarities(self, q: Sequence[float]) -> Dict[str, float]:
        if HAS_NUMPY:
            # Vectorized cosine via matrix multiply over a stacked matrix.
            ids = list(self.vectors.keys())
            mat = _np.asarray([self.vectors[i] for i in ids], dtype=_np.float64)
            qv = _np.asarray(q, dtype=_np.float64)
            dots = mat @ qv
            norms = _np.linalg.norm(mat, axis=1)
            qn = _np.linalg.norm(qv)
            if qn == 0.0:
                return {i: 0.0 for i in ids}
            norms[norms == 0.0] = 1.0
            sims = dots / (norms * qn)
            return {i: float(s) for i, s in zip(ids, sims)}
        return {i: _cosine(v, q) for i, v in self.vectors.items()}

    def search(self, query_vec: Sequence[float], n: int = 5) -> List[Tuple[str, float]]:
        sims = self._similarities(query_vec)
        ranked = sorted(sims.items(), key=lambda x: x[1], reverse=True)
        return ranked[:n]


# ---------------------------------------------------------------------------
# Reciprocal Rank Fusion + Cross-Encoder stand-in
# ---------------------------------------------------------------------------

def rrf(rankings: Sequence[Sequence[str]], k: int = 60) -> Dict[str, float]:
    """Reciprocal Rank Fusion.

    Each ranking is an ordered list of doc-ids. Score = sum(1 / (k + rank)).
    """
    scores: Dict[str, float] = {}
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking, start=1):
            scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + rank)
    return scores


def cross_scorer(query_emb: Sequence[float], doc_emb: Sequence[float]) -> float:
    """Cross-Encoder stand-in using the dot product of embedding pairs.

    With real embeddings this mirrors a scoring head; swap in
    sentence-transformers.compute_score(query, doc) for production.
    """
    return _dot(query_emb, doc_emb)


# ---------------------------------------------------------------------------
# Facade: HybridIndex
# ---------------------------------------------------------------------------

@dataclass
class Document:
    id: str
    text: str
    metadata: Optional[Dict[str, Any]] = None


@dataclass
class SearchResult:
    doc_id: str
    score: float
    sparse_score: float
    dense_score: float
    rrf_score: float
    text: str


class HybridIndex:
    def __init__(self,
                 bm25: Optional[BM25Index] = None,
                 dense: Optional[DenseIndex] = None):
        self.bm25 = bm25 or BM25Index()
        self.dense = dense or DenseIndex()
        self.docs: Dict[str, Document] = {}

    def add_documents(self, docs: Sequence[Document]) -> None:
        ids = [d.id for d in docs]
        texts = [d.text for d in docs]
        self.bm25.index_texts(ids, texts)
        for d in docs:
            self.dense.add(d.id, embed(d.text))
            self.docs[d.id] = d

    def search(self, query: str, top_k: int = 5) -> List[SearchResult]:
        k_sparse = max(top_k * 4, 20)
        k_dense = max(top_k * 4, 20)
        sparse_rank = self.bm25.search(query, n=k_sparse)
        q_emb = embed(query)
        dense_rank = self.dense.search(q_emb, n=k_dense)

        sparse_ids = [did for did, _ in sparse_rank]
        dense_ids = [did for did, _ in dense_rank]
        fused = rrf([sparse_ids, dense_ids], k=60)

        merged_ids = list(fused.keys())
        sparse_scores = dict(sparse_rank)
        dense_scores = dict(dense_rank)

        results: List[SearchResult] = []
        for did in merged_ids:
            doc = self.docs[did]
            ce = cross_scorer(q_emb, self.dense.vectors[did])
            # Combine: 0.35 * RRF score + 0.3 * CE + 0.2 * sparse + 0.15 * dense
            score = (0.35 * fused[did]
                     + 0.30 * ce
                     + 0.20 * sparse_scores.get(did, 0.0)
                     + 0.15 * dense_scores.get(did, 0.0))
            results.append(SearchResult(
                doc_id=did,
                score=score,
                sparse_score=sparse_scores.get(did, 0.0),
                dense_score=dense_scores.get(did, 0.0),
                rrf_score=fused[did],
                text=doc.text,
            ))
        results.sort(key=lambda r: r.score, reverse=True)
        return results[:top_k]


def index_documents(docs: Sequence[Document]) -> HybridIndex:
    idx = HybridIndex()
    idx.add_documents(docs)
    return idx


# ---------------------------------------------------------------------------
# Built-in demo corpus
# ---------------------------------------------------------------------------

CORPUS: List[Document] = [
    Document("d1", "Python is a general-purpose programming language "
                    "famous for its readable syntax and rich ecosystem."),
    Document("d2", "OpenAI-compatible APIs let applications call LLMs "
                    "over JSON-REST endpoints with async streaming."),
    Document("d3", "A vector database stores embeddings and supports "
                    "approximate nearest-neighbor search at scale."),
    Document("d4", "BM25 ranks documents by term frequency and inverse "
                    "document frequency for keyword retrieval."),
    Document("d5", "Reciprocal rank fusion combines multiple ranked lists "
                    "linearly into a single blended ranking."),
    Document("d6", "Semantic search maps queries and documents into a "
                    "shared embedding space to measure meaning, not tokens."),
]


def main() -> int:
    idx = index_documents(CORPUS)
    queries = [
        "how do you rank documents by relevance",
        "nearest neighbor embedding search",
        "language model api streaming",
    ]
    for q in queries:
        print("=" * 64)
        print("QUERY:", q)
        for r in idx.search(q, top_k=3):
            print(f"  {r.doc_id:>5} rrf={r.rrf_score:6.3f} "
                  f"sp={r.sparse_score:6.3f} dn={r.dense_score:6.3f} "
                  f"score={r.score:7.4f}  {r.text[:60]}")
    print("=" * 64)
    print(f"[numpy={HAS_NUMPY}] hybrid index demo complete.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

Expected demo output (informational)

================================================================
QUERY: how do you rank documents by relevance
    d6 rrf=0.061 sp=... dn=... score=...  Semantic search maps queries ...
    d4 rrf=0.058 ...  BM25 ranks documents by term frequency ...
QUERY: nearest neighbor embedding search
    d3 ...  A vector database stores embeddings ...
    d6 ...  Semantic search maps queries ...
QUERY: language model api streaming
    d2 ...  OpenAI-compatible APIs let applications call LLMs ...
================================================================
[numpy=False] hybrid index demo complete.

The exact float values vary with the hashing embedder and the optional numpy path, but the ranking structure is stable: keyword-heavy queries surface d4/d6, semantic-language queries surface d3/d6, and API queries surface d2.

4. Execution Protocol & Step-by-Step Workflow

1. Prepare documents — wrap each stored item in a Document(id, text, metadata).
2. Index — call index_documents(docs); this builds the BM25 posting lists and stores normalized dense vectors in parallel.
3. Query — call idx.search("my query", top_k=5).
4. Sparse pass — BM25 scores all docs for the query terms and produces a ranked id list (boost k_sparse to recall more candidates).
5. Dense pass — embed the query, cosine-rank candidates; the matrix paths use numpy when present.
6. Fuse — call rrf([sparse_ids, dense_ids], k=60) to merge the two rankings into a single id-score map.
7. Re-rank — compute the Cross-Encoder stand-in dot product for each fused id, combine the four signals with the weight vector (0.35 RRF, 0.30 CE, 0.20 BM25, 0.15 dense), sort, trim to top_k.
8. Review — inspect rrf_score, sparse_score and dense_score breakdowns to debug ranking regressions.
9. Scale out — swap BM25Index/DenseIndex bodies for Qdrant/ChromaDB clients behind the same interfaces, or replace embed() with sentence-transformers for production embeddings.

5. Edge Cases & Error Handling

• Zero vectors — _cosine and the numpy path return 0.0 for a zero-norm query, preventing division-by-zero.
• Empty corpus — BM25Index.avg_len guards against division by zero; search on an empty index returns [].
• Dimension mismatch — DenseIndex.add raises a descriptive ValueError; embed dimension is fixed at _VEC_DIM so mismatches surface at index time, not query time.
• No numpy — the pure-math fallback keeps full functionality; only throughput differs.
• Overlapping ids — the last Document with a given id wins in the docs map, mirroring upsert semantics; BM25 and dense indices are rebuilt on each index_documents call.
• Fusion on disjoint results — rrf handles id sets of differing sizes (one list can be much shorter), scoring only the union of ids.
• Duplicate rank lists — rrf tolerates identical list inputs; duplicates simply add more 1/(k+rank) mass, making the doc surface earlier.
• Query with only stopwords — BM25 produces an all-zero ranking; the dense pass and CE still rank meaningfully, leaving the hybrid result sensible.
• top_k larger than corpus — tag results[:top_k] clamps output to the available docs without error.