基于加權RRF的混合檢索與RAG系統融合方法詳解
- 概述
- RRF基本原理
傳統RRF公式
- 改進的加權得分RRF方法
1. 直接得分融合法
2. 標準化得分融合法
- 完整示例演示
- 公式變量總結表
- 權重策略設計
基于算法特性的權重分配
- 參數調優建議
常數k的選擇示例
方法選擇指南
- 總結
- 完整的代碼示例：
概述
在信息檢索領域，混合檢索（Hybrid Search）通過結合多種搜索算法的優勢，提供比單一算法更準確、更全面的搜索結果。倒數排名融合（Reciprocal Rank Fusion, RRF）是一種簡單而有效的混合檢索結果融合方法，特別適用于結合關鍵詞搜索（如BM25）和向量搜索等不同檢索技術。
RRF基本原理
傳統RRF公式
傳統RRF方法的核心公式為：
RRF_score(d) = Σ(1 / (k + rank_i(d)))變量解釋：
- ?
?d??：文檔（Document），表示搜索結果中的單個文檔 - ?
?rank_i(d)??：文檔d在第i個排序列表中的排名（Rank），從1開始計數 - ?
?k??：調和常數（通常取60），防止排名靠后的文檔得分過小 - ?
?Σ??：求和符號，表示對所有排序列表的得分進行累加
Python實現：
def traditional_rrf(ranked_lists, k=60):
scores = {}
for rank_list in ranked_lists:
for rank, doc_id in enumerate(rank_list, 1): # rank從1開始
if doc_id not in scores:
scores[doc_id] = 0
scores[doc_id] += 1 / (k + rank) # 核心RRF公式
return sorted(scores.items(), key=lambda x: x[1], reverse=True)改進的加權得分RRF方法
在實際應用中，我們往往需要更精細的控制，因此提出了兩種改進的加權RRF方法：
1. 直接得分融合法
公式：
score(d) = Σ(weight_i * original_score(d) / (k + rank_i(d)))變量解釋：
- ?
?original_score(d)??：文檔d在原始搜索算法中的得分 - ?
?weight_i??：第i個搜索算法的權重，用于調整不同算法的重要性 - ?
?k??：RRF常數，控制排名影響的強度 - ?
?rank_i(d)??：文檔d在第i個算法結果中的排名
Python實現：
def direct_score_fusion(ranked_lists_with_scores, weights, k=60):
"""
ranked_lists_with_scores: [
[{"doc_id": "doc1", "score": 0.95}, ...], # BM25結果
[{"doc_id": "doc2", "score": 0.88}, ...] # 向量搜索結果
]
weights: [0.6, 0.4] # 對應每個算法的權重
"""
scores = {}
for weight, score_list in zip(weights, ranked_lists_with_scores):
# 按得分排序獲取排名
sorted_list = sorted(score_list, key=lambda x: x["score"], reverse=True)
for rank, item in enumerate(sorted_list, 1):
doc_id = item["doc_id"]
original_score = item["score"]
if doc_id not in scores:
scores[doc_id] = 0
# 直接得分融合公式
scores[doc_id] += weight * (original_score / (k + rank))
return sorted(scores.items(), key=lambda x: x[1], reverse=True)2. 標準化得分融合法
標準化公式：
normalized_score(d) = (original_score(d) - min_score) / (max_score - min_score)融合公式：
score(d) = Σ(weight_i * normalized_score(d) / (k + rank_i(d)))變量解釋：
- ?
?normalized_score(d)??：標準化后的得分，范圍[0, 1] - ?
?min_score??：當前算法結果中的最低得分 - ?
?max_score??：當前算法結果中的最高得分 - ?
?original_score(d)??：原始得分 - 其他變量含義同上
Python實現：
def normalized_score_fusion(ranked_lists_with_scores, weights, k=60):
scores = {}
for weight, score_list in zip(weights, ranked_lists_with_scores):
# 第一步：得分標準化
normalized_list = normalize_scores(score_list)
# 第二步：按標準化得分排序獲取排名
sorted_list = sorted(normalized_list, key=lambda x: x["normalized_score"], reverse=True)
for rank, item in enumerate(sorted_list, 1):
doc_id = item["doc_id"]
normalized_score = item["normalized_score"]
if doc_id not in scores:
scores[doc_id] = 0
# 標準化得分融合公式
scores[doc_id] += weight * (normalized_score / (k + rank))
return sorted(scores.items(), key=lambda x: x[1], reverse=True)
def normalize_scores(score_list):
"""得分標準化函數"""
if not score_list:
return []
scores = [item["score"] for item in score_list]
min_score = min(scores)
max_score = max(scores)
# 處理所有得分相同的情況
if max_score == min_score:
return [{"doc_id": item["doc_id"], "normalized_score": 1.0}
for item in score_list]
normalized_list = []
for item in score_list:
# 最小-最大標準化公式
normalized_score = (item["score"] - min_score) / (max_score - min_score)
normalized_list.append({
"doc_id": item["doc_id"],
"normalized_score": normalized_score
})
return normalized_list完整示例演示
# 模擬數據
bm25_results = [
{"doc_id": "doc1", "score": 0.95, "content": "機器學習教程"},
{"doc_id": "doc3", "score": 0.85, "content": "深度學習指南"},
{"doc_id": "doc2", "score": 0.80, "content": "Python數據分析"}
]
vector_results = [
{"doc_id": "doc2", "score": 0.88, "content": "Python數據分析"},
{"doc_id": "doc1", "score": 0.82, "content": "機器學習教程"},
{"doc_id": "doc4", "score": 0.65, "content": "數據科學基礎"}
]
# 應用融合方法
ranked_lists = [bm25_results, vector_results]
weights = [0.6, 0.4]
print("直接得分融合結果:")
direct_results = direct_score_fusion(ranked_lists, weights, k=60)
for doc_id, score in direct_results:
print(f" {doc_id}: {score:.4f}")
print("\n標準化得分融合結果:")
normalized_results = normalized_score_fusion(ranked_lists, weights, k=60)
for doc_id, score in normalized_results:
print(f" {doc_id}: {score:.4f}")公式變量總結表
變量名 | 含義 | 取值范圍 | 說明 |
? | 文檔 | 任意文檔ID | 搜索結果的單個文檔 |
? | 排名 | 1, 2, 3, ... | 文檔在第i個列表中的位置 |
? | RRF常數 | 通常60 | 調節排名影響的強度 |
? | 原始得分 | 算法相關 | 搜索算法的原始評分 |
? | 標準化得分 | [0, 1] | 歸一化后的得分 |
? | 算法權重 | [0, 1] | 第i個算法的重要性 |
? | 最小得分 | 算法相關 | 當前算法結果的最低分 |
? | 最大得分 | 算法相關 | 當前算法結果的最高分 |
權重策略設計
基于算法特性的權重分配
- BM25主導策略（權重：[0.6, 0.4]）
ounter(lineounter(line
# 適用于關鍵詞敏感的場景
weights = [0.6, 0.4] # [BM25權重, 向量搜索權重]- 向量搜索主導策略（權重：[0.4, 0.6]）
ounter(lineounter(line
# 適用于語義理解重要的場景
weights = [0.4, 0.6] # [BM25權重, 向量搜索權重]- 平衡策略（權重：[0.5, 0.5]）
ounter(lineounter(line
# 適用于通用搜索場景
weights = [0.5, 0.5] # 平等對待兩種算法參數調優建議
常數k的選擇示例
# 不同k值的比較
k_values = [30, 60, 100]
for k in k_values:
results = direct_score_fusion(ranked_lists, weights, k=k)
print(f"k={k} 時的top結果:")
for doc_id, score in results[:3]:
print(f" {doc_id}: {score:.4f}")方法選擇指南
場景特征 | 推薦方法 | 代碼示例 |
得分尺度一致 | 直接得分融合 | ? |
得分尺度差異大 | 標準化得分融合 | ? |
結果穩定性要求高 | 標準化得分融合 | ? |
總結
通過清晰的變量解釋和對應的Python代碼實現，我們可以看到加權得分RRF方法為混合檢索提供了靈活而有效的解決方案。關鍵是根據具體需求選擇合適的融合方法、合理設置權重參數，并理解每個變量在公式中的作用。
在實際應用中，建議先進行小規模實驗，比較不同配置的效果：
# 實驗不同配置
configs = [
{"method": "direct", "weights": [0.6, 0.4], "k": 60},
{"method": "normalized", "weights": [0.5, 0.5], "k": 60},
{"method": "direct", "weights": [0.7, 0.3], "k": 30}
]
for config in configs:
# 測試每種配置的效果
pass通過系統的實驗和優化，可以找到最適合具體應用場景的RRF融合方案。
完整的代碼示例：
from typing import List, Dict, Any
import numpy as np
class WeightedScoreRRF:
def __init__(self, k: int = 60):
"""
初始化加權得分RRF
Args:
k: 常數，用于調整排名的影響，通常取60
"""
self.k = k
def direct_score_fusion(self,
ranked_lists_with_scores: List[List[Dict]],
weights: List[float] = None) -> List[Dict[str, Any]]:
"""
直接使用原始得分進行加權融合
Args:
ranked_lists_with_scores: 每個排序結果包含文檔ID和原始得分
weights: 每個搜索算法的權重
Returns:
融合后的排序結果
"""
if weights is None:
weights = [1.0] * len(ranked_lists_with_scores)
if len(weights) != len(ranked_lists_with_scores):
raise ValueError("權重數量必須與排序列表數量一致")
scores = {}
for weight, score_list in zip(weights, ranked_lists_with_scores):
# 按原始得分排序獲取排名
sorted_list = sorted(score_list, key=lambda x: x["score"], reverse=True)
for rank, item in enumerate(sorted_list, 1):
doc_id = item["doc_id"]
original_score = item["score"]
if doc_id not in scores:
scores[doc_id] = 0
# 直接使用原始得分 + 權重
scores[doc_id] += weight * (original_score / (self.k + rank))
# 按得分降序排序
sorted_results = sorted(
[{"doc_id": doc_id, "score": score}
for doc_id, score in scores.items()],
key=lambda x: x["score"],
reverse=True
)
return sorted_results
def normalized_score_fusion(self,
ranked_lists_with_scores: List[List[Dict]],
weights: List[float] = None) -> List[Dict[str, Any]]:
"""
使用標準化得分進行加權融合
Args:
ranked_lists_with_scores: 每個排序結果包含文檔ID和原始得分
weights: 每個搜索算法的權重
Returns:
融合后的排序結果
"""
if weights is None:
weights = [1.0] * len(ranked_lists_with_scores)
if len(weights) != len(ranked_lists_with_scores):
raise ValueError("權重數量必須與排序列表數量一致")
scores = {}
for weight, score_list in zip(weights, ranked_lists_with_scores):
# 標準化得分
normalized_list = self._normalize_scores(score_list)
# 按標準化得分排序獲取排名
sorted_list = sorted(normalized_list, key=lambda x: x["normalized_score"], reverse=True)
for rank, item in enumerate(sorted_list, 1):
doc_id = item["doc_id"]
normalized_score = item["normalized_score"]
if doc_id not in scores:
scores[doc_id] = 0
# 使用標準化得分 + 權重
scores[doc_id] += weight * (normalized_score / (self.k + rank))
# 按得分降序排序
sorted_results = sorted(
[{"doc_id": doc_id, "score": score}
for doc_id, score in scores.items()],
key=lambda x: x["score"],
reverse=True
)
return sorted_results
def _normalize_scores(self, score_list: List[Dict]) -> List[Dict]:
"""
對得分進行最小-最大標準化處理
Args:
score_list: 原始得分列表
Returns:
標準化后的得分列表
"""
if not score_list:
return []
scores = [item["score"] for item in score_list]
min_score = min(scores)
max_score = max(scores)
# 避免除零
if max_score == min_score:
return [{"doc_id": item["doc_id"], "normalized_score": 1.0}
for item in score_list]
normalized_list = []
for item in score_list:
# 最小-最大標準化到 [0, 1] 范圍
normalized_score = (item["score"] - min_score) / (max_score - min_score)
normalized_list.append({
"doc_id": item["doc_id"],
"normalized_score": normalized_score
})
return normalized_list
class HybridSearchSystem:
"""混合搜索系統"""
def __init__(self):
self.rrf = WeightedScoreRRF(k=60)
def bm25_search(self, query: str, top_k: int = 5) -> List[Dict]:
"""模擬BM25搜索"""
# 模擬BM25搜索結果 - 得分范圍 [0.6, 1.0]
return [
{"doc_id": "doc1", "score": 0.95, "title": "機器學習基礎教程", "content": "這是關于機器學習的基礎教程..."},
{"doc_id": "doc3", "score": 0.85, "title": "深度學習實踐", "content": "深度學習在圖像識別中的應用..."},
{"doc_id": "doc2", "score": 0.80, "title": "Python數據分析", "content": "使用Python進行數據分析的方法..."},
{"doc_id": "doc5", "score": 0.75, "title": "神經網絡原理", "content": "神經網絡的基本原理和結構..."},
{"doc_id": "doc4", "score": 0.70, "title": "統計學習方法", "content": "統計學習的基本概念和方法..."}
][:top_k]
def vector_search(self, query: str, top_k: int = 5) -> List[Dict]:
"""模擬向量搜索"""
# 模擬向量搜索結果 - 得分范圍 [0.3, 0.9] (不同的得分尺度)
return [
{"doc_id": "doc2", "score": 0.88, "title": "Python數據分析", "content": "使用Python進行數據分析的方法..."},
{"doc_id": "doc1", "score": 0.82, "title": "機器學習基礎教程", "content": "這是關于機器學習的基礎教程..."},
{"doc_id": "doc4", "score": 0.78, "title": "統計學習方法", "content": "統計學習的基本概念和方法..."},
{"doc_id": "doc6", "score": 0.65, "title": "數據挖掘技術", "content": "數據挖掘的基本技術和應用..."},
{"doc_id": "doc3", "score": 0.60, "title": "深度學習實踐", "content": "深度學習在圖像識別中的應用..."}
][:top_k]
def hybrid_search(self,
query: str,
method: str = "direct",
weights: List[float] = None) -> List[Dict[str, Any]]:
"""
執行混合搜索
Args:
query: 搜索查詢
method: 融合方法，'direct' 或 'normalized'
weights: 各搜索算法的權重
Returns:
融合后的搜索結果
"""
# 執行搜索
bm25_results = self.bm25_search(query)
vector_results = self.vector_search(query)
ranked_lists = [bm25_results, vector_results]
# 設置默認權重
if weights is None:
weights = [0.6, 0.4] # BM25權重更高
# 選擇融合方法
if method == "direct":
return self.rrf.direct_score_fusion(ranked_lists, weights)
elif method == "normalized":
return self.rrf.normalized_score_fusion(ranked_lists, weights)
else:
raise ValueError("方法必須是 'direct' 或 'normalized'")
def compare_methods(query: str = "機器學習教程"):
"""比較不同融合方法的效果"""
search_system = HybridSearchSystem()
# query = "機器學習教程"
print("=" * 60)
print(f"查詢: '{query}'")
print("=" * 60)
# 測試不同權重配置
weight_configs = [
([0.5, 0.5], "等權重"),
([0.6, 0.4], "BM25權重更高"),
([0.4, 0.6], "向量搜索權重更高"),
([0.8, 0.2], "主要依賴BM25")
]
for weights, desc in weight_configs:
print(f"\n?? 權重配置: {desc} {weights}")
print("-" * 40)
# 直接得分融合
direct_results = search_system.hybrid_search(
query, method="direct", weights=weights
)
# 標準化得分融合
normalized_results = search_system.hybrid_search(
query, method="normalized", weights=weights
)
print("直接得分融合結果:")
for i, result in enumerate(direct_results, 1):
print(f" {i}. {result['doc_id']} (得分: {result['score']:.4f})")
print("標準化得分融合結果:")
for i, result in enumerate(normalized_results, 1):
print(f" {i}. {result['doc_id']} (得分: {result['score']:.4f})")
if __name__ == "__main__":
# 運行比較不同權重的影響
query = "機器學習教程"
compare_methods(query=query)運行結果：
============================================================
查詢: '機器學習教程'
============================================================
?? 權重配置: 等權重 [0.5, 0.5]
----------------------------------------
直接得分融合結果:
1. doc1 (得分: 0.0144)
2. doc2 (得分: 0.0136)
3. doc4 (得分: 0.0116)
4. doc3 (得分: 0.0115)
5. doc5 (得分: 0.0059)
6. doc6 (得分: 0.0051)
標準化得分融合結果:
1. doc1 (得分: 0.0145)
2. doc2 (得分: 0.0114)
3. doc4 (得分: 0.0051)
4. doc3 (得分: 0.0048)
5. doc5 (得分: 0.0016)
6. doc6 (得分: 0.0014)
?? 權重配置: BM25權重更高 [0.6, 0.4]
----------------------------------------
直接得分融合結果:
1. doc1 (得分: 0.0146)
2. doc2 (得分: 0.0134)
3. doc3 (得分: 0.0119)
4. doc4 (得分: 0.0114)
5. doc5 (得分: 0.0070)
6. doc6 (得分: 0.0041)
標準化得分融合結果:
1. doc1 (得分: 0.0149)
2. doc2 (得分: 0.0104)
3. doc3 (得分: 0.0058)
4. doc4 (得分: 0.0041)
5. doc5 (得分: 0.0019)
6. doc6 (得分: 0.0011)
?? 權重配置: 向量搜索權重更高 [0.4, 0.6]
----------------------------------------
直接得分融合結果:
1. doc1 (得分: 0.0142)
2. doc2 (得分: 0.0137)
3. doc4 (得分: 0.0117)
4. doc3 (得分: 0.0110)
5. doc6 (得分: 0.0061)
6. doc5 (得分: 0.0047)
標準化得分融合結果:
1. doc1 (得分: 0.0142)
2. doc2 (得分: 0.0124)
3. doc4 (得分: 0.0061)
4. doc3 (得分: 0.0039)
5. doc6 (得分: 0.0017)
6. doc5 (得分: 0.0013)
?? 權重配置: 主要依賴BM25 [0.8, 0.2]
----------------------------------------
直接得分融合結果:
1. doc1 (得分: 0.0151)
2. doc2 (得分: 0.0130)
3. doc3 (得分: 0.0128)
4. doc4 (得分: 0.0111)
5. doc5 (得分: 0.0094)
6. doc6 (得分: 0.0020)
標準化得分融合結果:
1. doc1 (得分: 0.0156)
2. doc2 (得分: 0.0084)
3. doc3 (得分: 0.0077)
4. doc5 (得分: 0.0025)
5. doc4 (得分: 0.0020)
6. doc6 (得分: 0.0006)本文轉載自??AI小新??，作者：AI小新
已于2025-10-28 07:49:48修改
贊
收藏
回復
分享
微博
QQ
微信
舉報
回復
相關推薦
