windlass.providers.retrievers.bm25

bm25

BM25 lexical retrieval.

Okapi BM25, implemented from scratch over an inverted index — no rank_bm25, no dependencies. It earns its place next to vector search because the two fail in opposite ways: embeddings miss exact identifiers (error codes, product SKUs, function names) that BM25 nails, and BM25 misses paraphrases that embeddings handle. Hybrid retrieval combines both.

The scoring function is the standard one::

score(q, d) = Σ  IDF(t) · f(t,d)·(k1+1) / (f(t,d) + k1·(1 - b + b·|d|/avgdl))
             t∈q

Example

from windlass.core.types import Chunk r = BM25Retriever() r.index([ ... Chunk(content="error E1042 occurs when the socket closes"), ... Chunk(content="the network layer handles reconnection"), ... ]) 2 r.retrieve("E1042").hits[0].chunk.content.startswith("error E1042") True

BM25Retriever

BM25Retriever(
    *,
    top_k: int = 5,
    k1: float = 1.5,
    b: float = 0.75,
    remove_stopwords: bool = True,
    min_token_length: int = 1,
    vectorstore: VectorStore | None = None,
    **config: Any
)

Bases: Retriever

Sparse lexical retriever.

Parameters:

Name	Type	Description	Default
top_k	int	Default number of results.	5
k1	float	Term-frequency saturation. Higher values let repeated terms keep adding score; 1.2-1.5 is the usual range.	1.5
b	float	Length normalisation, in [0, 1]. 0 ignores document length, 1 fully normalises; 0.75 is the standard compromise.	0.75
remove_stopwords	bool	Drop :data:STOPWORDS from documents and queries.	True
min_token_length	int	Ignore tokens shorter than this.	1
vectorstore	VectorStore | None	Optional store to seed the index from at construction, so a hybrid retriever built on an existing collection works immediately.	None
**config	Any	Forwarded to :class:~windlass.interfaces.retriever.Retriever.	{}

Attributes:

Name	Type	Description
chunks	dict[str, Chunk]	Indexed chunks, keyed by id.

Performance

Indexing is O(total tokens). Query cost is proportional to the number of documents containing the query terms, not the corpus size, which makes it fast even on large collections. Memory is roughly proportional to the vocabulary.

Thread safety

Indexing and search are guarded by a re-entrant lock.

Source code in src\windlass\providers\retrievers\bm25.py

def __init__(
    self,
    *,
    top_k: int = 5,
    k1: float = 1.5,
    b: float = 0.75,
    remove_stopwords: bool = True,
    min_token_length: int = 1,
    vectorstore: VectorStore | None = None,
    **config: Any,
) -> None:
    super().__init__(top_k=top_k, **config)
    self.k1 = k1
    self.b = b
    self.remove_stopwords = remove_stopwords
    self.min_token_length = min_token_length

    self.chunks: dict[str, Chunk] = {}
    self._postings: dict[str, dict[str, int]] = {}
    self._lengths: dict[str, int] = {}
    self._total_length = 0
    self._lock = threading.RLock()

    if vectorstore is not None:
        seed = getattr(vectorstore, "all_chunks", None)
        if callable(seed):
            self.index(seed())

aindex async

aindex(chunks: Sequence[Chunk]) -> int

Add chunks to the inverted index.

Re-indexing a chunk id replaces its previous entry, so ingestion is idempotent.

Parameters:

Name	Type	Description	Default
chunks	Sequence[Chunk]	Chunks to index.	required

Returns:

Type	Description
int	How many chunks were indexed.

Source code in src\windlass\providers\retrievers\bm25.py

async def aindex(self, chunks: Sequence[Chunk]) -> int:
    """Add chunks to the inverted index.

    Re-indexing a chunk id replaces its previous entry, so ingestion is
    idempotent.

    Args:
        chunks: Chunks to index.

    Returns:
        How many chunks were indexed.
    """
    if not chunks:
        return 0
    with self._lock:
        for chunk in chunks:
            if chunk.id in self.chunks:
                self._remove(chunk.id)
            tokens = self._tokenize(chunk.content)
            if not tokens:
                self.chunks[chunk.id] = chunk
                self._lengths[chunk.id] = 0
                continue
            frequencies: dict[str, int] = {}
            for token in tokens:
                frequencies[token] = frequencies.get(token, 0) + 1
            for token, count in frequencies.items():
                self._postings.setdefault(token, {})[chunk.id] = count
            self.chunks[chunk.id] = chunk
            self._lengths[chunk.id] = len(tokens)
            self._total_length += len(tokens)
    return len(chunks)

