召回率-精度权衡：RAG 评测中的 NDCG 与 MRR 指标如何指导参数调优

一、深度引言与场景痛点

大家好，我是赵咕咕。

你的 RAG 系统能返回相关文档了，但总觉得不够好。调大了 top_k，无关文档也混进来了；调小了 top_k，有价值的文档被漏掉了。到底调到多少才算好？

这就是 RAG 评测的典型场景：你需要量化的指标来回答"我的检索系统到底好不好"。而且不能只看一个数字——有时候语义相近但排序靠后，有时候第一个结果就对了——你需要一套完整的评测指标体系。

NDCG 和 MRR 是两个最常用的检索评测指标。它们分别衡量"整体排序质量"和"首个相关结果的位置"，联手就能完整刻画你的 RAG 系统表现。

二、底层机制与原理深度剖析

评测 RAG 检索的核心思路：准备一组测试 query，为每个 query 标注哪些文档是"相关的"，然后让检索系统输出排序结果，用指标量化排序质量。

NDCG（Normalized Discounted Cumulative Gain）关注整体排序，强调相关文档越靠前越好。计算公式：

DCG@k = Σ(rel_i / log2(i+1)) # i 是排序位置 NDCG@k = DCG@k / IDCG@k # IDCG 是理想排序的 DCG

MRR（Mean Reciprocal Rank）关注首个命中的位置：越靠前分数越高。

MRR = (1/N) * Σ(1/rank_of_first_relevant)

评测流程如下：

flowchart TB A[评测数据集] --> B[测试 Query 集合] A --> C[相关性标注<br/>Relevance Judgments] B --> D[RAG 检索系统<br/>待评测] D --> E[检索结果<br/>Top-K 排序列表] C --> F[评测计算器<br/>Metric Calculator] E --> F F --> G1[NDCG@5<br/>整体排序质量] F --> G2[NDCG@10<br/>更长列表的质量] F --> G3[MRR<br/>首个命中位置] F --> G4[Recall@K<br/>召回率] G1 --> H[指标分析<br/>指导参数调优] G2 --> H G3 --> H G4 --> H H --> I{MRR 低?} I -->|是| J[增大 top_k 或<br/>调低相似度阈值] I -->|否| K{NDCG 低?} K -->|是| L[优化 reranker<br/>或调整权重] K -->|否| M[参数已达最优] style D fill:#e8f5e9 style F fill:#fff3e0 style H fill:#f3e5f5

这里的核心思想是：用指标诊断检索系统的弱点，然后有针对性地调整参数。

• MRR 低：说明第一个相关结果出现得太靠后，问题在初筛阶段，需要扩大检索范围。
• NDCG@10 低：说明整体排序不好，相关文档和无关文档混在一起，需要在重排阶段优化。
• Recall@K 低：说明有相关文档根本没被检索到，可能需要降低向量相似度阈值或优化 chunk 策略。

三、生产级代码实现

下面是 RAG 检索评测系统的完整实现：

