# RAG 混合检索实战：BM25 + 向量检索 + RRF 融合，从原理到代码全链路

RAG 混合检索实战：BM25 + 向量检索 + RRF 融合，从原理到代码全链路

导读：向量检索擅长语义理解，但面对精确数字、政策名称时容易"跑偏"；BM25 擅长关键词匹配，但无法理解同义改写。本文从项目真实代码出发，完整讲解 BM25 算法原理、中文分词实现、BM25 检索编码、RRF 倒数排名融合算法，以及如何用 asyncio.gather 并行执行双路检索。读完可直接落地到自己的 RAG 项目中。

适合读者：

• 正在做 RAG 系统但召回质量不达标的开发者
• 听说过 BM25 和混合检索但还没动手实现过的工程师
• 想了解 RRF 融合算法内部机制的算法爱好者
• 从向量检索单一路径升级到混合检索的团队

阅读收益：

• 理解 BM25 的 TF-IDF 升级版原理（不用背公式，看懂思想）
• 掌握 jieba 中文分词在 BM25 中的实战用法（保留数字、过滤标点）
• 学会编写 BM25 检索函数（从 MySQL 子块中召回关键词匹配结果）
• 理解 RRF（Reciprocal Rank Fusion）为什么比直接相加分数更科学
• 拿到可直接落地的 hybrid 检索模式代码（向量 + BM25 并行执行）

目录

1. 为什么单一路径检索不够

1.1 向量检索的局限

向量检索基于 Embedding 语义相似度，擅长处理：

• “办社保卡需要什么材料？” ↔ “申请社保卡要准备哪些证件？”（同义改写）
• “数字化补贴政策” ↔ “企业数字化转型扶持措施”（语义相近）

但向量检索在面对以下场景时表现不佳：

• 精确数字匹配：“2026年8月31日”——向量会把 “2026” 和 “2025” 视为相近
• 短关键词：“申报材料”——向量可能把它和 “报告材料” 混为一谈
• 政策名称/条款编号：“第二条”、“第三章”——这些短文本语义信息不足

1.2 BM25 的定位

BM25（Best Match 25）是基于词频的经典信息检索算法，属于 TF-IDF 的升级版。它的核心特点是：

• 只看字面词汇，不理解语义
• 精确匹配关键词，对数字、日期、短词敏感
• 分数基于词频和文档长度，不是概率值

1.3 两者互补

混合检索的思路：向量负责语义召回，BM25 负责关键词召回，RRF 负责融合两路结果。

2. BM25 算法原理：TF-IDF 的升级版

2.1 TF-IDF 的问题

传统 TF-IDF 的公式：

score = TF * IDF
TF = 词在文档中出现次数
IDF = log(总文档数 / 包含词的文档数)

TF-IDF 的问题是：

• 词频无限增长时分数无限增大（一篇文档重复 100 次 “社保”，分数远高于合理的 5 次）
• 没有考虑文档长度（长文档天然词频更高，不公平）

2.2 BM25 的改进

BM25 引入了三个改进：

（1）词频饱和函数

BM25 不直接用词频，而是用饱和函数：

TF_saturation = (freq * (k1 + 1)) / (freq + k1 * (1 – b + b * doc_len / avg_len))

• k1 控制饱和速度（通常 1.2~2.0）
• 词频越高，分数增长越慢，最终趋于饱和

（2）文档长度归一化

• b 控制长度惩罚（通常 0.75）
• doc_len / avg_len 是当前文档长度与平均长度的比值
• 长文档的 TF 会被适当降低

（3）IDF 改进

IDF_BM25 = log((N – n + 0.5) / (n + 0.5))

• N 是总文档数
• n 是包含该词的文档数
• 避免 IDF 为负数（当几乎所有文档都包含该词时）

2.3 核心洞察

BM25 分数不是概率，不适合直接和向量分数相加。这也是为什么要用 RRF 融合，而不是简单加权求和。

3. 中文分词：jieba + 正则过滤

3.1 为什么需要分词

BM25 基于词（token）计算词频。英文天然有空格分词，中文需要先做分词处理。

3.2 项目中的分词实现

import re
import jieba

def _tokenize(text: str) –> list[str]:
"""中文简单分词：jieba + 过滤纯标点/空格（保留数字和含数字的词）。"""
tokens = jieba.lcut(text.strip())
return [
t for t in tokens
if t.strip() and not re.match(r"^[\\s\\W]+$", t)
]

关键点：

• jieba.lcut()：精确模式分词，返回列表
• t.strip()：过滤纯空白字符
• re.match(r"^[\\s\\W]+$", t)：过滤纯标点符号（但保留含数字的词）