remove

remove(chunk_ids: Sequence[str]) -> int

Remove chunks from the index.

Parameters:

Name	Type	Description	Default
chunk_ids	Sequence[str]	Ids to drop.	required

Returns:

Type	Description
int	How many were actually removed.

Source code in src\windlass\providers\retrievers\bm25.py

def remove(self, chunk_ids: Sequence[str]) -> int:
    """Remove chunks from the index.

    Args:
        chunk_ids: Ids to drop.

    Returns:
        How many were actually removed.
    """
    with self._lock:
        return sum(1 for cid in chunk_ids if self._remove(cid))

clear

clear() -> None

Empty the index.

Source code in src\windlass\providers\retrievers\bm25.py

def clear(self) -> None:
    """Empty the index."""
    with self._lock:
        self.chunks.clear()
        self._postings.clear()
        self._lengths.clear()
        self._total_length = 0

aretrieve_chunks async

aretrieve_chunks(
    query: str, k: int, *, filters: MetadataFilter | None = None, **kwargs: Any
) -> list[ScoredChunk]

Score the corpus against query using BM25.

Parameters:

Name	Type	Description	Default
query	str	The search query.	required
k	int	How many results to return.	required
filters	MetadataFilter | None	Metadata constraints, applied to candidates before ranking.	None
**kwargs	Any	Ignored.	{}

Returns:

Type	Description
list[ScoredChunk]	Scored chunks, best first. Chunks matching no query term are omitted
list[ScoredChunk]	entirely rather than returned with a zero score.

Source code in src\windlass\providers\retrievers\bm25.py

async def aretrieve_chunks(
    self,
    query: str,
    k: int,
    *,
    filters: MetadataFilter | None = None,
    **kwargs: Any,
) -> list[ScoredChunk]:
    """Score the corpus against ``query`` using BM25.

    Args:
        query: The search query.
        k: How many results to return.
        filters: Metadata constraints, applied to candidates before ranking.
        **kwargs: Ignored.

    Returns:
        Scored chunks, best first. Chunks matching no query term are omitted
        entirely rather than returned with a zero score.
    """
    terms = self._tokenize(query)
    if not terms:
        return []

    with self._lock:
        corpus_size = len(self.chunks)
        if corpus_size == 0:
            return []
        average_length = (self._total_length / corpus_size) or 1.0

        scores: dict[str, float] = {}
        for term in set(terms):
            postings = self._postings.get(term)
            if not postings:
                continue
            idf = self._idf(len(postings), corpus_size)
            for chunk_id, frequency in postings.items():
                length = self._lengths.get(chunk_id, 0)
                denominator = frequency + self.k1 * (
                    1 - self.b + self.b * length / average_length
                )
                scores[chunk_id] = scores.get(chunk_id, 0.0) + idf * (
                    frequency * (self.k1 + 1) / denominator
                )

        hits = [
            ScoredChunk(chunk=self.chunks[cid], score=score, retriever=self.name)
            for cid, score in scores.items()
            if cid in self.chunks
            and VectorStore.match_filters(self.chunks[cid].metadata, filters)
        ]

    hits.sort(key=lambda h: h.score, reverse=True)
    return hits[:k]

stats

stats() -> dict[str, Any]

Return index statistics.

Returns:

Type	Description
dict[str, Any]	A dict with documents, vocabulary, total_tokens and
dict[str, Any]	average_length.

Source code in src\windlass\providers\retrievers\bm25.py

def stats(self) -> dict[str, Any]:
    """Return index statistics.

    Returns:
        A dict with ``documents``, ``vocabulary``, ``total_tokens`` and
        ``average_length``.
    """
    with self._lock:
        count = len(self.chunks)
        return {
            "documents": count,
            "vocabulary": len(self._postings),
            "total_tokens": self._total_length,
            "average_length": (self._total_length / count) if count else 0.0,
        }