from __future__ import annotations import numpy as np from dataclasses import dataclass, field from typing import Optional import math @dataclass class TestQuery: """单条测试 query""" query_id: str query_text: str relevant_doc_ids: list[str] # 标注的相关文档 ID 列表 relevance_grades: Optional[dict[str, int]] = None # 分级标注（0-4） @dataclass class SearchResult: """检索结果项""" doc_id: str score: float rank: int = 0 class RAGEvaluator: """RAG 检索评测器""" @staticmethod def dcg_at_k( relevance_scores: list[float], k: int ) -> float: """ 计算 DCG@k 使用对数折损：rel / log2(rank + 1) """ dcg = 0.0 for i in range(min(k, len(relevance_scores))): rel = relevance_scores[i] # rank 从 1 开始 dcg += rel / math.log2(i + 2) return dcg @staticmethod def ndcg_at_k( search_results: list[str], relevant_docs: dict[str, float], k: int, ) -> float: """ 计算 NDCG@k relevant_docs: {doc_id: relevance_score} """ # 实际 DCG scores = [ relevant_docs.get(doc_id, 0) for doc_id in search_results[:k] ] dcg = RAGEvaluator.dcg_at_k(scores, k) # 理想 DCG（按相关性降序排列） ideal_scores = sorted( relevant_docs.values(), reverse=True ) idcg = RAGEvaluator.dcg_at_k(ideal_scores, k) if idcg == 0: return 0.0 return dcg / idcg @staticmethod def mrr( search_results: list[str], relevant_doc_ids: set[str], ) -> float: """ 计算 MRR 返回首个相关文档排在倒数第几的倒数 """ for rank, doc_id in enumerate(search_results, 1): if doc_id in relevant_doc_ids: return 1.0 / rank return 0.0 @staticmethod def recall_at_k( search_results: list[str], relevant_doc_ids: set[str], k: int, ) -> float: """计算 Recall@k：前 k 个结果中召回了多少相关文档""" if not relevant_doc_ids: return 1.0 found = sum( 1 for doc_id in search_results[:k] if doc_id in relevant_doc_ids ) return found / len(relevant_doc_ids) @staticmethod def precision_at_k( search_results: list[str], relevant_doc_ids: set[str], k: int, ) -> float: """计算 Precision@k：前 k 个结果中相关文档的比例""" if k == 0: return 0.0 found = sum( 1 for doc_id in search_results[:k] if doc_id in relevant_doc_ids ) return found / k @staticmethod def evaluate_all( test_queries: list[TestQuery], search_results_map: dict[str, list[str]], ) -> dict[str, float]: """ 对一批测试 query 做完整评测 search_results_map: {query_id: [doc_id_rank1, doc_id_rank2, ...]} """ metrics = { "ndcg@5": [], "ndcg@10": [], "ndcg@20": [], "mrr": [], "recall@5": [], "recall@10": [], "precision@5": [], "precision@10": [], } for query in test_queries: results = search_results_map.get(query.query_id, []) relevant_set = set(query.relevant_doc_ids) # 相关性映射（未分级标注时默认为 1） rel_map = {doc: 1.0 for doc in query.relevant_doc_ids} metrics["ndcg@5"].append( RAGEvaluator.ndcg_at_k(results, rel_map, 5) ) metrics["ndcg@10"].append( RAGEvaluator.ndcg_at_k(results, rel_map, 10) ) metrics["ndcg@20"].append( RAGEvaluator.ndcg_at_k(results, rel_map, 20) ) metrics["mrr"].append( RAGEvaluator.mrr(results, relevant_set) ) metrics["recall@5"].append( RAGEvaluator.recall_at_k(results, relevant_set, 5) ) metrics["recall@10"].append( RAGEvaluator.recall_at_k(results, relevant_set, 10) ) metrics["precision@5"].append( RAGEvaluator.precision_at_k(results, relevant_set, 5) ) metrics["precision@10"].append( RAGEvaluator.precision_at_k(results, relevant_set, 10) ) # 均值汇总 summary = {} for name, values in metrics.items(): summary[name] = np.mean(values) if values else 0.0 return summary class ParameterTuner: """基于评测结果的参数调优建议器""" @staticmethod def suggest( ndcg5: float, ndcg10: float, mrr: float, recall5: float, recall10: float, ) -> list[str]: """根据指标给出参数调优建议""" suggestions = [] # MRR 低：初筛阶段问题 if mrr < 0.5: suggestions.append( "MRR 偏低：建议增大 top_k 或降低相似度阈值，" "确保相关文档能进入候选集" ) # NDCG 低但 MRR 高：重排阶段问题 if ndcg5 < 0.5 and mrr > 0.7: suggestions.append( "首个相关文档排在了前面，但整体排序不理想；" "建议优化 reranker 模型或引入更多排序信号" ) # Recall 低：漏召问题 if recall10 < 0.6: suggestions.append( "Recall@10 偏低：有大量相关文档未被召回；" "建议检查 chunk 策略，或降低 top_k 的硬阈值" ) # NDCG@5 远低于 NDCG@10：相关文档排得太靠后 if ndcg10 - ndcg5 > 0.2: suggestions.append( "NDCG@5 和 @10 差距大：相关文档集中在后半部分；" "建议优化 embedding 模型或引入 BM25 混合检索" ) if not suggestions: suggestions.append("各项指标表现良好，继续监控即可") return suggestions # 使用示例 def main(): evaluator = RAGEvaluator() # 构建测试数据 test_queries = [ TestQuery( query_id="q1", query_text="如何优化 Python 性能？", relevant_doc_ids=["doc_a", "doc_c", "doc_e"], ), TestQuery( query_id="q2", query_text="向量检索的 HNSW 算法原理", relevant_doc_ids=["doc_f", "doc_g"], ), TestQuery( query_id="q3", query_text="asyncio 协程调度机制", relevant_doc_ids=["doc_h", "doc_i", "doc_j"], ), ] # 模拟检索系统返回的结果 search_results_map = { "q1": ["doc_a", "doc_b", "doc_c", "doc_d", "doc_e", "doc_x", "doc_y", "doc_z", "doc_w", "doc_v"], "q2": ["doc_x", "doc_f", "doc_g", "doc_y", "doc_z", "doc_a", "doc_b", "doc_c", "doc_d", "doc_e"], "q3": ["doc_h", "doc_i", "doc_j", "doc_a", "doc_b", "doc_c", "doc_d", "doc_e", "doc_f", "doc_g"], } # 评测 results = evaluator.evaluate_all(test_queries, search_results_map) print("=== RAG 检索评测结果 ===") for metric, value in results.items(): print(f" {metric:>15s}: {value:.4f}") # 调优建议 suggestions = ParameterTuner.suggest( ndcg5=results["ndcg@5"], ndcg10=results["ndcg@10"], mrr=results["mrr"], recall5=results["recall@5"], recall10=results["recall@10"], ) print("\n=== 参数调优建议 ===") for i, s in enumerate(suggestions, 1): print(f" {i}. {s}") if __name__ == "__main__": main()

