检索增强生成中的混合检索策略稠密检索与稀疏检索的融合方案一、单一检索的召回盲区稠密检索与稀疏检索的互补性RAG 系统的检索质量直接决定生成质量。当前主流的稠密检索Dense Retrieval通过 Embedding 向量计算语义相似度擅长捕捉语义关联但容易遗漏关键词精确匹配的文档。例如用户查询PyTorch 2.0 compile 优化稠密检索可能返回语义相近但版本不对的文档。稀疏检索Sparse Retrieval如 BM25基于词频统计擅长关键词精确匹配但无法理解语义。用户查询深度学习框架性能优化稀疏检索可能遗漏只包含PyTorch 速度提升的文档因为词汇不重叠。混合检索Hybrid Retrieval将两种检索方式的结果融合取长补短。但融合策略的选择——如何加权、如何去重、如何排序——直接影响最终效果。二、混合检索的架构设计从并行检索到结果融合flowchart TD A[用户查询] -- B[查询预处理] B -- C[稠密检索: Embedding 向量搜索] B -- D[稀疏检索: BM25 关键词搜索] C -- E[稠密结果: TopK_D] D -- F[稀疏结果: TopK_S] E -- G[分数归一化: Min-Max / Z-Score] F -- G G -- H[加权融合: α × Dense β × Sparse] H -- I[去重: 基于文档 ID] I -- J[重排序: Cross-Encoder 精排] J -- K[最终结果: TopN] subgraph 融合策略 L[Reciprocal Rank Fusion: 基于排名的融合] M[Linear Combination: 基于分数的融合] N[Learned Fusion: 基于模型的融合] end H -- L H -- M H -- N混合检索的核心挑战是分数归一化。稠密检索的余弦相似度范围是 [-1, 1]稀疏检索的 BM25 分数范围是 [0, ∞)两者不可直接比较。归一化策略的选择直接影响融合效果。三、生产级代码实现与最佳实践 混合检索引擎 融合稠密检索和稀疏检索的结果 from dataclasses import dataclass from typing import List, Optional import numpy as np from rank_bm25 import BM25Okapi from sentence_transformers import SentenceTransformer dataclass class RetrievalResult: 检索结果 doc_id: str content: str dense_score: float 0.0 sparse_score: float 0.0 combined_score: float 0.0 rank: int 0 class HybridRetriever: 混合检索器 支持多种融合策略自动归一化分数 def __init__( self, embedding_model: str BAAI/bge-large-zh-v1.5, dense_weight: float 0.6, sparse_weight: float 0.4, top_k: int 20, fusion_method: str rrf, ): self.encoder SentenceTransformer(embedding_model) self.dense_weight dense_weight self.sparse_weight sparse_weight self.top_k top_k self.fusion_method fusion_method # 稀疏检索索引 self.bm25: Optional[BM25Okapi] None self.doc_ids: List[str] [] self.doc_contents: List[str] [] # 稠密检索索引简化实现生产环境用 Milvus/Qdrant self.doc_embeddings: Optional[np.ndarray] None def index(self, doc_ids: List[str], contents: List[str]): 构建双索引 self.doc_ids doc_ids self.doc_contents contents # 稀疏索引BM25 tokenized [self._tokenize(text) for text in contents] self.bm25 BM25Okapi(tokenized) # 稠密索引Embedding self.doc_embeddings self.encoder.encode( contents, normalize_embeddingsTrue, show_progress_barTrue ) def search(self, query: str, top_n: int 10) - List[RetrievalResult]: 混合检索 # 稠密检索 query_embedding self.encoder.encode( [query], normalize_embeddingsTrue ) dense_scores np.dot(self.doc_embeddings, query_embedding.T).flatten() dense_top_indices np.argsort(dense_scores)[::-1][:self.top_k] # 稀疏检索 tokenized_query self._tokenize(query) sparse_scores self.bm25.get_scores(tokenized_query) sparse_top_indices np.argsort(sparse_scores)[::-1][:self.top_k] # 合并候选集去重 candidate_indices set(dense_top_indices) | set(sparse_top_indices) # 构建结果列表 results [] for idx in candidate_indices: results.append(RetrievalResult( doc_idself.doc_ids[idx], contentself.doc_contents[idx], dense_scorefloat(dense_scores[idx]), sparse_scorefloat(sparse_scores[idx]), )) # 融合策略 if self.fusion_method rrf: results self._rrf_fusion(results) elif self.fusion_method linear: results self._linear_fusion(results) else: raise ValueError(f未知融合策略: {self.fusion_method}) # 排序并返回 TopN results.sort(keylambda r: r.combined_score, reverseTrue) for i, r in enumerate(results[:top_n]): r.rank i 1 return results[:top_n] def _rrf_fusion(self, results: List[RetrievalResult], k: int 60) - List[RetrievalResult]: Reciprocal Rank Fusion 基于排名的融合不依赖原始分数对分数分布差异鲁棒 RRF(d) Σ 1/(k rank_i(d)) # 按稠密分数排名 by_dense sorted(results, keylambda r: r.dense_score, reverseTrue) # 按稀疏分数排名 by_sparse sorted(results, keylambda r: r.sparse_score, reverseTrue) dense_rank {r.doc_id: i 1 for i, r in enumerate(by_dense)} sparse_rank {r.doc_id: i 1 for i, r in enumerate(by_sparse)} for r in results: dr dense_rank.get(r.doc_id, len(results) 1) sr sparse_rank.get(r.doc_id, len(results) 1) r.combined_score 1.0 / (k dr) 1.0 / (k sr) return results def _linear_fusion(self, results: List[RetrievalResult]) - List[RetrievalResult]: 线性加权融合 需要先归一化分数否则量纲不同导致偏向一方 # Min-Max 归一化 dense_scores [r.dense_score for r in results] sparse_scores [r.sparse_score for r in results] norm_dense self._minmax_normalize(dense_scores) norm_sparse self._minmax_normalize(sparse_scores) for i, r in enumerate(results): r.combined_score ( self.dense_weight * norm_dense[i] self.sparse_weight * norm_sparse[i] ) return results staticmethod def _minmax_normalize(scores: List[float]) - List[float]: Min-Max 归一化到 [0, 1] if not scores: return [] min_s, max_s min(scores), max(scores) if max_s min_s: return [0.5] * len(scores) return [(s - min_s) / (max_s - min_s) for s in scores] staticmethod def _tokenize(text: str) - List[str]: 中文分词 生产环境建议使用 jieba 或 pkuseg import jieba return list(jieba.cut(text))四、混合检索的调优权衡权重选择、延迟开销与索引维护权重选择。稠密和稀疏的权重比例没有通用最优值取决于数据集特征。关键词密集的技术文档适合提高稀疏权重如 0.4/0.6语义丰富的自然语言问答适合提高稠密权重如 0.7/0.3。建议在标注数据集上网格搜索最优权重。延迟开销。混合检索需要同时执行两次检索延迟约为单一检索的 1.5-2 倍。如果延迟敏感可以并行执行两次检索将延迟压缩到接近单次检索。但并行执行增加了系统复杂度。索引维护。稠密索引和稀疏索引需要同步更新。文档新增或修改时两个索引都要更新否则会出现检索结果不一致。建议将索引更新封装为事务性操作或使用异步更新 最终一致性策略。适用边界混合检索适用于对召回率要求高、查询类型多样的 RAG 场景。对于查询模式固定、关键词匹配即可满足需求的场景如日志搜索纯稀疏检索更高效。五、总结混合检索通过融合稠密检索的语义理解能力和稀疏检索的关键词精确匹配能力显著提升 RAG 系统的召回率。RRF 融合策略对分数分布差异鲁棒推荐作为默认选择线性加权融合在分数归一化后效果稳定适合权重调优。工程实践中建议在标注数据集上评估不同权重和融合策略的效果并行执行两次检索控制延迟并确保双索引的同步更新。