召回率 benchmark：为什么你的向量搜索结果看起来还行但评测很差
召回率 benchmark：为什么你的向量搜索结果看起来还行但评测很差
一、深度引言与场景痛点
"看起来还行"是 RAG 系统最大的陷阱。
我经历过这样一个项目：团队用了一个打分 0.92 的 Embedding 模型，配合 Milvus 向量数据库做检索。随便试了几个 query，返回的文档看起来都很相关——"Python 异步编程"搜出来的是 asyncio 教程，"Redis 缓存策略"搜出来的是缓存模式的文章。大家觉得效果不错，就上线了。
两个月后，产品经理做了一个系统的评测：找了 200 条业务方的查询，让人工标注了每道题的正确答案应该包含哪些文档。评测出来的 Recall@5 连 0.4 都不到——也就是说，60% 的查询中，前 5 个检索结果里没有应该出现的那篇文档。
为什么"看起来还行"实际却很差？原因有三：
第一，确认偏误。开发者往往只测试那些能搜出好结果的 query，跳过了真正的长尾和困难查询。你搜"Python 异步编程"，返回 asyncio 教程，你点头；你搜"怎么在 FastAPI 里用异步但不想用 await"，返回了一堆不相关的结果，但你没测试过这个 query。
第二，片段级 vs 文档级的相关性。向量相似度分数 0.85 看起来很高，但这个高分是因为 query 和文档片段共享了一些高频词（如"Python"、"异步"），而文档真正有效的答案可能在另一个段落里——那个段落被分到了另一个 chunk 里，向量分数只有 0.5。
第三，召回≠排序。即使相关的文档确实在 top-20 的检索结果里，但如果它排在 15-20 位，而 Rerank 也只看了 top-10，那这篇文档在最终的 Recall@5 里仍然是"未召回"。
本文就来拆解召回率评测的正确方式——怎么设计测试集、怎么算 Recall@K、以及怎么从评测结果里找到真正的优化方向。
二、底层机制与原理深度剖析
召回率评测的完整流程：
flowchart TD A[评测集构建] --> A1[收集 200+ 真实 query] A1 --> A2[人工标注每个 query 的相关文档] A2 --> A3[标注一致性校验: 至少 2 人标注] A3 --> B[检索评测执行] B --> B1[对每个 query 执行检索] B1 --> B2[记录 top-K 结果] B2 --> B3[与标注结果比对] B3 --> C{评测指标计算} C --> C1[Recall@K: top-K 中包含多少相关文档 / 总相关文档数] C --> C2[MRR: 第一个相关文档排第几] C --> C3[NDCG@K: 考虑排序位置的加权分数] C --> C4[Hit Rate: 至少命中一个的 query 比例] C1 --> D[诊断分析] C2 --> D C3 --> D C4 --> D D --> E{瓶颈在哪里?} E -->|Recall 低| F[检查 chunk 策略 / embedding 模型 / 检索配置] E -->|MRR 低| G[检查排序策略 / Rerank 效果] E -->|方差大| H[检查 query 分类 / 不同类别分别评测]关键概念区分：
- Recall@K是最朴素的指标：前 K 个结果里包含了多少个标注为相关的文档。适合评估"检索的覆盖能力"。
- MRR（平均倒数排名）关注第一个相关文档的排名：排名越靠前分数越高。适合评估"用户能否在第一屏看到相关结果"。
- NDCG@K同时考虑了位置和相关度等级：排在前面的高相关文档比排在后面的低相关文档更有价值。适合有 graded relevance（如 0=不相关，1=部分相关，2=高度相关）的场景。
大多数 RAG 系统的召回率问题，根源不在检索阶段，而在更上游的文档切分——一个被切断的答案段落，无论 Embedding 模型多好、检索算法多精细，都搜不回来。
三、生产级代码实现
import json import numpy as np from typing import List, Dict, Set, Tuple, Optional from dataclasses import dataclass, field from collections import defaultdict @dataclass class QueryCase: """评测用例""" query_id: str query_text: str relevant_doc_ids: Set[str] # 标注为相关的文档 ID partially_relevant_ids: Set[str] = field(default_factory=set) # 部分相关 category: str = "general" # 查询类别 @dataclass class RetrievalMetrics: """单条检索的评测指标""" recall_at_1: float = 0 recall_at_3: float = 0 recall_at_5: float = 0 recall_at_10: float = 0 mrr: float = 0 ndcg_at_5: float = 0 first_relevant_rank: int = -1 total_relevant: int = 0 total_retrieved_relevant: int = 0 class RecallBenchmark: """RAG 召回率评测框架""" def __init__(self, test_cases: List[QueryCase]): self.test_cases = test_cases self.retriever = None # 需注入实际检索器 # ===== 指标计算 ===== @staticmethod def compute_recall_at_k( retrieved_ids: List[str], relevant_ids: Set[str], k: int ) -> float: """计算 Recall@K""" if not relevant_ids: return 0.0 top_k = set(retrieved_ids[:k]) relevant_in_top_k = top_k & relevant_ids return len(relevant_in_top_k) / len(relevant_ids) @staticmethod def compute_mrr( retrieved_ids: List[str], relevant_ids: Set[str] ) -> float: """计算 MRR（Mean Reciprocal Rank）""" for rank, doc_id in enumerate(retrieved_ids, start=1): if doc_id in relevant_ids: return 1.0 / rank return 0.0 @staticmethod def compute_ndcg_at_k( retrieved_ids: List[str], relevance_map: Dict[str, float], # doc_id → relevance (0/1/2) k: int, ) -> float: """计算 NDCG@K""" # 实际相关性 rels = [] for doc_id in retrieved_ids[:k]: rels.append(relevance_map.get(doc_id, 0.0)) # DCG dcg = sum( (2**rel - 1) / np.log2(idx + 2) for idx, rel in enumerate(rels) ) # 理想 DCG（所有相关文档都排在前面） ideal_rels = sorted(relevance_map.values(), reverse=True)[:k] idcg = sum( (2**rel - 1) / np.log2(idx + 2) for idx, rel in enumerate(ideal_rels) ) return dcg / idcg if idcg > 0 else 0.0 # ===== 评测执行 ===== async def evaluate_single( self, case: QueryCase, retrieved_ids: List[str] ) -> RetrievalMetrics: """评测单条查询的检索结果""" metrics = RetrievalMetrics() metrics.total_relevant = len(case.relevant_doc_ids) # Recall@K metrics.recall_at_1 = self.compute_recall_at_k(retrieved_ids, case.relevant_doc_ids, 1) metrics.recall_at_3 = self.compute_recall_at_k(retrieved_ids, case.relevant_doc_ids, 3) metrics.recall_at_5 = self.compute_recall_at_k(retrieved_ids, case.relevant_doc_ids, 5) metrics.recall_at_10 = self.compute_recall_at_k(retrieved_ids, case.relevant_doc_ids, 10) # MRR metrics.mrr = self.compute_mrr(retrieved_ids, case.relevant_doc_ids) # NDCG@5（合并标注，全相关=2.0，部分相关=1.0） relevance_map = {} for doc_id in case.relevant_doc_ids: relevance_map[doc_id] = 2.0 for doc_id in case.partially_relevant_ids: relevance_map[doc_id] = 1.0 metrics.ndcg_at_5 = self.compute_ndcg_at_k(retrieved_ids, relevance_map, 5) # 第一个相关文档的排名 for rank, doc_id in enumerate(retrieved_ids, start=1): if doc_id in case.relevant_doc_ids: metrics.first_relevant_rank = rank break # 检索到的相关文档数 metrics.total_retrieved_relevant = len( set(retrieved_ids) & case.relevant_doc_ids ) return metrics async def evaluate_all( self, retriever_func ) -> Dict[str, Any]: """评测所有测试用例 Args: retriever_func: async func(query: str) -> List[str] 返回 doc_id 列表 """ results: List[Tuple[QueryCase, RetrievalMetrics]] = [] for case in self.test_cases: retrieved_ids = await retriever_func(case.query_text) metrics = await self.evaluate_single(case, retrieved_ids) results.append((case, metrics)) return self._aggregate(results) def _aggregate( self, results: List[Tuple[QueryCase, RetrievalMetrics]] ) -> Dict[str, Any]: """汇总评测结果""" n = len(results) # 总体指标 avg_recall_1 = sum(r[1].recall_at_1 for r in results) / n avg_recall_3 = sum(r[1].recall_at_3 for r in results) / n avg_recall_5 = sum(r[1].recall_at_5 for r in results) / n avg_recall_10 = sum(r[1].recall_at_10 for r in results) / n avg_mrr = sum(r[1].mrr for r in results) / n avg_ndcg = sum(r[1].ndcg_at_5 for r in results) / n # Hit Rate（至少召回一个相关文档的 query 比例） hit_count = sum( 1 for _, m in results if m.total_retrieved_relevant > 0 ) hit_rate = hit_count / n # 按类别分组统计 by_category = defaultdict(list) for case, metrics in results: by_category[case.category].append(metrics) category_breakdown = {} for cat, cat_metrics in by_category.items(): category_breakdown[cat] = { "count": len(cat_metrics), "Recall@5": round(sum(m.recall_at_5 for m in cat_metrics) / len(cat_metrics), 3), "MRR": round(sum(m.mrr for m in cat_metrics) / len(cat_metrics), 3), } # 分布分析：看看有多少 query 的 Recall@5 < 0.2 low_recall_queries = [ (results[i][0].query_text, results[i][1].recall_at_5) for i in range(n) if results[i][1].recall_at_5 < 0.2 ] return { "total_queries": n, "overall_metrics": { "Recall@1": round(avg_recall_1, 4), "Recall@3": round(avg_recall_3, 4), "Recall@5": round(avg_recall_5, 4), "Recall@10": round(avg_recall_10, 4), "MRR": round(avg_mrr, 4), "NDCG@5": round(avg_ndcg, 4), "Hit_Rate": round(hit_rate, 4), }, "category_breakdown": category_breakdown, "low_recall_count": len(low_recall_queries), "low_recall_examples": [ {"query": q[:80], "recall@5": r} for q, r in low_recall_queries[:5] ], } # ===== 诊断工具 ===== @staticmethod def diagnose_low_recall( case: QueryCase, retrieved_ids: List[str], doc_store: Dict[str, str], # doc_id → content ) -> str: """诊断单个低召回率 case 的原因""" issues = [] # 检查相关文档是否完全没被检索到 missed = case.relevant_doc_ids - set(retrieved_ids) if missed: issues.append(f"缺失 {len(missed)} 篇相关文档: {missed}") # 检查相关文档是否在 top-K 很靠后 for doc_id in case.relevant_doc_ids: if doc_id in retrieved_ids: rank = retrieved_ids.index(doc_id) + 1 if rank > 10: issues.append( f"文档 {doc_id} 排在第 {rank} 位，不在 top-10 中" ) # 检查检索到的 top 文档是否实际上不相关 top3_unrelated = [ rid for rid in retrieved_ids[:3] if rid not in case.relevant_doc_ids ] if top3_unrelated: # 显示部分内容 snippets = [ f"{rid}: {doc_store.get(rid, '未知')[:50]}..." for rid in top3_unrelated[:2] ] issues.append(f"top-3 中有不相关文档: {'; '.join(snippets)}") return "\n".join(issues) if issues else "未发现明显问题" # ================== 使用示例 ================== def demo(): # 构造评测数据集 test_cases = [ QueryCase( query_id="q1", query_text="如何在 Python 中使用异步 IO", relevant_doc_ids={"doc_003", "doc_007"}, category="programming", ), QueryCase( query_id="q2", query_text="Redis 缓存过期策略", relevant_doc_ids={"doc_015"}, partially_relevant_ids={"doc_022"}, category="database", ), QueryCase( query_id="q3", query_text="Milvus 向量检索性能优化", relevant_doc_ids={"doc_042", "doc_043", "doc_044"}, category="vector_db", ), ] # 模拟检索器 async def mock_retriever(query: str) -> List[str]: # 真实场景替换为实际的向量检索逻辑 mock_results = { "如何在 Python 中使用异步 IO": ["doc_003", "doc_001", "doc_007", "doc_010"], "Redis 缓存过期策略": ["doc_022", "doc_018", "doc_025"], "Milvus 向量检索性能优化": ["doc_042", "doc_050", "doc_060"], } return mock_results.get(query, []) # 模拟文档库 doc_store = { "doc_003": "Python asyncio 库提供了异步 IO 能力...", "doc_007": "使用 async/await 语法进行异步编程...", "doc_015": "Redis 支持多种过期策略，包括惰性删除...", } async def run(): benchmark = RecallBenchmark(test_cases) report = await benchmark.evaluate_all(mock_retriever) print("=== 召回率评测报告 ===") for metric, value in report["overall_metrics"].items(): print(f" {metric}: {value}") print(f"\n低召回率查询数量: {report['low_recall_count']}") print("\n=== 类别细分 ===") for cat, stats in report["category_breakdown"].items(): print(f" {cat}: {stats}") import asyncio asyncio.run(run()) if __name__ == "__main__": demo()四、边界分析与架构权衡
1. 评测集的质量决定了评测结果的可信度
一个常见的错误是：用 LLM 自动生成评测集（给 LLM 一篇文档，让它生成 10 个问题和答案）。问题是 LLM 生成的问题天然地更"友好"——它知道整篇文档的内容，生成的问题往往能和文档片段高度匹配。这种评测集上的 Recall@5 轻松过 0.9，但一到真实用户查询，效果就断崖式下跌。评测集的 query 必须来自真实用户——即使需要花时间收集。
2. 标注一致性（IAA）
如果两个标注者对同一篇文档的相关性判断不一致，说明评测集的定义就存在模糊性。建议计算 Cohen's Kappa 或 Krippendorff's Alpha，当 IAA < 0.6 时，需要重新审视标注指南。
3. Recall 不是唯一指标
高 Recall 但低 Precision（返回了大量相关文档但也夹杂了大量不相关文档）同样会让用户不满——用户在 top 结果里看到一堆无关内容。所以在看 Recall 的同时，也要关注 Precision@K 或 F1@K。
4. 动态评测 vs 静态评测
文档库是动态变化的——新增文档、修改文档都会改变检索结果。静态评测集（固定的 query-doc 配对）无法反映这种动态性。建议定期（如每周）重新执行评测，追踪指标变化趋势，在指标出现明显劣化时发出告警。
五、总结
召回率评测不是上线前做一次就完事的事情，而是一个持续运行的质量保障体系。核心要点：
- 用真实 query 做评测集——不要用 LLM 生成的 query，那是自己骗自己
- 同时看多个指标——Recall@K 看覆盖，MRR 看排名，NDCG 看排序质量
- 按类别拆开看——"技术类查询"和"闲聊类查询"的 Recall 可能差 2 倍，混在一起看平均数没有任何意义
- 低召回 query 是最宝贵的优化资源——每条低召回 query 都是一个优化机会，分析它们为什么搜不出，往往能发现 chunk 策略或 embedding 模型的问题
下一篇是本系列 0707 的倒数第二篇——技术原理可视化方法论：用时序图把 Agent 协作流程画得明明白白。