检索增强生成中的混合检索策略：稠密检索与稀疏检索的融合方案

检索增强生成中的混合检索策略：稠密检索与稀疏检索的融合方案

一、单一检索的"召回盲区"：稠密检索与稀疏检索的互补性

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_embeddings=True, show_progress_bar=True ) def search(self, query: str, top_n: int = 10) -> List[RetrievalResult]: """混合检索""" # 稠密检索 query_embedding = self.encoder.encode( [query], normalize_embeddings=True ) 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_id=self.doc_ids[idx], content=self.doc_contents[idx], dense_score=float(dense_scores[idx]), sparse_score=float(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(key=lambda r: r.combined_score, reverse=True) 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, key=lambda r: r.dense_score, reverse=True) # 按稀疏分数排名 by_sparse = sorted(results, key=lambda r: r.sparse_score, reverse=True) 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 融合策略对分数分布差异鲁棒，推荐作为默认选择；线性加权融合在分数归一化后效果稳定，适合权重调优。工程实践中，建议在标注数据集上评估不同权重和融合策略的效果，并行执行两次检索控制延迟，并确保双索引的同步更新。