Reciprocal rank fusion

Now we have two ranked lists: one from BM25, one from FAISS. Their raw scores live on different scales and cannot be compared directly. Reciprocal rank fusion (RRF) sidesteps that problem by ignoring scores entirely and working only with positions.

The formula is tiny: for each chunk, sum 1 divided by (rank + k) across every list it appears in, where rank is the zero-indexed position. Chunks that rank well in both lists get the biggest boost. The k constant (typically 60) softens the drop-off between ranks so results below position 10 still contribute meaningfully.

How RRF combines two ranked lists

Each chunk gets a fused score from its positions in both rankings.

retrieval_fusion.py
python
def reciprocal_rank_fusion(
    keyword_results: list[dict],
    vector_results: list[dict],
    k: int = 60,
) -> list[dict]:
    """Merge results using reciprocal rank fusion."""
    keyword_ranks = {}
    vector_ranks = {}

    # Use (start_time, end_time) as a unique chunk id across both lists
    for rank, chunk in enumerate(keyword_results):
        chunk_id = (chunk.get("start_time"), chunk.get("end_time"))
        keyword_ranks[chunk_id] = (rank, chunk)

    for rank, chunk in enumerate(vector_results):
        chunk_id = (chunk.get("start_time"), chunk.get("end_time"))
        vector_ranks[chunk_id] = (rank, chunk)

    rrf_scores = {}
    all_chunks = {}

    for chunk_id, (rank, chunk) in keyword_ranks.items():
        rrf_scores[chunk_id] = 1.0 / (rank + k)
        all_chunks[chunk_id] = chunk

    for chunk_id, (rank, chunk) in vector_ranks.items():
        contribution = 1.0 / (rank + k)
        if chunk_id in rrf_scores:
            rrf_scores[chunk_id] += contribution
        else:
            rrf_scores[chunk_id] = contribution
            all_chunks[chunk_id] = chunk

    sorted_items = sorted(
        all_chunks.items(),
        key=lambda item: rrf_scores[item[0]],
        reverse=True,
    )
    return [dict(chunk, rrf_score=rrf_scores[cid])
            for cid, chunk in sorted_items]


def fuse_and_get_top_k(
    keyword_results: list[dict],
    vector_results: list[dict],
    top_k: int = 5,
    rrf_k: int = 60,
) -> list[dict]:
    """Fuse both rankings and return the top_k chunks."""
    fused = reciprocal_rank_fusion(keyword_results, vector_results, k=rrf_k)
    return fused[:top_k]

The (start_time, end_time) tuple is a clean chunk identifier because both lists came from the same chunk set. Chunks appearing in both rankings get contributions from both sides and float to the top. fuse_and_get_top_k is the small entry point the rest of the app calls: fuse everything, then slice to the top_k chunks.

Quiz: Quiz

Loading practice…