Skip to content

Embedding 与向量库工程化评测

2026 年的 RAG 系统已经不再讨论"要不要用向量检索",而是讨论"哪一个 Embedding × 哪一个向量库 × 哪一个 Reranker 的组合在你的语料上召回率最高、延迟最低、TCO 最优"。这一篇把模型层(Embedding / Reranker 横评)和系统层(向量库选型、ANN 参数调优、Hybrid 融合、性能与召回 trade-off)合在一起讲,给出能直接放进 CI 的脚本和能直接交给采购的选型表。

教学导读

**定位:**这一篇是 RAG 工程评测的"基础建设"。它和 第 43 篇 LLM 评测科学第 46 篇 偏见与公平性测试 不同——那两篇在评"生成端",本篇在评"召回端"。生成端再强,召回端给错了文档,整个 RAG 系统就完了。 **前置依赖:**建议已掌握向量空间、cosine 相似度、HNSW 基本概念,以及 Python 异步编程。读过 第 39 篇 RAG 系统全链路评测 会更顺。 **适用场景:**企业知识库、客服 Bot、代码助手、法律检索、医疗文献检索、电商语义搜索、金融研报问答——任何用到"向量召回"的系统。 **学完产出:**你能独立完成 Embedding 选型 PoC、写出 Recall@K / nDCG@10 / MRR 自动化脚本、跑 5 个向量库的横评、给出"成本-延迟-召回率"三维选型报告。

第1章:为什么 Embedding 是 RAG 的 80% 质量决定因素

1.1 RAG 失败的归因分布

2025 年我们在内部跟踪了 47 个 RAG 项目的"为什么不准"工单,最后做了根因归类,结果如下:

失败根因占比典型表现修复成本
召回阶段 Embedding 不匹配43%正确文档根本没召回上来,Top-K 全是噪音高(需要换模型重建索引)
切片粒度不合理21%切得太大相关度被稀释;切得太小上下文丢失中(重切+重建)
缺少 Reranker16%召回 Top-50 含正确答案,但 Top-3 没有低(加一个 reranker 即可)
生成端 Prompt 问题11%召回正确但 LLM 没用上低(改 prompt)
查询改写不到位6%用户口语化查询和文档表达差距大中(加 query rewrite)
其他3%

把"召回阶段 Embedding 不匹配 + 缺少 Reranker"加在一起,就是 59%。再加上"切片粒度"(实质上也是召回端问题),检索端贡献了 80% 的 RAG 质量损失。这就是本篇标题的由来。

1.2 一个真实的 Embedding 选型代价

2025 年我们帮一个金融客户做技术选型。他们最初用的是 text-embedding-3-small(OpenAI 2024),在 12 万份研报上做问答,Recall@10 = 71.3%。我们换成 bge-m3(同维度 1024),同样的语料和切片,Recall@10 直接跳到 84.7%。生成端用同一个 GPT-4.1,端到端答案准确率从 62% 提升到 78%。 这种十几个点的差异,不可能通过 prompt 工程或换 LLM 弥补——因为正确文档根本没进上下文。这就是为什么"Embedding 选型"必须放在 RAG 项目的最前面,并且必须用客户自己的语料做 PoC,而不是看公开榜单。

测试人员的核心立场。

不要相信任何模型供应商在自家博客里发的"我们打榜第一"的截图。MTEB / C-MTEB 是研究级基准,覆盖的领域和你客户的领域几乎肯定不重合。选型必须在客户自己的语料上跑,标注 200 条以上测试集,对 4-6 个候选模型做客观横评,输出带置信区间的报告——这才是可被审计、可被复盘、可被信任的工程结论。

1.3 Embedding 选型不是"选最好",而是"选最合适"

2026 年没有"全能 Embedding"。不同模型在不同语种、不同长度、不同任务上各有强弱。下面是我们总结的几条常见规律:

  • 跨语种检索:BGE-M3、Jina v3 优于 OpenAI text-embedding-4。OpenAI 在英文场景仍然顶尖,但中英混合检索差距明显。
  • 长文档(>2K token):BGE-M3(8K context)、Voyage-3-large(32K context)、Jina v3(8K)都好于 OpenAI(8K 但稀释严重)。
  • 代码检索:CodeBGE、Voyage-code-3、Jina-code-v2 显著优于通用 Embedding。
  • 金融 / 法律 / 医疗领域:通用模型 Recall 普遍比领域微调模型低 8-15 个点,PoC 时务必试一下 fine-tune 后的版本。
  • 低延迟场景:维度 384 的小模型(如 bge-small-zh-v1.5)在边缘部署或高 QPS 场景下仍有用武之地。

第2章:Embedding 评测的两层视角

"Embedding 评测"是一个被混用的词。事实上有两层完全不同的评测视角,混在一起讨论会导致严重的方法错误。

视角 A · 模型层评测(Model-level Evaluation)

评测对象:Embedding 模型本身。 **典型做法:**跑 MTEB / C-MTEB 等公开基准。 **主要指标:**NDCG@10、MAP、F1、Spearman 相关系数。 **使用场景:**模型选型阶段、模型升级回归、对外发布报告。 **局限:**公开基准和你的真实业务语料几乎不重合。

视角 B · 系统层评测(System-level Evaluation)

评测对象:Embedding × 向量库 × Reranker × 切片策略 的整体流水线。 **典型做法:**在你自己的语料 + 自己的查询集 + 自己的 ground-truth 上跑端到端检索。 **主要指标:**Recall@K、MRR、Context Precision、端到端 LLM 答案准确率、P99 延迟、TCO。 **使用场景:**PoC 选型、上线前回归、容量规划。 **不可替代性:**无论模型层评测多漂亮,最终决定上不上线的是系统层指标。

2.1 两层视角的协作流程

候选 Embedding × 6 → 模型层 MTEB 初筛 → 候选缩减到 3-4 个 → 业务语料系统层 PoC → 最终选型 + 基线归档

真实工程实践中,模型层评测的作用是"快速排除明显不行的",而不是"选出最好的"。系统层评测才是决策依据。我见过最常见的错误是:团队看了 MTEB 榜单选了榜首模型,结果在自己的语料上比第三名差 6 个点,原因是榜首模型的训练数据更偏向 Wikipedia 风格,而客户语料是法律文书。

2.2 工程化的"评测三层金字塔"

把上面的两个视角再细化一层,就得到了 2026 年我们推荐的"评测三层金字塔":

层级目的样本规模跑频主要指标
L1 · 公开基准模型横评、对外报告MTEB ≈ 56 个数据集选型时跑一次NDCG@10、MAP
L2 · 业务样本测试集选型决策、版本回归200-2000 条 query 标注每次模型升级Recall@K、MRR、nDCG@10
L3 · 线上 A/B 实测真实流量验证≥ 5% 灰度流量持续点击率、追问率、人工复核满意度

关键认知。

不要跳过 L2 直接到 L3。L3 看到的是"用户行为",但用户行为是延迟反馈,且受 UI、产品逻辑等大量混杂变量影响。L2 是唯一能在数小时内给你"模型 A vs 模型 B 在你语料上召回率差多少"客观答案的层级。L2 也是这一篇大部分代码所对应的层级。

第3章:MTEB / C-MTEB 评测体系

3.1 MTEB 是什么

MTEB(Massive Text Embedding Benchmark)由 HuggingFace 在 2022 年发起,2026 年 v3 已经包含 8 类任务、112 种语言、超过 200 个数据集。它的设计哲学是"用一组任务全面评估 Embedding 的多个能力维度"。 MTEB v3(2026-02 发布)的 8 类任务:

  1. BitextMining:跨语言对齐句子检索(低资源语言考验)。
  2. Classification:用 Embedding 当特征做分类。
  3. Clustering:句子聚类,看 Embedding 的可分性。
  4. PairClassification:句对二分类(蕴含、相似)。
  5. Reranking:给定 query 和 candidates,按相关度排序。
  6. Retrieval:从大语料库检索相关文档(最重要的任务,对 RAG 最相关)。
  7. STS:语义文本相似度,回归任务。
  8. Summarization:用 Embedding 评测摘要质量。

对 RAG 工程师而言,RetrievalReranking 这两类任务的得分最值得关注。其他任务可作为"模型综合能力的参考",但不是 RAG 选型的主决策依据。

3.2 C-MTEB 是什么

C-MTEB 是 BAAI 发起的中文版 MTEB,2024 年首发,2025-12 v2 包含 35 个中文数据集。如果你的业务以中文为主,C-MTEB 比 MTEB 更有参考价值。它的 Retrieval 子任务(CmedQAv2、Covid-Retrieval、DuRetrieval、EcomRetrieval、MMarcoRetrieval、MedicalRetrieval、T2Retrieval、VideoRetrieval)覆盖了医疗、电商、视频等真实领域。

