Python后端AI专题33：评估召回与答案质量：Recall@K、MRR、nDCG 和引用正确率

Python后端AI专题33评估召回与答案质量RecallK、MRR、nDCG 和引用正确率回答错了可能是相关 chunk 没召回、召回了却排太后、模型没使用证据或引用了错误来源。只看“答案准确率 80%”无法定位该修分块、Embedding、Rerank 还是 Prompt。评测必须把检索、生成和引用拆开。坏标注测试与判断答案loader 现在完整覆盖重复 id、可回答无 relevant、不可回答却有 relevant、空文件评测单测结果8 passed in 0.17s。“退款多久到账”与“七天内申请”是不同事实。expected_facts七天会诱导模型把申请期限当到账时间应把该样本标为不可回答或加入真正说明到账时间的证据和正确事实。评测集本身错误时模型回答正确也会被判错所以失败样本必须回看原文。RecallK相关证据有没有进候选RecallK Top K 命中的相关项数 / 全部相关项数排名[a,b,c,d]相关{b,d}K3只命中 b所以 Recall31/2。Recall 高不代表顺序好把正确文档放第 50 也可能在 K100 时满分但上下文根本放不下。MRR第一个正确结果有多早RR 1 / 第一个相关项名次 MRR 所有问题 RR 的平均上例第一个相关 b 在第 2RR0.5。MRR 特别适合用户通常只看第一条、或 RAG 很依赖头部证据的场景它不关心第二个相关项是否找到。nDCG多个相关项的整体位置二元相关时DCGK Σ relevant(rank) / log2(rank1) nDCG DCG / 理想排序的 DCG上例 K3只有 rank2 的 b1/log2(3)0.63093。两个相关项的理想 DCG 是1/log2(2)1/log2(3)1.63093所以 nDCG≈0.38685。完整指标模块from__future__importannotationsimportmathfromcollections.abcimportSequence,Setdefrecall_at_k(ranked_ids:Sequence[str],relevant_ids:Set[str],*,k:int)-float:ifnotrelevant_ids:return0.0returnlen(set(ranked_ids[:k])set(relevant_ids))/len(relevant_ids)defmean_reciprocal_rank(ranked_ids:Sequence[str],relevant_ids:Set[str])-float:forrank,item_idinenumerate(ranked_ids,start1):ifitem_idinrelevant_ids:return1.0/rankreturn0.0defndcg_at_k(ranked_ids:Sequence[str],relevant_ids:Set[str],*,k:int)-float:ifnotrelevant_ids:return0.0dcgsum(1.0/math.log2(rank1)forrank,item_idinenumerate(ranked_ids[:k],start1)ifitem_idinrelevant_ids)ideal_hitsmin(len(relevant_ids),k)idealsum(1.0/math.log2(rank1)forrankinrange(1,ideal_hits1))returndcg/idealifidealelse0.0defcitation_precision(cited_numbers:Sequence[int],supported_numbers:Set[int])-float:ifnotcited_numbers:return0.0returnsum(numberinsupported_numbersfornumberincited_numbers)/len(cited_numbers)citation_precision([1,3,9], {1,3})2/3。还可增加 citation recall应引用的事实有多少带了出处当前项目先实现精度和“至少一个有效引用”的在线闸门。Runner 怎样处理不可回答样本不可回答且无召回结果时检索指标按 1.0 记录该样本行为但汇总检索分数优先只平均 answerable 样本避免不可回答比例改变掩盖召回退化。答案正确则查拒答关键词。当前字符串包含法简单可解释但无法理解同义表达。真实评测可加入结构化答案、规则归一化、人工评审或 judge modelJudge 也会偏差需要标注集校准不能当真理。Runner 输出每条明细不只总分{id:refund-deadline,ranked_ids:[refund-policy],recall_at_k:1.0,mrr:1.0,ndcg_at_k:1.0,citation_precision:1.0,answer_correct:true,latency_seconds:0.0000165}EvaluationRunner.run()在调用每个evaluate(case)前后使用perf_counter()计时逐条保存latency_seconds再用 nearest-rank 方法计算p95_seconds。这样下一篇的发布门禁所需 P95 来自真实评测执行不再是手工塞进示例字典的孤立字段。同时记录 config_hash、时间、K 和样本数。没有配置身份的两个 0.85 无法比较。计时覆盖本项目传入的完整evaluate函数若真实环境还包含网关排队或响应序列化就必须把计时边界放到包含这些步骤的调用处不能把局部耗时冒充端到端 P95。当前离线接线结果.\.venv\Scripts\python scripts\run_evaluation.py --output..\reports\rag-evaluation.json{summary: {recall_at_k: 1.0, mrr: 1.0, ndcg_at_k: 1.0, citation_precision: 1.0, answer_accuracy: 1.0, p95_seconds: 1.650000922381878e-05}, release_gate_failures: []}这是本次 Fake 运行的真实输出P95 会随机器调度变化不应照抄成性能结论。Fake 根据标注构造完美结果只证明报表链不证明语义真实模型报告必须另存并注明未运行时绝不伪造。需要比较候选版本时保存上一份报告并传入.\.venv\Scripts\python.exe scripts\run_evaluation.py --baseline..\reports\rag-evaluation-baseline.json --output..\reports\rag-evaluation-candidate.jsonCLI 会把候选summary交给下一篇的check_release_gate()。所有硬门槛通过时退出码为 0存在失败原因时打印完整release_gate_failures并以退出码 2 结束所以 CI 不会把“生成了报告”误当成“允许发布”。指标之间怎样做决策Recall 降、MRR 降先查解析/分块/召回Recall 不变、MRR 降融合或重排退化检索不变、答案准确降Prompt/模型/上下文组织答案准确不变、引用精度降引用协议或校验质量上升但 P95/成本翻倍需要产品权衡不是自动接受。本篇最终完整模块metrics.py前面的代码片段用于解释本次改动下面是本篇结束时可直接核对和替换的磁盘完整版本。from__future__importannotationsimportmathfromcollections.abcimportSequence,Setdefrecall_at_k(ranked_ids:Sequence[str],relevant_ids:Set[str],*,k:int)-float:ifnotrelevant_ids:return0.0returnlen(set(ranked_ids[:k])set(relevant_ids))/len(relevant_ids)defmean_reciprocal_rank(ranked_ids:Sequence[str],relevant_ids:Set[str])-float:forrank,item_idinenumerate(ranked_ids,start1):ifitem_idinrelevant_ids:return1.0/rankreturn0.0defndcg_at_k(ranked_ids:Sequence[str],relevant_ids:Set[str],*,k:int)-float:ifnotrelevant_ids:return0.0dcgsum(1.0/math.log2(rank1)forrank,item_idinenumerate(ranked_ids[:k],start1)ifitem_idinrelevant_ids)ideal_hitsmin(len(relevant_ids),k)idealsum(1.0/math.log2(rank1)forrankinrange(1,ideal_hits1))returndcg/idealifidealelse0.0defcitation_precision(cited_numbers:Sequence[int],supported_numbers:Set[int])-float:ifnotcited_numbers:return0.0returnsum(numberinsupported_numbersfornumberincited_numbers)/len(cited_numbers)本篇最终完整模块runner.py前面的代码片段用于解释本次改动下面是本篇结束时可直接核对和替换的磁盘完整版本。from__future__importannotationsfromcollections.abcimportAwaitable,Callable,Sequencefromdatetimeimportdatetime,timezonefrommathimportceilfromstatisticsimportfmeanfromtimeimportperf_counterfromapp.services.evaluation.datasetimportEvaluationCasefromapp.services.evaluation.metricsimportcitation_precision,mean_reciprocal_rank,ndcg_at_k,recall_at_k EvaluationFunctionCallable[[EvaluationCase],Awaitable[tuple[list[str],str,list[int],set[int]]]]classEvaluationRunner:def__init__(self,*,k:int5)-None:self.kkasyncdefrun(self,cases:Sequence[EvaluationCase],*,evaluate:EvaluationFunction,config_hash:str,)-dict[str,object]:results:list[dict[str,object]][]forcaseincases:startedperf_counter()ranked_ids,answer,cited,supportedawaitevaluate(case)latency_secondsperf_counter()-started relevantset(case.relevant_ids)recall1.0ifcase.unanswerableandnotranked_idselserecall_at_k(ranked_ids,relevant,kself.k)mrr1.0ifcase.unanswerableandnotranked_idselsemean_reciprocal_rank(ranked_ids,relevant)ndcg1.0ifcase.unanswerableandnotranked_idselsendcg_at_k(ranked_ids,relevant,kself.k)ifcase.unanswerable:answer_correctany(terminanswerfortermin(没有足够证据,无法回答,证据不足))else:answer_correctall(factinanswerforfactincase.expected_facts)precisioncitation_precision(cited,supported)ifcitedelse(1.0ifcase.unanswerableelse0.0)results.append({id:case.id,question:case.question,ranked_ids:ranked_ids,answer:answer,recall_at_k:recall,mrr:mrr,ndcg_at_k:ndcg,citation_precision:precision,answer_correct:answer_correct,latency_seconds:latency_seconds,})answerable[itemforitem,caseinzip(results,cases,strictTrue)ifnotcase.unanswerable]retrieval_populationanswerableorresults ordered_latenciessorted(float(item[latency_seconds])foriteminresults)p95_indexmax(0,ceil(len(ordered_latencies)*0.95)-1)return{generated_at:datetime.now(timezone.utc).isoformat(),config_hash:config_hash,k:self.k,case_count:len(results),summary:{recall_at_k:fmean(float(item[recall_at_k])foriteminretrieval_population),mrr:fmean(float(item[mrr])foriteminretrieval_population),ndcg_at_k:fmean(float(item[ndcg_at_k])foriteminretrieval_population),citation_precision:fmean(float(item[citation_precision])foriteminresults),answer_accuracy:fmean(bool(item[answer_correct])foriteminresults),p95_seconds:ordered_latencies[p95_index],},cases:results,}本篇最终完整模块run_evaluation.py前面的代码片段用于解释本次改动下面是本篇结束时可直接核对和替换的磁盘完整版本。from__future__importannotationsimportargparseimportasyncioimporthashlibimportjsonimportsysfrompathlibimportPath# Direct script execution puts only scripts/ on sys.path. Add the project root# explicitly so the README command behaves the same before and after install.sys.path.insert(0,str(Path(__file__).resolve().parents[1]))fromapp.services.evaluation.datasetimportEvaluationCase,load_casesfromapp.services.evaluation.gatesimportcheck_release_gatefromapp.services.evaluation.runnerimportEvaluationRunnerasyncdeffake_evaluate(case:EvaluationCase)-tuple[list[str],str,list[int],set[int]]:ifcase.unanswerable:return[],当前知识库没有足够证据回答。,[],set()answer.join(case.expected_facts)。[1]returnlist(case.relevant_ids),answer,[1],{1}defrelease_decision(baseline_path:Path,candidate_summary:dict[str,float])-list[str]:Compare the new run with a saved baseline using the projects hard gates.payloadjson.loads(baseline_path.read_text(encodingutf-8))baselinepayload.get(summary)ifnotisinstance(baseline,dict):raiseValueError(baseline report must contain a summary object)returncheck_release_gate(baseline,candidate_summary)asyncdefmain()-None:parserargparse.ArgumentParser()parser.add_argument(--dataset,typePath,defaultPath(evals/rag_cases.jsonl))parser.add_argument(--output,typePath,defaultPath(../reports/rag-evaluation.json))parser.add_argument(--baseline,typePath,defaultNone,helpoptional prior report; exit 2 when the candidate violates a hard gate,)parser.add_argument(--provider,choices[fake],defaultfake)argsparser.parse_args()casesload_cases(args.dataset)config_hashhashlib.sha256(args.dataset.read_bytes()args.provider.encode()).hexdigest()[:16]reportawaitEvaluationRunner(k5).run(cases,evaluatefake_evaluate,config_hashconfig_hash)args.output.parent.mkdir(parentsTrue,exist_okTrue)args.output.write_text(json.dumps(report,ensure_asciiFalse,indent2),encodingutf-8)summaryreport[summary]ifnotisinstance(summary,dict):raiseTypeError(evaluation report summary must be an object)reasonsrelease_decision(args.baseline,summary)ifargs.baselineelse[]print(json.dumps({summary:summary,release_gate_failures:reasons},ensure_asciiFalse,))ifreasons:raiseSystemExit(2)if__name____main__:asyncio.run(main())本篇练习做一次版本门禁判定基线Recall .90、MRR .76、nDCG .72、引用精度 .98、答案准确 .84、P95 2.0s。候选.92/.74/.75/.96/.87/2.8s。门槛Recall 不下降MRR 最多降 .01引用精度不低于 .97答案准确不下降P95 增幅不超过 20%。逐项计算并给最终是否发布。再写 Python 函数返回所有失败原因而不是遇到第一个就停止。下一篇给出完整门禁函数然后从一次慢请求开始把 embedding/vector/keyword/rerank/model 分段耗时、结构化日志和 Prometheus 指标串起来。