3.3 分词测试

test_cases = [
"2026年8月31日",
"社保缴纳记录",
"第二条 支持范围",
"数字化补贴申报材料",
"办理社保卡需要什么材料？",
]

for text in test_cases:
tokens = _tokenize(text)
print(f"原文: {text}")
print(f"分词: {tokens}")

输出示例：

原文: 2026年8月31日
分词: ['2026', '年', '8', '月', '31', '日']

原文: 社保缴纳记录
分词: ['社保', '缴纳', '记录']

原文: 第二条 支持范围
分词: ['第二', '条', '支持', '范围']

注意：数字被保留了！这是 BM25 的优势所在——精确匹配 “2026”、“8月”、“31日”。

3.4 分词踩坑：数字被过滤

如果正则写错，把数字也过滤了：

# 错误：过滤了所有非字母字符，包括数字
return [t for t in tokens if re.match(r"^[a-zA-Z\\u4e00-\\u9fa5]+$", t)]
# 结果：['年', '月', '日'] —— 数字没了！

正确做法：只过滤纯标点，保留数字和含数字的词。

4. BM25 检索函数实现

4.1 整体流程

BM25 检索的完整流程：

1. 从 MySQL 查询子块（只查 child 类型，信息更聚焦）
2. 应用过滤条件（文档ID、主题、生效日期）
3. 对所有子块内容分词，构建 BM25 语料库
4. 对用户问题分词，计算 BM25 分数
5. 按分数降序排序，截取 top_k
6. 返回与向量检索一致的数据结构

4.2 代码实现

from rank_bm25 import BM25Okapi
from sqlalchemy import select
from sqlalchemy.orm import Session

def _bm25_search(
question: str,
session: Session,
filters: RetrievalFilters,
top_k: int,
) –> list[dict]:
"""
BM25 关键词检索：从 MySQL 子块中查询，计算 BM25 分数。
只检索子块（chunk_type='child'），原因：子块更短，信息更聚焦。
"""
# 1. 构建 SQL 查询（只查子块）
stmt = select(DocumentChunk).where(DocumentChunk.is_parent == False)

# document_ids 过滤
if filters.document_ids:
stmt = stmt.where(DocumentChunk.document_id.in_(filters.document_ids))

# topic / effective_before 过滤（合并为一次 join）
needs_join = filters.topic or filters.effective_before
if needs_join:
stmt = stmt.join(PolicyDocument)

if filters.topic:
stmt = stmt.where(PolicyDocument.topic == filters.topic)

if filters.effective_before:
stmt = stmt.where(
(PolicyDocument.effective_date <= filters.effective_before)
| (PolicyDocument.effective_date.is_(None))
)

rows = list(session.scalars(stmt).all())
if not rows:
return []

# 2. 构建 BM25 语料库（对所有子块分词）
corpus = [_tokenize(r.content) for r in rows]
bm25 = BM25Okapi(corpus)

# 3. 计算问题与子块的 BM25 分数
query_tokens = _tokenize(question)
scores = bm25.get_scores(query_tokens)

# 4. 组装结果（保持与向量检索一致的数据结构）
results = []
for idx, row in enumerate(rows):
if scores[idx] <= 0:
continue
results.append({
"chunk_id": row.id,
"content": row.content,
"score": float(scores[idx]),
"metadata": {
"document_id": row.document_id,
"filename": row.document.filename if row.document else "",
"topic": row.document.topic if row.document else "",
"section": row.section or "",
"parent_id": row.parent_id or "",
},
"source": "bm25",
})

# 5. 按分数降序排序并截断
results.sort(key=lambda x: x["score"], reverse=True)
return results[:top_k]

4.3 数据结构一致性

BM25 返回的数据结构和向量检索保持一致：

{
"chunk_id": "uuid",
"content": "子块正文",
"score": 2.5, # BM25 分数（非概率，不可跨方法比较）
"metadata": {
"document_id": "uuid",
"filename": "文件名",
"topic": "主题",
"section": "章节",
"parent_id": "父块ID",
},
"source": "bm25", # 标记来源
}

4.4 过滤条件处理

BM25 的过滤直接写在 SQL 查询中：

# 主题过滤
stmt = stmt.where(PolicyDocument.topic == filters.topic)

# 生效日期过滤（包含 NULL 的情况）
stmt = stmt.where(
(PolicyDocument.effective_date <= filters.effective_before)
| (PolicyDocument.effective_date.is_(None))
)

注意：日期过滤需要 join PolicyDocument 表，但要避免笛卡尔积（多个 join 条件合并为一次 join）。

