RAG 混合检索实战：稠密向量 + 关键词，各取所长-拓冰建站

RAG 混合检索实战：稠密向量 + 关键词，各取所长

一、向量搜索找不到"HTTP 502"——它不知道这是错误码

向量搜索擅长语义匹配。
问"怎么修复服务器报错"，能召回相关文章。
但问"HTTP 502 怎么解决"，向量搜索可能返回 HTTP 协议的通用介绍。
因为它不理解 502 是一个具体的错误码。

反过来，关键词搜索（BM25）精准匹配"HTTP 502"。
但搜索"服务器连不上"，它只能匹配字面。
理解不了这等于"连接超时"、"网络错误"。

两种检索方式各有所长。
向量搜索的强项是语义理解和跨语言。
关键词搜索的强项是精确匹配和罕见术语。
混合检索并不复杂，设计好融合策略即可。

二、混合检索的并行-融合架构

混合检索的核心是两路召回、统一排序。

flowchart TB A[用户 Query: HTTP 502 怎么解决?] --> B[Query 预处理] B --> C[向量化 Query] B --> D[提取关键词] C --> E[向量检索 Top-50] D --> F[BM25 检索 Top-50] E --> G[结果合并与去重] F --> G G --> H[RRF 融合排序] H --> I[重排序模型精排] I --> J[取 Top-5 返回]

RRF（Reciprocal Rank Fusion）是常用的融合算法。
它不依赖原始分数的量纲。
公式：score(d) = Σ(1 / (k + rank_i(d)))
其中 k 是常数（通常 60），rank_i 是文档在第 i 个列表中的排位。

三、混合检索的 Python 实现

