从检索排名到 Recall 和 MRR：写一个口径固定的离线评分器
先把相关性判断、候选条数和测试分母固定下来，再比较排名。
1. 排名拿到之后，还缺一份能对照的答案
假设你正在做一个帮助文档助手。第一版检索返回三篇文档，第二版换了排序，又返回三篇。打开结果看一眼，两版似乎都有道理。要判断这次改动有没有帮助，先得说清楚：这条问题需要哪些文档？我们只给后续回答提供前几条？一次请求没返回结果，还算不算一次测试？
这篇教程从已经拿到的文档排名开始，用 Python 标准库写一个小评分器。它只算 Recall@2 和 MRR@2，并把容易让成绩变好看的错误写成测试。跑完你会得到每条 query 的明细、固定分母下的均值，以及被拒绝的非法输入。
实验包含六份原创虚构文档、五条中文问题和三十个二元相关性判断。A、B 都是作者手写的候选顺序。它们用于检验公式和输入协议，没有执行任何检索模型，也不能作为 FTS、RRF、EmbeddingGemma 或某个 RAG 系统的成绩。全文可离线复现，不需要账号、SDK、安装依赖或真实用户数据。
2. 先定信息需求，再把每个判断写下来
qrels 是 query-document relevance judgments 的常见简称，可以把它理解成“每条问题和每份文档之间的相关性表”。这里约定：文档能直接支持问题要求的操作，或明确支持其中一个子需求，就标 1；否则标 0。
例如 q2 问“有哪些导出格式，分别怎样导出”。d3 教 TXT 导出，d4 教 CSV 导出，两份都相关。d1 只讲下载目录，虽然也在操作文件，却不能回答导出问题，所以标 0。只找到 d3，覆盖了这个语料里一半的相关文档。
先写完整信息需求，再判断文档。遇到一句“导出怎么了”这样的模糊查询，要先界定场景或标记争议，不能等看见某个候选结果后，再修改需求使它显得相关。信息检索评估所需的文档集合、信息需求和相关性判断，可参见 Stanford《Introduction to Information Retrieval》8.1 节。
本例的三十个标签由作者为教学固定，没有独立标注员实验，也没有测过标注一致性。所有 0 都是明确判断，不是“没检查到就默认为 0”。这在六份文档的小库里能做到；真实大库往往只有部分文档经过判断，不能直接套用本例的穷尽分母。
3. 两项指标分别回答什么
Recall@k 回答“全部相关文档里，前 k 条找到了多少”。对一条有相关文档的 query，它等于 top-k 中不同相关文档的数量，除以该 query 在固定语料中的全部相关文档数量。分母不是返回条数，也不是 k。Recall 的集合定义见 教材 8.3 节。
RR@k 回答“前 k 条里的首个相关文档出现得早不早”。第一条相关，得 1；第二条才相关，得 1/2；前 k 条都没有相关文档，得 0。MRR@k 是固定 eligible query 集合上的 RR@k 算术平均。本例另把每条 Recall@k 做宏平均，每条 query 权重相同，而非把所有文档命中数合并后算一个比例。
一定保留 @k。第三条才命中，在 RR@2 中就是 0，不能拿未截断 RR 的 1/3 代替。ir-measures 的 RR 定义也明确列出 cutoff 和相关性阈值；本文只参考定义，没有安装该库或做跨实现对照。
只要命中一份相关文档就记 1，衡量的是这条 query 有没有成功命中，不是 Recall。q2 相关集合是 d3、d4，若前两条只有 d3，则 Recall@2=1/2，尽管它已经“命中过”。再放入 d4，Recall 才变成 1；只要 d3 仍在第一位，RR 就仍然是 1。两个指标关注不同问题，没必要为了让分数好看而互换名称。
4. 先写下缺结果和无答案的处理协议
评分器遍历预先固定的 test query ID，而不是遍历“这次成功返回的 query”。q4 明明有相关文档 d6，A 却缺了整条排名。我们将它视为空列表，两项得分都为 0，而且继续占据宏平均分母。显式提交空列表与完全漏交 query 的分数相同，但 missing_queries 会记录后者，方便区分系统没找到结果和流程漏跑。
q5 则不同：这六份文档根本没有注销账户的说明。其相关文档数为 0，不能做 0/0。本教程事先约定：逐 query 的 Recall 和 RR 都记 null，并从两个宏平均中共同排除。因此 total_queries=4，eligible_queries=3，no_relevant_queries=["q5"]。若所有 query 都没有相关文档，均值也为 null。
这是本文的评价协议，不是所有库的默认行为。q5 返回空列表还是胡乱返回 d1，在这两个宏平均里都不会体现。报告必须保留 q5 的 returned、missing 等明细；想考查“没有答案时能否保持克制”，需要另定适合该任务的评价，不要把这里的 null 当成系统通过了无答案测试。
5. 把实验数据保存到空目录
准备 Python 3.10 或以上版本；本次实际复现环境为 Python 3.12.14。建一个空目录，将下面四段 Python 分别保存为标明的文件名。源码只用标准库，同目录运行即可。文末还给出完整预期 JSON，保存为 expected.json 后可以做字节级输出对照。
fixture.py 把语料、查询、split、全部标签和候选排名放在一起，便于逐格审阅。公开的 dev/test 划分只演示“分母由事先指定的 query ID 决定”；所有数据在本文可见，不能声称这是隐藏测试集或证明了没有泄漏。
保存为 fixture.py：
# Invented documents and author-assigned exhaustive judgments, not user data.
CORPUS = {
'd1': '下载操作指南：在设置的下载目录字段选择保存位置。',
'd2': '下载故障指南：连接断开后可点击重试，重新下载文件。',
'd3': '导出文本指南：点击导出并选择 TXT，保存纯文本副本。',
'd4': '导出表格指南：点击导出并选择 CSV，保存表格副本。',
'd5': '同步指南：关闭自动同步后可点击立即同步进行手动同步。',
'd6': '界面指南：在外观设置中选择深色主题。',
}
QUERIES = {
'q1': {'text': '怎样重新下载失败的文件？', 'split': 'dev'},
'q2': {'text': '有哪些导出格式，分别怎样导出？', 'split': 'test'},
'q3': {'text': '关闭自动同步后，怎样手动同步？', 'split': 'test'},
'q4': {'text': '怎样切换到深色主题？', 'split': 'test'},
'q5': {'text': '怎样注销账户？', 'split': 'test'},
}
# Each query-doc pair is deliberately judged. A 0 is an explicit judgment.
QRELS = {
'q1': {'d1': 0, 'd2': 1, 'd3': 0, 'd4': 0, 'd5': 0, 'd6': 0},
'q2': {'d1': 0, 'd2': 0, 'd3': 1, 'd4': 1, 'd5': 0, 'd6': 0},
'q3': {'d1': 0, 'd2': 0, 'd3': 0, 'd4': 0, 'd5': 1, 'd6': 0},
'q4': {'d1': 0, 'd2': 0, 'd3': 0, 'd4': 0, 'd5': 0, 'd6': 1},
'q5': {'d1': 0, 'd2': 0, 'd3': 0, 'd4': 0, 'd5': 0, 'd6': 0},
}
RUNS = {
'fixture_A': {'q2': ['d3','d1','d4'], 'q3': ['d2','d5'], 'q5': []},
'fixture_B': {'q2': ['d1','d3','d4'], 'q3': ['d5','d2'], 'q4': ['d6'], 'q5': []},
'all_empty': {q: [] for q, m in QUERIES.items() if m['split'] == 'test'},
}
6. 先拒绝含糊输入，再计算分数
evaluate 的第一段检查 k、query 集和 qrels。随后对每条选定的 query 验证整份排名：文档 ID 必须在语料中，同一文档不能重复。即便重复项落在 k 之后，也会拒绝整个 run，而非悄悄切掉它。本例采用严格约束，让调用方明确处理脏数据。
未知 query、混入 dev 的 query、未知文档和没有判断的 query-document 对都会报错。qrels 只接受整数 0、1，Python 的 True、False 虽然能参与整数运算，也不能充当本协议的标签。少一格与明确写 0 是两回事。
如果实际检索返回的是 chunk，先确定评价粒度。要按文档评价，就事先规定 chunk 到文档的映射、同文档如何合并、如何保持首次出现顺序，以及 k 在哪个层级截断。本文拒绝重复 doc ID，不替你临时选择去重规则。
实现使用 Fraction 保存精确分数，避免在小样例里把 5/6 四舍五入后比较。protocol_sha256 覆盖语料、query 文本与 split、qrels、k；它用于记录评价协议版本，不包含 run，也不证明没人见过测试集。接入真实系统时还需记录 run 内容、检索器配置、模型和索引版本、过滤规则等信息。
保存为 metrics.py：
"""Research-only, exhaustive binary qrels, no models or dependencies."""
from fractions import Fraction
import hashlib
import json
def evaluate(corpus, queries, qrels, run, split='test', k=2):
if type(k) is not int or k < 1:
raise ValueError('k must be a positive integer')
if not corpus or not queries or not isinstance(run, dict):
raise ValueError('nonempty corpus and queries, mapping run required')
if set(qrels) != set(queries):
raise ValueError('qrels must cover every query')
selected = [q for q, meta in queries.items() if meta['split'] == split]
if not selected:
raise ValueError('split has no queries')
if set(run) - set(selected):
raise ValueError('unknown or wrong-split query')
for q in queries:
if set(qrels[q]) != set(corpus):
raise ValueError('unjudged or unknown documents: exhaustive qrels required')
if any(type(v) is not int or v not in (0, 1) for v in qrels[q].values()):
raise ValueError('qrels labels must be integer 0 or 1')
rows = []
for q in selected:
ranking = run.get(q, [])
if not isinstance(ranking, list) or any(not isinstance(d, str) for d in ranking):
raise ValueError('ranking must be a list of document IDs')
if len(set(ranking)) != len(ranking):
raise ValueError('duplicate document ID; deduplicate upstream explicitly')
if set(ranking) - set(corpus):
raise ValueError('unknown document ID')
relevant = {d for d, rel in qrels[q].items() if rel == 1}
top = ranking[:k]
hits = len(set(top) & relevant)
first = next((i for i, d in enumerate(top, 1) if d in relevant), None)
rows.append(dict(query=q, missing=q not in run, returned=len(ranking),
relevant=len(relevant), hits=hits, first_relevant_rank=first,
recall=Fraction(hits, len(relevant)) if relevant else None,
rr=Fraction(1, first) if first else (Fraction(0) if relevant else None)))
eligible = [r for r in rows if r['relevant'] > 0]
def mean(key):
return sum((r[key] for r in eligible), Fraction(0)) / len(eligible) if eligible else None
canonical = json.dumps(dict(corpus=corpus, queries=queries, qrels=qrels, split=split, k=k),
sort_keys=True, ensure_ascii=False, separators=(',', ':'))
return dict(protocol_sha256=hashlib.sha256(canonical.encode()).hexdigest(), split=split, k=k,
total_queries=len(rows), eligible_queries=len(eligible),
no_relevant_queries=[r['query'] for r in rows if not r['relevant']],
missing_queries=[r['query'] for r in rows if r['missing']],
recall_at_k=mean('recall'), mrr_at_k=mean('rr'), rows=rows)
7. 把错误算法作为负控一起跑
负控的作用是故意展示一个错误做法会怎样歪曲结果，再检查正式实现有没有挡住它。run_experiment.py 算三组人工排名，并保留三种错误计算：删除缺排名的 query，把 hit-rate 冒充 Recall，以及把重复文档按出现次数计算。
对 q2 的 [d3, d3]，错误代码把同一份相关文档算两遍，得到 1；严格评分器直接拒绝该 run。这里不是说“重复输入应得几分”，而是明确告诉上游：先按约定清理输入，再来评价。输出中的 contract_rejections 同时记录未知文档、错误 split 和未判断 qrel 的拒绝原因。
保存为 run_experiment.py：
import copy
import json
from fractions import Fraction
from metrics import evaluate
from fixture import CORPUS, QUERIES, QRELS, RUNS
def score(run): return evaluate(CORPUS, QUERIES, QRELS, run)
results = {name: score(run) for name,run in RUNS.items()}
a = results['fixture_A']
present = [r for r in a['rows'] if r['relevant'] and not r['missing']]
wrong_dropped_missing = sum(r['recall'] for r in present)/len(present)
assert wrong_dropped_missing == Fraction(3,4) and a['recall_at_k']==Fraction(1,2)
wrong_hit_rate = Fraction(1) # q2 finds one of two relevant documents
correct_recall = a['rows'][0]['recall']
assert wrong_hit_rate != correct_recall == Fraction(1,2)
wrong_duplicate_recall = sum(QRELS['q2'][d] for d in ['d3','d3'])/2
assert wrong_duplicate_recall == 1
rejections={}
for name,run in {'duplicate':{'q2':['d3','d3']},'unknown_doc':{'q2':['d7']},'wrong_split':{'q1':['d2']}}.items():
try: score(run)
except ValueError as e: rejections[name]=str(e)
else: raise AssertionError(name)
rels=copy.deepcopy(QRELS); del rels['q2']['d1']
try: evaluate(CORPUS,QUERIES,rels,RUNS['fixture_A'])
except ValueError as e: rejections['unjudged']=str(e)
else: raise AssertionError('unjudged')
output=dict(scope='Synthetic fixed rankings; metric implementation evidence, not model performance',
results=results,
negative_controls={'dropping_missing_wrong':wrong_dropped_missing,
'fixed_denominator_correct':a['recall_at_k'],
'hit_rate_mislabeled_as_recall_wrong':wrong_hit_rate,
'q2_recall_correct':correct_recall,
'duplicate_counting_wrong':wrong_duplicate_recall,
'contract_rejections':rejections})
print(json.dumps(output,default=lambda x:str(x) if isinstance(x,Fraction) else x,ensure_ascii=False,indent=2))
8. 用十八项测试把口径固定下来
测试不仅检查 A、B 的总分，也检查缺结果仍留在分母、没有相关文档时的 null、k 的截断、相关文档新增后两指标的差异，以及非法输入必须拒绝。这些断言防的是公式实现和数据契约回归，不验证某个检索系统的语义能力。
保存为 test_metrics.py：
import copy
import unittest
from fractions import Fraction as F
from metrics import evaluate
from fixture import CORPUS, QUERIES, QRELS, RUNS
class MetricsTests(unittest.TestCase):
def score(self, run=None, **kw):
return evaluate(CORPUS, QUERIES, QRELS, RUNS['fixture_A'] if run is None else run, **kw)
def test_exact_A(self):
r=self.score(); self.assertEqual((r['recall_at_k'],r['mrr_at_k']), (F(1,2),F(1,2)))
def test_exact_B(self):
r=self.score(RUNS['fixture_B']); self.assertEqual((r['recall_at_k'],r['mrr_at_k']), (F(5,6),F(5,6)))
def test_missing_stays_denominator(self):
r=self.score(); self.assertEqual((r['eligible_queries'],r['missing_queries']), (3,['q4']))
def test_no_relevant_explicit(self):
r=self.score(); self.assertEqual((r['total_queries'],r['no_relevant_queries']), (4,['q5'])); self.assertIsNone(r['rows'][-1]['recall'])
def test_empty(self):
r=self.score(RUNS['all_empty']); self.assertEqual((r['recall_at_k'],r['mrr_at_k']), (0,0))
def test_duplicate_rejected(self):
with self.assertRaisesRegex(ValueError,'duplicate'): self.score({'q2':['d3','d3']})
def test_unknown_doc(self):
with self.assertRaisesRegex(ValueError,'unknown document'): self.score({'q2':['d999']})
def test_unknown_query(self):
with self.assertRaisesRegex(ValueError,'query'): self.score({'q999':[]})
def test_wrong_split(self):
with self.assertRaisesRegex(ValueError,'split'): self.score({'q1':['d2']})
def test_unjudged_not_nonrelevant(self):
rel=copy.deepcopy(QRELS); del rel['q2']['d1']
with self.assertRaisesRegex(ValueError,'unjudged'): evaluate(CORPUS,QUERIES,rel,RUNS['fixture_A'])
def test_bool_label_rejected(self):
rel=copy.deepcopy(QRELS); rel['q2']['d1']=False
with self.assertRaisesRegex(ValueError,'integer'): evaluate(CORPUS,QUERIES,rel,{})
def test_cutoff(self):
r=self.score({'q2':['d1','d2','d3']}); self.assertEqual(r['rows'][0]['rr'],0)
def test_recall_not_hit_rate(self):
r=self.score({'q2':['d3']}); self.assertEqual(r['rows'][0]['recall'],F(1,2)); self.assertEqual(r['rows'][0]['rr'],1)
def test_second_relevant_changes_recall_not_rr(self):
a=self.score({'q2':['d3']})['rows'][0]; b=self.score({'q2':['d3','d4']})['rows'][0]
self.assertEqual((a['rr'],b['rr']),(1,1)); self.assertEqual((a['recall'],b['recall']),(F(1,2),1))
def test_k_validation(self):
for k in [True,0,-1,1.5]:
with self.assertRaises(ValueError): self.score(k=k)
def test_empty_eligible_mean_is_null(self):
queries={'q5':QUERIES['q5']}; r=evaluate(CORPUS,queries,{'q5':QRELS['q5']},{})
self.assertIsNone(r['mrr_at_k']); self.assertEqual(r['eligible_queries'],0)
def test_hash_changes_with_qrels(self):
rel=copy.deepcopy(QRELS); rel['q2']['d1']=1
self.assertNotEqual(self.score()['protocol_sha256'],evaluate(CORPUS,QUERIES,rel,{})['protocol_sha256'])
def test_missing_qrels_rejected(self):
rel=copy.deepcopy(QRELS); del rel['q4']
with self.assertRaisesRegex(ValueError,'every query'): evaluate(CORPUS,QUERIES,rel,{})
if __name__=='__main__': unittest.main()
在这四个文件所在目录运行：
python -m unittest -v
python run_experiment.py > actual.json
python -c "from pathlib import Path; assert Path('actual.json').read_bytes() == Path('expected.json').read_bytes(); print('output matches')"第三条命令需要先把下一节 JSON 保存成 expected.json，并保留文件末尾换行。前两条已经足以执行全部断言和生成结果。成功测试的结尾应是 Ran 18 tests 和 OK；出现失败时先看对应输入，不要为了通过测试改掉预期值。
9. 按 query 读结果，再看平均值
固定 k=2，A 的 q2 找到 d3，Recall=1/2、RR=1；q3 在第二位找到 d5，Recall=1、RR=1/2；q4 缺结果，两项都是 0。三个 eligible query 的均值分别是 (1/2+1+0)/3=1/2 和 (1+1/2+0)/3=1/2。
B 的 q2 把 d3 放到第二位，Recall 仍为 1/2，RR 退到 1/2。q3、q4 都在首位命中，所以两个均值都是 5/6。平均值变高，仍然有 q2 的首个命中退步；保留成对的逐 query 对照才能看见它。all_empty 的两个均值均为 0，分母仍然是三个有相关文档的 test query。
这些数字只说明两份手写排名在固定标签和协议下的算术关系。它们没有采样误差分析，没有代表真实流量的查询分布，也没有训练或真实检索过程，不能据此声称某个模型提高了质量。
下面是本次程序完整输出。Fraction 在 JSON 中写成字符串，null 是 JSON 空值，不是字符串 "null"：
保存为 expected.json：
{
"scope": "Synthetic fixed rankings; metric implementation evidence, not model performance",
"results": {
"fixture_A": {
"protocol_sha256": "9dd1a576cc830736a0884e847845f31f2390843c27a250abf3bafaf8fa72cadb",
"split": "test",
"k": 2,
"total_queries": 4,
"eligible_queries": 3,
"no_relevant_queries": [
"q5"
],
"missing_queries": [
"q4"
],
"recall_at_k": "1/2",
"mrr_at_k": "1/2",
"rows": [
{
"query": "q2",
"missing": false,
"returned": 3,
"relevant": 2,
"hits": 1,
"first_relevant_rank": 1,
"recall": "1/2",
"rr": "1"
},
{
"query": "q3",
"missing": false,
"returned": 2,
"relevant": 1,
"hits": 1,
"first_relevant_rank": 2,
"recall": "1",
"rr": "1/2"
},
{
"query": "q4",
"missing": true,
"returned": 0,
"relevant": 1,
"hits": 0,
"first_relevant_rank": null,
"recall": "0",
"rr": "0"
},
{
"query": "q5",
"missing": false,
"returned": 0,
"relevant": 0,
"hits": 0,
"first_relevant_rank": null,
"recall": null,
"rr": null
}
]
},
"fixture_B": {
"protocol_sha256": "9dd1a576cc830736a0884e847845f31f2390843c27a250abf3bafaf8fa72cadb",
"split": "test",
"k": 2,
"total_queries": 4,
"eligible_queries": 3,
"no_relevant_queries": [
"q5"
],
"missing_queries": [],
"recall_at_k": "5/6",
"mrr_at_k": "5/6",
"rows": [
{
"query": "q2",
"missing": false,
"returned": 3,
"relevant": 2,
"hits": 1,
"first_relevant_rank": 2,
"recall": "1/2",
"rr": "1/2"
},
{
"query": "q3",
"missing": false,
"returned": 2,
"relevant": 1,
"hits": 1,
"first_relevant_rank": 1,
"recall": "1",
"rr": "1"
},
{
"query": "q4",
"missing": false,
"returned": 1,
"relevant": 1,
"hits": 1,
"first_relevant_rank": 1,
"recall": "1",
"rr": "1"
},
{
"query": "q5",
"missing": false,
"returned": 0,
"relevant": 0,
"hits": 0,
"first_relevant_rank": null,
"recall": null,
"rr": null
}
]
},
"all_empty": {
"protocol_sha256": "9dd1a576cc830736a0884e847845f31f2390843c27a250abf3bafaf8fa72cadb",
"split": "test",
"k": 2,
"total_queries": 4,
"eligible_queries": 3,
"no_relevant_queries": [
"q5"
],
"missing_queries": [],
"recall_at_k": "0",
"mrr_at_k": "0",
"rows": [
{
"query": "q2",
"missing": false,
"returned": 0,
"relevant": 2,
"hits": 0,
"first_relevant_rank": null,
"recall": "0",
"rr": "0"
},
{
"query": "q3",
"missing": false,
"returned": 0,
"relevant": 1,
"hits": 0,
"first_relevant_rank": null,
"recall": "0",
"rr": "0"
},
{
"query": "q4",
"missing": false,
"returned": 0,
"relevant": 1,
"hits": 0,
"first_relevant_rank": null,
"recall": "0",
"rr": "0"
},
{
"query": "q5",
"missing": false,
"returned": 0,
"relevant": 0,
"hits": 0,
"first_relevant_rank": null,
"recall": null,
"rr": null
}
]
}
},
"negative_controls": {
"dropping_missing_wrong": "3/4",
"fixed_denominator_correct": "1/2",
"hit_rate_mislabeled_as_recall_wrong": "1",
"q2_recall_correct": "1/2",
"duplicate_counting_wrong": 1.0,
"contract_rejections": {
"duplicate": "duplicate document ID; deduplicate upstream explicitly",
"unknown_doc": "unknown document ID",
"wrong_split": "unknown or wrong-split query",
"unjudged": "unjudged or unknown documents: exhaustive qrels required"
}
}
}
10. 换成自己的 qrels 时，先做这几件事
先选一个能认真判断的小语料快照，并给每份文档稳定 ID。收集希望系统支持的查询，写下它们的具体信息需求，包含常见问法、容易混淆的问法和语料没有答案的情况。不要只从某个系统已返回的结果里挑“看起来不错”的 query。
给判断者同时提供需求说明和文档正文，先定相关标准，再逐项记录标签及有争议的原因。小库可穷尽判断；大库做不到时，要明确哪些文档尚未判断，以及评价方法允许哪些推断，不能把未知自动改成 0 来凑齐本评分器。本例只支持穷尽二元 qrels，宁可报错，也不伪造可靠的 Recall 分母。
在看 test 结果之前固定 k、相关阈值、文档粒度、去重规则和无相关样本口径。k 应对应你真正会送给后续步骤的候选规模。调参用 dev，同一意图的近义改写尽量成组划分，以免几乎相同的问题分到两边。若不断看 test 再改系统，这批 query 已经参与开发，应找新的留出查询，而不是继续把它当作未见测试集。
然后把真实检索器输出转换成 run：query ID 映射到按顺序排列的文档 ID 列表。让每个候选系统使用完全相同的语料、标签、query 集和 k。为每次运行保存配置与结果版本，先核对 missing_queries 和每条明细，再比较均值。遇到某类 query 退步，回到需求和文档检查原因；不要仅因为总分更高就抹掉局部损失。
最后把能证明的范围写进报告。本教程已经实际执行评分程序和十八项测试；没有验证真实检索模型、查询分布代表性、标注一致性、统计泛化、答案生成或外部指标 SDK。把这段评分器接到真实系统之后，仍需分别验证这些问题。眼下得到的是一把口径可检查的尺子。
11. 检索返回的是块时，Recall@2 的 2 到底数什么
前面的评分器只收文档 ID，实际检索却可能返回 chunk。接入前先问：要衡量“前两块覆盖了哪些文档”，还是“将已有候选按文档合并后取前两篇”？两种协议都可以成立，候选预算必须写清楚。
沿用 q2 的导出问题，相关文档是 d3 和 d4。固定输入为 [d3a, d3b, d4a]，前三个块依次映射到 d3、d3、d4。按首次出现合并为 [d3, d4]，文档级 Recall@2=2/2=1。若流水线只提供前两块，合并后只有 d3，Recall@2=1/2。适配器不能补回从未收到的 d4。错误在于隐去预算变化，把两种输入当成同一协议比较。
q3 更能看清排名单位：[d1a, d1b, d5a] 中，只有 d5 相关。它在块排名的第三位，在合并后的文档排名中却是第二位，故 RR@2=1/2；只提供前两块时，RR@2=0。块位置不能直接充当文档位置。这些次序都是手写夹具，没有运行检索器。
12. 按首次出现合并，并保留每一步的来路
chunk_to_doc 由应用从同一语料快照提供，值必须对应 qrels 标注的文档 ID。不要从块名、标题或相同正文猜身份。LangChain 的父文档检索说明明确区分原始整篇与较大的父块；有 parent_id 还不足以证明它对应当前标签。Elasticsearch 的 collapse 文档也提醒，折叠前命中数与分组后的结果数不同。这里仅参考这些概念，没有调用或模拟两者。
适配器先检查整份输入，再生成结果。不同块属于同一文档很正常；同一个块 ID 重复出现则拒收。未知块、未使用但指向语料外的映射、最终截止位之后的重复块，也必须报错。每篇文档保留首块 ID、首块排名和全部贡献排名，缺失的 query 不补成空列表。
函数故意没有 k 参数。它只转换收到的候选，文档级截止位和 qrels 检查仍由前文的 evaluate 负责。adapter_sha256 记录映射及合并规则版本；原来的 protocol_sha256 不包含块映射。映射变了，即使评分器哈希未变，上游协议也已改变。哈希只能帮助发现变化，不能认证映射来源。
保存为 chunk_fixture.py：
"""Author-written rank/identity fixture; no actual chunker or retriever."""
CHUNK_TO_DOC = {
'd3a': 'd3', 'd3b': 'd3', 'd4a': 'd4',
'd1a': 'd1', 'd1b': 'd1', 'd5a': 'd5',
'd6a': 'd6',
}
CHUNK_RUN = {
'q2': ['d3a', 'd3b', 'd4a'],
'q3': ['d1a', 'd1b', 'd5a'],
# q4 is deliberately missing, not silently fabricated as returned.
'q5': [],
}
保存为 chunk_adapter.py：
"""Synthetic offline adapter: first-occurrence document collapse, no retrieval."""
import hashlib
import json
def adapt_runs(corpus, queries, chunk_to_doc, chunk_runs, split='test'):
"""Validate all supplied ranks, then preserve first parent occurrence.
`chunk_to_doc` must come from the same trusted corpus snapshot as qrels.
IDs are exact strings. No prefix parsing, title matching or score sorting.
This function intentionally has no k: metrics.evaluate applies document k.
"""
if type(corpus) is not dict or not corpus:
raise ValueError('nonempty corpus dictionary required')
if any(type(d) is not str or not d for d in corpus):
raise ValueError('corpus document IDs must be nonempty strings')
if type(queries) is not dict or not queries:
raise ValueError('nonempty query dictionary required')
if type(split) is not str or not split:
raise ValueError('split must be a nonempty string')
for query, meta in queries.items():
if type(query) is not str or not query or type(meta) is not dict:
raise ValueError('query IDs and metadata are invalid')
if type(meta.get('split')) is not str or not meta['split']:
raise ValueError('query split must be a nonempty string')
if type(chunk_to_doc) is not dict or type(chunk_runs) is not dict:
raise ValueError('mapping and runs must be dictionaries')
for chunk, doc in chunk_to_doc.items():
if type(chunk) is not str or not chunk or type(doc) is not str or not doc:
raise ValueError('chunk and document IDs must be strings')
if doc not in corpus:
raise ValueError('mapping points outside the corpus')
selected = {q for q, meta in queries.items() if meta['split'] == split}
if not selected:
raise ValueError('split has no queries')
if set(chunk_runs) - selected:
raise ValueError('unknown or wrong-split query')
# Complete validation before creating a consumable adapted run.
for ranking in chunk_runs.values():
if type(ranking) is not list:
raise ValueError('chunk ranking must be a list')
if any(type(c) is not str or c not in chunk_to_doc for c in ranking):
raise ValueError('unknown or invalid chunk ID')
if len(set(ranking)) != len(ranking):
raise ValueError('duplicate chunk ID')
doc_runs, evidence = {}, {}
for query, ranking in chunk_runs.items():
rows = {}
for rank, chunk in enumerate(ranking, 1):
doc = chunk_to_doc[chunk]
if doc not in rows:
rows[doc] = {'document_id': doc, 'first_chunk_id': chunk,
'first_chunk_rank': rank, 'chunk_ranks': []}
rows[doc]['chunk_ranks'].append(rank)
doc_runs[query] = list(rows)
evidence[query] = {'received_chunks': len(ranking),
'distinct_documents': len(rows),
'documents': list(rows.values())}
config = {'adapter': 'first-occurrence-document-v1',
'chunk_to_doc': chunk_to_doc}
raw = json.dumps(config, sort_keys=True, ensure_ascii=False,
separators=(',', ':')).encode('utf-8')
return {'adapter_sha256': hashlib.sha256(raw).hexdigest(),
'document_runs': doc_runs, 'evidence': evidence}
13. 逐条对照，再检查固定分母和完整预期
把本节文件和前文四个 Python 文件、expected.json 放在同一目录，原文件无需修改。q4 仍缺排名，在三个有相关文档的 test query 中记 0；q5 保持显式空列表，因无相关文档而得 null，不参加两项均值。
三块输入合并后，q2 的 Recall、RR 都为 1，q3 为 1、1/2，q4 为 0、0。因此宏平均 Recall@2=(1+1+0)/3=2/3，MRR@2=(1+1/2+0)/3=1/2。两块预算那份输入的均值分别是 (1/2+0+0)/3=1/6、(1+0+0)/3=1/3。它们是不同预算下的固定算术结果。Recall 的分母始终是相关文档数，见 Stanford 教材 8.3 节。
程序还实际演示两种错误：只替换 ID 会留下重复文档，被原评分器拒绝；按出现次数计数会把 d3 的两块误当成两篇相关文档。expected-chunk.json 是事先写定的字面预期，测试不会用当前程序的输出重新生成它。
保存为 probe_chunk_cutoff.py：
"""Print deterministic observations. Expected answers live in another file."""
import json
from fractions import Fraction
from chunk_adapter import adapt_runs
from chunk_fixture import CHUNK_TO_DOC, CHUNK_RUN
from fixture import CORPUS, QUERIES, QRELS
from metrics import evaluate
def summary(run):
report = evaluate(CORPUS, QUERIES, QRELS, run, k=2)
return {'recall_at_2': str(report['recall_at_k']),
'mrr_at_2': str(report['mrr_at_k']),
'missing_queries': report['missing_queries'],
'eligible_queries': report['eligible_queries'],
'rows': {row['query']: {'recall': None if row['recall'] is None else str(row['recall']),
'rr': None if row['rr'] is None else str(row['rr']),
'returned': row['returned']}
for row in report['rows']}}
def experiment():
complete_pool = adapt_runs(CORPUS, QUERIES, CHUNK_TO_DOC, CHUNK_RUN)
# A legitimate alternative candidate budget; do not claim equal input pools.
two_chunk_input = {q: ranks[:2] for q, ranks in CHUNK_RUN.items()}
truncated = adapt_runs(CORPUS, QUERIES, CHUNK_TO_DOC, two_chunk_input)
# Merely replacing IDs leaves duplicate docs; the original evaluator refuses it.
uncollapsed = {q: [CHUNK_TO_DOC[c] for c in ranks]
for q, ranks in CHUNK_RUN.items()}
try:
evaluate(CORPUS, QUERIES, QRELS, uncollapsed, k=2)
except ValueError as error:
raw_mapping = {'accepted': False, 'reason': str(error)}
else:
raise AssertionError('original evaluator unexpectedly accepted duplicate docs')
# Deliberately wrong: count occurrences instead of distinct relevant docs.
repeated_hits = sum(QRELS['q2'][CHUNK_TO_DOC[c]] for c in CHUNK_RUN['q2'][:2])
relevant_count = sum(QRELS['q2'].values())
return {'document_runs': complete_pool['document_runs'],
'provenance': complete_pool['evidence'],
'collapse_then_document_cutoff': summary(complete_pool['document_runs']),
'two_chunk_budget_then_collapse': summary(truncated['document_runs']),
'raw_id_replacement': raw_mapping,
'naive_q2_duplicate_occurrence_recall': str(Fraction(repeated_hits, relevant_count))}
if __name__ == '__main__':
print(json.dumps(experiment(), ensure_ascii=False, indent=2))
保存为 expected-chunk.json：
{
"document_runs": {"q2": ["d3", "d4"], "q3": ["d1", "d5"], "q5": []},
"provenance": {
"q2": {"received_chunks": 3, "distinct_documents": 2, "documents": [
{"document_id": "d3", "first_chunk_id": "d3a", "first_chunk_rank": 1, "chunk_ranks": [1, 2]},
{"document_id": "d4", "first_chunk_id": "d4a", "first_chunk_rank": 3, "chunk_ranks": [3]}
]},
"q3": {"received_chunks": 3, "distinct_documents": 2, "documents": [
{"document_id": "d1", "first_chunk_id": "d1a", "first_chunk_rank": 1, "chunk_ranks": [1, 2]},
{"document_id": "d5", "first_chunk_id": "d5a", "first_chunk_rank": 3, "chunk_ranks": [3]}
]},
"q5": {"received_chunks": 0, "distinct_documents": 0, "documents": []}
},
"collapse_then_document_cutoff": {
"recall_at_2": "2/3", "mrr_at_2": "1/2", "missing_queries": ["q4"], "eligible_queries": 3,
"rows": {
"q2": {"recall": "1", "rr": "1", "returned": 2},
"q3": {"recall": "1", "rr": "1/2", "returned": 2},
"q4": {"recall": "0", "rr": "0", "returned": 0},
"q5": {"recall": null, "rr": null, "returned": 0}
}
},
"two_chunk_budget_then_collapse": {
"recall_at_2": "1/6", "mrr_at_2": "1/3", "missing_queries": ["q4"], "eligible_queries": 3,
"rows": {
"q2": {"recall": "1/2", "rr": "1", "returned": 1},
"q3": {"recall": "0", "rr": "0", "returned": 1},
"q4": {"recall": "0", "rr": "0", "returned": 0},
"q5": {"recall": null, "rr": null, "returned": 0}
}
},
"raw_id_replacement": {"accepted": false, "reason": "duplicate document ID; deduplicate upstream explicitly"},
"naive_q2_duplicate_occurrence_recall": "1"
}
保存为 test_chunk_adapter.py：
import copy
import json
import unittest
from pathlib import Path
from chunk_adapter import adapt_runs
from chunk_fixture import CHUNK_TO_DOC, CHUNK_RUN
from fixture import CORPUS, QUERIES, QRELS
from metrics import evaluate
from probe_chunk_cutoff import experiment
class ChunkAdapterTests(unittest.TestCase):
def adapt(self, run=None, mapping=None, corpus=None):
return adapt_runs(CORPUS if corpus is None else corpus, QUERIES,
CHUNK_TO_DOC if mapping is None else mapping,
CHUNK_RUN if run is None else run)
def test_exact_fixed_observations(self):
expected = json.loads(Path(__file__).with_name('expected-chunk.json').read_text())
self.assertEqual(experiment(), expected)
def test_nonadjacent_parent_repeats_collapse(self):
result = self.adapt({'q2': ['d3a', 'd4a', 'd3b']})
self.assertEqual(result['document_runs']['q2'], ['d3', 'd4'])
self.assertEqual(result['evidence']['q2']['documents'][0]['chunk_ranks'], [1, 3])
def test_first_occurrence_not_sorted_document_id(self):
result = self.adapt({'q2': ['d4a', 'd3a']})
self.assertEqual(result['document_runs']['q2'], ['d4', 'd3'])
def test_identical_text_is_not_identical_document(self):
corpus = dict(CORPUS, d3='same title and text', d4='same title and text')
self.assertEqual(self.adapt(corpus=corpus)['document_runs']['q2'], ['d3', 'd4'])
def test_incomplete_candidate_pool_cannot_be_backfilled(self):
result = self.adapt({'q2': ['d3a', 'd3b']})
self.assertEqual(result['document_runs'], {'q2': ['d3']})
row = evaluate(CORPUS, QUERIES, QRELS, result['document_runs'])['rows'][0]
self.assertEqual(str(row['recall']), '1/2')
def test_missing_and_empty_stay_different(self):
result = self.adapt()['document_runs']
self.assertNotIn('q4', result)
self.assertEqual(result['q5'], [])
def test_unknown_chunk_after_document_cutoff_rejected(self):
with self.assertRaisesRegex(ValueError, 'unknown'):
self.adapt({'q2': ['d3a', 'd4a', 'missing']})
def test_duplicate_chunk_after_document_cutoff_rejected(self):
with self.assertRaisesRegex(ValueError, 'duplicate chunk'):
self.adapt({'q2': ['d3a', 'd4a', 'd3a']})
def test_mapping_outside_corpus_rejected_even_if_unused(self):
with self.assertRaisesRegex(ValueError, 'outside'):
self.adapt(mapping=dict(CHUNK_TO_DOC, unused='missing-doc'))
def test_unknown_and_wrong_split_queries_rejected(self):
for q in ('q1', 'absent'):
with self.subTest(query=q), self.assertRaisesRegex(ValueError, 'split'):
self.adapt({q: []})
def test_strict_rank_and_id_types(self):
for rank in ('d3a', ('d3a',), [True], [1], [None], ['']):
with self.subTest(rank=rank), self.assertRaises(ValueError):
self.adapt({'q2': rank})
for mapping in ([], None, {'': 'd3'}, {1: 'd3'}, {'x': ''}, {'x': True}):
with self.subTest(mapping=mapping), self.assertRaises(ValueError):
adapt_runs(CORPUS, QUERIES, mapping, {})
for index, invalid in ((0, []), (0, {}), (0, {1: 'text'}),
(1, []), (1, {}), (1, {True: {'split': 'test'}}),
(1, {'q2': None}), (1, {'q2': {}}),
(1, {'q2': {'split': True}}), (3, [])):
args = [CORPUS, QUERIES, CHUNK_TO_DOC, CHUNK_RUN]
args[index] = invalid
with self.subTest(index=index, invalid=invalid), self.assertRaises(ValueError):
adapt_runs(*args)
for split in ('absent', '', None, True):
with self.subTest(split=split), self.assertRaises(ValueError):
adapt_runs(CORPUS, QUERIES, CHUNK_TO_DOC, {}, split=split)
def test_input_snapshots_not_mutated(self):
data = copy.deepcopy((CORPUS, QUERIES, CHUNK_TO_DOC, CHUNK_RUN))
before = copy.deepcopy(data)
adapt_runs(*data)
self.assertEqual(data, before)
def test_mapping_fingerprint_changes_while_metric_protocol_does_not(self):
changed = dict(CHUNK_TO_DOC, **{'d3b': 'd4'})
one, two = self.adapt(), self.adapt(mapping=changed)
self.assertNotEqual(one['adapter_sha256'], two['adapter_sha256'])
a = evaluate(CORPUS, QUERIES, QRELS, one['document_runs'])
b = evaluate(CORPUS, QUERIES, QRELS, two['document_runs'])
self.assertEqual(a['protocol_sha256'], b['protocol_sha256'])
def test_dictionary_mapping_order_does_not_change_fingerprint(self):
backward = dict(reversed(list(CHUNK_TO_DOC.items())))
self.assertEqual(self.adapt()['adapter_sha256'], self.adapt(mapping=backward)['adapter_sha256'])
if __name__ == '__main__':
unittest.main()
运行下面三条命令。第一条应显示 32 项测试通过，其中原有 18 项、新增 14 项；第二条生成完整观察记录；第三条再次比较完整 JSON。
python -m unittest -v
python probe_chunk_cutoff.py > observed-chunk.json
python -c "import json; from pathlib import Path; assert json.loads(Path('observed-chunk.json').read_text()) == json.loads(Path('expected-chunk.json').read_text()); print('chunk output matches')"14. 故意改坏六次，看测试能否指出差错
下面的脚本在临时目录中分别改坏文档排序、完整输入检查、重复块拒绝、未使用映射检查、缺失 query 保留和首块记录，再运行各自的目标测试。六次都应出现指定断言失败；语法错误或导入错误不算抓到了问题。它不会改动当前目录的正确实现。
保存为 run_mutation_checks.py：
"""Run six deliberately broken adapters in isolated temporary directories."""
import json
from pathlib import Path
import shutil
import subprocess
import sys
import tempfile
BASE = Path(__file__).resolve().parent
MUTANTS = [
('sort_document_ids', 'doc_runs[query] = list(rows)',
'doc_runs[query] = sorted(rows)',
'test_first_occurrence_not_sorted_document_id'),
('truncate_input_before_validation',
' # Complete validation before creating a consumable adapted run.',
' chunk_runs = {q: ranks[:2] for q, ranks in chunk_runs.items()}',
'test_unknown_chunk_after_document_cutoff_rejected'),
('allow_duplicate_chunk_id',
' if len(set(ranking)) != len(ranking):',
' if False:',
'test_duplicate_chunk_after_document_cutoff_rejected'),
('ignore_unused_bad_mapping', ' if doc not in corpus:',
' if False:',
'test_mapping_outside_corpus_rejected_even_if_unused'),
('fabricate_missing_query',
" config = {'adapter': 'first-occurrence-document-v1',",
" doc_runs.update({q: [] for q in selected if q not in doc_runs})\n"
" config = {'adapter': 'first-occurrence-document-v1',",
'test_missing_and_empty_stay_different'),
('overwrite_first_occurrence', ' if doc not in rows:',
' if True:',
'test_nonadjacent_parent_repeats_collapse'),
]
def main():
original = (BASE / 'chunk_adapter.py').read_text()
report = []
for name, old, new, test in MUTANTS:
if original.count(old) != 1:
raise RuntimeError('mutation anchor is no longer unique: ' + name)
with tempfile.TemporaryDirectory(prefix='chunk-mutant-') as tmp:
work = Path(tmp)
for source in BASE.iterdir():
if source.suffix in ('.py', '.json'):
shutil.copyfile(source, work / source.name)
(work / 'chunk_adapter.py').write_text(original.replace(old, new, 1))
command = [sys.executable, '-m', 'unittest', '-v',
'test_chunk_adapter.ChunkAdapterTests.' + test]
process = subprocess.run(command, cwd=work, capture_output=True, text=True)
# A syntax/import error is not evidence that the assertion caught a bug.
killed = (process.returncode == 1 and 'FAIL: ' + test in process.stderr
and 'ERROR:' not in process.stderr)
print(name + '\n' + process.stderr, file=sys.stderr)
if not killed:
raise AssertionError('mutant did not fail its intended assertion: ' + name)
report.append({'mutation': name, 'test': test, 'assertion_failed': True})
print(json.dumps(report, indent=2))
if __name__ == '__main__':
main()
python run_mutation_checks.py脚本成功结束时返回六条 assertion_failed: true。终端中的 FAILED 是故意改坏后的预期结果。新增测试还覆盖非相邻同文档块、相同正文但不同 ID、短候选池、缺失与空结果、严格类型，以及映射哈希对插入顺序不敏感。测试项数不代表覆盖率。
本补充于 2026-10-13 用 Python 3.12.14 执行，仅需标准库。32 项测试与六次故障注入均按预期完成。它证明的是固定数据的转换与计分关系，不验证模型质量、实际切块、分页补取、权限过滤或线上性能。文档相关，也不保证命中的那一块包含答案；要验收引用内容，还需要块级标注与正文复核。