开发者日记

外观

召回率从 61% 到 89%：BM25+向量 RRF 融合实战 ​

问题背景：为什么向量检索会漏掉正确答案 ​

在企业知识库问答中，很多 RAG 系统第一版都会采用“切片 + Embedding + 向量 TopK”的架构。它在语义相似问题上表现不错，例如用户问“怎么关闭自动续费”，文档写的是“取消连续包月”，向量模型仍可能找到相关段落。

但上线后常见另一类失败：文档中包含精确的错误码、接口名、产品型号、人名或缩写，而用户问题也使用同样的关键词。此时向量相似度未必最高，尤其是语料中存在大量泛化描述时，语义相近但不含关键字的内容会挤掉真正答案。反过来，纯 BM25 又难以处理同义改写、口语化提问和跨语言表达。

因此，知识库问答更稳妥的召回层通常采用混合检索：BM25 负责精确词项匹配，向量检索负责语义召回，再用 RRF（Reciprocal Rank Fusion，倒数排名融合）合并两路结果。本文给出一套不依赖特定云厂商、可直接替换到生产系统的最小实现，并用 Recall@K 说明如何验证收益。

方案对比：BM25、向量检索与 RRF 的适用边界 ​

BM25 的核心是词频、逆文档频率和文档长度归一化，优点是可解释、对精确词敏感、无需 GPU；缺点是词不匹配时几乎无法召回。向量检索将文本映射到稠密向量，优点是语义泛化强；缺点是对低频专有名词、数字编号和短查询不稳定，且效果受 Embedding 模型、切片策略影响明显。

RRF 不直接比较 BM25 分数和向量余弦分数，因为两者数值尺度不同。它只使用每个文档在各自排序列表中的名次，常见公式为：

score(d) = Σ 1 / (k + rank_i(d))

其中 rank_i(d) 从 1 开始，k 常取 60。这样可以避免强行归一化 BM25 与向量分数，也降低了调参成本。

用户问题QueryBM25 稀疏检索关键词 / 错误码向量稠密检索语义 / 同义改写RRF 排名融合只看名次，不混合原始分

数据准备：文档切片与评测集设计 ​

先构造一个小型知识库。生产环境可以把 docs 替换为数据库或对象存储中的切片结果，但要保留稳定的 chunk_id，因为 RRF 融合依赖文档唯一标识。

python
# pip install rank-bm25 numpy scikit-learn

from __future__ import annotations

import math
import re
from dataclasses import dataclass
from rank_bm25 import BM25Okapi
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity

# 为了让示例完全离线可运行，这里用 TF-IDF 模拟稠密向量。
# 生产环境可替换为 OpenAI、BGE、M3E 等 Embedding 模型。
docs = [
    {"id": "c1", "text": "订单 E1043 表示优惠券已过期，下单时需要更换有效优惠券。"},
    {"id": "c2", "text": "用户可以在账户设置中关闭自动续费，关闭后下个周期不再扣费。"},
    {"id": "c3", "text": "发票抬头填写错误后，可在订单完成前作废并重新提交申请。"},
    {"id": "c4", "text": "连续包月服务支持随时取消，取消后当前周期仍可继续使用。"},
    {"id": "c5", "text": "接口返回 429 时代表请求过于频繁，客户端应执行指数退避重试。"},
]

# query 是用户问题；gold_ids 是人工标注的答案切片
qa_dataset = [
    {"query": "下单提示 E1043 怎么办？", "gold_ids": {"c1"}},
    {"query": "我想取消连续包月", "gold_ids": {"c4"}},
    {"query": "接口频繁请求报 429 怎么处理？", "gold_ids": {"c5"}},
]

def tokenize(text: str) -> list[str]:
    """极简中文分词：按单字、英文单词和数字切分。生产建议用 jieba 或 IK。"""
    return re.findall(r"[a-zA-Z0-9]+|[\u4e00-\u9fff]", text.lower())

@dataclass
class RetrievalResult:
    doc_id: str
    rank: int
    score: float