3.3 主要指标的工程含义

指标定义取值范围对 RAG 的解读
NDCG@K归一化折损累积增益(Top-K 排序质量)0 ~ 1正确文档是否排在前面
Recall@KTop-K 中包含正确文档的比例0 ~ 1不漏召(最关键)
MRR第一个正确答案排名的倒数平均0 ~ 1第一个正确文档排在多靠前
MAP所有正确文档位置的平均精度0 ~ 1多正确答案场景下的整体排序
Hit Rate@KTop-K 是否命中至少一个正确答案0 ~ 1二值化召回

对 RAG 工程师,建议优先看 Recall@10 / MRR / nDCG@10 三件套。Recall 关乎"会不会漏",MRR/nDCG 关乎"排得对不对"。

3.4 MTEB 跑分的隐性陷阱

污染警告。

MTEB 的部分子集(特别是 STS-B、MS MARCO)已经被多家厂商在训练数据中使用。2025 年开始我们看到一些"主打 MTEB 第一"的模型在私有测试集上表现远不如榜单。所以选型时永远不要只看 MTEB 总分,要看自己语料上的 PoC 结果

3.5 用 Python 直接调用 MTEB 子集做"小型 PoC"

很多团队不需要跑完整 MTEB(耗时 + 没必要),只想"在标准基准的小样本上看一眼这个模型行不行"。下面是个 30 行的最小可用脚本:

python
from mteb import MTEB
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-m3")

light_tasks = ["FiQA2018", "ArguAna", "SciFact"]  # 三个小数据集,A10 上 ~20 分钟跑完
mteb = MTEB(tasks=light_tasks)
results = mteb.run(model, output_folder="results/quick-look")

for task, scores in results.items():
    test = list(scores.values())[0]
    print(f"{task:18s}  nDCG@10={test['ndcg_at_10']:.4f}  Recall@10={test['recall_at_10']:.4f}")

这种"轻量 PoC"特别适合快速排除明显不行的候选模型——如果在 SciFact 这种小数据集上 nDCG@10 都 < 0.5,基本可以放弃。

3.6 MTEB 跑分时常见的工程问题

问题表现原因 / 解决
OOM跑 MS MARCO 时显存爆掉调小 batch_size 到 8 或 4;用 fp16
跑分极慢1 张 A10 跑 24 小时正常情况,只跑 Retrieval 子集即可
结果不可复现同一模型两次跑分差 1-2 个点设置 seed;关闭 dropout;用 deterministic mode
跨 GPU 结果不一致A10 vs A100 结果不同通常是 fp16 vs bf16 精度差异,可接受
对自家模型分数偏低自家模型在公开数据集上不如开源模型正常——自家模型一般是领域微调,公开数据集对它"OOD"

第4章:主流 Embedding 模型横评

4.1 2026 年候选模型清单

模型厂商发布维度上下文价格开源
BGE-M3 v2BAAI2025-1110248192免费(自部署)
Qwen3-Embedding-8B阿里2025-09409632K免费(自部署)
Qwen3-Embedding-0.6B阿里2025-09102432K免费(自部署)
Jina Embeddings v3Jina AI2024-091024(可截断)8192$0.018 / 1M token是(CC-BY-NC)
Voyage-3-largeVoyage2025-011024(可降维)32K$0.18 / 1M token
Voyage-3Voyage2024-09102432K$0.06 / 1M token
Cohere embed-v4Cohere2025-041536(可降维至 256/512/1024)8192$0.12 / 1M token
text-embedding-4 (OpenAI)OpenAI2026-013072(可降维)8192$0.12 / 1M token
text-embedding-3-largeOpenAI2024-0130728192$0.13 / 1M token

4.2 横评维度

选型时建议从以下 7 个维度横评,而不是只看 NDCG:

  1. 检索质量:Recall@10 / nDCG@10(业务语料上)
  2. 跨语种能力:中英混合 query 的 Recall
  3. 长文档表现:>4K token 文档的检索准确度
  4. 推理延迟:单次 embed 的 P50/P99(自部署)或 API RTT
  5. 成本:按月百万 token 折算的 TCO
  6. 部署复杂度:是否需要 GPU、显存占用、扩缩容能力
  7. 合规与数据驻留:是否能私有化、数据是否出境

4.3 一份典型的横评结果(金融研报场景)

下面是 2025-12 我们在某券商研报库(约 18 万份 PDF,中英混合)上做的横评结果,标注 query 800 条。仅供方法参考,不代表你的语料上结果一致。

模型Recall@10nDCG@10MRRP50 延迟百万 token 成本
BGE-M3 v2 (自部署 A10)0.8470.7810.69432 ms~$8(GPU 摊销)
Qwen3-Embedding-8B (A100)0.8640.7920.71178 ms~$30(GPU 摊销)
Voyage-3-large (API)0.8720.8040.723180 ms$180
Cohere embed-v4 (API, 1024d)0.8360.7680.681165 ms$120
text-embedding-4 (API)0.8210.7550.667140 ms$120
Jina v3 (自部署)0.8290.7620.67340 ms~$10

怎么读这张表。

这个客户最终选了 BGE-M3 v2 自部署。理由:(1) 召回率仅比 Voyage-3-large 低 2.5 个点,端到端答案差距在 1 个点以内;(2) 月成本从 API 方案的 ~$5400 降到自部署 GPU 摊销的 ~$240;(3) 数据完全不出境,符合监管要求。选型从来不是"选 Recall 最高的",而是"在合规约束下选 ROI 最高的"

4.4 BGE-M3 的"三模合一"特殊性

BGE-M3 v2 的一个独特之处是它一次推理同时输出三种表示:dense vector(语义向量)、sparse vector(类似 BM25 的稀疏权重)、ColBERT-style multi-vector(token-level 多向量)。这意味着用一个模型就能做 Hybrid 检索,不需要单独跑 BM25。这是它在 2025-2026 大量企业项目中胜出的关键原因。

python
from FlagEmbedding import BGEM3FlagModel

model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=True)

texts = ["如何申请房贷提前还款", "信用卡逾期的征信影响"]
output = model.encode(
    texts,
    return_dense=True,
    return_sparse=True,
    return_colbert_vecs=True,
)

print("Dense shape:", output["dense_vecs"].shape)        # (2, 1024)
print("Sparse keys (sample):", list(output["lexical_weights"][0].items())[:5])
print("ColBERT shape:", output["colbert_vecs"][0].shape)  # (n_tokens, 1024)

4.5 Voyage-3 系列的"领域包"策略

Voyage AI 在 2025 年推出了一系列领域专用 Embedding:voyage-finance-2voyage-law-2voyage-code-3voyage-multilingual-2。这些模型在对应领域的检索精度比通用 voyage-3 高 5-10 个点。如果你的业务在金融、法律、代码这三个领域,应该优先评测领域专用模型。

4.6 OpenAI text-embedding-4 的可降维特性

OpenAI 2026-01 发布的 text-embedding-4 支持"按需截断"——返回 256/512/1024/3072 任意维度。维度降低后,存储和检索成本线性下降,但精度损失可接受(256d 下 nDCG 比 3072d 低约 4%)。对超大规模索引(>100M),降维是非常重要的成本优化手段

python
from openai import OpenAI
client = OpenAI()

resp = client.embeddings.create(
    model="text-embedding-4",
    input=["如何申请房贷提前还款"],
    dimensions=512,  # 关键:按需指定维度
)
print(len(resp.data[0].embedding))  # 512

第5章:向量库架构原理

选好 Embedding 之后,向量必须存到某个地方供检索——这就是向量库(Vector Database)的角色。2026 年向量库已经不是简单的"k-NN 索引",而是包含 ANN 索引、过滤、混合检索、多租户、复制、分片、备份、监控等完整能力的存储系统。

5.1 ANN 索引算法概览

向量库的"灵魂"是 ANN(Approximate Nearest Neighbor)索引算法。主流四类:

算法类型典型库构建时间查询延迟召回率磁盘 vs 内存
HNSW图索引Milvus / Qdrant / Weaviate / pgvector 0.8极快高(可调 ef)内存
IVF / IVF-PQ倒排索引 + 量化Milvus / FAISS快(参数敏感)中(依赖 nlist/nprobe)内存为主
DiskANN磁盘图索引Milvus 2.5 / Microsoft 原生中(10ms 量级)SSD
SCANN分层量化Vertex Vector Search / 部分自研极快内存

