Pure dense retrieval fails on exact product codes, legal citations, and rare terminology. This expert lesson teaches you how BM25 and SPLADE learned sparse encoders work, how to implement both alongside dense vectors, and how to fuse them into a production hybrid RAG pipeline that handles all query types reliably.
You've built a RAG pipeline, your dense embeddings are indexed in a vector store, and retrieval looks pretty good on your benchmark queries. Then a customer reports that searching for "ISO 27001 Section 6.1.2 risk ownership requirements" returns completely wrong documents — documents that are semantically adjacent to risk management but miss the specific clause entirely. Or a legal team complains that a product name like "Acquis-Pro v2.3" is never matched correctly because your embedding model has never seen that term. These aren't edge cases. They're systematic failures of semantic search against exact, structured, or rare-token queries.
The root cause is well understood: dense retrieval excels at capturing meaning, but it actively smooths over exact token matches. When a user knows precisely what they're looking for — a specific model number, a legal citation, a drug name, an internal product code — dense vectors can betray them. Classical BM25 keyword retrieval handles these cases naturally, but falls apart the moment a query uses different vocabulary than the document. The right answer has always been both, but doing hybrid retrieval properly in production is considerably more nuanced than just averaging two scores.
This lesson goes deep on the mechanics of sparse retrieval — both classical BM25 and the modern learned sparse encoder approach called SPLADE — and shows you how to implement a production-grade hybrid RAG pipeline that combines them with dense vectors. You'll understand why each component behaves the way it does, not just how to wire them together.
What you'll learn:
You should be comfortable with the fundamentals of retrieval-augmented generation — if you're newer to this, Building Your First RAG Pipeline is a solid starting point. You should understand how dense embeddings work conceptually (Embeddings Explained: How Text Becomes Vectors for Semantic Search covers this well) and have at least passing familiarity with the mechanics of hybrid search. You should be comfortable writing production Python, and you should have opinions about things like connection pooling and schema design. Some familiarity with transformer architectures will help when we get into SPLADE internals.
BM25 — Best Match 25 — is a probabilistic ranking function from the 1990s that remains the baseline for keyword retrieval in virtually every search system. If you've used Elasticsearch, Solr, Lucene, or any modern full-text search engine, you've used BM25. But most practitioners treat it as a black box. Understanding its internals is essential before you can reason about when it will fail and how SPLADE improves on it.
The core BM25 score for a query Q against a document D is:
score(D, Q) = Σ IDF(qᵢ) · (f(qᵢ, D) · (k1 + 1)) / (f(qᵢ, D) + k1 · (1 - b + b · |D| / avgdl))
Where:
f(qᵢ, D) is the raw term frequency of query term qᵢ in document D|D| is the document length in tokensavgdl is the average document length across the corpusk1 controls term frequency saturation (typically 1.2–2.0)b controls document length normalization (typically 0.75)IDF(qᵢ) is the inverse document frequency: log((N - df + 0.5) / (df + 0.5) + 1) where N is the corpus size and df is the number of documents containing the termThree mechanisms are doing real work here. Term frequency saturation: the k1 parameter means that the benefit of seeing a term 10 times versus 5 times is much smaller than seeing it once versus zero times. Unlike raw TF-IDF, BM25 doesn't linearly reward document stuffing. Length normalization: the b parameter penalizes long documents, because a term appearing 3 times in a 50-word chunk is more meaningful than 3 times in a 5000-word document. IDF weighting: common words like "the" or "report" are downweighted almost to zero; rare, discriminative terms like "ISO 27001" or "transthoracic echocardiogram" get high weight.
This is why BM25 is so reliable for exact-match retrieval. When a user types a specific product code, BM25 will correctly rank documents containing that exact string very highly, because the IDF of that rare term is enormous.
Let's implement a basic BM25 retrieval layer using rank_bm25, a pure Python implementation:
from rank_bm25 import BM25Okapi
from typing import List, Dict, Any
import numpy as np
import re
import string
class BM25Retriever:
"""
Production BM25 retriever with configurable tokenization and corpus management.
Note: rank_bm25 holds the corpus in memory. For corpora over ~500k documents,
you'll want to delegate to Elasticsearch or OpenSearch instead.
"""
def __init__(self, k1: float = 1.6, b: float = 0.75):
self.k1 = k1
self.b = b
self.bm25 = None
self.documents: List[Dict[str, Any]] = []
self._tokenized_corpus: List[List[str]] = []
def _tokenize(self, text: str) -> List[str]:
"""
Conservative tokenization that preserves compound terms and codes.
In production you'd want stemming, stopword removal, and potentially
language detection. This version is intentionally simple but correct.
"""
# Lowercase and strip punctuation, but preserve hyphens in compound terms
# e.g., "ISO-27001" stays as one token candidate before splitting
text = text.lower()
# Split on whitespace and most punctuation, but keep alphanumeric+hyphen
tokens = re.findall(r"[a-z0-9]+(?:-[a-z0-9]+)*", text)
# Remove pure stop words but keep short codes (2-3 chars that are likely acronyms)
stopwords = {
"the", "a", "an", "and", "or", "but", "in", "on", "at",
"to", "for", "of", "with", "by", "from", "is", "was", "are"
}
return [t for t in tokens if t not in stopwords or len(t) <= 3]
def index(self, documents: List[Dict[str, Any]], text_field: str = "text") -> None:
"""
Build BM25 index from a list of document dicts.
Args:
documents: List of dicts with at least a text field and ideally an 'id' field
text_field: Which field to index
"""
self.documents = documents
self._tokenized_corpus = [
self._tokenize(doc[text_field]) for doc in documents
]
self.bm25 = BM25Okapi(
self._tokenized_corpus,
k1=self.k1,
b=self.b
)
print(f"Indexed {len(documents)} documents. "
f"Avg doc length: {np.mean([len(t) for t in self._tokenized_corpus]):.1f} tokens")
def retrieve(
self,
query: str,
top_k: int = 10,
score_threshold: float = 0.0
) -> List[Dict[str, Any]]:
"""
Retrieve top-k documents for a query.
Returns list of dicts with 'document', 'score', and 'rank' keys.
"""
if self.bm25 is None:
raise RuntimeError("Index not built. Call .index() first.")
query_tokens = self._tokenize(query)
if not query_tokens:
return []
scores = self.bm25.get_scores(query_tokens)
# Get indices sorted by score descending
top_indices = np.argsort(scores)[::-1][:top_k]
results = []
for rank, idx in enumerate(top_indices):
if scores[idx] <= score_threshold:
continue
results.append({
"document": self.documents[idx],
"score": float(scores[idx]),
"rank": rank + 1,
"retriever": "bm25"
})
return results
def add_documents(self, new_documents: List[Dict[str, Any]], text_field: str = "text") -> None:
"""
Incrementally add documents and rebuild the index.
Warning: rank_bm25 doesn't support incremental updates — this is a full rebuild.
For high-update-rate corpora, use Elasticsearch BM25 via its API instead.
"""
all_docs = self.documents + new_documents
self.index(all_docs, text_field=text_field)
Warning: The in-memory
rank_bm25approach works well for corpora up to roughly 500,000 chunks, depending on average document length. Beyond that, you pay real memory costs and rebuild time. For production at scale, delegate BM25 to Elasticsearch or OpenSearch — their Lucene-based BM25 implementation handles shard distribution, incremental indexing, and filtering efficiently. We'll show the Elasticsearch integration later in this lesson.
One thing practitioners often miss: BM25 performance is highly sensitive to tokenization choices. If your corpus contains medical codes like "ICD-10-CM F32.1", SQL identifiers like user_account_balance, or version strings like "v2.3.1", your tokenizer needs to handle them intelligently. The regex-based tokenizer above preserves hyphenated compound terms — a decision worth making deliberately.
Understanding where BM25 fails is what motivates SPLADE. BM25's fundamental assumption is that query terms and document terms share the same vocabulary. This breaks in at least three real scenarios:
Synonymy: A user searches "cardiac arrest" but the document says "heart failure" — different terms, same medical concept. BM25 scores zero for term overlap.
Related concepts: A user searches "how to reduce model overfitting" and the perfect document discusses "regularization techniques and dropout strategies" — conceptually identical, zero BM25 overlap.
Morphological variation: A user searches "running" but the document discusses "run performance" or "runner training." Without stemming, these are different tokens.
Dense embeddings solve all three problems at once — that's their strength. But they create their own failure mode: they over-generalize. A dense search for "Apple MacBook Pro M3" might return documents about iPad peripherals or competing laptops, because the embedding captures "Apple laptop high-performance" rather than the specific product. BM25 would never make that mistake.
This is the fundamental tension that SPLADE was designed to resolve.
SPLADE (Sparse Lexical and Expansion model) was introduced by researchers at Naver Labs Europe in 2021 and has seen rapid adoption since. The key insight is elegant: use a BERT-style transformer to produce sparse, token-weighted vectors in the vocabulary space, not in a dense embedding space. This gives you the semantic understanding of transformers with the efficiency and interpretability of sparse vectors.
When SPLADE encodes a document, it produces a vector of length equal to the vocabulary size (typically ~30,000 for BERT-base). Most entries are zero. Non-zero entries represent vocabulary tokens that are relevant to this document — including tokens that weren't explicitly in the document but the model has learned are semantically related.
The process works like this:
The crucial difference from BM25: SPLADE learns which vocabulary expansions are relevant. A document about "myocardial infarction" might have non-zero weights for "heart attack," "cardiac event," and "chest pain" — terms that never appeared in the document — because the pre-trained transformer has encoded these relationships from training data.
At query time, the same process applies to the query. Matching is then done using an inverted index lookup (identical to keyword search) over these learned sparse vectors. The score for a query-document pair is the dot product of their sparse vectors.
Key insight: SPLADE gives you semantic term expansion without dense vector arithmetic. Because the representation lives in vocabulary space, you can use a standard inverted index for retrieval. This means SPLADE gets the scalability and interpretability of BM25 infrastructure with the semantic coverage of transformers. You can literally inspect what terms SPLADE expanded a query to — something impossible with dense embeddings.
The most practical way to use SPLADE in Python is through the splade library or via the beir benchmarking library. The Naver Labs team publishes pre-trained checkpoints on HuggingFace. Let's use their naver/splade-cocondenser-ensembledistil model, which is a strong general-purpose checkpoint:
from transformers import AutoTokenizer, AutoModelForMaskedLM
import torch
import numpy as np
from typing import Dict, List, Tuple
import scipy.sparse as sp
class SPLADEEncoder:
"""
SPLADE sparse encoder using HuggingFace transformers.
Produces sparse vectors in vocabulary space for use with inverted index retrieval.
"""
def __init__(
self,
model_name: str = "naver/splade-cocondenser-ensembledistil",
device: str = None,
max_length: int = 512
):
self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
self.max_length = max_length
print(f"Loading SPLADE model {model_name} on {self.device}...")
self.tokenizer = AutoTokenizer.from_pretrained(model_name)
self.model = AutoModelForMaskedLM.from_pretrained(model_name)
self.model.to(self.device)
self.model.eval()
self.vocab_size = self.tokenizer.vocab_size
print(f"Vocabulary size: {self.vocab_size}")
def encode(
self,
texts: List[str],
batch_size: int = 32,
show_progress: bool = False
) -> sp.csr_matrix:
"""
Encode a list of texts into sparse SPLADE vectors.
Returns a CSR sparse matrix of shape (len(texts), vocab_size).
CSR format is efficient for dot product computation and storage.
"""
all_vectors = []
for i in range(0, len(texts), batch_size):
batch = texts[i:i + batch_size]
# Tokenize the batch
inputs = self.tokenizer(
batch,
return_tensors="pt",
padding=True,
truncation=True,
max_length=self.max_length
).to(self.device)
with torch.no_grad():
# Get MLM logits: shape (batch, seq_len, vocab_size)
outputs = self.model(**inputs)
logits = outputs.logits
# SPLADE activation: ReLU + log(1 + x), then max-pool over sequence
# This is the core SPLADE aggregation formula
activated = torch.log(1 + torch.relu(logits))
# Max pool over the sequence dimension (dim=1)
# Result shape: (batch, vocab_size)
sparse_vecs, _ = activated.max(dim=1)
# Mask out padding tokens to avoid polluting the representation
# attention_mask shape: (batch, seq_len)
attention_mask = inputs["attention_mask"].unsqueeze(-1)
activated_masked = torch.log(1 + torch.relu(logits)) * attention_mask
sparse_vecs, _ = activated_masked.max(dim=1)
all_vectors.append(sparse_vecs.cpu().numpy())
if show_progress:
print(f"Encoded {min(i + batch_size, len(texts))}/{len(texts)} texts")
# Stack and convert to sparse matrix
dense_matrix = np.vstack(all_vectors)
return sp.csr_matrix(dense_matrix)
def decode_vector(
self,
vector: np.ndarray,
top_k: int = 20
) -> List[Tuple[str, float]]:
"""
Decode a SPLADE vector into interpretable (token, weight) pairs.
This is one of SPLADE's killer features: you can inspect what the model
actually thinks a document or query is about.
"""
if sp.issparse(vector):
vector = vector.toarray().flatten()
top_indices = np.argsort(vector)[::-1][:top_k]
results = []
for idx in top_indices:
if vector[idx] > 0:
token = self.tokenizer.convert_ids_to_tokens([idx])[0]
results.append((token, float(vector[idx])))
return results
Let's see SPLADE's term expansion in action before we build the full retrieval layer:
encoder = SPLADEEncoder()
# Encode a query about cardiac events
query = "heart attack symptoms in elderly patients"
query_vec = encoder.encode([query])
expanded_terms = encoder.decode_vector(query_vec[0])
print("SPLADE expansion for:", query)
for token, weight in expanded_terms:
print(f" {token}: {weight:.4f}")
Running this on a real model produces output like:
SPLADE expansion for: heart attack symptoms in elderly patients
heart: 2.8341
attack: 2.1203
symptoms: 1.9842
cardiac: 1.7234
##iac: 1.6891 # subword for "cardiac"
myocardial: 1.5102
infarction: 1.4823
elderly: 1.3901
patients: 1.3211
chest: 1.2445
pain: 1.1983
stroke: 0.9821
coronary: 0.9334
artery: 0.8901
disease: 0.8234
Notice that "myocardial," "infarction," "coronary," and "artery" all appeared with significant weights even though they weren't in the original query. This is semantic expansion happening in vocabulary space — and because it's in vocabulary space, a BM25-style inverted index can retrieve it efficiently.
Note: SPLADE subword tokens (like
##iac) will appear in your expansion vectors because BERT tokenizers use WordPiece tokenization. These subwords can match document subwords correctly through the inverted index. However, you may want to filter them out when displaying expansions to users, since they're not human-readable. The retrieval math is unaffected.
The efficiency of SPLADE depends on using an inverted index — not dense vector similarity search. Each non-zero entry in a SPLADE vector maps to a vocabulary token ID. When you build an inverted index over SPLADE-encoded documents, each posting list contains the document IDs that have a non-zero weight for that vocabulary token, along with the weight value.
from collections import defaultdict
import pickle
from pathlib import Path
class SPLADEIndex:
"""
In-memory inverted index for SPLADE sparse vectors.
For production deployments over millions of documents, consider:
- Elasticsearch with sparse_vector field type (8.x+)
- Qdrant with sparse vector support
- Vespa with WAND operator for efficient top-k retrieval
"""
def __init__(self):
# posting_list[token_id] = [(doc_id, weight), ...]
self.posting_lists: Dict[int, List[Tuple[int, float]]] = defaultdict(list)
self.documents: List[Dict] = []
self.doc_vectors: sp.csr_matrix = None
def build(
self,
documents: List[Dict],
sparse_matrix: sp.csr_matrix,
weight_threshold: float = 0.0
) -> None:
"""
Build inverted index from pre-computed SPLADE sparse matrix.
Args:
documents: Original document dicts
sparse_matrix: CSR matrix of shape (n_docs, vocab_size) from SPLADEEncoder
weight_threshold: Ignore weights below this value (reduces index size)
"""
self.documents = documents
self.doc_vectors = sparse_matrix
self.posting_lists = defaultdict(list)
# CSR matrix: iterate over non-zero entries efficiently
cx = sp.coo_matrix(sparse_matrix)
for doc_id, token_id, weight in zip(cx.row, cx.col, cx.data):
if weight > weight_threshold:
self.posting_lists[int(token_id)].append(
(int(doc_id), float(weight))
)
# Sort each posting list by weight descending for early termination
for token_id in self.posting_lists:
self.posting_lists[token_id].sort(key=lambda x: x[1], reverse=True)
n_tokens = len(self.posting_lists)
n_postings = sum(len(pl) for pl in self.posting_lists.values())
print(f"Index built: {len(documents)} docs, {n_tokens} unique tokens, "
f"{n_postings} total postings, "
f"avg sparsity: {n_postings / (len(documents) * sparse_matrix.shape[1]):.6f}")
def retrieve(
self,
query_vector: sp.csr_matrix,
top_k: int = 10
) -> List[Dict]:
"""
Retrieve top-k documents using the inverted index.
Computes dot products only for documents that share at least one
non-zero token with the query — much faster than dense retrieval
over the full vocabulary.
"""
# Accumulate scores for candidate documents
scores = defaultdict(float)
# Get non-zero entries in query vector
query_coo = sp.coo_matrix(query_vector)
for token_id, query_weight in zip(query_coo.col, query_coo.data):
token_id = int(token_id)
if token_id in self.posting_lists:
for doc_id, doc_weight in self.posting_lists[token_id]:
scores[doc_id] += query_weight * doc_weight
if not scores:
return []
# Sort by score and return top-k
sorted_doc_ids = sorted(scores.keys(), key=lambda d: scores[d], reverse=True)[:top_k]
results = []
for rank, doc_id in enumerate(sorted_doc_ids):
results.append({
"document": self.documents[doc_id],
"score": scores[doc_id],
"rank": rank + 1,
"retriever": "splade"
})
return results
def save(self, path: str) -> None:
"""Persist the index to disk."""
save_path = Path(path)
save_path.mkdir(parents=True, exist_ok=True)
with open(save_path / "posting_lists.pkl", "wb") as f:
pickle.dump(dict(self.posting_lists), f)
sp.save_npz(str(save_path / "doc_vectors.npz"), self.doc_vectors)
with open(save_path / "documents.pkl", "wb") as f:
pickle.dump(self.documents, f)
print(f"Index saved to {path}")
@classmethod
def load(cls, path: str) -> "SPLADEIndex":
"""Load a persisted index."""
load_path = Path(path)
index = cls()
with open(load_path / "posting_lists.pkl", "rb") as f:
index.posting_lists = defaultdict(list, pickle.load(f))
index.doc_vectors = sp.load_npz(str(load_path / "doc_vectors.npz"))
with open(load_path / "documents.pkl", "rb") as f:
index.documents = pickle.load(f)
return index
The weight_threshold parameter deserves attention. SPLADE vectors are technically sparse after the ReLU activation, but they can still have thousands of non-zero entries per document, especially at lower weights. Setting weight_threshold=0.01 or 0.05 can cut index size by 30-50% with minimal recall impact. Tune this on your validation set.
Now we need to bring together BM25, SPLADE, and dense retrieval into a coherent hybrid system. The fusion strategy we use matters a lot here.
The two main approaches are:
Reciprocal Rank Fusion (RRF): Combines ranked lists from each retriever using 1 / (k + rank) where k=60 is typical. No score normalization needed. Score-agnostic — only ranks matter. This is robust and easy to implement. See Reciprocal Rank Fusion: Merging Multiple Retrieval Results into a Single Ranked List for Hybrid RAG Pipelines for a deep dive on this technique.
Linear combination: α · score_dense + β · score_bm25 + γ · score_splade. Requires score normalization (min-max or softmax across results) to make scores from different retrievers comparable. More expressive than RRF but sensitive to normalization choices and requires tuning α, β, γ.
from typing import Optional
import numpy as np
class HybridRetriever:
"""
Production hybrid retriever combining BM25, SPLADE, and dense retrieval.
Fusion strategies:
- 'rrf': Reciprocal Rank Fusion (robust, no tuning required)
- 'linear': Weighted linear combination (requires normalized scores + tuning)
"""
def __init__(
self,
bm25_retriever: BM25Retriever,
splade_index: SPLADEIndex,
splade_encoder: SPLADEEncoder,
dense_retriever, # Any retriever implementing .retrieve(query, top_k) -> List[Dict]
fusion_strategy: str = "rrf",
rrf_k: int = 60,
weights: Dict[str, float] = None # For 'linear' strategy
):
self.bm25 = bm25_retriever
self.splade_index = splade_index
self.splade_encoder = splade_encoder
self.dense = dense_retriever
self.fusion_strategy = fusion_strategy
self.rrf_k = rrf_k
self.weights = weights or {"bm25": 0.2, "splade": 0.3, "dense": 0.5}
def retrieve(
self,
query: str,
top_k: int = 10,
retriever_top_k: int = 50, # How many each sub-retriever fetches before fusion
include_scores: bool = False
) -> List[Dict]:
"""
Full hybrid retrieval with fusion.
retriever_top_k should be significantly larger than top_k to give
fusion enough candidates to work with. 50 is a reasonable default;
for high-recall use cases, use 100-200.
"""
# Parallel retrieval from all three sources
bm25_results = self.bm25.retrieve(query, top_k=retriever_top_k)
query_sparse_vec = self.splade_encoder.encode([query])
splade_results = self.splade_index.retrieve(
query_sparse_vec, top_k=retriever_top_k
)
dense_results = self.dense.retrieve(query, top_k=retriever_top_k)
# Fuse results
if self.fusion_strategy == "rrf":
fused = self._reciprocal_rank_fusion(
bm25_results, splade_results, dense_results
)
elif self.fusion_strategy == "linear":
fused = self._linear_combination(
bm25_results, splade_results, dense_results
)
else:
raise ValueError(f"Unknown fusion strategy: {self.fusion_strategy}")
return fused[:top_k]
def _reciprocal_rank_fusion(self, *result_lists) -> List[Dict]:
"""RRF fusion across arbitrary number of ranked lists."""
doc_scores = defaultdict(float)
doc_objects = {}
for result_list in result_lists:
for item in result_list:
# Use document ID or text as the unique key
doc_key = item["document"].get("id") or item["document"].get("text", "")[:100]
rank = item["rank"]
doc_scores[doc_key] += 1.0 / (self.rrf_k + rank)
doc_objects[doc_key] = item["document"]
sorted_keys = sorted(doc_scores.keys(), key=lambda k: doc_scores[k], reverse=True)
return [
{
"document": doc_objects[key],
"score": doc_scores[key],
"rank": rank + 1,
"retriever": "hybrid_rrf"
}
for rank, key in enumerate(sorted_keys)
]
def _linear_combination(self, *result_lists) -> List[Dict]:
"""
Weighted linear combination with min-max normalization per retriever.
"""
retriever_names = ["bm25", "splade", "dense"]
all_scores = {}
doc_objects = {}
for name, result_list in zip(retriever_names, result_lists):
if not result_list:
continue
raw_scores = {
r["document"].get("id", r["document"]["text"][:100]): r["score"]
for r in result_list
}
# Min-max normalize within this retriever
min_s = min(raw_scores.values())
max_s = max(raw_scores.values())
score_range = max_s - min_s or 1.0
for key, score in raw_scores.items():
normalized = (score - min_s) / score_range
if key not in all_scores:
all_scores[key] = 0.0
all_scores[key] += self.weights[name] * normalized
for r in result_list:
key = r["document"].get("id", r["document"]["text"][:100])
doc_objects[key] = r["document"]
sorted_keys = sorted(all_scores.keys(), key=lambda k: all_scores[k], reverse=True)
return [
{
"document": doc_objects[key],
"score": all_scores[key],
"rank": rank + 1,
"retriever": "hybrid_linear"
}
for rank, key in enumerate(sorted_keys)
]
Tip: Run both RRF and linear combination on your evaluation set before committing to one. In practice, RRF outperforms linear combination more often than practitioners expect, especially when score distributions from different retrievers are wildly different — which is common when mixing BM25 (unbounded scores based on corpus statistics) with SPLADE (bounded by log activation) with cosine similarity (bounded to [-1, 1]).
For most production deployments, you won't run pure Python in-memory indices. Let's show how to integrate with real infrastructure. The pattern is: Elasticsearch for BM25 (and optionally SPLADE via sparse_vector), Qdrant for dense vectors.
from elasticsearch import Elasticsearch, helpers
from qdrant_client import QdrantClient
from qdrant_client.models import Distance, VectorParams, PointStruct
import uuid
class ProductionHybridRetriever:
"""
Production hybrid retriever using Elasticsearch (BM25/SPLADE) + Qdrant (dense).
This architecture separates concerns cleanly:
- ES handles sparse retrieval with inverted index expertise
- Qdrant handles ANN for dense retrieval
- Python layer handles fusion
"""
def __init__(
self,
es_url: str,
es_index: str,
qdrant_url: str,
qdrant_collection: str,
embedding_fn, # Callable: str -> List[float]
splade_encoder: Optional[SPLADEEncoder] = None,
):
self.es = Elasticsearch(es_url)
self.es_index = es_index
self.qdrant = QdrantClient(url=qdrant_url)
self.qdrant_collection = qdrant_collection
self.embedding_fn = embedding_fn
self.splade_encoder = splade_encoder
def create_es_index(self, vector_dim: int = 30522) -> None:
"""
Create Elasticsearch index with BM25 text field and optional sparse_vector for SPLADE.
Note: sparse_vector type requires ES 8.x+
"""
mapping = {
"mappings": {
"properties": {
"text": {
"type": "text",
"analyzer": "english" # Stemming + stopwords for BM25
},
"doc_id": {"type": "keyword"},
"metadata": {"type": "object", "enabled": False}
}
},
"settings": {
"index": {
"similarity": {
"default": {
"type": "BM25",
"k1": 1.6,
"b": 0.75
}
}
}
}
}
# Add SPLADE sparse vector field if encoder is provided
if self.splade_encoder:
mapping["mappings"]["properties"]["splade_vector"] = {
"type": "sparse_vector"
}
self.es.indices.create(index=self.es_index, body=mapping, ignore=400)
def index_documents(
self,
documents: List[Dict],
batch_size: int = 100
) -> None:
"""Index documents into both ES and Qdrant."""
# Pre-compute SPLADE vectors if encoder is available
splade_vectors = None
if self.splade_encoder:
texts = [doc["text"] for doc in documents]
splade_matrix = self.splade_encoder.encode(
texts, batch_size=32, show_progress=True
)
# Index into Elasticsearch
def es_actions():
for i, doc in enumerate(documents):
action = {
"_index": self.es_index,
"_id": doc.get("id", str(uuid.uuid4())),
"_source": {
"text": doc["text"],
"doc_id": doc.get("id", ""),
"metadata": doc.get("metadata", {})
}
}
# Add SPLADE vector if available
if self.splade_encoder and splade_matrix is not None:
vec = splade_matrix[i]
cx = sp.coo_matrix(vec)
# ES sparse_vector format: {token_id_str: weight}
sparse_dict = {
str(int(tid)): float(w)
for tid, w in zip(cx.col, cx.data)
if w > 0.01 # Threshold to control index size
}
action["_source"]["splade_vector"] = sparse_dict
yield action
helpers.bulk(self.es, es_actions(), chunk_size=batch_size)
# Index into Qdrant (dense vectors)
points = []
for i, doc in enumerate(documents):
vector = self.embedding_fn(doc["text"])
point = PointStruct(
id=doc.get("id", str(uuid.uuid4())),
vector=vector,
payload={
"text": doc["text"],
"metadata": doc.get("metadata", {})
}
)
points.append(point)
# Upsert in batches
for i in range(0, len(points), batch_size):
self.qdrant.upsert(
collection_name=self.qdrant_collection,
points=points[i:i + batch_size]
)
print(f"Indexed {len(documents)} documents into ES and Qdrant")
def retrieve(self, query: str, top_k: int = 10) -> List[Dict]:
"""Full hybrid retrieval against production infrastructure."""
# BM25 retrieval via ES
bm25_response = self.es.search(
index=self.es_index,
body={
"query": {"match": {"text": query}},
"size": top_k * 5 # Over-fetch for fusion
}
)
bm25_results = [
{
"document": {
"id": hit["_id"],
"text": hit["_source"]["text"],
"metadata": hit["_source"].get("metadata", {})
},
"score": hit["_score"],
"rank": rank + 1,
"retriever": "bm25"
}
for rank, hit in enumerate(bm25_response["hits"]["hits"])
]
# Dense retrieval via Qdrant
query_vector = self.embedding_fn(query)
qdrant_results = self.qdrant.search(
collection_name=self.qdrant_collection,
query_vector=query_vector,
limit=top_k * 5
)
dense_results = [
{
"document": {
"id": str(r.id),
"text": r.payload["text"],
"metadata": r.payload.get("metadata", {})
},
"score": r.score,
"rank": rank + 1,
"retriever": "dense"
}
for rank, r in enumerate(qdrant_results)
]
# RRF fusion
return self._rrf_fusion(bm25_results, dense_results, k=60)[:top_k]
This architecture scales independently: you can add ES nodes for BM25 capacity and Qdrant shards for dense capacity without coupling the two. See Retrieval Latency Optimization: Indexing Strategies, ANN Tuning, and Caching Layers for Sub-100ms RAG in Production for a deep discussion on keeping the total pipeline under 100ms.
Let's be concrete about the performance profile of each retriever. This matters for deciding how much weight to give each component and whether SPLADE is worth the inference cost for your use case.
| Query Type | BM25 | SPLADE | Dense |
|---|---|---|---|
| Exact product code / ID | ✅ Excellent | ✅ Good | ❌ Often misses |
| Medical terminology | ❌ Vocab gap | ✅ Excellent | ✅ Good |
| Paraphrase / synonym | ❌ Poor | ✅ Good | ✅ Excellent |
| Short keyword query (2-3 words) | ✅ Good | ✅ Excellent | ⚠️ Variable |
| Long natural language question | ⚠️ Partial | ✅ Good | ✅ Excellent |
| Cross-lingual | ❌ | ❌ (model-dependent) | ✅ (multilingual models) |
| Out-of-domain vocabulary | ❌ | ⚠️ Degrades | ⚠️ Degrades |
The practical implication: SPLADE largely obsoletes pure BM25 as a sparse retriever if you can afford the inference cost at indexing time. At query time, SPLADE encoding adds roughly 10-30ms on CPU (or 1-3ms on GPU) per query — often acceptable in production. The indexing cost is where SPLADE gets expensive: a corpus of 10 million documents requires 10 million SPLADE encodings. For a reasonably-sized GPU, that's on the order of hours to days.
Key insight: SPLADE gives its biggest gains on knowledge-domain queries where the vocabulary gap between queries and documents is large — medical, legal, scientific, and technical corpora. If your corpus is primarily conversational text and your queries are natural language questions, the marginal benefit over good dense retrieval plus BM25 narrows considerably.
This consideration is why Adaptive Retrieval: Dynamically Adjusting Chunk Count and Search Strategy Based on Query Complexity at Runtime is compelling: you can route exact-match queries to BM25 only, natural-language questions to dense-only, and domain-vocabulary queries to full SPLADE+dense. The Query Routing in RAG article covers how to implement that classification layer.
In this exercise, you'll build a hybrid retrieval system over a realistic legal document corpus and compare BM25, SPLADE, and hybrid retrieval quality.
Setup:
# Sample legal document corpus — realistic domain with vocabulary gap challenges
documents = [
{
"id": "doc_001",
"text": """The petitioner contends that the respondent's failure to disclose
material facts constitutes fraudulent misrepresentation under Section 17 of
the Indian Contract Act, 1872. The burden of proof lies with the party
alleging fraud, requiring clear and convincing evidence."""
},
{
"id": "doc_002",
"text": """Promissory estoppel operates to prevent a party from going back
on a promise that another party has relied upon to their detriment. Unlike
consideration in classical contract theory, reliance is the operative element."""
},
{
"id": "doc_003",
"text": """The doctrine of res judicata bars relitigation of claims that were
or could have been raised in prior proceedings between the same parties.
Issue preclusion and claim preclusion are its two primary components."""
},
{
"id": "doc_004",
"text": """Quantum meruit allows a party to recover the reasonable value of
services rendered when no enforceable contract exists. Courts assess this
under the theory of unjust enrichment."""
},
{
"id": "doc_005",
"text": """GDPR Article 17 establishes the right to erasure, colloquially
known as the 'right to be forgotten.' Data controllers must delete personal
data upon request when the original processing purpose no longer applies."""
}
]
# Test queries that stress-test vocabulary gaps
test_queries = [
"can I sue someone for lying during contract negotiations", # Paraphrase of doc_001
"Section 17 Indian Contract Act", # Exact match for doc_001
"blocking old lawsuits from being refiled", # Paraphrase of doc_003
"res judicata", # Exact match for doc_003
"delete my data from a company's records", # Paraphrase of doc_005
]
Your tasks:
Build a BM25Retriever and index the five documents. Run all five queries and record the top result for each. Notice which paraphrase queries return wrong results.
Build a SPLADEEncoder and SPLADEIndex. Run the same queries and compare. Use decode_vector() to inspect what terms SPLADE expanded each query to.
Build a HybridRetriever combining BM25 and SPLADE (skip dense for this exercise). Use RRF fusion. Compare results against each individual retriever.
Extension: Add a dense retrieval layer using sentence-transformers/all-MiniLM-L6-v2. Compare RRF versus linear combination on your test queries. Which fusion strategy produces better results on your query set?
Evaluation: Implement a simple MRR (Mean Reciprocal Rank) evaluator and score each configuration. Define ground truth as: query 1 → doc_001, query 2 → doc_001, query 3 → doc_003, query 4 → doc_003, query 5 → doc_005.
def mean_reciprocal_rank(results_list, ground_truth_ids):
"""
Calculate MRR across a set of queries.
Args:
results_list: List of result lists (one per query)
ground_truth_ids: List of correct document IDs (one per query)
Returns:
MRR score between 0 and 1
"""
reciprocal_ranks = []
for results, gt_id in zip(results_list, ground_truth_ids):
rr = 0.0
for rank, result in enumerate(results, start=1):
if result["document"]["id"] == gt_id:
rr = 1.0 / rank
break
reciprocal_ranks.append(rr)
return np.mean(reciprocal_ranks)
# Expected: SPLADE and hybrid should significantly outperform BM25 on paraphrase queries
For a thorough understanding of how to evaluate your retrieval system more formally, the Evaluating RAG Systems: Precision, Recall, and Faithfulness article covers the full evaluation stack including context precision, context recall, and answer faithfulness.
SPLADE vectors can have 3,000-8,000 non-zero entries per document before thresholding. Without a threshold, your inverted index becomes enormous and retrieval slows dramatically. Setting weight_threshold=0.01 cuts this to 200-500 entries with negligible recall impact in most corpora.
BM25 scores for a typical document corpus might range from 0 to 15. SPLADE dot product scores might range from 0 to 50. Cosine similarity for dense retrieval ranges from -1 to 1. Combining these raw scores without normalization means BM25 and SPLADE will completely dominate, and your "hybrid" system is effectively just BM25+SPLADE with some ignored dense retrieval. Always normalize per-retriever before weighting.
If you use different tokenization at index time versus query time, your BM25 scores will be wrong. A common mistake is stemming during indexing but not during querying (or vice versa). Your _tokenize method must be identical in both paths. This sounds obvious but breaks in surprising ways when you update your tokenizer after initial indexing.
If each sub-retriever returns only top_k=10 documents and you're fusing three retrievers, your final pool has at most 30 unique candidates. The "best" document might rank 15th in BM25, 3rd in SPLADE, and 8th in dense — RRF would rank it highly, but if you only fetched 10 from each retriever, you'd never see it in BM25. As a rule of thumb, set retriever_top_k = max(50, top_k * 5).
Warning: Over-fetching for fusion adds latency. If you're fetching 200 candidates from three retrievers before returning 10, you're doing 3x the retrieval work. For latency-sensitive applications, profile your retrieval times and consider whether a two-stage approach (fast rough retrieval, then a reranker) is more efficient. See Reranking Retrieved Results: Implementing Cross-Encoders to Improve RAG Accuracy for the reranking layer.
SPLADE models trained on MS MARCO (general web queries) may perform poorly on highly specialized domains like clinical notes or legal filings. If you're in a specialized domain, fine-tune SPLADE on domain-relevant query-document pairs, or at minimum verify recall on a domain-representative evaluation set before committing to the model. Fine-Tuning Embedding Models on Domain-Specific Data to Improve Retrieval Accuracy in RAG Pipelines covers the fine-tuning methodology, and the same principles apply to SPLADE.
BM25 scores are not normalized across queries. A score of 8.3 for query A means something completely different than 8.3 for query B, because IDF values depend on which terms appear in that query. Never threshold results on absolute BM25 score; always use relative rank or normalize within the result set.
When your hybrid system returns a wrong document, diagnose by running each retriever independently and checking:
def diagnose_retrieval(hybrid_retriever, query, expected_doc_id, top_k=20):
"""Diagnose which retrievers are succeeding or failing for a query."""
print(f"Query: '{query}'")
print(f"Expected: {expected_doc_id}")
print()
# BM25
bm25_results = hybrid_retriever.bm25.retrieve(query, top_k=top_k)
bm25_rank = next(
(r["rank"] for r in bm25_results if r["document"]["id"] == expected_doc_id),
"NOT FOUND"
)
print(f"BM25 rank: {bm25_rank}")
print(f"BM25 top result: {bm25_results[0]['document']['id'] if bm25_results else 'none'}")
# SPLADE
q_vec = hybrid_retriever.splade_encoder.encode([query])
expanded = hybrid_retriever.splade_encoder.decode_vector(q_vec[0], top_k=10)
print(f"SPLADE expanded terms: {[t for t, w in expanded[:5]]}")
splade_results = hybrid_retriever.splade_index.retrieve(q_vec, top_k=top_k)
splade_rank = next(
(r["rank"] for r in splade_results if r["document"]["id"] == expected_doc_id),
"NOT FOUND"
)
print(f"SPLADE rank: {splade_rank}")
If BM25 fails but SPLADE succeeds, you have a vocabulary gap that SPLADE is bridging — this is the expected case. If both fail, check whether your chunking strategy might be the issue (see Chunking Strategies: How to Split Documents for Better Retrieval). If only hybrid fails but both individual retrievers succeed at ranking it within top-20, you have a fusion issue — likely your retriever_top_k is too small.
You now have a complete mental model and implementation path for production hybrid sparse-dense retrieval. Let's consolidate what you've built:
BM25 is a probabilistic keyword retrieval function with elegant term frequency saturation and length normalization. It's reliable, interpretable, and fast for exact-match queries, but fails on vocabulary gaps.
SPLADE uses a BERT-style transformer to produce sparse vocabulary-space vectors, enabling semantic term expansion while maintaining the efficiency of inverted index retrieval. It largely supersedes BM25 for specialized domains, at the cost of inference time at indexing and query time.
Hybrid fusion with Reciprocal Rank Fusion is robust and requires no hyperparameter tuning. Linear combination can outperform RRF but requires careful score normalization and tuning. Always over-fetch before fusion.
In production, delegate BM25 and SPLADE to Elasticsearch (sparse_vector type in ES 8.x) and dense retrieval to a dedicated ANN store like Qdrant. This keeps concerns separated and lets you scale each layer independently.
The natural next steps from here:
Implement a reranking layer on top of your hybrid retrieval. A cross-encoder reranker typically recovers significant precision lost during the coarse retrieval phase — Reranking Retrieved Results: Implementing Cross-Encoders to Improve RAG Accuracy shows you how.
Add query routing logic to select retriever configurations dynamically based on query characteristics — keyword-heavy queries can skip SPLADE inference at query time for lower latency.
Evaluate your system rigorously using BEIR benchmarks if your domain is represented there, and build domain-specific evaluation sets using annotated query-document pairs.
Consider whether Contextual Compression in RAG: Filtering and Compressing Retrieved Chunks Before Passing to the LLM applies to your pipeline — better retrieval and better compression compound to produce significantly better LLM outputs.
Sparse retrieval is one of those areas where the gap between "it works" and "it works correctly under production conditions" is large. The implementations in this lesson are production-oriented, but they're starting points — your corpus characteristics, query distribution, and latency requirements will push you toward specific configurations that no general tutorial can prescribe. The debugging and evaluation patterns you've learned here are how you find those configurations for your specific system.