Merge Ranked Lists

1"""merge-ranked-lists — generic Reciprocal Rank Fusion of two ranked lists.23PURE python (no gais calls, no network): fuses two ranked lists of dicts into4one by RRF — ``score(item) = sum over the lists containing it of51/(k + rank)`` with 1-based ranks — so items surfacing in BOTH lists rank6highest. The fusion step of the unified-retrieval flow (Ontology Search7results × Ontology From Documents entities), but deliberately generic:8any two ranked lists sharing an id field fuse.910Semantics (pinned by tests):11- Items that are not dicts or lack ``idKey`` are SKIPPED with a counted12  stderr warning (the ``skipped`` output), never a crash. Ranks are assigned13  1-based over the VALID items of each list.14- Ids are compared as strings (an int 7 and a string "7" merge).15- Dedup keeps the FIRST-seen payload (A before B), merged with a ``sources``16  annotation like ``[{"list": "A", "rank": 3}]`` and the ``fusedScore``.17  A within-list duplicate id keeps its first (best) rank but still consumes18  a rank slot.19- Exact score ties are stable: A-rank ascending, then B-rank ascending20  (items absent from a list sort after those present).21- Both lists empty / one empty degrade cleanly (the survivor's order wins).22- ``anchorFirst`` (1.4.1, default OFF) moves the entity-intent anchor to the23  head of the ranking for entity-lookup callers, by the same two tiers topId24  uses (lexical list-A hit, else list-A rank 1). OFF reproduces 1.3.0 exactly.25"""2627from __future__ import annotations2829import json30import sys31import traceback32from typing import Any3334MAX_LIMIT = 100035# Sort sentinel for "not present in this list" — after any real rank.36_ABSENT = float("inf")373839def _parse_list(raw: Any, field: str) -> list:40    """Lenient list parse: None/'' -> []; JSON string -> list; list ok."""41    if raw is None or raw == "":42        return []43    if isinstance(raw, str):44        try:45            raw = json.loads(raw)46        except json.JSONDecodeError as e:47            raise ValueError(f"'{field}' is not valid JSON: {e}") from e48    if not isinstance(raw, (list, tuple)):49        raise ValueError(f"'{field}' must be a JSON array")50    return list(raw)515253def _select_anchor(entries: list[dict[str, Any]]) -> dict[str, Any] | None:54    """The entity-intent anchor, by the SAME two tiers topId has used since 1.3.0.5556    Why an anchor is needed at all: RRF scores an item by SUM over the lists57    containing it, so an item that is mediocre in both lists outranks one that is58    FIRST in the authoritative list and absent from the other —59    1/(k+5) + 1/(k+1) > 1/(k+1). Correct when both lists rank the same kind of60    thing, wrong for entity lookup, where list A (entity search) is authoritative61    and list B is a thematic bridge. Measured on the E1 bank: fused rank 1 was a62    double-lister on 22/30 questions, demoting list A's own rank-1 entity on 9/30.6364    Tier 1 — a LEXICAL list-A hit: the query terms are in the entity's name, so65    it is the named entity rather than a document that merely mentions it.66    Tier 2 — list-A rank 1: used when nothing is tagged lexical. This is the67    common case in practice, because `lexical` is computed by ontology-search as68    "every query token appears in the name", which no full-sentence question ever69    satisfies. A caller that passes a raw question therefore gets tier 2.70    """71    for entry in entries:72        payload = entry["payload"]73        if entry["rankA"] is not _ABSENT and isinstance(payload, dict) \74                and payload.get("lexical") is True:75            return entry76    for entry in entries:77        if entry["rankA"] == 1:78            return entry79    return None808182def process(inputs: dict[str, Any]) -> dict[str, Any]:83    list_a = _parse_list(inputs.get("listA"), "listA")84    list_b = _parse_list(inputs.get("listB"), "listB")8586    raw_id_key = inputs.get("idKey")87    id_key = (88        raw_id_key.strip()89        if isinstance(raw_id_key, str) and raw_id_key.strip()90        else "id"91    )9293    try:94        k = float(inputs.get("k") if inputs.get("k") is not None else 60)95    except (TypeError, ValueError):96        k = 60.097    k = max(0.0, k)9899    try:100        limit = int(inputs.get("limit") or 25)101    except (TypeError, ValueError):102        limit = 25103    limit = max(1, min(limit, MAX_LIMIT))104105    # anchorFirst (1.4.1): OFF by default — plain RRF order, byte-identical to106    # 1.3.0 for every existing consumer. See _select_anchor for why it exists.107    anchor_first = inputs.get("anchorFirst") is True108109    skipped = 0110    # merge key -> {"payload", "score", "sources", "rankA", "rankB"}111    entries: dict[str, dict[str, Any]] = {}112113    for label, items in (("A", list_a), ("B", list_b)):114        rank = 0115        for item in items:116            if not isinstance(item, dict) or item.get(id_key) in (None, ""):117                skipped += 1118                print(119                    f"[merge-ranked-lists] skipping list{label} item without "120                    f"{id_key!r}: {str(item)[:120]}",121                    file=sys.stderr,122                )123                continue124            rank += 1125            key = str(item[id_key])126            entry = entries.get(key)127            if entry is None:128                entry = {129                    "payload": item,130                    "score": 0.0,131                    "sources": [],132                    "rankA": _ABSENT,133                    "rankB": _ABSENT,134                }135                entries[key] = entry136            rank_field = "rankA" if label == "A" else "rankB"137            if entry[rank_field] is not _ABSENT:138                # Within-list duplicate id: first (best) rank already counted.139                continue140            entry[rank_field] = rank141            entry["score"] += 1.0 / (k + rank)142            entry["sources"].append({"list": label, "rank": rank})143144    # Exact ties resolve by A-rank then B-rank; insertion order (A first)145    # backs even that up via sort stability.146    ranked = sorted(147        entries.values(),148        key=lambda entry: (-entry["score"], entry["rankA"], entry["rankB"]),149    )150151    # Entity-intent reorder: move the anchor to the head, leaving every other152    # item at its plain-RRF position. Deliberately a single-item promotion, not153    # a re-sort — it fixes the measured defect (the named entity losing rank 1154    # to a double-lister) without inventing a second ranking policy. Two155    # consequences worth relying on: fused[0] becomes exactly the topId anchor156    # computed below, so the two rules collapse into one; and the promotion runs157    # BEFORE the limit slice, so an anchor sitting past `limit` on fused score is158    # no longer truncated away. A no-op when list A is empty.159    anchor = _select_anchor(ranked) if anchor_first else None160    if anchor is not None:161        ranked.remove(anchor)162        ranked.insert(0, anchor)163164    fused = []165    for entry in ranked[:limit]:166        item = dict(entry["payload"])167        item["fusedScore"] = entry["score"]168        item["sources"] = entry["sources"]169        fused.append(item)170171    if fused:172        # Only annotate when an anchor was actually promoted, so that "flag on173        # but nothing to anchor" stays a true no-op end to end.174        anchor_note = "; entity-anchored" if anchor is not None else ""175        lines = [176            f"Fused ranking (RRF k={k:g}{anchor_note}; "177            f"top {len(fused)} of {len(ranked)} candidates):"178        ]179        for position, item in enumerate(fused, 1):180            display = next(181                (182                    item[field]183                    for field in ("displayName", "name", "title")184                    if isinstance(item.get(field), str) and item[field].strip()185                ),186                str(item.get(id_key)),187            )188            type_name = item.get("typeName")189            type_suffix = f" ({type_name})" if isinstance(type_name, str) and type_name else ""190            source_note = ", ".join(191                f"{source['list']}#{source['rank']}" for source in item["sources"]192            )193            lines.append(194                f"{position}. {display}{type_suffix} — score "195                f"{item['fusedScore']:.4f} [{source_note}]"196            )197        items_text = "\n".join(lines)198    else:199        items_text = ""200201    print(202        f"[merge-ranked-lists] listA={len(list_a)} listB={len(list_b)} "203        f"fused={len(fused)} skipped={skipped}",204        file=sys.stderr,205    )206207    # topId (1.3.0): the ENTITY-INTENT anchor for ontology-entity-360 — the208    # entity the question NAMES. Anchor to the highest-ranked FUSED item that was209    # a LEXICAL list-A hit (`lexical: true`, set by ontology-search when the query210    # terms are in the entity's name/keywords). This skips vector-only mentions:211    # B-prime's profile_vector embeds description_en, so a doc that merely MENTIONS212    # the entity (a certification "issued at the Ministry of Foreign Affairs") can213    # rank A#1 without being the entity — and top=list-A-rank-1 then profiled the214    # wrong object. Fallbacks preserve the 1.2.0 behavior: list-A rank 1 (when no215    # lexical flag is present — older search node), then fused rank 1 (list A216    # empty). RRF still rewards double-listers, which is why we do NOT fall217    # straight to fused rank 1: a substring cousin at A#4+B#14 could outscore the218    # true entity (observed live: Ufficio di controllo beat MAECI on "chi lavora219    # presso il MAECI"). The fused ORDER for items/itemsText is unchanged.220    top = None221    for item in fused:222        if item.get("lexical") is True and any(223            source.get("list") == "A" for source in item["sources"]224        ):225            top = item226            break227    if top is None:  # no lexical A-hit (or pre-1.3.0 search node) -> old anchor228        for item in fused:229            if any(s.get("list") == "A" and s.get("rank") == 1 for s in item["sources"]):230                top = item231                break232    if top is None and fused:233        top = fused[0]234    top_id = top.get(id_key) if isinstance(top, dict) else None235    return {236        "items": fused,237        "itemsText": items_text,238        "skipped": skipped,239        "topId": top_id if isinstance(top_id, str) else "",240    }241242243def main() -> None:244    try:245        envelope = json.loads(sys.stdin.read() or "{}")246        if not isinstance(envelope, dict):247            envelope = {}248        inputs = envelope.get("inputs", {}) or {}249        json.dump(process(inputs), sys.stdout)250    except Exception as e:251        print(252            json.dumps({253                "error": str(e),254                "errorType": type(e).__name__,255                "traceback": traceback.format_exc(),256            }),257            file=sys.stderr,258        )259        sys.exit(1)260261262if __name__ == "__main__":263    main()