| """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 |
| ) |
|
|