四、边界分析与架构权衡

RAG 评测有几个常被忽略的边界：

标注成本。每个 query 标注哪些文档是相关的，需要领域专家人工完成。100 个 query × 10 个相关文档 = 1000 次标注。降低标注成本的方法是使用"成对偏好"标注：不标注所有文档，只标注文档 A 和文档 B 哪个更相关。或者使用 LLM 辅助标注，但需要人工抽检校准。

NDCG 对分级标注的依赖。如果相关性标注只有 0/1（相关/不相关），NDCG 退化为平均精度的变体。鼓励使用 0～4 的分级标注（不相关/略微相关/相关/很相关/完美匹配），这样 NDCG 才能体现排序质量的细粒度差异。

评测数据分布偏移。如果评测集的 query 和生产实际的 query 分布不同，评测结果没有参考意义。需要定期从生产日志中抽样真实 query 更新评测集，而不是永远用最初的 100 个标注 query。

k 值的选择。NDCG@5 和 NDCG@10 关注的是用户"第一页"体验，但如果你的场景是返回推荐文章（用户可能往下翻很多页），则需要 NDCG@50 甚至 @100。k 值根据业务场景确定，没有通用标准。

指标之间的冲突。提高 Recall 可能需要更多候选（牺牲 Precision），提高 MRR（首个命中）可能在 NDCG 上失分。不存在一个最优参数同时满足所有指标，需要根据业务优先级做权衡。

五、总结

NDCG 告诉你整体的排序好不好，MRR 告诉你用户最先看到的是不是正确的答案。两者联合使用，才能完整诊断 RAG 检索系统的健康状况。

核心要点：

1. NDCG 关注整体排序质量，适合评估"用户会看多个结果"的场景
2. MRR 关注首个命中的位置，适合评估"只看第一个就够了"的场景
3. 通过指标诊断定位问题阶段：初筛、重排还是 chunk 策略
4. 定期更新评测集，防止分布偏移

没有度量就没有优化。NDCG 和 MRR 就是你优化 RAG 时的"指南针"。