RRF与检索指标 — AI系统工程

沿着对象与状态理解本章

检索排名只说明候选顺序，不说明事实已经成立。下图固定两份候选排名逐项融合，再加入版本与证据判断；如果答案需要两个条件，只召回其中一个还不够。

下方图解的固定输入在“输入与适用条件”中声明。它用于解释本章机制，不与本章其他独立例子的数值暗中接续。

动态图解 · F29

从两个候选列表，走到可用证据

“相关”与“足以回答”中间还差哪几步？

多个检索入口可能互补。融合处理“哪些候选值得优先看”；重排对少量候选作更细比较；证据核验处理“这段话是否真的支持当前结论”。三个阶段目标不同，不能用一个相似度分数替代全部判断。

RRF通过名次而非原始分数融合候选列表的方法。

重排对召回后的候选重新作相关性排序。

证据充分性材料能否支持问题要求的结论和条件。

输入与适用条件：本图独立排名算例：关键词[A,B,C]、语义[B,C,A]、k=60。所有候选始终不变，不接用其他章节的排名数据。

1 / 5 · 两种搜索给出不同名次

停留时间

步骤第1步，共5步

窄屏可左右滑动图形，或点击“放大图形”。文字解释紧随图后。

固定三个候选：关键词排名A、B、C；语义排名B、C、A。两种原始分数不在同一量纲，不能盲目相加。

本步状态由图形与说明共同给出。

默认静止。用“下一步”看完解释再继续；播放到最后一步自动停止。键盘聚焦播放器后，可用左右键逐步查看、空格播放/暂停。图中颜色辅助区分，含义同时用文字注明。

为什么需要

精确词项与语义线索各有盲区，组合能提供互补候选。

怎样实现

先融合候选，再核实原文与元数据，必要时补查缺失证据。

可能的代价与误解

提高Top-k可能把更多无关内容塞入上下文；检索器成绩提升不一定带来任务成功率提升。

如何判断是否做对

分别测召回、排名和最终回答支持度；不要只测最终一项或只测搜索分数。

怎样观察与验证

需要A、B却只返回A、X、Y时，Precision@3为1/3，Recall@3为1/2。增加噪声或交换排序，观察不同指标反应。

运行命令

python3 reading-notes/experiments/04_retrieval.py

完整教学代码

"""合成排名上的 RRF、Precision、Recall、MRR 和 nDCG。"""
import math


def unique(seq):
    return list(dict.fromkeys(seq))


def rrf(rankings: list[list[str]], constant: float = 60.0) -> list[tuple[str, float]]:
    if not math.isfinite(constant) or constant < 0:
        raise ValueError("平滑常数必须非负且有限")
    scores = {}
    for ranking in rankings:
        for rank, doc in enumerate(unique(ranking), 1):
            scores[doc] = scores.get(doc, 0.0) + 1 / (constant + rank)
    return sorted(scores.items(), key=lambda pair: (-pair[1], pair[0]))


def metrics(ranking: list[str], relevance: dict[str, int], k: int) -> dict:
    if not isinstance(k, int) or isinstance(k, bool) or k <= 0:
        raise ValueError("k 必须为正整数")
    if any(v < 0 for v in relevance.values()):
        raise ValueError("相关性等级不能为负")
    top = unique(ranking)[:k]
    relevant = {doc for doc, grade in relevance.items() if grade > 0}
    hits = sum(doc in relevant for doc in top)
    # 按请求的 k 作分母；结果不足 k 时空缺算未命中。
    precision = hits / k
    recall = hits / len(relevant) if relevant else None
    reciprocal_rank = next((1/i for i,d in enumerate(top,1) if d in relevant), 0.0)
    dcg = sum((2**relevance.get(doc,0)-1)/math.log2(i+1) for i,doc in enumerate(top,1))
    ideal = sorted(relevance.values(), reverse=True)[:k]
    idcg = sum((2**grade-1)/math.log2(i+1) for i,grade in enumerate(ideal,1))
    return {"precision_at_k": precision, "recall_at_k": recall,
            "reciprocal_rank_at_k": reciprocal_rank,
            "ndcg_at_k": dcg/idcg if idcg else None}


def self_test() -> None:
    m = metrics(["A","X","Y"], {"A":1,"B":1}, 3)
    assert m["precision_at_k"] == 1/3 and m["recall_at_k"] == 1/2
    assert metrics(["B","A"], {"A":1,"B":2},2)["ndcg_at_k"] == 1
    assert rrf([["A","A"],["B"]],0) == [("A",1.0),("B",1.0)]
    assert metrics([], {}, 3)["recall_at_k"] is None


def run_demo() -> dict:
    return {"性质":"人工相关性标签，非真实搜索基准",
            "融合":rrf([["A","B","C"],["B","D","A"]]),
            "缺少必要证据B":metrics(["A","X","Y"],{"A":1,"B":1},3),
            "说明":"单查询倒数排名；MRR需要在多个查询上取平均"}


if __name__ == "__main__":
    import json
    self_test()
    print(json.dumps(run_demo(), ensure_ascii=False, indent=2))

运行结果示例

以下结果由随包脚本产生。它们不是LLM或产品性能数据。

{
  "性质": "人工相关性标签，非真实搜索基准",
  "融合": [
    [
      "B",
      0.03252247488101534
    ],
    [
      "A",
      0.032266458495966696
    ],
    [
      "D",
      0.016129032258064516
    ],
    [
      "C",
      0.015873015873015872
    ]
  ],
  "缺少必要证据B": {
    "precision_at_k": 0.3333333333333333,
    "recall_at_k": 0.5,
    "reciprocal_rank_at_k": 1.0,
    "ndcg_at_k": 0.6131471927654584
  },
  "说明": "单查询倒数排名；MRR需要在多个查询上取平均"
}