这里有一个容易被忽视的点：不要只凭几个主观问题判断召回效果。至少准备 50 到 200 条带标注的问答，覆盖精确关键词、同义词、缩写、跨语言、否定表达和多跳问题。指标优先看 Recall@K，即正确切片是否出现在前 K 条；生成答案是否正确则属于后续端到端评测。

实操步骤：实现 BM25、向量检索与 RRF 融合 ​

下面代码同时建立 BM25 索引和向量索引。BM25 返回按相关性降序的文档 ID；向量检索返回余弦相似度最高的文档 ID。RRF 只消费两个排名列表。

python
class BM25Retriever:
    def __init__(self, documents: list[dict]):
        self.docs = {d["id"]: d["text"] for d in documents}
        corpus = [tokenize(d["text"]) for d in documents]
        self.bm25 = BM25Okapi(corpus)
        self.ids = [d["id"] for d in documents]

    def search(self, query: str, top_k: int = 5) -> list[RetrievalResult]:
        scores = self.bm25.get_scores(tokenize(query))
        ranked = sorted(zip(self.ids, scores), key=lambda x: x[1], reverse=True)
        return [
            RetrievalResult(doc_id=doc_id, rank=i + 1, score=score)
            for i, (doc_id, score) in enumerate(ranked[:top_k])
        ]

class VectorRetriever:
    def __init__(self, documents: list[dict]):
        self.ids = [d["id"] for d in documents]
        texts = [d["text"] for d in documents]
        # char analyzer 适配示例中的中文单字分词；真实 Embedding 不需要这一步
        self.vectorizer = TfidfVectorizer(analyzer="char", ngram_range=(1, 2))
        self.matrix = self.vectorizer.fit_transform(texts)

    def search(self, query: str, top_k: int = 5) -> list[RetrievalResult]:
        qv = self.vectorizer.transform([query])
        scores = cosine_similarity(qv, self.matrix)[0]
        ranked = sorted(zip(self.ids, scores), key=lambda x: x[1], reverse=True)
        return [
            RetrievalResult(doc_id=doc_id, rank=i + 1, score=float(score))
            for i, (doc_id, score) in enumerate(ranked[:top_k])
        ]

def rrf_fuse(
    rank_lists: list[list[RetrievalResult]],
    top_k: int = 5,
    k: int = 60,
    weights: list[float] | None = None,
) -> list[tuple[str, float]]:
    """
    RRF 融合。
    k 越大，头部名次优势越平缓；默认 60 来自经典论文与工程经验。
    weights 可设置不同检索通道权重，例如 [1.0, 1.2]。
    """
    if weights is None:
        weights = [1.0] * len(rank_lists)

    fused_scores: dict[str, float] = {}
    for results, channel_weight in zip(rank_lists, weights):
        for item in results:
            fused_scores[item.doc_id] = fused_scores.get(item.doc_id, 0.0)
            fused_scores[item.doc_id] += channel_weight / (k + item.rank)

    return sorted(fused_scores.items(), key=lambda x: x[1], reverse=True)[:top_k]

接下来执行检索并计算 Recall@5。可以看到，某一路漏掉的文档，如果在另一路排名靠前，RRF 仍可能把它拉回最终候选集。

python
bm25_retriever = BM25Retriever(docs)
vector_retriever = VectorRetriever(docs)

def recall_at_k(retrieved_ids: list[str], gold_ids: set[str], k: int) -> float:
    hit = len(set(retrieved_ids[:k]) & gold_ids)
    return hit / len(gold_ids) if gold_ids else 0.0

def evaluate(top_k: int = 5) -> None:
    bm25_hits = vec_hits = hybrid_hits = 0

    for sample in qa_dataset:
        query = sample["query"]
        gold_ids = sample["gold_ids"]

        bm25_results = bm25_retriever.search(query, top_k=top_k)
        vec_results = vector_retriever.search(query, top_k=top_k)
        hybrid_results = rrf_fuse([bm25_results, vec_results], top_k=top_k)

        bm25_ids = [r.doc_id for r in bm25_results]
        vec_ids = [r.doc_id for r in vec_results]
        hybrid_ids = [doc_id for doc_id, _ in hybrid_results]

        bm25_hits += recall_at_k(bm25_ids, gold_ids, top_k)
        vec_hits += recall_at_k(vec_ids, gold_ids, top_k)
        hybrid_hits += recall_at_k(hybrid_ids, gold_ids, top_k)

    n = len(qa_dataset)
    print(f"BM25 Recall@{top_k}: {bm25_hits / n:.2%}")
    print(f"Vector Recall@{top_k}: {vec_hits / n:.2%}")
    print(f"Hybrid Recall@{top_k}: {hybrid_hits / n:.2%}")

