What is RRF?
RRF stands for Reciprocal Rank Fusion. It is an algorithm that merges multiple ranked result lists from different search systems into a single unified ranking.
全称是 Reciprocal Rank Fusion（倒数排名融合,e.g. ‘A’ –>’1/A’）。它是一种将来自不同检索系统的多个排名结果列表合并成一个统一排名的算法。
Imagine you’re hiring for a job. You ask two different recruiters to each give you their top 10 candidates ranked from best to worst. Recruiter A uses resume keywords; Recruiter B uses cultural fit interviews. Both give you a ranked list, but their scores are totally different scales (one uses 1-10, the other uses A-F). How do you combine them? RRF says: “Don’t look at the scores — look at the positions. If a candidate is #1 on both lists, they’re clearly the best. If someone is #1 on one list but #10 on the other, they’re good but not universal.
想象你在招聘。你让两个不同的猎头分别给你他们心中从最好到最差排名的前10名候选人。猎头A看简历关键词，猎头B看文化契合度面试。两个人都给你一份排名列表，但他们的打分标准完全不同（一个用1-10分，一个用A-F评级）。怎么合并？RRF的做法是：“别看分数，看位置。如果一个候选人在两份名单上都排第1，那Ta显然是最好的。如果某人在一份名单排第1但在另一份排第10，那Ta不错但不是公认的。
Why learn RRF? Why does it matter?
In hybrid search, we run two searches in parallel:
- BM25 (keyword/lexical search): Excels at exact keyword matching
- Vector search (semantic search): Excels at understanding meaning, synonyms, and context
The problem: BM25 scores and vector similarity scores are on completely different scales. BM25 scores have no upper bound; cosine similarity is 0-1. You can’t just add them together — one would dominate the other.
问题在于：BM25分数和向量相似度分数处于完全不同的量纲上。BM25分数没有上限；余弦相似度是0-1。你不能直接把它们加在一起——其中一个会主导另一个
RRF solves this by ignoring the raw scores entirely and working only with rank positions. It treats both retrieval methods equally, giving each a fair vote in the final ranking.
RRF通过完全忽略原始分数、只使用排名位置来解决这个问题。它平等对待两种检索方法，在最终排名中给每种方法公平的投票权。
RRF Workflow
- Parallel Retrieval: Execute BM25 and vector search simultaneously on the same query
并行检索：对同一个查询同时执行BM25和向量检索 - Get Ranked Lists: Each returns a ranked list of top-N results (e.g., top 50)
获取排名列表：各自返回top-N结果的排名列表（如top 50） - Assign Reciprocal Scores: For each document in each list, compute
1/(rank + k)
分配倒数分数：对每个列表中的每个文档，计算1/(rank + k) - Sum Scores: For each document, sum its reciprocal scores from all lists
汇总分数：对每个文档，将所有列表中的倒数分数相加 - Final Ranking: Sort documents by total RRF score descending最
终排名：按RRF总分降序排列文档
Code implementation
Pure Python RRF Implementation
"""
Reciprocal Rank Fusion (RRF) - Pure Python Implementation
倒数排名融合(RRF) - 纯Python实现
This module implements the RRF algorithm to merge multiple ranked result lists.
本模块实现RRF算法，用于合并多个排名结果列表。
Key concept: RRF uses rank positions, not raw scores, to fuse results.
核心概念：RRF使用排名位置而非原始分数来融合结果。
"""
from typing import List, Dict, Any, Optional
from collections import defaultdict
import math
def reciprocal_rank_fusion(
ranked_lists: List[List[Dict[str, Any]]],
k: int = 60,
id_key: str = "id"
) -> List[Dict[str, Any]]:
"""
Merge multiple ranked result lists using Reciprocal Rank Fusion.
使用倒数排名融合合并多个排名结果列表。
Args:
ranked_lists: List of ranked result lists from different search systems.
每个元素是一个排名列表，来自不同的检索系统。
Each list is sorted by relevance descending (best first).
每个列表按相关性降序排列（最好的在前）。
k: Smoothing constant (default 60). Prevents high ranks from dominating.
平滑常数（默认60）。防止高排名过度主导。
id_key: The key used to identify documents in each result dict.
每个结果字典中用于标识文档的键名。
Returns:
A single merged and ranked list of results.
一个合并并排序后的结果列表。
"""
# Step 1: Collect all unique documents across all lists
# 第一步：收集所有列表中出现的所有唯一文档
# We use a dict to store each document's accumulated RRF score.
# 使用字典存储每个文档累积的RRF分数。
rrf_scores: Dict[str, float] = defaultdict(float)
# We also store the full document data for each ID.
# 同时为每个ID存储完整的文档数据。
doc_data: Dict[str, Dict[str, Any]] = {}
# Step 2: Iterate through each ranked list
# 第二步：遍历每个排名列表
for list_idx, ranked_list in enumerate(ranked_lists):
# Iterate through each document in this list with its rank position
# 遍历该列表中的每个文档及其排名位置
# rank starts at 1 (1 = best / highest relevance)
# rank从1开始（1 = 最好 / 相关性最高）
for rank, doc in enumerate(ranked_list, start=1):
# Extract the document ID
# 提取文档ID
doc_id = doc.get(id_key)
if doc_id is None:
# Skip documents without an ID
# 跳过没有ID的文档
continue
# Store the full document data (first time we see it)
# 存储完整的文档数据（首次遇到时）
if doc_id not in doc_data:
doc_data[doc_id] = doc
# Calculate reciprocal rank score for this document in this list
# 计算该文档在此列表中的倒数排名分数
# Formula: score = 1 / (rank + k)
# 公式: score = 1 / (rank + k)
reciprocal_score = 1.0 / (rank + k)
# Accumulate the score across all lists
# 在所有列表中累积分数
rrf_scores[doc_id] += reciprocal_score
# Optional: Store which lists found this document (for debugging)
# 可选：存储哪些列表找到了该文档（用于调试）
if "found_in_lists" not in doc_data[doc_id]:
doc_data[doc_id]["found_in_lists"] = []
doc_data[doc_id]["found_in_lists"].append(list_idx)
# Step 3: Build final results with RRF scores
# 第三步：构建带RRF分数的最终结果
results = []
for doc_id, score in rrf_scores.items():
# Copy the document data and add the RRF score
# 复制文档数据并添加RRF分数
result = doc_data[doc_id].copy()
result["rrf_score"] = score
# Add consensus information: how many lists found this doc
# 添加共识信息：多少个列表找到了该文档
result["consensus_count"] = len(result.get("found_in_lists", []))
results.append(result)
# Step 4: Sort by RRF score descending (highest score first)
# 第四步：按RRF分数降序排列（最高分在前）
results.sort(key=lambda x: x["rrf_score"], reverse=True)
return results
# ============================================================
# Example Usage with Simulated Search Results
# 带模拟搜索结果的示例用法
# ============================================================
def run_rrf_demo():
"""
Demonstrate RRF with simulated BM25 and vector search results.
用模拟的BM25和向量检索结果演示RRF。
"""
# Simulated BM25 search results (keyword-based)
# 模拟BM25检索结果（基于关键词）
# Each dict has: id, title, and bm25_score
# 每个字典包含: id, title, 和 bm25_score
bm25_results = [
{"id": "doc_1", "title": "Python List Comprehension Guide", "bm25_score": 12.5},
{"id": "doc_2", "title": "Python Programming Basics", "bm25_score": 10.2},
{"id": "doc_3", "title": "Advanced Python Techniques", "bm25_score": 8.7},
{"id": "doc_4", "title": "Data Science with Python", "bm25_score": 6.1},
{"id": "doc_5", "title": "Machine Learning 101", "bm25_score": 4.3},
]
# Simulated Vector search results (semantic-based)
# 模拟向量检索结果（基于语义）
# Each dict has: id, title, and vector_similarity
# 每个字典包含: id, title, 和 vector_similarity
vector_results = [
{"id": "doc_3", "title": "Advanced Python Techniques", "vector_similarity": 0.92},
{"id": "doc_1", "title": "Python List Comprehension Guide", "vector_similarity": 0.88},
{"id": "doc_6", "title": "Functional Programming in Python", "vector_similarity": 0.85},
{"id": "doc_7", "title": "Python Decorators Deep Dive", "vector_similarity": 0.79},
{"id": "doc_2", "title": "Python Programming Basics", "vector_similarity": 0.72},
]
print("=" * 60)
print("BM25 Results (Keyword Search) / BM25结果（关键词检索）")
print("=" * 60)
for i, doc in enumerate(bm25_results, 1):
print(f" {i}. {doc['title']} (score: {doc['bm25_score']})")
print("\n" + "=" * 60)
print("Vector Results (Semantic Search) / 向量结果（语义检索）")
print("=" * 60)
for i, doc in enumerate(vector_results, 1):
print(f" {i}. {doc['title']} (similarity: {doc['vector_similarity']})")
# Fuse the two lists using RRF
# 使用RRF融合两个列表
print("\n" + "=" * 60)
print("Fusing with RRF (k=60) / 使用RRF融合 (k=60)")
print("=" * 60)
fused_results = reciprocal_rank_fusion(
ranked_lists=[bm25_results, vector_results],
k=60,
id_key="id"
)
# Display fused results
# 显示融合结果
print("\nFinal RRF-Ranked Results / 最终RRF排名结果:")
print("-" * 60)
for i, doc in enumerate(fused_results, 1):
# Show which lists found this document (consensus)
# 显示哪些列表找到了该文档（共识）
found_in = doc.get("found_in_lists", [])
found_desc = "BM25" if 0 in found_in else ""
if 1 in found_in:
found_desc += " + Vector" if found_desc else "Vector"
if len(found_in) == 2:
found_desc += " ✓ CONSENSUS (both found this!)"
found_desc += " ✓ 共识（两种检索都找到了！）"
print(f" {i}. {doc['title']}")
print(f" RRF Score: {doc['rrf_score']:.6f}")
print(f" Found by: {found_desc}")
print()
if __name__ == "__main__":
run_rrf_demo()Complete Hybrid Search with rank_bm25 + sentence-transformers
"""
Complete Hybrid Search Implementation with RRF Fusion
使用RRF融合的完整混合检索实现
This implementation combines:
1. BM25 keyword search (using rank_bm25 library)
2. Dense vector semantic search (using sentence-transformers)
3. RRF fusion to merge results
本实现结合了：
1. BM25关键词检索（使用rank_bm25库）
2. 稠密向量语义检索（使用sentence-transformers）
3. RRF融合来合并结果
"""
from typing import List, Dict, Any, Optional, Tuple
import numpy as np
from rank_bm25 import BM25Okapi
from sentence_transformers import SentenceTransformer
import math
class HybridRetriever:
"""
Hybrid retriever combining BM25 and vector search with RRF fusion.
结合BM25和向量检索并使用RRF融合的混合检索器。
This class manages:
- Document corpus storage
- BM25 index building and searching
- Vector embedding generation and similarity search
- RRF fusion of results from both systems
该类管理：
- 文档语料库存储
- BM25索引构建和检索
- 向量嵌入生成和相似度检索
- 两种系统结果的RRF融合
"""
def __init__(
self,
documents: List[str],
doc_metadata: Optional[List[Dict[str, Any]]] = None,
embedding_model_name: str = "all-MiniLM-L6-v2",
rrf_k: int = 60
):
"""
Initialize the hybrid retriever with documents.
用文档初始化混合检索器。
Args:
documents: List of document text strings.
文档文本字符串列表。
doc_metadata: Optional metadata for each document (e.g., titles, IDs).
每个文档的可选元数据（如标题、ID）。
embedding_model_name: Name of the sentence-transformers model to use.
要使用的sentence-transformers模型名称。
rrf_k: RRF smoothing constant (default 60).
RRF平滑常数（默认60）。
"""
# Store documents and metadata
# 存储文档和元数据
self.documents = documents
self.doc_metadata = doc_metadata or [{"id": i} for i in range(len(documents))]
# RRF constant / RRF常数
self.rrf_k = rrf_k
# Step 1: Build BM25 index
# 第一步：构建BM25索引
# Tokenize each document into words (lowercase, split on whitespace)
# 将每个文档分词为单词（小写，按空格分割）
# In production, you'd use a proper tokenizer (e.g., NLTK, spaCy)
# 生产环境中应使用专业分词器（如NLTK、spaCy）
tokenized_docs = [doc.lower().split() for doc in documents]
self.bm25 = BM25Okapi(tokenized_docs)
# Step 2: Initialize embedding model and generate embeddings
# 第二步：初始化嵌入模型并生成嵌入向量
self.embedding_model = SentenceTransformer(embedding_model_name)
# Generate embeddings for all documents / 为所有文档生成嵌入
# This may take time for large corpora / 大规模语料可能需要较长时间
self.doc_embeddings = self.embedding_model.encode(
documents,
convert_to_numpy=True,
show_progress_bar=True
)
print(f"Initialized HybridRetriever with {len(documents)} documents")
print(f"使用 {len(documents)} 个文档初始化混合检索器")
print(f"Embedding dimension: {self.doc_embeddings.shape[1]}")
print(f"嵌入维度: {self.doc_embeddings.shape[1]}")
def bm25_search(
self,
query: str,
top_k: int = 10
) -> List[Dict[str, Any]]:
"""
Perform BM25 keyword search.
执行BM25关键词检索。
Args:
query: Search query string.
检索查询字符串。
top_k: Number of top results to return.
返回的top结果数量。
Returns:
List of results with document content, metadata, and BM25 score.
包含文档内容、元数据和BM25分数的结果列表。
"""
# Tokenize the query / 对查询进行分词
tokenized_query = query.lower().split()
# Get BM25 scores for all documents / 获取所有文档的BM25分数
bm25_scores = self.bm25.get_scores(tokenized_query)
# Get top-k indices sorted by score descending / 按分数降序获取top-k索引
# argsort sorts ascending, so we reverse / argsort升序排列，所以反转
top_indices = np.argsort(bm25_scores)[::-1][:top_k]
results = []
for idx in top_indices:
# Skip documents with zero score / 跳过零分文档
if bm25_scores[idx] <= 0:
continue
results.append({
"id": self.doc_metadata[idx].get("id", idx),
"content": self.documents[idx],
"metadata": self.doc_metadata[idx],
"bm25_score": float(bm25_scores[idx]),
"index": idx
})
return results
def vector_search(
self,
query: str,
top_k: int = 10
) -> List[Dict[str, Any]]:
"""
Perform dense vector semantic search.
执行稠密向量语义检索。
Args:
query: Search query string.
检索查询字符串。
top_k: Number of top results to return.
返回的top结果数量。
Returns:
List of results with document content, metadata, and similarity score.
包含文档内容、元数据和相似度分数的结果列表。
"""
# Generate embedding for the query / 为查询生成嵌入
query_embedding = self.embedding_model.encode(
[query],
convert_to_numpy=True
)[0]
# Compute cosine similarity between query and all documents
# 计算查询与所有文档之间的余弦相似度
# Normalize embeddings for cosine similarity / 归一化嵌入以计算余弦相似度
query_norm = query_embedding / np.linalg.norm(query_embedding)
doc_norms = self.doc_embeddings / np.linalg.norm(
self.doc_embeddings,
axis=1,
keepdims=True
)
similarities = np.dot(doc_norms, query_norm)
# Get top-k indices sorted by similarity descending
# 按相似度降序获取top-k索引
top_indices = np.argsort(similarities)[::-1][:top_k]
results = []
for idx in top_indices:
results.append({
"id": self.doc_metadata[idx].get("id", idx),
"content": self.documents[idx],
"metadata": self.doc_metadata[idx],
"vector_similarity": float(similarities[idx]),
"index": idx
})
return results
def hybrid_search(
self,
query: str,
top_k: int = 10,
bm25_top_k: int = 50,
vector_top_k: int = 50
) -> List[Dict[str, Any]]:
"""
Perform hybrid search: BM25 + Vector + RRF fusion.
执行混合检索：BM25 + 向量 + RRF融合。
This is the main method to use for production.
这是生产环境中使用的主方法。
Args:
query: Search query string.
检索查询字符串。
bm25_top_k: Number of results to fetch from BM25 before fusion.
融合前从BM25获取的结果数量。
vector_top_k: Number of results to fetch from vector search before fusion.
融合前从向量检索获取的结果数量。
top_k: Final number of results to return after RRF fusion.
RRF融合后返回的最终结果数量。
Returns:
Final ranked list of results after RRF fusion.
RRF融合后的最终排名结果列表。
"""
# Step 1: Execute both searches in parallel (conceptually)
# 第一步：并行执行两种检索（概念上）
# In production, you could use asyncio.gather() for true parallelism
# 生产环境中可使用asyncio.gather()实现真正的并行
bm25_results = self.bm25_search(query, top_k=bm25_top_k)
vector_results = self.vector_search(query, top_k=vector_top_k)
# Step 2: Apply RRF fusion
# 第二步：应用RRF融合
# We need to pass lists with consistent ID keys
# 需要传递具有一致ID键的列表
fused = self._rrf_fusion(
ranked_lists=[bm25_results, vector_results],
id_key="id"
)
# Return top-k after fusion / 返回融合后的top-k
return fused[:top_k]
def _rrf_fusion(
self,
ranked_lists: List[List[Dict[str, Any]]],
id_key: str = "id"
) -> List[Dict[str, Any]]:
"""
Internal RRF fusion implementation.
内部RRF融合实现。
This is the core fusion algorithm used by hybrid_search.
这是hybrid_search使用的核心融合算法。
"""
# Dictionary to accumulate RRF scores for each document
# 用于累积每个文档RRF分数的字典
rrf_scores: Dict[str, float] = {}
# Dictionary to store full document data
# 用于存储完整文档数据的字典
doc_data: Dict[str, Dict[str, Any]] = {}
# Iterate through each ranked list
# 遍历每个排名列表
for list_idx, ranked_list in enumerate(ranked_lists):
# rank starts at 1 / rank从1开始
for rank, doc in enumerate(ranked_list, start=1):
doc_id = str(doc.get(id_key))
if doc_id is None:
continue
# Store document data first time we see it
# 首次遇到时存储文档数据
if doc_id not in doc_data:
doc_data[doc_id] = doc.copy()
doc_data[doc_id]["found_in"] = []
# Calculate reciprocal score / 计算倒数分数
# RRF formula: score = 1 / (rank + k)
# RRF公式: score = 1 / (rank + k)
reciprocal_score = 1.0 / (rank + self.rrf_k)
rrf_scores[doc_id] = rrf_scores.get(doc_id, 0.0) + reciprocal_score
# Track which list found this document / 追踪哪个列表找到了该文档
doc_data[doc_id]["found_in"].append(list_idx)
# Build final results with RRF scores
# 构建带RRF分数的最终结果
results = []
for doc_id, score in rrf_scores.items():
result = doc_data[doc_id].copy()
result["rrf_score"] = score
# Consensus = found in both lists (0 and 1)
# 共识 = 在两个列表中都找到（0和1）
result["consensus"] = len(set(result["found_in"])) == len(ranked_lists)
results.append(result)
# Sort by RRF score descending / 按RRF分数降序排列
results.sort(key=lambda x: x["rrf_score"], reverse=True)
return results
# ============================================================
# Example Usage / 示例用法
# ============================================================
def run_hybrid_search_demo():
"""
Demonstrate the complete hybrid search system.
演示完整的混合检索系统。
"""
# Sample documents / 示例文档
documents = [
"Python list comprehension is a concise way to create lists.",
"Python programming basics include variables, loops, and functions.",
"Advanced Python techniques include decorators, generators, and context managers.",
"Data science with Python uses pandas, numpy, and matplotlib.",
"Machine learning basics cover supervised and unsupervised learning.",
"Functional programming in Python uses map, filter, and reduce.",
"Python decorators are functions that modify other functions.",
"Deep learning with Python uses frameworks like TensorFlow and PyTorch.",
"Natural language processing with Python uses NLTK and spaCy.",
"Web development with Python uses Django and Flask.",
]
# Metadata for each document (titles) / 每个文档的元数据（标题）
metadata = [
{"id": f"doc_{i}", "title": f"Document {i}"}
for i in range(len(documents))
]
# Initialize hybrid retriever / 初始化混合检索器
retriever = HybridRetriever(
documents=documents,
doc_metadata=metadata,
embedding_model_name="all-MiniLM-L6-v2",
rrf_k=60
)
# Test query / 测试查询
query = "Python list creation methods"
print("=" * 70)
print(f"Query: '{query}'")
print(f"查询: '{query}'")
print("=" * 70)
# Get hybrid search results / 获取混合检索结果
results = retriever.hybrid_search(
query=query,
top_k=5,
bm25_top_k=10,
vector_top_k=10
)
print("\nFinal Hybrid Search Results (RRF Fused) / 最终混合检索结果（RRF融合）:")
print("-" * 70)
for i, result in enumerate(results, 1):
title = result.get("metadata", {}).get("title", "Unknown")
consensus = "✓ CONSENSUS" if result.get("consensus", False) else ""
print(f"{i}. {title}")
print(f" Content: {result['content'][:60]}...")
print(f" RRF Score: {result['rrf_score']:.6f} {consensus}")
print()
if __name__ == "__main__":
run_hybrid_search_demo()Key Takeaways
| 要点 (Key Point) | EN | CN |
|---|---|---|
| RRF使用排名而非分数 | RRF uses rank positions, not raw scores | RRF使用排名位置，而非原始分数 |
| 公式：score = 1/(rank + k) | Formula: score = 1/(rank + k) | 公式：score = 1/(rank + k) |
| k默认值为60 | Default k value is 60 | k默认值为60 |
| 处理不同量纲的分数 | Handles scores on different scales | 处理不同量纲的分数 |
| 共识效应：两个列表都排名高者得分更高 | Consensus effect: documents high in both lists get higher scores | 共识效应：在两个列表中都排名高的文档得分更高 |
| 无需分数归一化 | No score normalization needed | 无需分数归一化 |
| 并行检索后融合 | Parallel retrieval then fusion | 并行检索后融合 |
| 对两种检索方法一视同仁 | Treats both retrieval methods equally | 对两种检索方法一视同仁 |