From Single Retrieval to Hybrid Strategies
In traditional RAG systems, the retrieval step typically relies on vector similarity search. While this approach excels at capturing semantic similarity, pure semantic retrieval often fails when user queries contain precise terms, proper nouns, or specific formats.
Have you ever encountered a scenario where querying for "IRR calculation formula" returns numerous documents discussing the "concept of return on investment," while the documents actually containing the formula details are ranked lower? This happens because vector embeddings lose the ability to perform exact keyword matching during encoding.
Solving this problem requires a smarter retrieval strategy—Hybrid Retrieval—which leverages the strengths of both keyword retrieval and semantic retrieval simultaneously.
Implementing BM25 Keyword Retrieval
BM25 (Best Matching 25) is a classic keyword ranking algorithm widely used in information retrieval. Compared to simple TF-IDF, BM25 introduces two key improvements:
- Term Frequency Saturation: Controls the upper limit of term frequency influence via parameter k1, preventing high-frequency words from dominating excessively.
- Document Length Normalization: Balances preferences for short and long documents via parameter b.
class BM25:
"""BM25 Keyword Retrieval Algorithm
Implementation of the Okapi BM25 algorithm for keyword retrieval.
"""
def __init__(
self,
k1: float = 1.5,
b: float = 0.75,
epsilon: float = 0.25,
):
self.k1 = k1
self.b = b
self.epsilon = epsilon
self.doc_freqs: dict[str, int] = {}
self.doc_len: list[int] = []
self.avgdl: float = 0
self.doc_count: int = 0
self.doc_term_freqs: list[dict[str, int]] = []
self.idf: dict[str, float] = {}
self.documents: list[Document] = []
def _tokenize(self, text: str) -> list[str]:
"""Tokenization"""
text = text.lower()
tokens = re.findall(r"\w+", text)
return tokens
def fit(self, documents: list[Document]):
"""Train the BM25 model"""
self.documents = documents
self.doc_count = len(documents)
self.doc_len = []
self.doc_term_freqs = []
self.doc_freqs = {}
for doc in documents:
tokens = self._tokenize(doc.page_content)
self.doc_len.append(len(tokens))
term_freqs = Counter(tokens)
self.doc_term_freqs.append(dict(term_freqs))
for term in term_freqs:
if term not in self.doc_freqs:
self.doc_freqs[term] = 0
self.doc_freqs[term] += 1
self.avgdl = sum(self.doc_len) / self.doc_count if self.doc_count > 0 else 0
self._calc_idf()
The core of this implementation lies in the fit method: it iterates through all documents, counting term frequencies, while also calculating Inverse Document Frequency (IDF). The IDF calculation uses the standard form of the Okapi formula:
def _calc_idf(self):
"""Calculate IDF"""
idf_sum = 0
negative_idfs = []
for term, freq in self.doc_freqs.items():
idf = math.log(self.doc_count - freq + 0.5) - math.log(freq + 0.5)
self.idf[term] = idf
idf_sum += idf
if idf < 0:
negative_idfs.append(term)
avg_idf = idf_sum / len(self.idf) if self.idf else 0
eps = self.epsilon * avg_idf
for term in negative_idfs:
self.idf[term] = eps
Note the handling of negative IDF values here: when a word appears in almost all documents, its IDF may become negative. The author uses the epsilon parameter as a lower bound to prevent these common words from negatively impacting the score.
Three Fusion Strategies
The key challenge in hybrid retrieval is how to fuse the results of keyword retrieval and semantic retrieval. The HybridRetriever class provides three different fusion strategies:
Weighted Fusion
This is the most intuitive method: normalize the results of both retrievals separately, then add them together according to weights.
def _weighted_fusion(
self,
keyword_results: list[tuple[Document, float]],
semantic_results: list[tuple[Document, float]],
k: int,
keyword_weight: float,
semantic_weight: float,
) -> list[tuple[Document, float]]:
"""Weighted Fusion"""
doc_scores: dict[str, tuple[Document, float]] = {}
max_kw_score = max((score for _, score in keyword_results), default=1.0) or 1.0
for doc, score in keyword_results:
doc_id = self._get_doc_id(doc)
normalized_score = score / max_kw_score
doc_scores[doc_id] = (doc, normalized_score * keyword_weight)
max_sem_score = max((score for _, score in semantic_results), default=1.0) or 1.0
for doc, score in semantic_results:
doc_id = self._get_doc_id(doc)
normalized_score = 1 - (score / max_sem_score)
if doc_id in doc_scores:
existing_doc, existing_score = doc_scores[doc_id]
doc_scores[doc_id] = (
existing_doc,
existing_score + normalized_score * semantic_weight,
)
else:
doc_scores[doc_id] = (doc, normalized_score * semantic_weight)
results = sorted(doc_scores.values(), key=lambda x: x[1], reverse=True)
return results[:k]
A key point is that semantic retrieval returns distance scores (smaller means more similar), so they need to be converted into similarity scores (1 - score/max).
Reciprocal Rank Fusion (RRF)
RRF is a fusion method that does not rely on specific scores; it only focuses on the ranking position of documents in the two lists. This method is more robust because the score scales of different retrievers can be completely different.
def _rrf_fusion(
self,
keyword_results: list[tuple[Document, float]],
semantic_results: list[tuple[Document, float]],
k: int,
) -> list[tuple[Document, float]]:
"""Reciprocal Rank Fusion"""
doc_scores: dict[str, tuple[Document, float]] = {}
for rank, (doc, _) in enumerate(keyword_results):
doc_id = self._get_doc_id(doc)
rrf_score = 1 / (self.rrf_k + rank + 1)
doc_scores[doc_id] = (doc, rrf_score)
for rank, (doc, _) in enumerate(semantic_results):
doc_id = self._get_doc_id(doc)
rrf_score = 1 / (self.rrf_k + rank + 1)
if doc_id in doc_scores:
existing_doc, existing_score = doc_scores[doc_id]
doc_scores[doc_id] = (existing_doc, existing_score + rrf_score)
else:
doc_scores[doc_id] = (doc, rrf_score)
results = sorted(doc_scores.values(), key=lambda x: x[1], reverse=True)
return results[:k]
The core formula for RRF is 1 / (k + rank), where k is a tuning parameter (default is 60). The characteristics of this formula are:
- Higher rankings (smaller rank) result in higher scores.
- Documents appearing with high ranks in both lists receive significantly higher combined scores.
- No normalization of raw scores is required.
Score Normalization Fusion
This is a variant of weighted fusion, but it uses Min-Max normalization to handle inconsistent score distributions.
def _score_fusion(
self,
keyword_results: list[tuple[Document, float]],
semantic_results: list[tuple[Document, float]],
k: int,
keyword_weight: float,
semantic_weight: float,
) -> list[tuple[Document, float]]:
"""Score Normalization Fusion"""
doc_scores: dict[str, tuple[Document, float]] = {}
kw_scores = [score for _, score in keyword_results]
kw_min = min(kw_scores) if kw_scores else 0
kw_max = max(kw_scores) if kw_scores else 1
kw_range = kw_max - kw_min if kw_max != kw_min else 1
for doc, score in keyword_results:
doc_id = self._get_doc_id(doc)
normalized = (score - kw_min) / kw_range if kw_range > 0 else 0.5
doc_scores[doc_id] = (doc, normalized * keyword_weight)
sem_scores = [score for _, score in semantic_results]
sem_min = min(sem_scores) if sem_scores else 0
sem_max = max(sem_scores) if sem_scores else 1
sem_range = sem_max - sem_min if sem_max != sem_min else 1
for doc, score in semantic_results:
doc_id = self._get_doc_id(doc)
normalized = 1 - (score - sem_min) / sem_range if sem_range > 0 else 0.5
if doc_id in doc_scores:
existing_doc, existing_score = doc_scores[doc_id]
doc_scores[doc_id] = (
existing_doc,
existing_score + normalized * semantic_weight,
)
else:
doc_scores[doc_id] = (doc, normalized * semantic_weight)
results = sorted(doc_scores.values(), key=lambda x: x[1], reverse=True)
return results[:k]
The advantage of this method is that it is less sensitive to outliers, but the disadvantage is that if a retrieval result returns no documents, the normalization degrades to the default value of 0.5.
Integrating Hybrid RAG System
The HybridRAGSystem class integrates hybrid retrieval into the RAG workflow, providing a unified interface:
class HybridRAGSystem:
"""Hybrid RAG System
A RAG system combining keyword retrieval and semantic retrieval.
"""
def __init__(
self,
rag_system,
fusion_method: str = "rrf",
keyword_weight: float = 0.3,
):
self.rag = rag_system
self.retriever = HybridRetriever(
fusion_method=fusion_method,
keyword_weight=keyword_weight,
)
self._fitted = False
def fit(self):
"""Train the hybrid retrieval model"""
if not self.rag.vector_store:
raise ValueError("Vector store of the RAG system is not initialized")
if hasattr(self.rag.vector_store, "similarity_search"):
docs = self.rag.vector_store.similarity_search("", k=1000)
self.retriever.fit(docs)
self._fitted = True
def generate_answer(
self,
query: str,
k: int = 3,
alpha: float | None = None,
max_context_length: int = 4000,
):
"""Generate Answer"""
results = self.retrieve(query, k, alpha)
relevant_docs = [doc for doc, _ in results]
if not relevant_docs:
return "Sorry, no related document information found.", []
prompt = self.rag.prompt_builder.build_qa_prompt(
query=query,
documents=relevant_docs,
max_context_length=max_context_length,
)
answer = self.rag.llm_integration.generate(prompt)
return answer, relevant_docs
The design of the fit method is clever: it extracts a batch of documents (default 1000) from the vector store to train the BM25 index. This avoids the performance overhead of indexing the entire document library while ensuring recall rate.
Source Code Navigation
| Module | File | Description |
|---|---|---|
| BM25 Keyword Retrieval | hybrid_retriever.py | BM25 algorithm implementation and hybrid retriever |
| LLM Integration | llm_integration.py | LLM calls, retries, caching, rate limiting |
| RAG System | rag.py | Complete RAG system implementation |
Quick Start
Install dependencies:
pip install langchain-llm-toolkit
Use the hybrid retrieval RAG system:
from langchain_llm_toolkit.hybrid_retriever import HybridRAGSystem
from langchain_llm_toolkit.rag import RAGSystem
# Initialize basic RAG system
rag = RAGSystem(vector_store_type="qdrant")
# Load documents and create vector store
documents = rag.load_and_process_documents(["document.pdf"])
rag.create_vector_store(documents)
# Initialize hybrid RAG system
hybrid_rag = HybridRAGSystem(rag_system=rag, fusion_method="rrf")
# Generate answer
answer, docs = hybrid_rag.generate_answer("What is IRR?")
print(answer)
You can also use the hybrid retrieval method directly within RAGSystem:
# Use built-in hybrid retrieval
relevant_docs = rag.retrieve_hybrid(query="IRR calculation formula", k=5, bm25_weight=0.3)
This method defaults to using 30% BM25 weight and 70% semantic retrieval weight. You can adjust the bm25_weight parameter according to your actual needs.