retrieval.consolidation

Merges per-retriever SearchHit rankings into a single deduplicated, explainable ranking — a starting ranking a following conversation/agent verifies and explores from. Confidence and score reflect cross-retriever agreement on query-text match, not currency or authority — a "high" hit can still point at legacy or deprecated code, and consumers must verify each span against the live source before relying on it.

Consolidate per-retriever SearchHit rankings into a single, deduplicated, explainable ranking — a starting ranking a following conversation/agent verifies and explores from.

Pure stdlib. Given {retriever_name: [SearchHit, ...]} (one ranked list per retriever already run against the same query), consolidate merges hits whose spans overlap (or are line-adjacent) into a single canonical ConsolidatedHit, fuses each group's per-retriever ranks with a (optionally weighted) Reciprocal Rank Fusion, and returns the fused list sorted best first — with provenance (which retrievers found it), agreement (how many), and a confidence label baked in so the handoff is self-explanatory without re-deriving any of this from the raw per-retriever output. Confidence and score reflect cross-retriever agreement on query-text match, not currency or authority — a "high" hit can still point at legacy or deprecated code, and consumers must verify each span against the live source before relying on it.

ConsolidatedHit dataclass

One deduplicated, ranked, explainable result in the consolidated list.

Source code in engine/retrieval/consolidation.py

@dataclass
class ConsolidatedHit:
    """One deduplicated, ranked, explainable result in the consolidated list."""

    source_path: str
    start_line: Optional[int]
    end_line: Optional[int]
    docid: str
    context: str
    score: float
    rank: int
    provenance: List[str] = field(default_factory=list)
    agreement: int = 0
    confidence: str = "low"
    contributors: List[dict] = field(default_factory=list)

consolidate(per_retriever_hits, *, k=60, weights=None, merge_adjacent=True)

Merge, fuse, and rank per-retriever SearchHit lists into a single deduplicated, explainable ConsolidatedHit list.

Parameters

per_retriever_hits : {retriever_name: [SearchHit, ...]}, one ranked list per retriever already run against the same query. Retrievers may be a subset of the six strategies (graceful skip) and may be supplied in any dict order — iteration is always in fixed canonical order internally, so the result is reproducible regardless of input dict order. k : RRF damping constant, forwarded to reciprocal_rank_fusion. weights : Optional {retriever_name: weight}; retrievers absent from this mapping default to weight 1.0. merge_adjacent : When True (default), same-path spans whose end/start lines are exactly adjacent (end + 1 == start) are merged into one group in addition to spans that overlap.

Returns

ConsolidatedHit list sorted by fused score descending (ties broken by source_path, start_line, end_line, docid), with rank assigned 1-based over that final order.

Source code in engine/retrieval/consolidation.py

def consolidate(
    per_retriever_hits: Dict[str, List[SearchHit]],
    *,
    k: int = 60,
    weights: Optional[Dict[str, float]] = None,
    merge_adjacent: bool = True,
) -> List[ConsolidatedHit]:
    """Merge, fuse, and rank per-retriever ``SearchHit`` lists into a single
    deduplicated, explainable ``ConsolidatedHit`` list.

    Parameters
    ----------
    per_retriever_hits :
        ``{retriever_name: [SearchHit, ...]}``, one ranked list per retriever
        already run against the same query. Retrievers may be a subset of
        the six strategies (graceful skip) and may be supplied in any dict
        order — iteration is always in fixed canonical order internally, so
        the result is reproducible regardless of input dict order.
    k :
        RRF damping constant, forwarded to ``reciprocal_rank_fusion``.
    weights :
        Optional ``{retriever_name: weight}``; retrievers absent from this
        mapping default to weight ``1.0``.
    merge_adjacent :
        When ``True`` (default), same-path spans whose end/start lines are
        exactly adjacent (``end + 1 == start``) are merged into one group in
        addition to spans that overlap.

    Returns
    -------
    ``ConsolidatedHit`` list sorted by fused score descending (ties broken by
    ``source_path``, ``start_line``, ``end_line``, ``docid``), with ``rank``
    assigned 1-based over that final order.
    """
    ordered_names = _ordered_retriever_names(per_retriever_hits.keys())
    all_hits: List[SearchHit] = []
    retriever_of: Dict[int, str] = {}
    for name in ordered_names:
        for hit in per_retriever_hits[name]:
            all_hits.append(hit)
            retriever_of[id(hit)] = name

    groups = _group_hits_by_span(all_hits, merge_adjacent)

    # Per-group, per-retriever best (lowest) single-arm rank -> group index.
    group_index_of_hit: Dict[int, int] = {}
    for group_index, group in enumerate(groups):
        for hit in group:
            group_index_of_hit[id(hit)] = group_index

    rankings, resolved_weights = _build_rankings(
        ordered_names, per_retriever_hits, group_index_of_hit, weights
    )
    fused = reciprocal_rank_fusion(rankings, k=k, weights=resolved_weights)

    rank_lookup = _rank_lookup_by_group(ordered_names, per_retriever_hits, group_index_of_hit)
    return _assemble_consolidated_hits(fused, groups, rank_lookup, ordered_names, retriever_of)