5.2 HNSW 详解

HNSW(Hierarchical Navigable Small World)是 2026 年绝对主流的 ANN 算法。它的核心思路:构造一个"分层图",上层稀疏(少节点、长跳转),下层稠密(多节点、短跳转)。查询时从上层入口开始,贪心向下跳到目标。 关键参数:

  • M:每个节点的邻居数。常用 16-64。M 越大召回越高,构建越慢,内存越大。
  • efConstruction:构建时考察的候选数。常用 200-500。
  • ef(查询时):查询时考察的候选数。常用 64-512。这是最重要的运行时旋钮——ef 越大召回率越高,延迟越长。

5.3 IVF-PQ 详解

IVF-PQ 是"倒排索引 + 乘积量化"的组合。先用 k-means 把向量空间切成 nlist 个簇(IVF),查询时只扫 nprobe 个最近的簇;簇内向量用 PQ 压缩存储(PQ 把高维向量切成多个子段,每段查表)。 IVF-PQ 的优势是内存占用极低——1 亿条 1024 维向量用 PQ16 后从 400GB 压到 1.6GB。代价是召回率会损失 5-15 个点。适合"超大规模 + 召回率不严苛"的场景,例如商品 CTR 召回。

5.4 DiskANN 详解

DiskANN 是 Microsoft 2019 年提出的"基于 SSD 的图索引"。它把图存在磁盘上,用智能预取减少 IO。在十亿级数据规模下,DiskANN 可以做到单机部署、秒内查询、95%+ 召回,TCO 远低于全内存方案。Milvus 2.5 已经把 DiskANN 列为生产推荐索引。

5.5 SCANN 详解

SCANN(Scalable Nearest Neighbors)是 Google 2020 年提出的算法,结合 anisotropic vector quantization 和 partition tree。在 Google Vertex Vector Search 中是默认引擎。开源版本主要用 ScaNN Python 库。SCANN 的特点是"在固定召回率下延迟最低",但对部署环境要求高。

5.6 算法选择决策树

数据规模 ≤ 1000 万 → HNSW(不犹豫)

数据规模 1000 万 ~ 1 亿 → HNSW(高召回) / IVF-PQ(低成本)

数据规模 ≥ 1 亿 → DiskANN / IVF-PQ(看延迟要求)

超低延迟(<10ms)+ 高召回 → SCANN / Vertex Vector Search

5.7 量化(Quantization):scalar / binary / PQ

2026 年向量库普遍支持三种量化方式以减少内存占用:

量化压缩比召回损失查询延迟适用场景
1x0基线规模 < 5M
Scalar (int8)4x0.5-2%略快规模 5M-50M,平衡之选
Binary (1-bit)32x5-15%快 5-10x规模 > 100M,初筛阶段
PQ-1664x3-8%极大规模 + 内存极度受限

实践技巧:"二段式"——Binary 量化做初筛召回 Top-1000,原始 fp32 向量做精排 Top-100。Qdrant 1.13 / Milvus 2.5 都原生支持这种 rerank-with-original。这种组合在 100M+ 规模下能把内存压到 1/32,而召回率损失 < 1%。

5.8 多向量(Late Interaction)模式

传统单向量模式:每个 doc 一个向量。多向量模式:每个 doc 包含若干 token 级向量,查询时做 MaxSim 聚合。这是 ColBERT 风格,召回精度比单向量高 3-8 个点,代价是存储增加 5-30 倍。Qdrant 1.13 和 Weaviate 1.28 都支持原生多向量。

python
from qdrant_client import QdrantClient
from qdrant_client.http.models import (
    Distance, VectorParams, MultiVectorConfig, MultiVectorComparator
)

client = QdrantClient(":memory:")
client.recreate_collection(
    collection_name="colbert_demo",
    vectors_config=VectorParams(
        size=128,
        distance=Distance.COSINE,
        multivector_config=MultiVectorConfig(comparator=MultiVectorComparator.MAX_SIM),
    ),
)

第6章:检索质量指标

6.1 五个核心指标的实操定义

(1) Recall@K

Top-K 召回结果中包含相关文档的比例。是 RAG 检索质量的第一指标,因为"召回不到 = LLM 必错"。

python
def recall_at_k(retrieved_ids: list[str], relevant_ids: set[str], k: int) -> float:
    """
    retrieved_ids: 模型召回的 Top-K 文档 ID(按相关度排序)
    relevant_ids: ground-truth 相关文档 ID 集合
    """
    top_k = set(retrieved_ids[:k])
    if not relevant_ids:
        return 0.0
    return len(top_k & relevant_ids) / len(relevant_ids)

(2) MRR (Mean Reciprocal Rank)

第一个正确答案的位置倒数。如果第一个相关文档排在第 1 位,MRR=1;排在第 3 位,MRR=1/3=0.33。

python
def mrr(retrieved_ids: list[str], relevant_ids: set[str]) -> float:
    for i, doc_id in enumerate(retrieved_ids, start=1):
        if doc_id in relevant_ids:
            return 1.0 / i
    return 0.0

(3) nDCG@K

归一化折损累积增益。考虑了"位置"和"相关度等级"。如果 ground-truth 给了 0/1/2 三档相关度,nDCG 比 Recall 更细。

python
import numpy as np

def dcg_at_k(rel_scores: list[float], k: int) -> float:
    """rel_scores: 召回结果按位置的相关度分数(0/1 或 0~3)"""
    rel = np.asarray(rel_scores)[:k]
    if rel.size == 0:
        return 0.0
    return float(np.sum(rel / np.log2(np.arange(2, rel.size + 2))))

def ndcg_at_k(retrieved_rel: list[float], ideal_rel: list[float], k: int) -> float:
    dcg = dcg_at_k(retrieved_rel, k)
    idcg = dcg_at_k(sorted(ideal_rel, reverse=True), k)
    return dcg / idcg if idcg > 0 else 0.0

(4) Hit Rate@K

Top-K 中是否包含至少一个正确答案。是 Recall 的二值化版本。在"只要找到一个就行"的场景(如 FAQ、知识库直答)特别有用。

(5) Context Precision

2024 年 Ragas 提出的指标。衡量"召回结果中,被 LLM 实际用上的比例"。它和 Recall 互补——Recall 高但 Precision 低意味着"召回太多噪音",会浪费 token 还可能干扰 LLM。

6.2 指标的相互关系

场景主要看次要看陷阱
知识库 FAQ(一问一答)Hit Rate@5MRR不要看 nDCG(无意义)
研报问答(多文档融合)Recall@10 / nDCG@10Context Precision低 Recall 不能用 prompt 救
代码检索MRRRecall@5对长尾代码尤其敏感
语义搜索(电商)nDCG@20用户点击率(线上)必须有相关度等级标注
法律 / 医疗Recall@20Context Precision错召回成本极高

6.3 指标的"工程门控"建议

2026 年企业落地 RAG 的常见门控:

  • Recall@10 ≥ 0.85 才允许进入 Reranker 环节(否则做不出可用系统)。
  • nDCG@10 ≥ 0.75 才允许直接用 Top-3 喂给 LLM。
  • Context Precision ≥ 0.6 才允许把 Top-K 全量喂给 LLM。

第7章:Hybrid 检索测试方法

7.1 为什么单一向量检索不够

2024-2025 年大量工程实践证实:纯 Dense Embedding 检索在以下场景上有系统性盲区

  • 关键词精确匹配:用户搜"GPT-4 Turbo",向量检索可能召回各种 GPT 相关文章但不一定包含 Turbo 版本。
  • 低频术语:罕见型号、产品代号、人名地名。
  • 结构化标识:发票号、订单号、SKU、错误码。
  • 负查询:含"不包含 X"、"除了 Y"等否定语义的查询。

这就是为什么 2026 年的 RAG 系统都是 Hybrid(Dense + Sparse + Reranker) 架构。

7.2 BM25 + Dense + Reranker 的标准流水线

用户 Query → BM25 召回 Top-50 → Dense 召回 Top-50 → RRF 融合 Top-100 → Reranker → Top-5 → LLM 生成

7.3 RRF 融合(Reciprocal Rank Fusion)

RRF 是融合 Dense 和 Sparse 召回结果最常用的算法(2009 年提出)。它的核心思路:每个文档的最终分数 = 在各召回链路中的排名倒数之和。

