企业官网建设流程全解析

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

回答错了，可能是相关 chunk 没召回、召回了却排太后、模型没使用证据，或引用了错误来源。只看“答案准确率 80%”无法定位该修分块、Embedding、Rerank 还是 Prompt。评测必须把检索、生成和引用拆开。

坏标注测试与判断答案

loader 现在完整覆盖：重复 id、可回答无 relevant、不可回答却有 relevant、空文件；评测单测结果8 passed in 0.17s。

“退款多久到账”与“七天内申请”是不同事实。expected_facts=七天会诱导模型把申请期限当到账时间；应把该样本标为不可回答，或加入真正说明到账时间的证据和正确事实。评测集本身错误时，模型回答正确也会被判错，所以失败样本必须回看原文。

Recall@K：相关证据有没有进候选

Recall@K = Top K 命中的相关项数 / 全部相关项数

排名[a,b,c,d]，相关{b,d}，K=3，只命中 b，所以 Recall@3=1/2。Recall 高不代表顺序好；把正确文档放第 50 也可能在 K=100 时满分，但上下文根本放不下。

MRR：第一个正确结果有多早

RR = 1 / 第一个相关项名次 MRR = 所有问题 RR 的平均

上例第一个相关 b 在第 2，RR=0.5。MRR 特别适合用户通常只看第一条、或 RAG 很依赖头部证据的场景；它不关心第二个相关项是否找到。

nDCG：多个相关项的整体位置

二元相关时：

DCG@K = Σ relevant(rank) / log2(rank+1) nDCG = DCG / 理想排序的 DCG

上例 K=3，只有 rank2 的 b：1/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,start=1):ifitem_idinrelevant_ids:return1.0/rankreturn0.0defndcg_at_k(ranked_ids:Sequence[str],relevant_ids:Set[str],*,k:int)->float:ifnotrelevant_ids:return0.0dcg=sum(1.0/math.log2(rank+1)forrank,item_idinenumerate(ranked_ids[:k],start=1)ifitem_idinrelevant_ids)ideal_hits=min(len(relevant_ids),k)ideal=sum(1.0/math.log2(rank+1)forrankinrange(1,ideal_hits+1))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 model；Judge 也会偏差，需要标注集校准，不能当真理。

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.json

CLI 会把候选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,start=1):ifitem_idinrelevant_ids:return1.0/rankreturn0.0defndcg_at_k(ranked_ids:Sequence[str],relevant_ids:Set[str],*,k:int)->float:ifnotrelevant_ids:return0.0dcg=sum(1.0/math.log2(rank+1)forrank,item_idinenumerate(ranked_ids[:k],start=1)ifitem_idinrelevant_ids)ideal_hits=min(len(relevant_ids),k)ideal=sum(1.0/math.log2(rank+1)forrankinrange(1,ideal_hits+1))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 EvaluationFunction=Callable[[EvaluationCase],Awaitable[tuple[list[str],str,list[int],set[int]]]]classEvaluationRunner:def__init__(self,*,k:int=5)->None:self.k=kasyncdefrun(self,cases:Sequence[EvaluationCase],*,evaluate:EvaluationFunction,config_hash:str,)->dict[str,object]:results:list[dict[str,object]]=[]forcaseincases:started=perf_counter()ranked_ids,answer,cited,supported=awaitevaluate(case)latency_seconds=perf_counter()-started relevant=set(case.relevant_ids)recall=1.0ifcase.unanswerableandnotranked_idselserecall_at_k(ranked_ids,relevant,k=self.k)mrr=1.0ifcase.unanswerableandnotranked_idselsemean_reciprocal_rank(ranked_ids,relevant)ndcg=1.0ifcase.unanswerableandnotranked_idselsendcg_at_k(ranked_ids,relevant,k=self.k)ifcase.unanswerable:answer_correct=any(terminanswerfortermin("没有足够证据","无法回答","证据不足"))else:answer_correct=all(factinanswerforfactincase.expected_facts)precision=citation_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,strict=True)ifnotcase.unanswerable]retrieval_population=answerableorresults ordered_latencies=sorted(float(item["latency_seconds"])foriteminresults)p95_index=max(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 project's hard gates."""payload=json.loads(baseline_path.read_text(encoding="utf-8"))baseline=payload.get("summary")ifnotisinstance(baseline,dict):raiseValueError("baseline report must contain a summary object")returncheck_release_gate(baseline,candidate_summary)asyncdefmain()->None:parser=argparse.ArgumentParser()parser.add_argument("--dataset",type=Path,default=Path("evals/rag_cases.jsonl"))parser.add_argument("--output",type=Path,default=Path("../reports/rag-evaluation.json"))parser.add_argument("--baseline",type=Path,default=None,help="optional prior report; exit 2 when the candidate violates a hard gate",)parser.add_argument("--provider",choices=["fake"],default="fake")args=parser.parse_args()cases=load_cases(args.dataset)config_hash=hashlib.sha256(args.dataset.read_bytes()+args.provider.encode()).hexdigest()[:16]report=awaitEvaluationRunner(k=5).run(cases,evaluate=fake_evaluate,config_hash=config_hash)args.output.parent.mkdir(parents=True,exist_ok=True)args.output.write_text(json.dumps(report,ensure_ascii=False,indent=2),encoding="utf-8")summary=report["summary"]ifnotisinstance(summary,dict):raiseTypeError("evaluation report summary must be an object")reasons=release_decision(args.baseline,summary)ifargs.baselineelse[]print(json.dumps({"summary":summary,"release_gate_failures":reasons},ensure_ascii=False,))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 指标串起来。