File size: 1,723 Bytes
d61821a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 | """Deterministic reciprocal-rank fusion for repository evidence."""
from __future__ import annotations
from dataclasses import replace
from typing import Sequence
from .components import Candidate
def unique_files(candidates: Sequence[Candidate]) -> tuple[Candidate, ...]:
seen: set[str] = set()
result: list[Candidate] = []
for candidate in candidates:
if candidate.path not in seen:
seen.add(candidate.path)
result.append(candidate)
return tuple(result)
def reciprocal_rank_fusion(
rankings: Sequence[Sequence[Candidate]],
limit: int,
k: int = 60,
) -> tuple[Candidate, ...]:
if len(rankings) < 2:
raise ValueError("RRF requires at least two rankings")
scores: dict[str, float] = {}
representatives: dict[str, Candidate] = {}
sources: dict[str, list[str]] = {}
for ranking in rankings:
for rank, candidate in enumerate(unique_files(tuple(ranking)), start=1):
scores[candidate.path] = scores.get(candidate.path, 0.0) + 1.0 / (k + rank)
sources.setdefault(candidate.path, []).append(candidate.source)
if candidate.path not in representatives:
representatives[candidate.path] = candidate
ordered = sorted(scores, key=lambda path: (-scores[path], path))[:limit]
return tuple(
replace(
representatives[path],
source="rrf:" + "+".join(sorted(set(sources[path]))),
score=scores[path],
metadata={
**(representatives[path].metadata or {}),
"fused_sources": tuple(sorted(set(sources[path]))),
"rrf_k": k,
},
)
for path in ordered
)
|