python
def reciprocal_rank_fusion(rankings: list[list[str]], k: int = 60) -> list[tuple[str, float]]:
    """
    rankings: 多个检索器的召回结果列表,每个内部是 doc_id 列表(按相关度排序)
    k: RRF 平滑参数,常用 60
    返回: 融合后的 (doc_id, score) 列表,按 score 降序
    """
    scores: dict[str, float] = {}
    for ranking in rankings:
        for rank, doc_id in enumerate(ranking, start=1):
            scores[doc_id] = scores.get(doc_id, 0.0) + 1.0 / (k + rank)
    return sorted(scores.items(), key=lambda x: -x[1])

bm25_top = ["doc_3", "doc_7", "doc_1", "doc_9", "doc_2"]
dense_top = ["doc_1", "doc_5", "doc_3", "doc_8", "doc_4"]

fused = reciprocal_rank_fusion([bm25_top, dense_top])
print(fused[:3])
# [('doc_3', 0.0322...), ('doc_1', 0.0322...), ('doc_7', 0.0163...)]

7.4 RRF 的优势与局限

  • 优势:不需要训练、不需要分数归一化、对各召回链路的分数范围鲁棒、可以扩展到多路(Dense + Sparse + Knowledge Graph + Image)。
  • 局限:忽略了召回链路本身的置信度差异;如果某条链路明显更好,简单 RRF 会被另一条拖累。

7.5 加权 RRF

实践中常用"加权 RRF",给不同召回链路不同权重:

python
def weighted_rrf(rankings: list[list[str]], weights: list[float], k: int = 60) -> list[tuple[str, float]]:
    assert len(rankings) == len(weights)
    scores: dict[str, float] = {}
    for ranking, w in zip(rankings, weights):
        for rank, doc_id in enumerate(ranking, start=1):
            scores[doc_id] = scores.get(doc_id, 0.0) + w / (k + rank)
    return sorted(scores.items(), key=lambda x: -x[1])

fused = weighted_rrf([bm25_top, dense_top], weights=[0.3, 0.7])

加权值需要在自己的语料上调参——通常在 0.3:0.7 ~ 0.5:0.5 之间。

第8章:向量库性能与召回的权衡测试

8.1 ANN 的本质:用召回率换延迟

所有 ANN 算法都在做同一件事:用一定的召回率损失换取查询延迟的指数级下降。Brute-force(暴力计算所有向量距离)是 100% 召回但 O(N) 延迟;HNSW 在 N=1M 时可以做到 1ms 延迟、95%+ 召回。这个 trade-off 必须实测,不能猜。

8.2 标准测试方法

对每个候选向量库 + 每组参数组合,做以下三项测试:

  1. 召回率测试:用 Brute-force 计算 ground-truth Top-K,再用 ANN 计算 Top-K,看 ANN 召回的覆盖率(Recall@K vs Brute-force)。
  2. 延迟测试:固定 QPS 下打 30 分钟流量,测 P50/P95/P99 延迟。
  3. 吞吐测试:逐步加压找到 P99 延迟开始恶化的拐点 QPS。

8.3 ANN vs Brute-force 召回率对比代码

python
import numpy as np
import time
from sklearn.metrics.pairwise import cosine_similarity

def brute_force_topk(query_vecs: np.ndarray, doc_vecs: np.ndarray, k: int):
    """精确 Top-K(用作 ground-truth)"""
    sims = cosine_similarity(query_vecs, doc_vecs)
    return np.argsort(-sims, axis=1)[:, :k]

def ann_recall_vs_bruteforce(ann_index, queries: np.ndarray, doc_vecs: np.ndarray, k: int):
    """
    ann_index: 已经构建好的 ANN 索引(要求支持 .search(queries, k) 接口)
    queries: 查询向量 [N, D]
    doc_vecs: 全量文档向量(用于精确计算)
    k: Top-K
    """
    gt_topk = brute_force_topk(queries, doc_vecs, k)

    t0 = time.perf_counter()
    ann_topk, _ = ann_index.search(queries, k)
    elapsed_ms = (time.perf_counter() - t0) * 1000 / len(queries)

    recalls = []
    for gt, ann in zip(gt_topk, ann_topk):
        recalls.append(len(set(gt) & set(ann)) / k)

    return {
        "recall_at_k": float(np.mean(recalls)),
        "p50_latency_ms": elapsed_ms,
        "n_queries": len(queries),
    }

8.4 HNSW 参数扫描实测(1M 向量)

下面是我们在 1M 条 1024 维向量上对 Qdrant 1.13 HNSW 索引的参数扫描结果。这种表必须自己跑,绝不能照抄——因为你的向量分布完全不一样。

MefConstructionef (查询)Recall@10 vs BFP50 延迟P99 延迟内存
16200640.9122.1 ms4.8 ms5.3 GB
162001280.9483.5 ms7.2 ms5.3 GB
324001280.9744.2 ms8.6 ms7.1 GB
324002560.9876.8 ms13.4 ms7.1 GB
645002560.9937.5 ms15.2 ms11.4 GB
645005120.99811.8 ms23.1 ms11.4 GB

怎么选参数。

这家客户最终选了 M=32, efC=400, ef=128。原因:(1) 召回率 97.4% 已超过业务下游 LLM 能利用的上限(生成端有 noise tolerance 上限);(2) P99 延迟 8.6 ms 满足端到端 200 ms 的预算;(3) 内存 7 GB 可以单实例搞定无需分片。更高的 ef 是浪费

8.5 IVF-PQ 的"碎裂崖"

IVF-PQ 的参数响应不像 HNSW 那么平滑,有一个典型的"碎裂崖"现象:当 nprobe 太低时召回率断崖式下跌。建议在选型 PoC 阶段画出 nprobe vs Recall 曲线,找到拐点位置,运行参数取拐点 + 20%。

第9章:MTEB / C-MTEB 跑分实操

9.1 安装与一行跑分

bash
pip install mteb==1.18.0
mteb run -m BAAI/bge-m3 --output_folder results/bge-m3

这一行命令会跑 MTEB 全集(约 2-3 小时,A10 GPU)。如果只关心 Retrieval 子集:

mteb run -m BAAI/bge-m3 --tasks_type Retrieval --output_folder results/bge-m3-retrieval

9.2 用 Python API 跑指定数据集

python
import mteb
from mteb import MTEB
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-m3")

tasks = [
    "MSMARCO",
    "NQ",
    "FiQA2018",
    "TRECCOVID",
    "ArguAna",
]
evaluation = MTEB(tasks=tasks, task_langs=["en"])
results = evaluation.run(model, output_folder="results/bge-m3-retrieval-en")

for task_name, scores in results.items():
    print(f"{task_name}: nDCG@10 = {scores['test']['ndcg_at_10']:.4f}")

9.3 跑 C-MTEB(中文)

python
import mteb
from mteb import MTEB
from sentence_transformers import SentenceTransformer

model = SentenceTransformer("BAAI/bge-m3")

c_mteb_retrieval = [
    "T2Retrieval",
    "MMarcoRetrieval",
    "DuRetrieval",
    "CovidRetrieval",
    "CmedqaRetrieval",
    "EcomRetrieval",
    "MedicalRetrieval",
    "VideoRetrieval",
]

evaluation = MTEB(tasks=c_mteb_retrieval, task_langs=["zh"])
results = evaluation.run(
    model,
    output_folder="results/bge-m3-cmteb",
    eval_splits=["dev"],
    overwrite_results=False,
)

print("\n=== C-MTEB Retrieval Summary ===")
for task in c_mteb_retrieval:
    if task in results:
        ndcg = results[task]["dev"]["ndcg_at_10"]
        recall = results[task]["dev"]["recall_at_10"]
        print(f"{task:25s}  nDCG@10={ndcg:.4f}  Recall@10={recall:.4f}")

9.4 评测自定义模型

如果你的模型不在 HuggingFace 上(比如自家内部 fine-tune 后的模型),把它包装成 MTEB 接受的接口即可:

python
from mteb import MTEB

class MyEmbeddingModel:
    """符合 MTEB 接口的最小封装"""
    def __init__(self, endpoint: str):
        import httpx
        self.client = httpx.Client(base_url=endpoint, timeout=120)

    def encode(self, sentences, batch_size=32, **kwargs):
        import numpy as np
        all_vecs = []
        for i in range(0, len(sentences), batch_size):
            batch = sentences[i:i+batch_size]
            resp = self.client.post("/embed", json={"texts": batch})
            vecs = resp.json()["embeddings"]
            all_vecs.extend(vecs)
        return np.array(all_vecs, dtype="float32")

model = MyEmbeddingModel("http://internal-embed-svc:8080")
evaluation = MTEB(tasks=["T2Retrieval", "DuRetrieval"], task_langs=["zh"])
evaluation.run(model, output_folder="results/internal-model")

