memory.ranking

memory.ranking

Rank-fusion helpers shared across search surfaces.

Pure ranking math with no storage or embedding dependencies, so both the memory and zettelkasten servers can import it.

rrf_fuse

rrf_fuse(rankings: list[list[str]], k: int | None = None, weights: list[float] | None = None) -> dict[str, float]

Fuse several ranked id-lists into one score map via Reciprocal Rank Fusion.

Each entry in rankings is a list of ids already ordered best-first (rank 1 is the top hit). An id's fused score is the sum over the lists it appears in of weight * 1 / (k + rank), with rank starting at 1. Ids appearing in multiple lists naturally rise, without the input score scales needing to be comparable.

Parameters:

Name	Type	Description	Default
rankings	list[list[str]]	One id-list per retrieval channel, each best-first.	required
k	int | None	Damping constant (default 60, or ANGELO_RRF_K). Larger k flattens the contribution of top ranks.	None
weights	list[float] | None	Optional per-channel multipliers, one per entry in rankings. None (the default) weights every channel equally at 1.0 and reproduces the classic RRF behaviour exactly. When supplied it MUST have the same length as rankings (a mismatch raises ValueError). A channel with weight 0.0 contributes nothing — identical to omitting that channel entirely — which lets a caller switch a signal off without restructuring the call.	None

Returns:

Type	Description
dict[str, float]	Mapping of id -> fused score. Sort descending to get the fused ranking.

Source code in memory/ranking.py

def rrf_fuse(
    rankings: list[list[str]],
    k: int | None = None,
    weights: list[float] | None = None,
) -> dict[str, float]:
    """Fuse several ranked id-lists into one score map via Reciprocal Rank Fusion.

    Each entry in ``rankings`` is a list of ids already ordered best-first (rank
    1 is the top hit). An id's fused score is the sum over the lists it appears
    in of ``weight * 1 / (k + rank)``, with ``rank`` starting at 1. Ids appearing
    in multiple lists naturally rise, without the input score scales needing to be
    comparable.

    Args:
        rankings: One id-list per retrieval channel, each best-first.
        k: Damping constant (default 60, or ``ANGELO_RRF_K``). Larger k flattens
            the contribution of top ranks.
        weights: Optional per-channel multipliers, one per entry in ``rankings``.
            ``None`` (the default) weights every channel equally at ``1.0`` and
            reproduces the classic RRF behaviour exactly. When supplied it MUST
            have the same length as ``rankings`` (a mismatch raises ``ValueError``).
            A channel with weight ``0.0`` contributes nothing — identical to
            omitting that channel entirely — which lets a caller switch a signal
            off without restructuring the call.

    Returns:
        Mapping of id -> fused score. Sort descending to get the fused ranking.
    """
    if k is None:
        k = _rrf_k()
    if weights is not None and len(weights) != len(rankings):
        raise ValueError(
            f"weights length ({len(weights)}) must match rankings length "
            f"({len(rankings)})"
        )
    scores: dict[str, float] = {}
    for idx, ranking in enumerate(rankings):
        weight = 1.0 if weights is None else weights[idx]
        # A zero-weight channel contributes nothing, so skipping it entirely is
        # equivalent to (and cheaper than) multiplying every term by 0 — and it
        # keeps ids that appear ONLY in a switched-off channel out of the map.
        if weight == 0.0:
            continue
        for rank, item_id in enumerate(ranking, start=1):
            if not item_id:
                continue
            scores[item_id] = scores.get(item_id, 0.0) + weight * (1.0 / (k + rank))
    return scores

rrf_order

rrf_order(rankings: list[list[str]], k: int | None = None, weights: list[float] | None = None) -> list[str]

Return ids fused by :func:rrf_fuse, ordered best-first.

Ties (equal fused score) are broken by best rank achieved in any single channel, then by id, so the order is deterministic across runs. weights is threaded through to :func:rrf_fuse (see there); a switched-off (weight 0.0) channel contributes nothing to the tie-break either, staying consistent with the fused scores.

Source code in memory/ranking.py

def rrf_order(
    rankings: list[list[str]],
    k: int | None = None,
    weights: list[float] | None = None,
) -> list[str]:
    """Return ids fused by :func:`rrf_fuse`, ordered best-first.

    Ties (equal fused score) are broken by best rank achieved in any single
    channel, then by id, so the order is deterministic across runs. ``weights``
    is threaded through to :func:`rrf_fuse` (see there); a switched-off (weight
    ``0.0``) channel contributes nothing to the tie-break either, staying
    consistent with the fused scores.
    """
    scores = rrf_fuse(rankings, k=k, weights=weights)
    best_rank: dict[str, int] = {}
    for idx, ranking in enumerate(rankings):
        if weights is not None and weights[idx] == 0.0:
            continue
        for rank, item_id in enumerate(ranking, start=1):
            if item_id and rank < best_rank.get(item_id, 1 << 30):
                best_rank[item_id] = rank
    return sorted(
        scores,
        key=lambda i: (-scores[i], best_rank.get(i, 1 << 30), i),
    )