""" hybrid_search.py - 向量 + BM25 混合检索 """ import math import numpy as np from typing import List, Dict, Optional, Tuple from dataclasses import dataclass from collections import Counter import chromadb from chromadb.utils import embedding_functions @dataclass class SearchResult: """单条检索结果""" doc_id: str content: str score: float = 0.0 # 最终得分 vector_score: float = 0.0 # 向量得分 bm25_score: float = 0.0 # BM25 得分 metadata: Dict = None def __post_init__(self): if self.metadata is None: self.metadata = {} class BM25Scorer: """BM25 关键词检索器""" def __init__(self, k1: float = 1.5, b: float = 0.75): self.k1 = k1 self.b = b self.corpus: List[str] = [] self.doc_lengths: List[int] = [] self.avgdl: float = 0 self.df: Dict[str, int] = {} # 文档频率 def index(self, corpus: List[str]): """构建 BM25 索引""" self.corpus = corpus self.doc_lengths = [len(doc.split()) for doc in corpus] self.avgdl = ( sum(self.doc_lengths) / max(len(corpus), 1) ) # 计算 DF self.df = Counter() for doc in corpus: terms = set(doc.lower().split()) for t in terms: self.df[t] += 1 def search( self, query: str, doc_ids: List[str], top_k: int = 50, ) -> List[Tuple[str, float]]: """批量计算 BM25 得分""" scores = [] query_terms = query.lower().split() doc_count = len(self.corpus) for i, doc_id in enumerate(doc_ids): doc = self.corpus[i] doc_terms = doc.lower().split() doc_len = self.doc_lengths[i] tf = Counter(doc_terms) score = 0.0 for term in query_terms: if term not in tf: continue idf = math.log( (doc_count - self.df.get(term, 0) + 0.5) / (self.df.get(term, 0) + 0.5) + 1.0 ) numerator = tf[term] * (self.k1 + 1) denominator = ( tf[term] + self.k1 * (1 - self.b + self.b * doc_len / max(self.avgdl, 1)) ) score += idf * numerator / max(denominator, 0.001) scores.append((doc_id, score)) scores.sort(key=lambda x: x[1], reverse=True) return scores[:top_k] class HybridRetriever: """混合检索器""" def __init__( self, vector_weight: float = 0.5, bm25_weight: float = 0.5, rrf_k: int = 60, ): """ vector_weight: 向量检索权重 bm25_weight: BM25 检索权重 rrf_k: RRF 平滑常数 """ self.vector_weight = vector_weight self.bm25_weight = bm25_weight self.rrf_k = rrf_k # 初始化向量数据库 self.ef = embedding_functions.DefaultEmbeddingFunction() self.chroma_client = chromadb.Client() self.collection = None # BM25 索引 self.bm25 = BM25Scorer() def index(self, documents: List[str], metadata: List[Dict] = None): """构建双路索引""" if metadata is None: metadata = [{}] * len(documents) doc_ids = [f"doc_{i}" for i in range(len(documents))] # 向量索引 self.collection = self.chroma_client.create_collection( name="hybrid_index", embedding_function=self.ef, ) self.collection.add( ids=doc_ids, documents=documents, metadatas=metadata, ) # BM25 索引 self.bm25.index(documents) def search( self, query: str, top_k: int = 5, ) -> List[SearchResult]: """混合检索""" if self.collection is None: return [] # 1. 向量检索 vector_results = self.collection.query( query_texts=[query], n_results=50, include=["documents", "metadatas", "distances"], ) # 2. BM25 检索（在全量文档上计算得分） all_docs = self.collection.get(include=["documents"]) bm25_results = self.bm25.search( query, all_docs["ids"], top_k=50, ) # 3. RRF 融合 fused = self._rrf_fusion( vector_results, bm25_results, top_k ) return fused def _rrf_fusion( self, vector_results: Dict, bm25_results: List[Tuple[str, float]], top_k: int, ) -> List[SearchResult]: """RRF 多路融合排序""" scores: Dict[str, float] = {} # 向量检索贡献 for rank, doc_id in enumerate(vector_results["ids"][0]): rrf_score = 1.0 / (self.rrf_k + rank + 1) scores[doc_id] = ( scores.get(doc_id, 0) + self.vector_weight * rrf_score ) # BM25 贡献 for rank, (doc_id, _) in enumerate(bm25_results): rrf_score = 1.0 / (self.rrf_k + rank + 1) scores[doc_id] = ( scores.get(doc_id, 0) + self.bm25_weight * rrf_score ) # 排序取 Top-K ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True) ranked = ranked[:top_k] # 构造结果 results = [] id_to_idx = { doc_id: idx for idx, doc_id in enumerate(vector_results["ids"][0]) } bm25_map = dict(bm25_results) for doc_id, score in ranked: idx = id_to_idx.get(doc_id, 0) content = ( vector_results["documents"][0][idx] if idx < len(vector_results["documents"][0]) else "" ) results.append(SearchResult( doc_id=doc_id, content=content, score=score, vector_score=vector_results["distances"][0].get(idx, 0), bm25_score=bm25_map.get(doc_id, 0), )) return results def explain(self, query: str, doc_id: str) -> str: """解释某条结果的得分来源""" results = self.search(query, top_k=10) for r in results: if r.doc_id == doc_id: vec_pct = r.vector_score / max(r.score, 0.001) * 100 bm25_pct = r.bm25_score / max(r.score, 0.001) * 100 return ( f"文档 {doc_id} 的得分来源:\n" f" 向量贡献: {vec_pct:.0f}%\n" f" BM25贡献: {bm25_pct:.0f}%\n" f" 融合得分: {r.score:.4f}" ) return f"文档 {doc_id} 未在前 10 结果中" # ---- 使用示例 ---- def demo(): documents = [ "HTTP 502 Bad Gateway 错误的排查与修复方法", "Nginx 反向代理配置详解", "Docker 容器中部署 HTTP 服务的常见问题", "502 错误码表示网关收到了无效响应", "如何监控服务器状态和响应时间", ] retriever = HybridRetriever(vector_weight=0.6, bm25_weight=0.4) retriever.index(documents) query = "HTTP 502 怎么解决" results = retriever.search(query, top_k=3) print(f"查询: {query}\n") for i, r in enumerate(results): print(f"[{i+1}] {r.content}") print(f" 融合得分: {r.score:.4f}") # 解释得分来源 if results: print(f"\n{retriever.explain(query, results[0].doc_id)}")