if __name__ == "__main__":
    evaluate(top_k=5)

在真实知识库中，建议每路召回 20 到 100 条，再融合保留 20 到 50 条交给 Cross-Encoder 重排序。若最终只给大模型 5 到 8 个上下文片段，可以采用“多路宽召回、RRF 粗融合、Rerank 精排”的链路。

参数调优：TopK、RRF k 值与通道权重 ​

RRF 的 k=60 通常是稳健起点，但不是不可改变。k 较小时，排名第一的文档优势更大，适合两路结果质量差异明显、希望强信任头部结果的场景；k 较大时，分数差异更平滑，适合召回阶段希望保留更多潜在相关文档。

通道权重也不应凭感觉设置。若评测发现 BM25 在错误码、合同条款、接口名查询上显著更好，可以给 BM25 权重 1.1 到 1.3；若用户问题大量口语化、同义词改写，可提高向量通道权重。每次只改一个变量，并按查询类型分层统计，否则整体平均值可能掩盖某类问题的退化。

此外，RRF 解决的是“召回融合”而不是“最终相关性排序”。当两个切片都包含相同关键词时，BM25 可能返回陈旧但词面匹配的文档，向量也可能召回语义泛泛的说明页。此时加入 Rerank 模型，用“问题 + 候选切片”成对打分，通常能进一步提高上下文精度。

避坑总结：生产落地最常见的 5 个问题 ​

  1. 只看生成回答，不评测召回命中。 常见现象是答案偶发错误，但无法判断是检索没找到、上下文被截断，还是 LLM 没有使用资料。修正方法是为每条问答标注 gold chunk_id，先统计 Recall@K，再统计答案准确率和引用正确率。

  2. 中文 BM25 没有分词或分词粒度错误。 错误现象是搜索“自动续费”只能命中逐字重合的文档，召回异常混乱；英文错误码和数字也可能被粘连。修正方法是为中文引入 jieba、IK、HanLP 等分词器，并统一小写、全半角、同义词和停用词处理，写入与查询两端必须使用同一套分析逻辑。

  3. 直接把 BM25 分数和余弦相似度相加。 这会导致某一路分数尺度更大，另一路形同虚设。例如 BM25 可能是十几分，余弦只有 0 到 1。修正方法是使用 RRF 基于排名融合；若要加权，也应在 RRF 项上设置通道权重，而不是相加原始分数。

  4. 向量切片过大或丢失标题上下文。 一个数百字甚至上千字的切片会稀释专有名词和语义信号，导致正确片段相似度偏低。修正方法是按语义结构切到 200 到 500 字左右，并在切片前拼接文档标题、章节路径、产品名等元数据；检索命中后再回溯相邻上下文。

  5. TopK 设置过小，融合后再 Rerank 没有候选空间。 如果 BM25 和向量各自只取 5 条，一旦正确答案排在某一路第 8 名，RRF 根本看不到它。修正方法是召回阶段扩大到每路 30 条左右，融合后保留 20 条以上交给重排序，最终再选择 5 到 8 条进入提示词。

混合检索不是替代向量数据库的“银弹”，而是用两种互补信号降低漏召风险。真正可复制的收益来自高质量切片、稳定文档 ID、分层评测集和持续参数回归；RRF 则是其中工程成本最低、鲁棒性最高的融合基线。

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。

本文来自开发者日记（devdiary.cn）· 记录开发者实战笔记与免费在线工具箱

免责声明

本工具与文章内容仅用于学习测试场景，优化建议仅供参考，请勿直接用于生产环境，由此产生的风险自行承担。