手写 RRF，先补齐重复 ID、Top-K=0 和并列排序

RRF的公式很短：同一文档在每路结果中的贡献为1/(c+rank)，再把贡献相加。实现却可能在三个地方出问题：重复文档重复加分、Top-K等于零访问空堆、同分结果随map遍历顺序变化。

Elasticsearch文档把rank_constant默认值列为60，并定义rank_window_size。默认值不是所有数据的最优值。尤其不要把返回数量k和公式里的平滑常数c用同一个变量名。

概念示意图，非控制台截图、跑分或真实事件记录。

先用全排序写一个可检查的版本

这里定义每路结果是唯一文档排名：重复ID先保留首次出现，再重新连续编号。另一种协议可以保留原始名次，但必须说明，不能由循环偶然决定。为避免浮点同分比较受加法次序干扰，教学实现使用Fraction精确有理数。

from collections import defaultdict from fractions import Fraction def fuse(channels, top_k, rank_constant=60): if top_k < 0 or rank_constant < 1: raise ValueError("invalid top_k/rank_constant") if top_k == 0: return [] scores = defaultdict(Fraction) for hits in channels: unique = list(dict.fromkeys(hits)) for rank, doc_id in enumerate(unique, 1): scores[doc_id] += Fraction(1, rank_constant + rank) return sorted(scores.items(), key=lambda pair: (-pair[1], pair[0]))[:top_k] assert fuse([["A"]], 0) == [] assert fuse([["A","A","B"]], 2) == fuse([["A","B"]], 2) assert [x[0] for x in fuse([["A","B"],["B","A"]], 2)] == ["A","B"] assert fuse([], 10) == [] assert fuse([[],["A"]], 1)[0][0] == "A" try: fuse([["A"]], -1) except ValueError: pass else: raise AssertionError("negative top_k accepted") print("PASS zero/negative K, deduplication, ties, empty channels")

这个版本的用途是参考答案，不是极限吞吐实现。D个候选全排序约O(D log D)，Fraction也有额外开销。以后换成浮点、堆和对象池时，先让新实现与它在约定容差和排序协议下对齐，再谈分配次数。

一条重复ID到底算几票

若一路返回A、A、B，不去重就会让同一路给A投两票。它不等于A被两个独立检索器召回。上面的定义将它变成A、B；若后端按chunk返回，两个chunk是否归并到同一document也要先定义。

同分排序同样是产品契约。例子中两路是A、B与B、A，贡献完全对称。按文档ID升序打破平局，才能让重复请求、分页和回归测试得到稳定顺序。生产ID未必是字符串；比较规则要在各语言中一致。

候选窗口之外，融合器没有魔法

RRF只能给已经召回的候选排序。某文档在各路都被窗口截掉，后面调整c也找不回来。官方文档还要求关注分页时固定rank_window_size，否则可见候选集合变化会改变排序。

本地运行已覆盖零/负Top-K、去重、对称同分、空通道和单路结果。没有接入搜索后端，没有测NDCG或网络时延。封面沿用的k=60只是示意参数，图上的机制图也不代表一次相关性评测；内图展示本次实际核对的三个边界。比起“高性能融合引擎”，先做到同一输入给同一答案，比较好交差。

本文的本地检查仅覆盖明确列出的合成示例；外部研究与产品文档不作为本人性能实测。

复现命令：将完整Python块保存为 rrf_demo.py，运行 python3 rrf_demo.py；本轮使用Python 3.9.6。