5. RRF 倒数排名融合

5.1 为什么不用简单加权

假设向量检索返回：

• chunk_A: score=0.85
• chunk_B: score=0.72

BM25 返回：

• chunk_B: score=3.5
• chunk_C: score=2.1

问题：0.85 和 3.5 是不同量纲的分数，不能直接相加。

5.2 RRF 的核心思想

RRF（Reciprocal Rank Fusion）不看原始分数，只看排名位置：

RRF_score = Σ 1 / (k + rank)

• rank：该结果在某一路检索中的排名（从 0 开始）
• k：平滑常数，通常取 60
• 排名越靠前，加分越多

5.3 代码实现

def reciprocal_rank_fusion(
vector_results: list[dict],
bm25_results: list[dict],
k: int = 60,
) –> list[dict]:
"""
RRF 倒数排名融合。

原理：不直接比较分数，只看排名。某结果在两路检索中排名越靠前，得分越高。
score = Σ 1 / (k + rank)
"""
scores: dict[str, float] = {} # chunk_id -> rrf_score
sources: dict[str, set] = {} # chunk_id -> {source_names}
all_items: dict[str, dict] = {} # chunk_id -> 原始数据

# 处理向量结果
for rank, item in enumerate(vector_results):
cid = item["chunk_id"]
scores[cid] = scores.get(cid, 0.0) + 1.0 / (k + rank)
sources.setdefault(cid, set()).add("vector")
if cid not in all_items:
all_items[cid] = item

# 处理 BM25 结果
for rank, item in enumerate(bm25_results):
cid = item["chunk_id"]
scores[cid] = scores.get(cid, 0.0) + 1.0 / (k + rank)
sources.setdefault(cid, set()).add("bm25")
if cid not in all_items:
all_items[cid] = item

# 组装融合结果
ranked = []
for cid, item in all_items.items():
ranked.append({
**item,
"rrf_score": scores.get(cid, 0.0),
"sources": list(sources.get(cid, set())),
})

# 按 RRF 分数降序排序
ranked.sort(key=lambda x: x["rrf_score"], reverse=True)
return ranked

5.4 RRF 的优势

5.5 sources 字段的意义

融合结果增加了 sources 字段，标记该结果来自哪些检索方式：

["vector"] # 仅向量检索命中
["bm25"] # 仅 BM25 命中
["vector", "bm25"] # 两路同时命中（高可信度）

两路同时命中的结果通常更可靠，在后续排序中应该被优先考虑。

6. 混合检索：并行执行向量 + BM25

6.1 整体设计

async def _hybrid_search():
"""并行执行向量检索和 BM25 检索。"""
vector_task = asyncio.to_thread(
vector_store.search,
query=question,
top_k=top_k * 3, # 向量多召回一些，交给融合逻辑排序
document_ids=doc_ids,
)
bm25_task = asyncio.to_thread(
_bm25_search,
question=question,
session=session,
filters=filters,
top_k=top_k,
)
return await asyncio.gather(vector_task, bm25_task)

vector_candidates, bm25_candidates = asyncio.run(_hybrid_search())

6.2 为什么 top_k * 3

向量检索召回 top_k * 3 条，BM25 召回 top_k 条：

• 向量检索召回更多候选，让 RRF 有更充分的选择空间
• BM25 分数本身不精确，召回太多低分候选反而干扰
• 最终融合后截取 top_k 条交给下游

6.3 混合检索完整流程

用户问题
↓
并行执行 [向量检索] + [BM25检索]
↓
向量结果 (30条) BM25结果 (10条)
↓ ↓
└────── RRF 融合 ────┘
↓
融合结果 (去重后 <= 40条)
↓
按 rrf_score 降序排序
↓
截取 top_k (10条)
↓
父块扩展（子块 → 父块）
↓
分数过滤 + 证据截断
↓
最终证据 (5条)

6.4 trace 调试信息

混合检索在 trace 中保留每阶段的中间结果：

trace = {
"vector_results": [...], # 向量检索原始结果
"bm25_results": [...], # BM25 检索原始结果
"rrf_results": [...], # RRF 融合结果
"steps": [
"[vector] 召回 30 条候选",
"[bm25] 召回 8 条候选",
"[rrf] 融合后 35 条候选",
]
}

trace 的作用：

• 排查问题时查看哪一路召回出了问题
• 评估时对比 vector-only vs hybrid 的效果差异
• 监控两路召回的覆盖率和重叠率

7. 结果对比：vector vs hybrid

7.1 测试问题分类

7.2 观察指标

