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

一、为什么我们需要混合检索？

在上篇中，我们实现了基于 BGE+Chroma 的语义检索系统，它能很好地理解文本的语义，解决了传统检索 "字面匹配、语义不匹配" 的问题。但单一的语义检索存在致命短板：

1.1 单一语义检索的缺陷

1. 对精确关键词不敏感：当用户查询包含特定专业术语、产品型号、人名时，语义检索可能会漏检包含这些关键词的文档
   • 例子：用户查询 "iPhone 15 Pro Max 的电池容量"，语义检索可能会返回 "iPhone 15 的参数"，但漏检了明确包含 "iPhone 15 Pro Max" 的文档
2. 对短文本效果差：当查询非常短（如 "RAG"、"BM25"）时，语义检索的效果不稳定
3. 对嵌入模型质量依赖极高：如果嵌入模型没有见过某个专业术语，就无法生成准确的向量

1.2 单一关键词检索的缺陷

传统的关键词检索（如 BM25）擅长精确匹配，但也有明显的缺点：

1. 无法理解语义：只能匹配字面相同的词，无法理解同义词、近义词、转述句
   • 例子：用户查询 "什么是检索增强生成？"，关键词检索无法匹配包含 "RAG" 的文档
2. 无法处理歧义：同一个词有多个意思时，无法区分
3. 对长文本效果差：长文本中关键词出现频率高，但可能与查询语义无关

1.3 混合检索：取长补短

混合检索（Hybrid Retrieval）是目前 RAG 系统的标准配置，它将语义检索（向量）和关键词检索（BM25）结合起来，同时发挥两者的优势：

• 语义检索负责捕捉语义相关性
• 关键词检索负责捕捉精确匹配
• 两者融合后，召回率和精确率都能得到大幅提升

行业数据：混合检索比单一语义检索的召回率平均提升 20%-30%，是 RAG 效果提升性价比最高的优化手段之一。

二、BM25 关键词检索原理与实战

BM25（Best 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 jieba

• rank_bm25：BM25 算法实现
• 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, k1=1.5, b=0.75) print("BM25索引构建完成") def search(self, query, top_k=5): """ 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', encoding='utf-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_k=5) print(f"查询：{query}") print(f"找到{len(results)}个相关文档：\n") for i, result in enumerate(results): print(f"【结果{i+1}】") print(f"ID：{result['id']}") print(f"BM25得分：{result['score']:.4f}") print(f"文档内容：{result['text'][:200]}...") print("-" * 100)

2.3 BM25 vs 语义检索效果对比

我们来对比一下 BM25 和语义检索在不同查询下的效果：

可以看到，两者正好互补，这就是为什么我们需要混合检索。

三、混合检索技术

混合检索的核心是将语义检索和 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_weight=0.7, bm25_weight=0.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(), key=lambda x: x["final_score"], reverse=True) return fused_results

3.1.3 权重调优

权重的选择对融合效果有很大影响，中文场景下的最佳实践是：

• 语义检索权重：0.7
• BM25 检索权重：0.3

这个权重是经过大量实践验证的，在大多数中文场景下效果最好。你可以根据自己的数据集调整权重，找到最优值。

3.2 融合策略二：RRF 融合（Reciprocal Rank Fusion）

RRF（倒数排名融合）是一种基于排名的融合策略，它不关心得分的绝对值，只关心结果的排名。

3.2.1 RRF 原理

RRF 的计算公式为：

RRF 的优势是：

• 不需要归一化得分，避免了得分尺度不同的问题
• 对异常值不敏感，鲁棒性更好
• 在多个检索系统融合时效果优于权重融合

3.2.2 RRF 融合实现

def rrf_fusion(semantic_results, bm25_results, k=60): """ 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(), key=lambda x: x["rrf_score"], reverse=True) return fused_results

3.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_name="rag_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(path=db_path) self.bge_embedding = embedding_functions.SentenceTransformerEmbeddingFunction( model_name="BAAI/bge-large-zh-v1.5" ) self.collection = self.client.get_or_create_collection( name=collection_name, embedding_function=self.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(ids=ids, documents=documents, metadatas=metadatas) # 初始化BM25检索器 print("正在初始化BM25检索器...") self.bm25_retriever = BM25Retriever(chunks) print("混合检索器初始化完成") def semantic_search(self, query, top_k=20): """语义检索""" results = self.collection.query( query_texts=[query], n_results=top_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_k=20): """BM25检索""" return self.bm25_retriever.search(query, top_k=top_k) def hybrid_search(self, query, top_k=5, fusion_method="weighted", semantic_weight=0.7, bm25_weight=0.3): """ 混合检索 :param fusion_method: 融合方法，"weighted"或"rrf" :return: 融合后的Top-K结果 """ # 先召回更多结果（一般是最终Top-K的4-5倍） semantic_results = self.semantic_search(query, top_k=20) bm25_results = self.bm25_search(query, top_k=20) # 融合 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', encoding='utf-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_k=5) 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_k=5) 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_k=5, fusion_method="weighted") 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 构建评估数据集

评估的准确性取决于评估数据集的质量。一个好的评估数据集应该满足以下要求：

1. 代表性：覆盖真实业务中的各种查询类型
2. 多样性：包含简单查询、复杂查询、精确查询、语义查询
3. 准确性：每个查询的相关文档标注准确
4. 规模：至少包含 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 标注方法

1. 收集 20-30 个真实用户会问的问题
2. 对于每个问题，手动找出所有相关的文档块，记录它们的 ID
3. 可以邀请 2-3 个人一起标注，取交集作为最终的相关文档，提高标注准确性

4.3 评估代码实现

def evaluate_retriever(retriever, test_dataset, top_k=5): """ 评估检索器的效果 :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_k=top_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', encoding='utf-8') as f: test_dataset = json.load(f) # 读取分块数据 chunks = [] with open("processed_chunks.jsonl", 'r', encoding='utf-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_k=5): return self.hybrid_retriever.semantic_search(query, top_k=top_k) class BM25RetrieverWrapper: def __init__(self, hybrid_retriever): self.hybrid_retriever = hybrid_retriever def search(self, query, top_k=5): return self.hybrid_retriever.bm25_search(query, top_k=top_k) class HybridRetrieverWrapper: def __init__(self, hybrid_retriever): self.hybrid_retriever = hybrid_retriever def search(self, query, top_k=5): return self.hybrid_retriever.hybrid_search(query, top_k=top_k) # 评估语义检索 print("评估语义检索：") semantic_retriever = SemanticRetrieverWrapper(hybrid_retriever) semantic_result = evaluate_retriever(semantic_retriever, test_dataset, top_k=5) # 评估BM25检索 print("\n评估BM25检索：") bm25_retriever = BM25RetrieverWrapper(hybrid_retriever) bm25_result = evaluate_retriever(bm25_retriever, test_dataset, top_k=5) # 评估混合检索 print("\n评估混合检索：") hybrid_retriever_wrapper = HybridRetrieverWrapper(hybrid_retriever) hybrid_result = evaluate_retriever(hybrid_retriever_wrapper, test_dataset, top_k=5) # 对比结果 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)