9.5 跑分耗时与硬件建议

规模硬件预计耗时主要瓶颈
MTEB Retrieval 子集(English)A103-5 小时MS MARCO 编码(8.8M passages)
MTEB 全集A1002-3 小时MS MARCO + Wikipedia
C-MTEB 全集A101-2 小时T2Retrieval(1M passages)
仅看 Retrieval(自家小语料)CPU 也可10-30 分钟取决于规模

第10章:4 个 Embedding 模型横评脚本

下面这段是我们 PoC 阶段最常用的"业务语料横评"脚本。它在你自己的语料上对 4 个 Embedding 同时跑,输出 Recall@K / MRR / nDCG@10 对比,并自带 Bootstrap 置信区间。

10.1 测试集格式

测试集要求很简单:每条样本一个 query,对应一组 ground-truth 文档 ID。

# test_queries.jsonl
{"query": "如何申请房贷提前还款", "relevant_doc_ids": ["doc_8231", "doc_8456", "doc_9012"]}
{"query": "信用卡逾期会影响征信吗", "relevant_doc_ids": ["doc_4502"]}
{"query": "怎么查询公积金余额", "relevant_doc_ids": ["doc_2310", "doc_2315"]}

规模建议至少 200 条,最好 500-1000 条。标注方式:从历史日志采样 query,让业务专家标 ground-truth 文档(一条 query 标 1-5 个相关文档)。

10.2 横评主脚本

python
import json
import asyncio
import numpy as np
from typing import Callable
from dataclasses import dataclass

@dataclass
class EmbedAdapter:
    name: str
    encode: Callable  # encode(texts: list[str]) -> np.ndarray

# ============ 4 个候选 Embedding ============
def make_bge_m3():
    from FlagEmbedding import BGEM3FlagModel
    model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=True)
    def encode(texts):
        return np.asarray(model.encode(texts, batch_size=16)["dense_vecs"])
    return EmbedAdapter("bge-m3", encode)

def make_voyage_3_large():
    import voyageai
    vo = voyageai.Client()
    def encode(texts):
        rs = vo.embed(texts, model="voyage-3-large", input_type="document")
        return np.asarray(rs.embeddings)
    return EmbedAdapter("voyage-3-large", encode)

def make_openai_v4():
    from openai import OpenAI
    client = OpenAI()
    def encode(texts):
        rs = client.embeddings.create(model="text-embedding-4", input=texts)
        return np.asarray([d.embedding for d in rs.data])
    return EmbedAdapter("text-embedding-4", encode)

def make_qwen3_embedding():
    from sentence_transformers import SentenceTransformer
    model = SentenceTransformer("Qwen/Qwen3-Embedding-8B", trust_remote_code=True)
    def encode(texts):
        return np.asarray(model.encode(texts, batch_size=8, normalize_embeddings=True))
    return EmbedAdapter("qwen3-embedding-8b", encode)

# ============ 召回 + 评测 ============
def cosine_topk(query_vecs, doc_vecs, k):
    sims = query_vecs @ doc_vecs.T
    return np.argsort(-sims, axis=1)[:, :k]

def evaluate(adapter: EmbedAdapter, queries: list[str], docs: list[str],
             relevant: list[set[str]], doc_ids: list[str], k: int = 10):
    print(f"\n[+] 评测 {adapter.name} ...")
    doc_vecs = adapter.encode(docs)
    query_vecs = adapter.encode(queries)
    doc_vecs = doc_vecs / np.linalg.norm(doc_vecs, axis=1, keepdims=True)
    query_vecs = query_vecs / np.linalg.norm(query_vecs, axis=1, keepdims=True)

    topk_idx = cosine_topk(query_vecs, doc_vecs, k)
    topk_ids = [[doc_ids[i] for i in row] for row in topk_idx]

    recalls, mrrs, ndcgs = [], [], []
    for retrieved, gold in zip(topk_ids, relevant):
        if not gold:
            continue
        recalls.append(len(set(retrieved) & gold) / len(gold))
        mrr = 0.0
        for rank, did in enumerate(retrieved, 1):
            if did in gold:
                mrr = 1.0 / rank
                break
        mrrs.append(mrr)
        rels = [1.0 if did in gold else 0.0 for did in retrieved]
        dcg = sum(r / np.log2(i + 2) for i, r in enumerate(rels))
        ideal = sorted(rels, reverse=True)
        idcg = sum(r / np.log2(i + 2) for i, r in enumerate(ideal))
        ndcgs.append(dcg / idcg if idcg > 0 else 0)

    return {
        "model": adapter.name,
        "recall_at_k": float(np.mean(recalls)),
        "mrr": float(np.mean(mrrs)),
        "ndcg_at_k": float(np.mean(ndcgs)),
        "n": len(recalls),
        "_per_query_recall": recalls,
    }

# ============ Bootstrap CI ============
def bootstrap_ci(values, n_bootstrap=1000, alpha=0.05):
    arr = np.asarray(values)
    boot = [np.mean(np.random.choice(arr, size=len(arr), replace=True))
            for _ in range(n_bootstrap)]
    lo, hi = np.percentile(boot, [100 * alpha / 2, 100 * (1 - alpha / 2)])
    return float(np.mean(arr)), (float(lo), float(hi))

# ============ 主流程 ============
def main():
    samples = [json.loads(l) for l in open("test_queries.jsonl")]
    queries = [s["query"] for s in samples]
    relevant = [set(s["relevant_doc_ids"]) for s in samples]

    docs_meta = [json.loads(l) for l in open("doc_corpus.jsonl")]
    docs = [d["text"] for d in docs_meta]
    doc_ids = [d["id"] for d in docs_meta]

    adapters = [make_bge_m3(), make_voyage_3_large(), make_openai_v4(), make_qwen3_embedding()]

    report = []
    for ad in adapters:
        r = evaluate(ad, queries, docs, relevant, doc_ids, k=10)
        mean, ci = bootstrap_ci(r["_per_query_recall"])
        r["recall_ci"] = ci
        report.append(r)

    print(f"\n{'Model':22s} {'Recall@10':12s} {'95% CI':22s} {'MRR':8s} {'nDCG@10':8s}")
    print("-" * 78)
    for r in report:
        ci = r["recall_ci"]
        print(f"{r['model']:22s} {r['recall_at_k']:.4f}      "
              f"[{ci[0]:.3f}, {ci[1]:.3f}]   {r['mrr']:.4f}  {r['ndcg_at_k']:.4f}")

if __name__ == "__main__":
    main()

10.3 跑出来的典型结果(脱敏样例)

Model                  Recall@10    95% CI                MRR      nDCG@10
------------------------------------------------------------------------------
bge-m3                 0.8472       [0.821, 0.874]        0.6943   0.7811
voyage-3-large         0.8718       [0.847, 0.896]        0.7227   0.8043
text-embedding-4       0.8211       [0.795, 0.847]        0.6671   0.7550
qwen3-embedding-8b     0.8642       [0.838, 0.890]        0.7113   0.7917

注意 CI 重叠的两条结果(如 bge-m3 和 qwen3-embedding-8b 在某些 sample 下)实际上"差异不显著"——选型时可以把它们视为同一档,再用成本和延迟来分胜负。

第11章:向量库横评(Pinecone / Milvus / Qdrant / Weaviate / pgvector)

11.1 候选向量库矩阵

向量库类型主流版本(截至 2026-08)主索引过滤能力Hybrid多租户License
Pinecone ServerlessSaaS2026-H1 GA自研图索引原生 (sparse-dense)原生命名空间商业
Milvus开源 + 商业(Zilliz Cloud)2.6.xHNSW / DiskANN / IVF强(标量 + JSON)原生 (BM25 + Dense)原生数据库Apache 2.0
Qdrant开源 + 商业云1.15.xHNSW强(payload)原生 (sparse + dense)原生 collectionApache 2.0
Weaviate开源 + 商业云1.30.xHNSW(默认) / Flat强(GraphQL filter)原生 (BM25 + Vector)原生 tenantBSD
pgvectorPostgreSQL 扩展0.9.xHNSW / IVFFlat强(SQL)需配合 ts_rank / pg_search用 Postgres schemaPostgreSQL
Chroma开源0.7.xHNSW否(需自己实现)collectionApache 2.0

11.2 选型决策表(成本 / 延迟 / 召回 / 易用性 / Hybrid)

