RAG系统的多路召回（Multi-Retrieval）详解

在RAG检索增强生成系统中多路召回是一种通过多种检索策略并行获取候选文档再进行结果融合的机制。它的核心目的是提高召回率确保不同类型的查询都能被有效检索。一、为什么需要多路召回单一检索方式存在明显局限检索方式优势局限语义向量检索理解语义、处理同义词对专有名词、关键词不敏感关键词检索(BM25)精确匹配关键词无法理解语义混合检索兼顾语义和关键词权重配置复杂多路召回通过组合多种策略取长补短提升整体检索效果。二、常见的多路召回策略1.混合检索Hybrid Search结合语义检索和关键词检索# 语义向量检索 vector_results vector_retriever.retrieve(query, top_k10) # 关键词检索 bm25_results bm25_retriever.retrieve(query, top_k10) # 分数融合 def fuse_scores(vector_results, bm25_results, alpha0.5): alpha: 语义检索权重1-alpha为关键词检索权重 fused {} for doc, score in vector_results: fused[doc.id] alpha * score for doc, score in bm25_results: if doc.id in fused: fused[doc.id] (1 - alpha) * score else: fused[doc.id] (1 - alpha) * score return sorted(fused.items(), keylambda x: x[1], reverseTrue)2.多向量模型召回使用不同的Embedding模型class MultiEmbeddingRetriever: def __init__(self): self.embeddings { dense: DenseEmbeddingModel(text-embedding-3-large), bge: BGEEmbeddingModel(bge-large), e5: E5EmbeddingModel(e5-base) } self.vectorstores { dense: VectorStore(dense_index), bge: VectorStore(bge_index), e5: VectorStore(e5_index) } def retrieve(self, query, top_k5): results [] # 并行获取各模型检索结果 for model_name, embed_model in self.embeddings.items(): query_vector embed_model.embed(query) docs self.vectorstores[model_name].search( query_vector, top_ktop_k ) results.append((model_name, docs)) # RRF融合Reciprocal Rank Fusion return self._rrf_fusion(results, k60) def _rrf_fusion(self, results_list, k60): RRF融合算法 fused_scores defaultdict(float) for model_name, docs in results_list: for rank, (doc, score) in enumerate(docs): fused_scores[doc.id] 1 / (k rank 1) return sorted( fused_scores.items(), keylambda x: x[1], reverseTrue )[:top_k]3.多粒度召回不同 chunk size 和检索范围class MultiGranularityRetriever: def __init__(self): self.vectorstores { chunk_128: VectorStore(chunks_128), chunk_512: VectorStore(chunks_512), chunk_1024: VectorStore(chunks_1024), document: VectorStore(full_documents) } def retrieve(self, query, top_k5): # 细粒度精准匹配小段落 fine_results self.vectorstores[chunk_128].search(query, top_k20) # 中粒度完整段落 medium_results self.vectorstores[chunk_512].search(query, top_k10) # 粗粒度文档级别 coarse_results self.vectorstores[chunk_1024].search(query, top_k5) # 文档整体 doc_results self.vectorstores[document].search(query, top_k3) # 融合策略细粒度权重高保留上下文 return self._weighted_fusion({ fine: (fine_results, 0.4), medium: (medium_results, 0.3), coarse: (coarse_results, 0.2), doc: (doc_results, 0.1) })4.多索引召回根据query类型路由到不同索引class MultiIndexRetriever: def __init__(self): self.indexes { technical: VectorStore(technical_docs), business: VectorStore(business_docs), faq: VectorStore(faq_index), code: VectorStore(code_snippets) } def retrieve(self, query): # Query分类 query_type self.classify_query(query) if query_type code: # 代码相关查询 code_results self.indexes[code].search(query, top_k5) technical_results self.indexes[technical].search(query, top_k3) return self.fuse(code_results, technical_results) elif query_type faq: # FAQ精确匹配 faq_results self.indexes[faq].search(query, top_k5) return faq_results else: # 通用查询搜索所有索引 all_results [] for name, index in self.indexes.items(): results index.search(query, top_k5) all_results.append((name, results)) return self._rrf_fusion(all_results) def classify_query(self, query): 简单分类器 if any(kw in query.lower() for kw in [代码, function, def , import]): return code if any(kw in query.lower() for kw in [常见问题, FAQ, 如何, 怎么]): return faq return general三、结果融合方法1.RRFReciprocal Rank Fusiondef rrf_fusion(results_dict, k60): results_dict: {检索器名: [(doc, score), ...]} k: 平滑参数通常60 fused defaultdict(float) for retriever_name, results in results_dict.items(): for rank, (doc, score) in enumerate(results): # RRF公式1/(krank) fused[doc.id] 1 / (k rank) return sorted(fused.items(), keylambda x: x[1], reverseTrue)2.Comore Scoredef comore_score(results_dict): 基于TF-IDF的融合方法 doc_scores defaultdict(float) all_docs set() for retriever_name, results in results_dict.items(): docs [doc for doc, _ in results] all_docs.update(docs) # 归一化分数 max_score max(score for _, score in results) min_score min(score for _, score in results) for doc, score in results: if max_score ! min_score: normalized (score - min_score) / (max_score - min_score) else: normalized 1.0 doc_scores[doc.id] normalized return sorted(doc_scores.items(), keylambda x: x[1], reverseTrue)四、实际架构示例┌─────────────────────────────────────────────────────────┐ │ User Query │ └─────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ Query Understanding │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ 意图识别 │ │ 关键词 │ │ 语义扩展 │ │ │ │ │ │ 提取 │ │ │ │ │ └──────────┘ └──────────┘ └──────────┘ │ └─────────────────────────────────────────────────────────┘ │ ┌──────────────────┼──────────────────┐ ▼ ▼ ▼ ┌───────────────┐ ┌───────────────┐ ┌───────────────┐ │ 语义向量检索 │ │ BM25检索 │ │ 知识图谱检索 │ │ (Dense) │ │ (Sparse) │ │ (Knowledge) │ └───────────────┘ └───────────────┘ └───────────────┘ │ │ │ └──────────────────┼──────────────────┘ ▼ ┌─────────────────────────────────────────────────────────┐ │ 结果融合 (Rerank) │ │ ┌─────────────────────────────────────────────────┐ │ │ │ RRF / Comore Score / 加权融合 │ │ │ └─────────────────────────────────────────────────┘ │ └─────────────────────────────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ Top-K 候选文档 │ └─────────────────────────────────────────────────────────┘五、实现建议1.根据场景选择策略通用问答混合检索 RRF专业领域多索引 领域适配代码检索多粒度 语法分析2.权重调优通过A/B测试或线上指标优化各路召回的权重3.性能考虑并行执行各路检索使用异步IO限制每路检索的候选数量4.效果评估关注 RecallK、MRR、NDCG 等指标