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 返回]RRFReciprocal Rank Fusion是常用的融合算法。它不依赖原始分数的量纲。公式score(d) Σ(1 / (k rank_i(d)))其中 k 是常数通常 60rank_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(keylambda x: x[1], reverseTrue) 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 [fdoc_{i} for i in range(len(documents))] # 向量索引 self.collection self.chroma_client.create_collection( namehybrid_index, embedding_functionself.ef, ) self.collection.add( idsdoc_ids, documentsdocuments, metadatasmetadata, ) # 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_results50, include[documents, metadatas, distances], ) # 2. BM25 检索在全量文档上计算得分 all_docs self.collection.get(include[documents]) bm25_results self.bm25.search( query, all_docs[ids], top_k50, ) # 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(), keylambda x: x[1], reverseTrue) 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_iddoc_id, contentcontent, scorescore, vector_scorevector_results[distances][0].get(idx, 0), bm25_scorebm25_map.get(doc_id, 0), )) return results def explain(self, query: str, doc_id: str) - str: 解释某条结果的得分来源 results self.search(query, top_k10) 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_weight0.6, bm25_weight0.4) retriever.index(documents) query HTTP 502 怎么解决 results retriever.search(query, top_k3) print(f查询: {query}\n) for i, r in enumerate(results): print(f[{i1}] {r.content}) print(f 融合得分: {r.score:.4f}) # 解释得分来源 if results: print(f\n{retriever.explain(query, results[0].doc_id)})四、混合检索的调优空间权重配比需要针对场景调优。代码和错误码类问题BM25 权重应更高0.6。自然语言问答向量权重应更高0.6。可通过标注数据集做网格搜索确定最佳权重。BM25 对中文分词敏感。如果分词不准HTTP502被切为一个词搜索502可能匹配不到。建议使用 jieba 分词并保留原始字符 n-gram。不适合混合检索的场景已有成熟 ES 搜索体系的系统直接用 ES 的向量BM25 融合纯语义匹配场景如聊天机器人BM25 帮助有限延迟要求极高的场景两路检索增加约 20-50ms。五、总结向量检索和关键词检索各有所长混合检索取长补短。RRF 算法无需关心原始得分的量纲简单有效。向量权重和 BM25 权重需要根据场景调优。代码和专有名词多的场景BM25 权重要调高。实现上保持两路检索独立融合层灵活可配置。