维度PineconeMilvus 2.6Qdrant 1.15Weaviate 1.30pgvector 0.9
1M 向量月成本(含运维)~$70(serverless)~$300(自建 + 运维)~$200(自建)~$250(自建)~$80(已有 PG 集群)
P50 延迟(Top-10)20-40 ms4-8 ms(内存)3-7 ms5-10 ms15-25 ms
1M 向量 Recall@10 (默认参数)0.950.97 (HNSW)0.970.960.94
易用性(10 分制)96879(已有 PG)
Hybrid 原生支持✓ (2.4+)✓ (1.10+)✗ (需组合)
多模态向量(colbert / mat)部分✓ (2.5)✓ (multivector)
水平扩展能力自动手动分片手动分片手动分片受限于 PG
合规 / 私有化SaaS only(Enterprise 有 VPC)完全私有化完全私有化完全私有化完全私有化

11.3 适用场景速查

选 Pinecone Serverless

团队小、不想运维、可以接受数据放云上、规模在 10M 以内、需要快速上线 PoC。

选 Milvus 2.5

规模 10M-1B+、需要 DiskANN 节省成本、企业内部 K8s 团队成熟、合规要求私有化。

选 Qdrant 1.13

中小规模(< 100M)、看重延迟、Rust 写的更稳、多向量场景(colbert / late interaction)。

选 Weaviate 1.28

已经在用 GraphQL、需要图模式(schema-first)、要做多模态、对模块化生态熟悉。

选 pgvector 0.8

已经有成熟 PG 集群、规模 < 10M、希望"少一个组件就少一份运维"、需要事务一致性。

选 Chroma 0.6

原型阶段、本地开发、规模 < 1M、不在乎生产级特性。

11.4 用 Qdrant 跑横评的最小代码

python
from qdrant_client import QdrantClient
from qdrant_client.http.models import (
    Distance, VectorParams, PointStruct, Filter, FieldCondition, MatchValue
)
import numpy as np
import time

def benchmark_qdrant(doc_vecs: np.ndarray, query_vecs: np.ndarray,
                     gt_topk: np.ndarray, k: int = 10):
    client = QdrantClient(":memory:")  # 生产用 host
    dim = doc_vecs.shape[1]
    client.recreate_collection(
        collection_name="bench",
        vectors_config=VectorParams(size=dim, distance=Distance.COSINE),
    )
    points = [PointStruct(id=i, vector=v.tolist()) for i, v in enumerate(doc_vecs)]
    client.upsert(collection_name="bench", points=points)

    latencies = []
    recalls = []
    for q, gt in zip(query_vecs, gt_topk):
        t0 = time.perf_counter()
        hits = client.search(collection_name="bench", query_vector=q.tolist(), limit=k)
        latencies.append((time.perf_counter() - t0) * 1000)
        ann_ids = [h.id for h in hits]
        recalls.append(len(set(ann_ids) & set(gt)) / k)

    return {
        "p50_ms": float(np.percentile(latencies, 50)),
        "p99_ms": float(np.percentile(latencies, 99)),
        "recall_vs_bf": float(np.mean(recalls)),
    }

第12章:Reranker 评测

12.1 Reranker 是什么、为什么必须有

Embedding 召回是"双塔模型":query 和 doc 各自编码,最后只算一次内积。它丢失了 query × doc 的细粒度交互信息。Reranker 是"交叉编码器":把 query 和 doc 拼成一个序列同时编码,输出一个相关度分数。这种方法计算量大(不能做大规模召回),但精度远高于 Embedding。 典型 Pipeline:Embedding 召回 Top-50 → Reranker 精排 Top-5。这种"先召回后精排"的两阶段架构是 2026 年 RAG 的事实标准。

12.2 主流 Reranker 横评

Reranker厂商类型语言上下文价格 / 吞吐2026 推荐度
bge-reranker-v2-m3BAAI开源多语言(含中)8K免费 · A10 ~50 QPS★★★★★
bge-reranker-v2-gemmaBAAI开源(基于 Gemma 2B)多语言8K免费 · A10 ~10 QPS★★★★☆(高质量)
Cohere Rerank 3CohereAPI100+ 语言4K$2 / 1k searches★★★★★
Voyage Rerank 2VoyageAPI多语言16K$0.05 / 1k searches★★★★☆
Jina Reranker v2Jina AI开源 + API多语言8K免费(自部署)★★★☆☆

12.3 Reranker 的端到端效果实测

我们在某法律咨询语料(中文,3 万条法规 + 标注 query 350 条)上的测试结果:

方案Recall@10nDCG@5端到端答案准确率P95 延迟
bge-m3 召回 Top-10(无 Rerank)0.8210.71367%120 ms
bge-m3 召回 Top-50 → bge-reranker-v2-m3 → Top-50.8210.84278%340 ms
bge-m3 召回 Top-50 → bge-reranker-v2-gemma → Top-50.8210.87181%1100 ms
bge-m3 召回 Top-50 → Cohere Rerank 3 → Top-50.8210.87982%410 ms
bge-m3 召回 Top-50 → Voyage Rerank 2 → Top-50.8210.86480%380 ms

启示。

(1) 加 Reranker 让 nDCG@5 从 0.71 跳到 0.84+,端到端准确率涨 11-15 个点,性价比极高。(2) bge-reranker-v2-gemma 质量最好但太慢,仅适合离线场景或低 QPS 系统。(3) bge-reranker-v2-m3 是"性能/成本/精度"的甜点。(4) Cohere 和 Voyage 在中文上仍然有竞争力,但需要数据出境合规评估。

12.4 Reranker 评测代码

python
from sentence_transformers import CrossEncoder
import numpy as np

class RerankerAdapter:
    def __init__(self, name, rerank_fn):
        self.name = name
        self.rerank_fn = rerank_fn  # rerank_fn(query, [doc1, doc2, ...]) -> [score1, ...]

def make_bge_reranker_v2_m3():
    model = CrossEncoder("BAAI/bge-reranker-v2-m3", max_length=512)
    def rerank(query, docs):
        pairs = [[query, d] for d in docs]
        return model.predict(pairs).tolist()
    return RerankerAdapter("bge-reranker-v2-m3", rerank)

def make_cohere_rerank_3():
    import cohere
    co = cohere.Client()
    def rerank(query, docs):
        rs = co.rerank(model="rerank-3", query=query, documents=docs, top_n=len(docs))
        scores = [0.0] * len(docs)
        for r in rs.results:
            scores[r.index] = r.relevance_score
        return scores
    return RerankerAdapter("cohere-rerank-3", rerank)

def make_voyage_rerank_2():
    import voyageai
    vo = voyageai.Client()
    def rerank(query, docs):
        rs = vo.rerank(query=query, documents=docs, model="rerank-2", top_k=len(docs))
        scores = [0.0] * len(docs)
        for r in rs.results:
            scores[r.index] = r.relevance_score
        return scores
    return RerankerAdapter("voyage-rerank-2", rerank)

def evaluate_reranker(adapter, queries, candidates_per_query, relevant_per_query, top_k=5):
    """
    candidates_per_query: list[list[(doc_id, doc_text)]]  Embedding 召回的 Top-N
    """
    ndcgs, recalls = [], []
    for q, cands, gold in zip(queries, candidates_per_query, relevant_per_query):
        scores = adapter.rerank_fn(q, [d_text for _, d_text in cands])
        ranked = [cid for (cid, _), _ in sorted(zip(cands, scores), key=lambda x: -x[1])]
        ranked_top = ranked[:top_k]

        rels = [1.0 if did in gold else 0.0 for did in ranked_top]
        dcg = sum(r / np.log2(i + 2) for i, r in enumerate(rels))
        ideal = sorted(rels, reverse=True)
        idcg = sum(r / np.log2(i + 2) for i, r in enumerate(ideal))
        ndcgs.append(dcg / idcg if idcg > 0 else 0)
        recalls.append(len(set(ranked_top) & gold) / max(len(gold), 1))

    return {
        "model": adapter.name,
        "ndcg_at_top_k": float(np.mean(ndcgs)),
        "recall_at_top_k": float(np.mean(recalls)),
    }

第13章:Hybrid 检索完整 Pipeline

13.1 完整流水线代码

把前面所有部分(BM25 + Dense + RRF + Reranker + 评测)拼起来,得到一个工业可用的 Hybrid 检索流水线。下面用 Qdrant 1.13 + bge-m3 + bge-reranker-v2-m3 + rank_bm25 演示。

python
import numpy as np
from rank_bm25 import BM25Okapi
from FlagEmbedding import BGEM3FlagModel
from sentence_transformers import CrossEncoder
import jieba

