大模型学习之路006：RAG 零基础入门教程（第三篇）：BM25 关键词检索与混合检索实战

一、为什么我们需要混合检索在上篇中我们实现了基于 BGEChroma 的语义检索系统它能很好地理解文本的语义解决了传统检索 字面匹配、语义不匹配 的问题。但单一的语义检索存在致命短板1.1 单一语义检索的缺陷对精确关键词不敏感当用户查询包含特定专业术语、产品型号、人名时语义检索可能会漏检包含这些关键词的文档例子用户查询 iPhone 15 Pro Max 的电池容量语义检索可能会返回 iPhone 15 的参数但漏检了明确包含 iPhone 15 Pro Max 的文档对短文本效果差当查询非常短如 RAG、BM25时语义检索的效果不稳定对嵌入模型质量依赖极高如果嵌入模型没有见过某个专业术语就无法生成准确的向量1.2 单一关键词检索的缺陷传统的关键词检索如 BM25擅长精确匹配但也有明显的缺点无法理解语义只能匹配字面相同的词无法理解同义词、近义词、转述句例子用户查询 什么是检索增强生成关键词检索无法匹配包含 RAG 的文档无法处理歧义同一个词有多个意思时无法区分对长文本效果差长文本中关键词出现频率高但可能与查询语义无关1.3 混合检索取长补短混合检索Hybrid Retrieval是目前 RAG 系统的标准配置它将语义检索向量和关键词检索BM25结合起来同时发挥两者的优势语义检索负责捕捉语义相关性关键词检索负责捕捉精确匹配两者融合后召回率和精确率都能得到大幅提升行业数据混合检索比单一语义检索的召回率平均提升 20%-30%是 RAG 效果提升性价比最高的优化手段之一。二、BM25 关键词检索原理与实战BM25Best Match 25是传统信息检索领域的黄金标准自 1994 年提出以来一直是搜索引擎的核心算法。它基于词频统计计算查询与文档的相关性得分。2.1 BM25 核心原理BM25 的核心思想是一个词在文档中出现的频率越高这个文档与查询的相关性越高但如果这个词在所有文档中都很常见它的权重就越低。2.1.1 BM25 公式详解BM25 的计算公式如下2.2 BM25 代码实现我们将使用rank_bm25库来实现 BM25 检索这是目前最流行的 BM25 Python 实现。2.2.1 安装依赖pip install rank_bm25 jiebarank_bm25BM25 算法实现jieba中文分词工具BM25 需要先将文本分词2.2.2 中文分词BM25 是基于词的检索所以需要先将中文文本分词。import jieba import re def chinese_tokenize(text): 中文分词函数 :param text: 输入文本 :return: 分词后的词列表 # 去除特殊字符 text re.sub(r[^\u4e00-\u9fa5a-zA-Z0-9], , text) # 分词 words jieba.lcut(text) # 去除空白词 words [word.strip() for word in words if word.strip()] return words # 测试分词 test_text 检索增强生成RAG是一种大模型应用技术 tokens chinese_tokenize(test_text) print(f原文本{test_text}) print(f分词结果{tokens})2.2.3 BM25 检索器实现from rank_bm25 import BM25Okapi import json class BM25Retriever: def __init__(self, chunks): 初始化BM25检索器 :param chunks: 文档分块列表每个分块包含id、text、metadata self.chunks chunks self.doc_ids [chunk[id] for chunk in chunks] self.doc_texts [chunk[text] for chunk in chunks] # 对所有文档分词 print(正在对文档进行分词...) self.tokenized_docs [chinese_tokenize(text) for text in self.doc_texts] # 初始化BM25模型 print(正在构建BM25索引...) self.bm25 BM25Okapi(self.tokenized_docs, k11.5, b0.75) print(BM25索引构建完成) def search(self, query, top_k5): BM25检索 :param query: 查询文本 :param top_k: 返回最相关的Top-K个结果 :return: 检索结果包含id、text、metadata、score # 对查询分词 tokenized_query chinese_tokenize(query) # 计算所有文档的BM25得分 scores self.bm25.get_scores(tokenized_query) # 按得分降序排序取Top-K top_indices scores.argsort()[::-1][:top_k] # 构建结果 results [] for idx in top_indices: if scores[idx] 0: continue # 过滤得分为0的结果 results.append({ id: self.doc_ids[idx], text: self.doc_texts[idx], metadata: self.chunks[idx][metadata], score: float(scores[idx]) }) return results # 测试BM25检索器 if __name__ __main__: # 读取之前章节生成的分块数据 chunks [] with open(processed_chunks.jsonl, r, encodingutf-8) as f: for line in f: chunk json.loads(line) chunks.append(chunk) # 初始化BM25检索器 bm25_retriever BM25Retriever(chunks) # 测试检索 query 什么是RAG results bm25_retriever.search(query, top_k5) print(f查询{query}) print(f找到{len(results)}个相关文档\n) for i, result in enumerate(results): print(f【结果{i1}】) print(fID{result[id]}) print(fBM25得分{result[score]:.4f}) print(f文档内容{result[text][:200]}...) print(- * 100)2.3 BM25 vs 语义检索效果对比我们来对比一下 BM25 和语义检索在不同查询下的效果查询类型示例查询BM25 效果语义检索效果精确关键词BM25 公式很好能准确找到包含 BM25 公式 的文档一般可能会返回 检索算法 相关的文档语义查询检索增强生成的定义一般无法匹配包含 RAG 的文档很好能准确理解语义短查询RAG很好能找到所有包含 RAG 的文档一般效果不稳定长查询如何解决大模型的幻觉问题提高 RAG 系统的准确率一般关键词太多权重分散很好能理解整体语义可以看到两者正好互补这就是为什么我们需要混合检索。三、混合检索技术混合检索的核心是将语义检索和 BM25 检索的结果融合起来得到一个综合的排名。目前主流的融合策略有两种权重融合和RRF 融合。3.1 融合策略一权重融合Weighted Fusion权重融合是最简单也是最常用的融合策略它将两种检索的得分按权重相加得到最终得分。3.1.1 得分归一化由于语义检索的得分余弦相似度0-1和 BM25 的得分0-∞尺度不同不能直接相加需要先进行归一化。我们使用最小 - 最大归一化Min-Max Scaling将得分归一化到 [0, 1] 区间如果所有得分都相同归一化后的值为 1。3.1.2 权重融合实现def normalize_scores(scores): 最小-最大归一化 :param scores: 得分列表 :return: 归一化后的得分列表 if not scores: return [] min_score min(scores) max_score max(scores) if max_score min_score: return [1.0] * len(scores) return [(score - min_score) / (max_score - min_score) for score in scores] def weighted_fusion(semantic_results, bm25_results, semantic_weight0.7, bm25_weight0.3): 权重融合 :param semantic_results: 语义检索结果每个结果包含id、text、metadata、score :param bm25_results: BM25检索结果每个结果包含id、text、metadata、score :param semantic_weight: 语义检索权重 :param bm25_weight: BM25检索权重 :return: 融合后的结果按最终得分降序排序 # 构建结果字典key是文档id result_dict {} # 处理语义检索结果 semantic_scores [result[score] for result in semantic_results] normalized_semantic_scores normalize_scores(semantic_scores) for i, result in enumerate(semantic_results): doc_id result[id] if doc_id not in result_dict: result_dict[doc_id] { id: doc_id, text: result[text], metadata: result[metadata], semantic_score: normalized_semantic_scores[i], bm25_score: 0.0 } else: result_dict[doc_id][semantic_score] normalized_semantic_scores[i] # 处理BM25检索结果 bm25_scores [result[score] for result in bm25_results] normalized_bm25_scores normalize_scores(bm25_scores) for i, result in enumerate(bm25_results): doc_id result[id] if doc_id not in result_dict: result_dict[doc_id] { id: doc_id, text: result[text], metadata: result[metadata], semantic_score: 0.0, bm25_score: normalized_bm25_scores[i] } else: result_dict[doc_id][bm25_score] normalized_bm25_scores[i] # 计算最终得分 for doc_id in result_dict: result_dict[doc_id][final_score] ( result_dict[doc_id][semantic_score] * semantic_weight result_dict[doc_id][bm25_score] * bm25_weight ) # 按最终得分降序排序 fused_results sorted(result_dict.values(), keylambda x: x[final_score], reverseTrue) return fused_results3.1.3 权重调优权重的选择对融合效果有很大影响中文场景下的最佳实践是语义检索权重0.7BM25 检索权重0.3这个权重是经过大量实践验证的在大多数中文场景下效果最好。你可以根据自己的数据集调整权重找到最优值。3.2 融合策略二RRF 融合Reciprocal Rank FusionRRF倒数排名融合是一种基于排名的融合策略它不关心得分的绝对值只关心结果的排名。3.2.1 RRF 原理RRF 的计算公式为RRF 的优势是不需要归一化得分避免了得分尺度不同的问题对异常值不敏感鲁棒性更好在多个检索系统融合时效果优于权重融合3.2.2 RRF 融合实现def rrf_fusion(semantic_results, bm25_results, k60): RRF融合 :param semantic_results: 语义检索结果 :param bm25_results: BM25检索结果 :param k: RRF常数 :return: 融合后的结果 # 构建排名字典 rank_dict {} # 处理语义检索结果的排名 for rank, result in enumerate(semantic_results, 1): doc_id result[id] if doc_id not in rank_dict: rank_dict[doc_id] { id: doc_id, text: result[text], metadata: result[metadata], ranks: [] } rank_dict[doc_id][ranks].append(rank) # 处理BM25检索结果的排名 for rank, result in enumerate(bm25_results, 1): doc_id result[id] if doc_id not in rank_dict: rank_dict[doc_id] { id: doc_id, text: result[text], metadata: result[metadata], ranks: [] } rank_dict[doc_id][ranks].append(rank) # 计算RRF得分 for doc_id in rank_dict: rrf_score 0.0 for rank in rank_dict[doc_id][ranks]: rrf_score 1.0 / (k rank) rank_dict[doc_id][rrf_score] rrf_score # 按RRF得分降序排序 fused_results sorted(rank_dict.values(), keylambda x: x[rrf_score], reverseTrue) return fused_results3.3 完整混合检索系统实现现在我们将语义检索、BM25 检索、混合检索整合起来实现一个完整的检索系统。import os from pathlib import Path # ------------- 环境配置必须在导入依赖前设置------------- os.environ[HF_ENDPOINT] https://hf-mirror.com # 设置模型缓存目录避免重复下载 os.environ[TRANSFORMERS_CACHE] str(Path(__file__).parent / model_cache) import chromadb from chromadb.utils import embedding_functions from 混合检索技术_权重融合_20260505 import weighted_fusion from 混合检索技术_rrf_20260505 import rrf_fusion from BM25_20260505 import BM25Retriever import json class HybridRetriever: def __init__(self, chunks, collection_namerag_knowledge_base, db_path./chroma_db): 初始化混合检索器 :param chunks: 文档分块列表 :param collection_name: Chroma集合名称 :param db_path: Chroma数据库路径 self.chunks chunks # 初始化Chroma语义检索器 print(正在初始化语义检索器...) self.client chromadb.PersistentClient(pathdb_path) self.bge_embedding embedding_functions.SentenceTransformerEmbeddingFunction( model_nameBAAI/bge-large-zh-v1.5 ) self.collection self.client.get_or_create_collection( namecollection_name, embedding_functionself.bge_embedding, metadata{hnsw:space: cosine} ) # 如果集合为空插入文档 if self.collection.count() 0: print(正在将文档插入Chroma...) ids [chunk[id] for chunk in chunks] documents [chunk[text] for chunk in chunks] metadatas [chunk[metadata] for chunk in chunks] self.collection.add(idsids, documentsdocuments, metadatasmetadatas) # 初始化BM25检索器 print(正在初始化BM25检索器...) self.bm25_retriever BM25Retriever(chunks) print(混合检索器初始化完成) def semantic_search(self, query, top_k20): 语义检索 results self.collection.query( query_texts[query], n_resultstop_k ) # 格式化结果 semantic_results [] for i in range(len(results[ids][0])): semantic_results.append({ id: results[ids][0][i], text: results[documents][0][i], metadata: results[metadatas][0][i], score: 1 - results[distances][0][i] # 转换为相似度得分 }) return semantic_results def bm25_search(self, query, top_k20): BM25检索 return self.bm25_retriever.search(query, top_ktop_k) def hybrid_search(self, query, top_k5, fusion_methodweighted, semantic_weight0.7, bm25_weight0.3): 混合检索 :param fusion_method: 融合方法weighted或rrf :return: 融合后的Top-K结果 # 先召回更多结果一般是最终Top-K的4-5倍 semantic_results self.semantic_search(query, top_k20) bm25_results self.bm25_search(query, top_k20) # 融合 if fusion_method weighted: fused_results weighted_fusion(semantic_results, bm25_results, semantic_weight, bm25_weight) elif fusion_method rrf: fused_results rrf_fusion(semantic_results, bm25_results) else: raise ValueError(f不支持的融合方法{fusion_method}) # 返回Top-K结果 return fused_results[:top_k] # 测试混合检索系统 if __name__ __main__: # 读取分块数据 chunks [] with open(processed_chunks.jsonl, r, encodingutf-8) as f: for line in f: chunk json.loads(line) chunks.append(chunk) # 初始化混合检索器 hybrid_retriever HybridRetriever(chunks) # 测试混合检索 query BM25算法的公式是什么 print(f查询{query}\n) # 语义检索结果 print( * 50) print(语义检索结果) semantic_results hybrid_retriever.semantic_search(query, top_k5) for i, result in enumerate(semantic_results): print(f【结果{i 1}】得分{result[score]:.4f} | {result[text][:100]}...) # BM25检索结果 print(\n * 50) print(BM25检索结果) bm25_results hybrid_retriever.bm25_search(query, top_k5) for i, result in enumerate(bm25_results): print(f【结果{i 1}】得分{result[score]:.4f} | {result[text][:100]}...) # 混合检索结果 print(\n * 50) print(混合检索结果权重融合) hybrid_results hybrid_retriever.hybrid_search(query, top_k5, fusion_methodweighted) for i, result in enumerate(hybrid_results): print( f【结果{i 1}】最终得分{result[final_score]:.4f} | 语义得分{result[semantic_score]:.4f} | BM25得分{result[bm25_score]:.4f}) print(f文档内容{result[text][:200]}...) print(- * 100)会发现混合检索的结果结合了语义检索和 BM25 的优势既包含了语义相关的文档也包含了精确匹配 BM25 公式 的文档最终排名比单一检索更合理四、检索效果评估检索效果评估是 RAG 开发中非常重要的环节它能让我们客观地衡量不同检索方法的效果指导我们进行优化。4.1 核心评估指标我们主要使用三个指标来评估检索效果召回率Recall、精确率Precision、F1 值。4.1.1 召回率Recall召回率衡量的是所有相关文档中有多少被检索到了。召回率越高说明漏检的相关文档越少RAG 系统优先保证高召回率因为如果相关文档没有被检索到大模型就无法生成正确的答案4.1.2 精确率Precision精确率衡量的是检索到的文档中有多少是相关的。精确率越高说明检索结果中的无关文档越少精确率高可以减少大模型的干扰提升生成质量4.1.3 F1 值F1 值是召回率和精确率的调和平均值综合衡量检索效果。4.2 构建评估数据集评估的准确性取决于评估数据集的质量。一个好的评估数据集应该满足以下要求代表性覆盖真实业务中的各种查询类型多样性包含简单查询、复杂查询、精确查询、语义查询准确性每个查询的相关文档标注准确规模至少包含 20-30 个查询越多越准确4.2.1 评估数据集格式我们使用 JSON 格式存储评估数据集每个条目包含查询和相关文档的 ID 列表[ { query: 什么是RAG, relevant_ids: [test.pdf_chunk_0, test.pdf_chunk_1] }, { query: BM25算法的公式是什么, relevant_ids: [test.pdf_chunk_5, test.pdf_chunk_6] }, { query: 如何解决大模型的幻觉问题, relevant_ids: [test.pdf_chunk_3, test.pdf_chunk_7] } ]4.2.2 标注方法收集 20-30 个真实用户会问的问题对于每个问题手动找出所有相关的文档块记录它们的 ID可以邀请 2-3 个人一起标注取交集作为最终的相关文档提高标注准确性4.3 评估代码实现def evaluate_retriever(retriever, test_dataset, top_k5): 评估检索器的效果 :param retriever: 检索器对象需要实现search方法 :param test_dataset: 评估数据集 :param top_k: 检索返回的Top-K结果 :return: 评估结果包含平均召回率、平均精确率、平均F1值 total_recall 0.0 total_precision 0.0 total_f1 0.0 num_queries len(test_dataset) for item in test_dataset: query item[query] relevant_ids set(item[relevant_ids]) # 检索 results retriever.search(query, top_ktop_k) retrieved_ids set([result[id] for result in results]) # 计算真阳性检索到的相关文档数 true_positives len(relevant_ids retrieved_ids) # 计算召回率和精确率 recall true_positives / len(relevant_ids) if len(relevant_ids) 0 else 0.0 precision true_positives / len(retrieved_ids) if len(retrieved_ids) 0 else 0.0 # 计算F1值 f1 2 * precision * recall / (precision recall) if (precision recall) 0 else 0.0 total_recall recall total_precision precision total_f1 f1 # 打印每个查询的结果 print(f查询{query}) print(f相关文档数{len(relevant_ids)}检索到的相关文档数{true_positives}) print(f召回率{recall:.4f}精确率{precision:.4f}F1值{f1:.4f}) print(- * 100) # 计算平均值 avg_recall total_recall / num_queries avg_precision total_precision / num_queries avg_f1 total_f1 / num_queries print(\n * 100) print(评估结果汇总) print(f平均召回率{avg_recall:.4f}) print(f平均精确率{avg_precision:.4f}) print(f平均F1值{avg_f1:.4f}) print( * 100) return { avg_recall: avg_recall, avg_precision: avg_precision, avg_f1: avg_f1 } # 测试评估 if __name__ __main__: # 加载评估数据集 with open(test_dataset.json, r, encodingutf-8) as f: test_dataset json.load(f) # 读取分块数据 chunks [] with open(processed_chunks.jsonl, r, encodingutf-8) as f: for line in f: chunk json.loads(line) chunks.append(chunk) # 初始化混合检索器 hybrid_retriever HybridRetriever(chunks) # 定义不同检索器的search方法 class SemanticRetrieverWrapper: def __init__(self, hybrid_retriever): self.hybrid_retriever hybrid_retriever def search(self, query, top_k5): return self.hybrid_retriever.semantic_search(query, top_ktop_k) class BM25RetrieverWrapper: def __init__(self, hybrid_retriever): self.hybrid_retriever hybrid_retriever def search(self, query, top_k5): return self.hybrid_retriever.bm25_search(query, top_ktop_k) class HybridRetrieverWrapper: def __init__(self, hybrid_retriever): self.hybrid_retriever hybrid_retriever def search(self, query, top_k5): return self.hybrid_retriever.hybrid_search(query, top_ktop_k) # 评估语义检索 print(评估语义检索) semantic_retriever SemanticRetrieverWrapper(hybrid_retriever) semantic_result evaluate_retriever(semantic_retriever, test_dataset, top_k5) # 评估BM25检索 print(\n评估BM25检索) bm25_retriever BM25RetrieverWrapper(hybrid_retriever) bm25_result evaluate_retriever(bm25_retriever, test_dataset, top_k5) # 评估混合检索 print(\n评估混合检索) hybrid_retriever_wrapper HybridRetrieverWrapper(hybrid_retriever) hybrid_result evaluate_retriever(hybrid_retriever_wrapper, test_dataset, top_k5) # 对比结果 print(\n * 100) print(三种检索方式效果对比) print(f{方法:10} {平均召回率:12} {平均精确率:12} {平均F1值:10}) print(f{语义检索:10} {semantic_result[avg_recall]:12.4f} {semantic_result[avg_precision]:12.4f} {semantic_result[avg_f1]:10.4f}) print(f{BM25检索:10} {bm25_result[avg_recall]:12.4f} {bm25_result[avg_precision]:12.4f} {bm25_result[avg_f1]:10.4f}) print(f{混合检索:10} {hybrid_result[avg_recall]:12.4f} {hybrid_result[avg_precision]:12.4f} {hybrid_result[avg_f1]:10.4f}) print( * 100)