# sources 分布统计
only_vector = sum(1 for c in rrf_results if c.get("sources") == ["vector"])
only_bm25 = sum(1 for c in rrf_results if c.get("sources") == ["bm25"])
both = sum(1 for c in rrf_results if set(c.get("sources", [])) == {"vector", "bm25"})

print(f"仅 Vector: {only_vector}")
print(f"仅 BM25: {only_bm25}")
print(f"两者都有: {both}")

如果 both 比例高，说明两路召回有良好互补；如果 only_bm25 高，说明知识库中有大量精确关键词信息。

8. 踩坑清单：混合检索的 10 个常见错误

坑1：BM25 分数和向量分数直接相加

# 错误：量纲不同，不能直接加
combined_score = vector_score + bm25_score

# 正确：用 RRF 只看排名
rrf_score = 1/(60 + vector_rank) + 1/(60 + bm25_rank)

坑2：分词过滤了数字

# 错误：正则把数字也过滤了
tokens = [t for t in jieba.lcut(text) if re.match(r"^[\\u4e00-\\u9fa5]+$", t)]

# 正确：保留数字和含数字的词
tokens = [t for t in jieba.lcut(text) if t.strip() and not re.match(r"^[\\s\\W]+$", t)]

坑3：BM25 查的是父块不是子块

# 错误：父块太长，BM25 词频稀释
stmt = select(DocumentChunk).where(DocumentChunk.is_parent == True)

# 正确：子块更短，信息更聚焦
stmt = select(DocumentChunk).where(DocumentChunk.is_parent == False)

坑4：忘记 asyncio.to_thread

# 错误：同步函数直接 await，阻塞事件循环
vector_task = vector_store.search(...) # 这是同步函数！

# 正确：放到线程池执行
vector_task = asyncio.to_thread(vector_store.search, ...)

坑5：RRF 的 k 值太小

# 错误：k=1 导致排名差异被过度放大
reciprocal_rank_fusion(v, b, k=1)

# 正确：k=60 是论文推荐值，平滑效果好
reciprocal_rank_fusion(v, b, k=60)

坑6：过滤条件导致笛卡尔积

# 错误：多次 join 同一个表
stmt = stmt.join(PolicyDocument).where(...)
stmt = stmt.join(PolicyDocument).where(...) # 又 join 一次！

# 正确：合并为一次 join
needs_join = filters.topic or filters.effective_before
if needs_join:
stmt = stmt.join(PolicyDocument)

坑7：BM25 结果为空时没有降级处理

# 错误：BM25 为空时直接返回空
if not bm25_results:
return [] # 应该降级为只用向量结果

# 正确：RRF 融合时，一路为空不影响另一路
fused = reciprocal_rank_fusion(vector_results, bm25_results or [])

坑8：没有保留 trace 导致无法排查

# 错误：只返回最终结果，中间过程丢了
trace = {} # 空

# 正确：每阶段都记录
trace["vector_results"] = vector_candidates
trace["bm25_results"] = bm25_candidates
trace["rrf_results"] = candidates
trace["steps"].append(f"[vector] 召回 {len(vector_candidates)} 条")

坑9：混合模式用 vector 的 min_score

# 错误：BM25 分数范围是 0~10，向量是 0~1，用同一个阈值不合理
min_score = 0.5 # 对 BM25 来说太高了

# 正确：混合模式用更低的阈值，或不设阈值
min_score = 0.01 # 让 RRF 分数决定排序

坑10：没有测试 sources 字段

# 错误：不检查两路是否真的有互补
# 应该定期统计 sources 分布，评估混合检索效果

# 正确：在检查脚本中统计
print(f"仅 Vector: {only_vector}")
print(f"仅 BM25: {only_bm25}")
print(f"两者都有: {both}")

9. 总结

9.1 核心要点回顾

9.2 学习路径建议

9.3 延伸方向

• 重排序（Rerank）：在 RRF 后接交叉编码器（Cross-Encoder）精排
• 查询扩展：用 LLM 扩展查询词，提升 BM25 召回率
• 混合权重：尝试 Learned Sparse Retrieval（如 SPLADE）
• 多路融合：向量 + BM25 + 关键词规则 + 知识图谱

10. 文末互动

思考题：

如果本文对你有帮助：

• 点赞支持，让更多 RAG 开发者看到
• 收藏备用，实现混合检索时翻出来参考
• 评论交流，说说你的检索架构选型

作者的话：从单一路径到混合检索，是 RAG 系统从"能用"到"好用"的关键一步。向量检索和 BM25 不是竞争关系，而是互补关系。理解两者的边界，设计合理的融合策略，才能让系统既能理解用户的口语化表达，又能精确命中政策条文中的关键词。