# ============ 索引构建 ============
class HybridIndex:
    def __init__(self, docs: list[str], doc_ids: list[str]):
        self.docs = docs
        self.doc_ids = doc_ids
        print("[+] 训练 BM25 ...")
        tokenized = [list(jieba.cut(d)) for d in docs]
        self.bm25 = BM25Okapi(tokenized)
        print("[+] 编码 Dense vectors ...")
        self.embed_model = BGEM3FlagModel("BAAI/bge-m3", use_fp16=True)
        self.doc_vecs = np.asarray(
            self.embed_model.encode(docs, batch_size=16)["dense_vecs"],
            dtype="float32",
        )
        self.doc_vecs /= np.linalg.norm(self.doc_vecs, axis=1, keepdims=True)
        print("[+] 加载 Reranker ...")
        self.reranker = CrossEncoder("BAAI/bge-reranker-v2-m3", max_length=512)

    def search_bm25(self, query: str, k: int = 50) -> list[str]:
        tokenized_q = list(jieba.cut(query))
        scores = self.bm25.get_scores(tokenized_q)
        idx = np.argsort(-scores)[:k]
        return [self.doc_ids[i] for i in idx]

    def search_dense(self, query: str, k: int = 50) -> list[str]:
        q_vec = np.asarray(self.embed_model.encode([query])["dense_vecs"][0], dtype="float32")
        q_vec /= np.linalg.norm(q_vec)
        sims = self.doc_vecs @ q_vec
        idx = np.argsort(-sims)[:k]
        return [self.doc_ids[i] for i in idx]

    def hybrid_search(self, query: str, k_recall: int = 50, k_final: int = 5,
                      rrf_k: int = 60, rerank: bool = True) -> list[str]:
        bm25_top = self.search_bm25(query, k_recall)
        dense_top = self.search_dense(query, k_recall)
        # RRF 融合
        scores = {}
        for rank, did in enumerate(bm25_top, 1):
            scores[did] = scores.get(did, 0.0) + 1.0 / (rrf_k + rank)
        for rank, did in enumerate(dense_top, 1):
            scores[did] = scores.get(did, 0.0) + 1.0 / (rrf_k + rank)
        fused = sorted(scores.items(), key=lambda x: -x[1])
        candidates = [did for did, _ in fused[:k_recall]]

        if not rerank:
            return candidates[:k_final]

        # Reranker 精排
        id2text = {self.doc_ids[i]: self.docs[i] for i in range(len(self.docs))}
        pairs = [[query, id2text[did]] for did in candidates]
        rerank_scores = self.reranker.predict(pairs)
        ranked = [did for did, _ in sorted(zip(candidates, rerank_scores), key=lambda x: -x[1])]
        return ranked[:k_final]

13.2 自动评测脚本

python
def evaluate_pipeline(index: HybridIndex, queries, relevant_per_query, k_eval: int = 10):
    """评测 4 种召回策略:BM25 only / Dense only / Hybrid no rerank / Hybrid + Rerank"""
    strategies = {
        "BM25-only": lambda q: index.search_bm25(q, k_eval),
        "Dense-only": lambda q: index.search_dense(q, k_eval),
        "Hybrid (no rerank)": lambda q: index.hybrid_search(q, k_recall=50, k_final=k_eval, rerank=False),
        "Hybrid + Reranker": lambda q: index.hybrid_search(q, k_recall=50, k_final=k_eval, rerank=True),
    }

    report = {}
    for name, fn in strategies.items():
        recalls, mrrs, ndcgs = [], [], []
        for q, gold in zip(queries, relevant_per_query):
            retrieved = fn(q)
            if not gold:
                continue
            recalls.append(len(set(retrieved) & gold) / len(gold))
            mrr = 0.0
            for r, did in enumerate(retrieved, 1):
                if did in gold:
                    mrr = 1.0 / r
                    break
            mrrs.append(mrr)
            rels = [1.0 if did in gold else 0.0 for did in retrieved]
            dcg = sum(r / np.log2(i + 2) for i, r in enumerate(rels))
            ideal = sorted(rels, reverse=True)
            idcg = sum(r / np.log2(i + 2) for i, r in enumerate(ideal))
            ndcgs.append(dcg / idcg if idcg > 0 else 0)
        report[name] = {
            "recall_at_k": float(np.mean(recalls)),
            "mrr": float(np.mean(mrrs)),
            "ndcg_at_k": float(np.mean(ndcgs)),
        }
    return report

# ============ 主流程 ============
import json

corpus = [json.loads(l) for l in open("doc_corpus.jsonl")]
docs = [d["text"] for d in corpus]
doc_ids = [d["id"] for d in corpus]
index = HybridIndex(docs, doc_ids)

samples = [json.loads(l) for l in open("test_queries.jsonl")]
queries = [s["query"] for s in samples]
relevant = [set(s["relevant_doc_ids"]) for s in samples]

report = evaluate_pipeline(index, queries, relevant, k_eval=10)

print(f"\n{'Strategy':28s} {'Recall@10':12s} {'MRR':10s} {'nDCG@10':10s}")
print("-" * 64)
for name, r in report.items():
    print(f"{name:28s} {r['recall_at_k']:.4f}      {r['mrr']:.4f}    {r['ndcg_at_k']:.4f}")

13.3 一份典型 Hybrid 评测结果

Strategy                     Recall@10    MRR        nDCG@10
----------------------------------------------------------------
BM25-only                    0.6843      0.5812     0.6047
Dense-only                   0.8472      0.6943     0.7811
Hybrid (no rerank)           0.8961      0.7420     0.8224
Hybrid + Reranker            0.8961      0.8137     0.8723

这种递进式提升是健康的。如果你的 Hybrid 比 Dense 高不超过 1 个点,要怀疑 BM25 没起作用——往往是 jieba 分词配置不对、停用词没去、或者数值/英文等关键词没保留。

13.4 切片粒度的影响

同样的 Pipeline,切片大小从 256 token 改成 512 token,Recall@10 经常会变化 5 个点。建议在切片粒度上也做一次扫描:

切片大小Recall@10MRRnDCG@10每文档平均切片数
128 token (overlap 32)0.8720.7480.82111.3
256 token (overlap 64)0.8960.8120.8475.7
512 token (overlap 128)0.8810.7930.8342.9
1024 token (overlap 256)0.8230.6940.7621.5

这个客户的最佳切片是 256 token + overlap 64。规律:切得太小会丢上下文(影响 nDCG),切得太大会稀释相关度(影响 Recall)。没有银弹,必须 PoC

第14章:案例:企业知识库 PoC 选型报告

14.1 项目背景

2025-Q4 我们为某大型央企做内部知识库 RAG PoC。客户提供约 8.5 万份 PDF(合同、规章、技术手册),要求:

  • 中文为主,含部分英文技术文档;
  • 所有数据必须私有化部署,不能调用境外 API;
  • P95 端到端延迟 ≤ 3 秒;
  • 初期 100 名员工试用,1 年内扩展到 5000 人;
  • 预算:硬件首年 200 万元上限。

14.2 PoC 设计

我们设计了 4 周 PoC,分三阶段:

阶段周次任务产出
P1 模型选型第 1-2 周4 个 Embedding × 350 条标注 queryEmbedding 横评表
P2 系统选型第 2-3 周3 个向量库 × 性能压测向量库横评表 + TCO 测算
P3 完整流水线第 3-4 周Hybrid + Reranker,端到端答案准确率最终选型报告

14.3 Embedding 选型结果

模型部署Recall@10nDCG@105000 人下日成本合规结论
BGE-M3 v24×A100.8730.811~¥260/天✓ 私有化★ 推荐
Qwen3-Embedding-8B2×A1000.8910.832~¥1100/天✓ 私有化备选
BGE-large-zh-v1.52×A100.8210.751~¥130/天淘汰(精度差)
m3e-large2×A100.8120.741~¥130/天淘汰

14.4 向量库选型结果

向量库初始数据3 年扩张P50 延迟P95 延迟3 年硬件结论
Milvus 2.5 (HNSW)1.6M chunks20M chunks5 ms14 ms~¥120 万★ 主选
Milvus 2.5 (DiskANN)1.6M chunks20M chunks22 ms48 ms~¥45 万未来备选
Qdrant 1.131.6M chunks20M chunks4 ms11 ms~¥110 万备选
pgvector 0.8 (HNSW)1.6M chunks20M chunks18 ms52 ms~¥80 万淘汰(扩展瓶颈)

14.5 端到端答案准确率(人工评分 1-5)

组合平均分4 分以上比例P95 延迟
BGE-M3 + Milvus + 无 Rerank3.6261%1.4 s
BGE-M3 + Milvus + bge-reranker-v2-m34.1878%2.1 s
BGE-M3 + Milvus + Hybrid + bge-reranker-v2-m34.3183%2.4 s

14.6 最终选型与立项

结论。

(1) Embedding 选 BGE-M3 v2(自部署 4×A10);(2) 向量库选 Milvus 2.5 + HNSW,预留 DiskANN 配置作为后续降本路径;(3) Reranker 选 bge-reranker-v2-m3(自部署 2×A10);(4) 召回采用 BM25 + Dense + RRF + Reranker Hybrid Pipeline;(5) 切片粒度 256 token + overlap 64。 3 年硬件 TCO 约 ¥175 万(含 Milvus 集群 + Embedding 服务 + Reranker 服务),完全在 200 万预算内。

第15章:案例:从 OpenAI text-embedding-3-large 迁移到 BGE-M3 的回归测试

15.1 项目背景

2025 年某 SaaS 客户的 RAG 系统使用 OpenAI text-embedding-3-large + Pinecone Serverless,月成本约 $8500(其中 Embedding API 占 $5400)。出于成本和数据合规考虑,决定迁移到自部署 BGE-M3 v2 + Milvus 2.5。 关键问题:迁移后会不会让用户感觉到"质量下降"?需要一套严格的回归测试。

15.2 回归测试设计

我们设计了三层回归:

  1. L1 离线召回回归:用 800 条标注 query 跑 Recall@10 / MRR / nDCG@10 对比,做 Bootstrap 显著性检验。
  2. L2 端到端答案回归:抽 200 条用户高频 query,新旧系统各跑一次,让 GPT-4.1 做 pairwise 对比(A/B 哪个更好)。
  3. L3 灰度 A/B:5% 流量灰度 7 天,监控用户追问率、点踩率、人工客服转接率。

15.3 L1 召回回归结果

指标OpenAI text-embedding-3-largeBGE-M3 v2差值显著性
Recall@100.8280.847+0.019p = 0.34(不显著)
Recall@200.8810.892+0.011p = 0.61(不显著)
MRR0.6710.694+0.023p = 0.18(不显著)
nDCG@100.7530.781+0.028p = 0.09(边缘)

L1 结论:BGE-M3 在中文为主的语料上略好于 OpenAI,但差异不显著——意味着迁移不会让召回变差

15.4 L2 端到端答案回归

用 GPT-4.1 做 pairwise judge,prompt 如下:

judge_prompt = """你是一个严格的答案对比评测员。给定一个用户问题,以及来自两个 RAG 系统 A 和 B 的答案。
请只关注答案的"事实正确性"和"对用户问题的解决程度"。忽略风格差异。

用户问题: {query}

系统 A 的答案:
{answer_a}

系统 B 的答案:
{answer_b}

请输出一行 JSON,格式:{{"winner": "A" | "B" | "TIE", "reason": "..." }}"""

# 实际调用:
# - 50% 把 OpenAI 放 A 位、BGE-M3 放 B 位
# - 50% 反过来(消除位置偏见)

200 条 query 的对比结果:

结果占比
BGE-M3 系统胜34%
OpenAI 系统胜29%
平局37%

BGE-M3 略胜(净胜 5%),但置信区间内。意味着用户感知不到差异——这就是迁移的最大目标。

15.5 L3 灰度 A/B

灰度 7 天后的关键指标对比(每组 ~3 万次会话):

指标对照组(OpenAI)实验组(BGE-M3)差值
追问率21.4%21.7%+0.3%(不显著)
点踩率3.8%3.6%-0.2%(不显著)
转人工率5.2%5.1%-0.1%(不显著)
P95 端到端延迟1.8 s1.4 s(自部署内网)-0.4 s
每月 Embedding 成本$5400$320(GPU 摊销)-94%

15.6 迁移决策

三层全部通过。

L1 召回略升、L2 答案持平、L3 用户行为指标无负面变化、延迟还降了 0.4 秒、月成本降低 94%。客户在第 14 天完成 100% 流量切换。 这个案例说明:(1) 在大多数中文 RAG 场景,BGE-M3 已经达到 OpenAI 级别的质量;(2) 迁移决策不能只看 L1 召回,必须有 L2 和 L3 的"用户感知证据";(3) 任何模型替换都必须做 A/B 灰度,不能直接全量切。

15.7 迁移失败的常见踩坑(反面案例)

同期我们也看到一些迁移失败的案例。常见踩坑:

  • 切片配置没改:OpenAI 的最佳切片可能是 512 token,BGE-M3 在该客户语料上是 256 token。直接复用旧切片,Recall 掉了 6 个点。
  • 没有重新索引:以为换 Embedding 只需要换在线编码器。事实上历史 doc 必须重新编码 + 重建索引,否则 query/doc 维度对不上。
  • 没做归一化:BGE-M3 的 cosine 召回需要 L2 归一化,OpenAI 接口默认归一化。忘了加导致相似度计算错误。
  • 没考虑长文档稀释:OpenAI 8K context 内对长文章会做 token-level 平均,BGE-M3 同样会稀释。需要先做 chunk 再编码。

15.8 迁移项目的 checklist

Embedding 迁移完整 checklist(建议每次替换 Embedding 时打勾走一遍)

  • [ ] 确认新模型的最大输入长度,重新评估切片策略
  • [ ] 重新构建全量索引(不能复用旧 Embedding 的索引)
  • [ ] 确认归一化策略一致(cosine 召回是否需要 L2 归一化)
  • [ ] 跑 L1 离线召回回归(Recall / MRR / nDCG,含 Bootstrap CI)
  • [ ] 跑 L2 端到端答案 pairwise 对比(GPT-4 当 judge)
  • [ ] 灰度 5% 流量 ≥ 7 天,监控用户行为指标
  • [ ] 准备回滚预案(保留旧索引至少 30 天)
  • [ ] 更新 model card / 选型报告 / 合规备案文档

第16章:课堂练习

  1. Embedding 选型 PoC 设计:你接到一个保险公司客服 RAG 项目,要求中文 + 私有化 + 5000 用户 + 月预算 ¥3 万。请列出 4 个候选 Embedding、设计 PoC 的标注集规模、横评指标、判定阈值、最终决策表的列字段。
  2. Hybrid 检索调参:你的系统当前 BM25-only Recall@10 = 0.62,Dense-only Recall@10 = 0.81,简单 RRF 融合后 Recall@10 = 0.83,提升不到 2 个点。请列出至少 4 种可能原因和对应的诊断方法。
  3. ANN 参数对召回率的影响:你的 Qdrant HNSW 索引在 100 万向量上 Recall@10 vs Brute-force 仅有 0.89,目标是 0.97+。请给出至少 3 个可调参数和它们的代价(延迟/内存/构建时间)。
  4. Reranker 性价比分析:你的系统 P95 延迟预算 600 ms,召回 50 候选 + Reranker 精排需要 380 ms,是否值得加入 Reranker?请定量论证。
  5. 向量库选型决策:以下三个场景,选择最合适的向量库并说明理由:(a) 100M 向量、需要 PG 事务一致性、可接受 50 ms 延迟;(b) 5M 向量、纯私有化、要求 5 ms 延迟;(c) 50M 向量、SaaS 形态、需要快速 PoC 上线。
  6. 显著性陷阱:模型 A 的 Recall@10 = 0.84,模型 B 的 Recall@10 = 0.86,你能直接得出"B 优于 A"的结论吗?请用 Bootstrap 给出方法论说明。
  7. 迁移回归测试设计:你的系统要从 Pinecone 迁移到 Milvus,请设计三层回归测试方案(离线召回、端到端答案、灰度 A/B),每一层的样本规模、关键指标、判定门槛是多少?

本章小结。

Embedding 与向量库的工程化评测有四个核心信念:(1) 不要相信公开榜单——必须用客户自己的语料做 PoC;(2) 不要分开评测模型层和系统层——必须从端到端答案准确率倒推选型;(3) 不要省略 Reranker——它是 RAG 工程性价比最高的一环;(4) 不要省略统计显著性——单点指标差异往往不可信。本篇配套的 6 段代码已经覆盖了选型 PoC、Hybrid Pipeline、ANN 参数扫描、Reranker 横评的核心动作,你可以直接拿去改造成自家 CI 中的回归脚本。 下一篇我们将进入 第 51 篇 模型升级与微调回归测试——主题从"如何选模型"转向"如何在升级模型时保证质量不退化"。

Embedding 与向量库工程化评测 大模型测试体系教程 · 第 50 篇 · 内部培训资料