scripts/eval_quality.py

"""Embedding quality eval: cosine drift and nfcorpus retrieval, fp32 vs INT8.

Speed benches in this repo already show dynamic INT8 is slower on this hardware.
This script answers the missing question: does the quantized graph still retrieve?

Two measurements, same encode path as production (`model.encode` via app.model.load_model):

1. Cosine drift of each backend vs PyTorch CPU (short queries and long corpus docs).
2. BeIR/nfcorpus test retrieval: nDCG / recall / MRR at 10 and 100, plus top-k rank
   agreement vs PyTorch (does INT8 preserve nearest neighbors, not just vector values?).

Usage:
    uv sync --extra bench
    uv run python scripts/export_onnx.py          # once
    uv run python scripts/quantize_onnx.py        # once, for INT8
    uv run python scripts/eval_quality.py
    uv run python scripts/eval_quality.py --qa    # also fail if drift/nDCG gates trip

Writes benchmarks/quality-nfcorpus.json by default.
"""

from __future__ import annotations

import argparse
import sys
import time
from pathlib import Path
from typing import Any

import numpy as np

REPO_ROOT = Path(__file__).resolve().parent.parent
if str(REPO_ROOT) not in sys.path:
    sys.path.insert(0, str(REPO_ROOT))
SCRIPTS_DIR = Path(__file__).resolve().parent
if str(SCRIPTS_DIR) not in sys.path:
    sys.path.insert(0, str(SCRIPTS_DIR))

from quality_metrics import (  # noqa: E402
    cosine_drift_report,
    cosine_matrix,
    mean_rank_agreement,
    mean_retrieval_metrics,
    ranked_doc_ids,
)
from utils import base_metadata, write_json  # noqa: E402

from app.config import settings  # noqa: E402
from app.model import load_model  # noqa: E402

MAX_CHARS = settings.max_chars_per_text
DEFAULT_OUTPUT = REPO_ROOT / "benchmarks" / "quality-nfcorpus.json"
DEFAULT_KS = (10, 100)
DEFAULT_AGREEMENT_KS = (1, 10)

# QA gates from the committed nfcorpus run (benchmarks/quality-nfcorpus.json).
# onnx-fp32 must match pytorch geometrically. INT8 is allowed vector-space drift
# (measured mean cosine ~0.95) as long as nDCG@10 stays within 1.5 points.
DEFAULT_MIN_COSINE = {
    "onnx-fp32": 0.995,
    "onnx-int8": 0.94,
}
DEFAULT_MAX_NDCG10_DROP = {
    "onnx-fp32": 0.005,
    "onnx-int8": 0.015,
}


def _row_text(title: str | None, text: str | None) -> str:
    raw = " ".join(part for part in (title, text) if part)
    collapsed = " ".join(raw.split())
    return collapsed[:MAX_CHARS]


def load_nfcorpus(qrels_split: str) -> tuple[list[str], list[str], list[str], list[str], dict[str, dict[str, int]]]:
    """Return (query_ids, query_texts, doc_ids, doc_texts, qrels) for the BeIR test split."""
    from datasets import load_dataset

    corpus_ds = load_dataset("BeIR/nfcorpus", "corpus")["corpus"]
    queries_ds = load_dataset("BeIR/nfcorpus", "queries")["queries"]
    qrels_ds = load_dataset("BeIR/nfcorpus-qrels", split=qrels_split)

    qrels: dict[str, dict[str, int]] = {}
    for row in qrels_ds:
        query_id = str(row["query-id"])
        doc_id = str(row["corpus-id"])
        score = int(row["score"])
        if score <= 0:
            continue
        qrels.setdefault(query_id, {})[doc_id] = score

    query_map = {
        str(row["_id"]): _row_text(row.get("title"), row.get("text")) for row in queries_ds
    }
    doc_map = {
        str(row["_id"]): _row_text(row.get("title"), row.get("text")) for row in corpus_ds
    }

    query_ids = [qid for qid in qrels if qid in query_map and query_map[qid]]
    query_texts = [query_map[qid] for qid in query_ids]
    doc_ids = [did for did, text in doc_map.items() if text]
    doc_texts = [doc_map[did] for did in doc_ids]
    return query_ids, query_texts, doc_ids, doc_texts, qrels


def onnx_file_path(onnx_dir: Path, file_name: str) -> Path:
    return onnx_dir / file_name


def load_backend(name: str, onnx_dir: Path) -> Any | None:
    """Load a backend or return None if its ONNX artifact is missing."""
    if name == "pytorch":
        print(f"loading {name} ({settings.model_name}, cpu)")
        return load_model(settings.model_name, device="cpu", backend=None)
    if name == "onnx-fp32":
        file_name = "onnx/model_O3.onnx"
    elif name == "onnx-int8":
        file_name = "onnx/model_int8.onnx"
    else:
        raise ValueError(f"unknown backend {name}")
    path = onnx_file_path(onnx_dir, file_name)
    if not path.exists():
        print(f"skip {name}: {path} not found (export/quantize first)")
        return None
    print(f"loading {name} ({path})")
    return load_model(str(onnx_dir), device="cpu", backend="onnx", onnx_file_name=file_name)


def encode_texts(model: Any, texts: list[str], batch_size: int, label: str) -> np.ndarray:
    print(f"  encode {label}: {len(texts)} texts, batch_size={batch_size}")
    start = time.perf_counter()
    embeddings = model.encode(texts, batch_size=batch_size, show_progress_bar=True)
    elapsed = time.perf_counter() - start
    arr = np.asarray(embeddings, dtype=np.float32)
    print(f"  done {label} in {elapsed:.1f}s ({len(texts) / elapsed:.1f} seq/s)")
    return arr


def fmt(value: float, digits: int = 4) -> str:
    return f"{value:.{digits}f}"


def print_table(title: str, headers: list[str], rows: list[list[str]]) -> None:
    widths = [len(h) for h in headers]
    for row in rows:
        for i, cell in enumerate(row):
            widths[i] = max(widths[i], len(cell))
    print()
    print(title)
    line = " | ".join(h.ljust(widths[i]) for i, h in enumerate(headers))
    rule = "-|-".join("-" * widths[i] for i in range(len(headers)))
    print(line)
    print(rule)
    for row in rows:
        print(" | ".join(row[i].ljust(widths[i]) for i in range(len(headers))))


def evaluate_backend(
    query_emb: np.ndarray,
    doc_emb: np.ndarray,
    query_ids: list[str],
    doc_ids: list[str],
    qrels: dict[str, dict[str, int]],
    ks: tuple[int, ...],
) -> dict[str, Any]:
    scores = cosine_matrix(query_emb, doc_emb)
    rankings_list = ranked_doc_ids(scores, doc_ids)
    rankings = {qid: ranking for qid, ranking in zip(query_ids, rankings_list)}
    retrieval = mean_retrieval_metrics(rankings, qrels, ks=ks)
    return {"retrieval": retrieval, "rankings": rankings}


def qa_failures(
    report: dict[str, Any],
    min_cosine: dict[str, float],
    max_ndcg10_drop: dict[str, float],
) -> list[str]:
    """Return human-readable gate failures. Empty list means pass."""
    failures: list[str] = []
    backends = report["backends"]
    if "pytorch" not in backends:
        failures.append("QA gate requires pytorch as the reference backend")
        return failures
    ref_ndcg = backends["pytorch"]["retrieval"]["ndcg@10"]
    for name, backend in backends.items():
        if name == "pytorch":
            continue
        overall = backend["drift"]["overall"]
        mean_cos = overall["cosine_mean"]
        floor = min_cosine.get(name)
        if floor is not None and mean_cos < floor:
            failures.append(
                f"{name} mean cosine vs pytorch is {mean_cos:.4f} (floor {floor:.4f})"
            )
        ndcg = backend["retrieval"]["ndcg@10"]
        drop = ref_ndcg - ndcg
        cap = max_ndcg10_drop.get(name)
        if cap is not None and drop > cap:
            failures.append(
                f"{name} nDCG@10 drop vs pytorch is {drop:.4f} "
                f"({ref_ndcg:.4f} -> {ndcg:.4f}, cap {cap:.4f})"
            )
    return failures


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="fp32 vs INT8 embedding quality on nfcorpus")
    parser.add_argument(
        "--backends",
        default="pytorch,onnx-fp32,onnx-int8",
        help="Comma-separated backends to load.",
    )
    parser.add_argument(
        "--onnx-dir",
        type=Path,
        default=REPO_ROOT / settings.onnx_model_path,
        help="Local SentenceTransformers ONNX export directory.",
    )
    parser.add_argument("--qrels-split", default="test", help="BeIR qrels split (default: test).")
    parser.add_argument("--batch-size", type=int, default=32, help="Encode batch for fp32 backends.")
    parser.add_argument(
        "--int8-batch-size",
        type=int,
        default=8,
        help="Encode batch for onnx-int8 (smaller: avoids the large-batch INT8 slowdown).",
    )
    parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT)
    parser.add_argument(
        "--qa",
        action="store_true",
        help="Exit 1 if cosine/nDCG gates fail (use in a quality-check job).",
    )
    parser.add_argument(
        "--min-cosine-fp32",
        type=float,
        default=DEFAULT_MIN_COSINE["onnx-fp32"],
        help="QA: minimum mean cosine of onnx-fp32 vs pytorch.",
    )
    parser.add_argument(
        "--min-cosine-int8",
        type=float,
        default=DEFAULT_MIN_COSINE["onnx-int8"],
        help="QA: minimum mean cosine of onnx-int8 vs pytorch.",
    )
    parser.add_argument(
        "--max-ndcg10-drop-fp32",
        type=float,
        default=DEFAULT_MAX_NDCG10_DROP["onnx-fp32"],
    )
    parser.add_argument(
        "--max-ndcg10-drop-int8",
        type=float,
        default=DEFAULT_MAX_NDCG10_DROP["onnx-int8"],
    )
    return parser.parse_args()


def main() -> int:
    args = parse_args()
    wanted = [name.strip() for name in args.backends.split(",") if name.strip()]
    unknown = [name for name in wanted if name not in {"pytorch", "onnx-fp32", "onnx-int8"}]
    if unknown:
        raise SystemExit(f"unknown backends: {unknown}")

    print("loading BeIR/nfcorpus ...")
    query_ids, query_texts, doc_ids, doc_texts, qrels = load_nfcorpus(args.qrels_split)
    print(
        f"nfcorpus {args.qrels_split}: {len(query_ids)} queries, {len(doc_ids)} docs, "
        f"{sum(len(v) for v in qrels.values())} qrels"
    )

    onnx_dir = args.onnx_dir.resolve()
    models: dict[str, Any] = {}
    for name in wanted:
        model = load_backend(name, onnx_dir)
        if model is not None:
            models[name] = model
    if "pytorch" not in models:
        raise SystemExit("pytorch backend is required as the quality reference")
    if len(models) < 2:
        raise SystemExit("need at least one ONNX backend to compare against pytorch")

    embeddings: dict[str, dict[str, np.ndarray]] = {}
    encode_meta: dict[str, dict[str, Any]] = {}
    for name, model in models.items():
        batch_size = args.int8_batch_size if name == "onnx-int8" else args.batch_size
        print(f"\n== {name} ==")
        t0 = time.perf_counter()
        query_emb = encode_texts(model, query_texts, batch_size, "queries")
        doc_emb = encode_texts(model, doc_texts, batch_size, "corpus")
        encode_meta[name] = {
            "batch_size": batch_size,
            "encode_s": round(time.perf_counter() - t0, 2),
        }
        embeddings[name] = {"queries": query_emb, "corpus": doc_emb}

    ks = DEFAULT_KS
    per_backend: dict[str, Any] = {}
    ranking_store: dict[str, dict[str, list[str]]] = {}
    ref_q = embeddings["pytorch"]["queries"]
    ref_d = embeddings["pytorch"]["corpus"]

    for name, emb in embeddings.items():
        result = evaluate_backend(
            emb["queries"], emb["corpus"], query_ids, doc_ids, qrels, ks
        )
        ranking_store[name] = result["rankings"]
        drift: dict[str, Any] = {}
        if name != "pytorch":
            drift["queries"] = cosine_drift_report(ref_q, emb["queries"])
            drift["corpus"] = cosine_drift_report(ref_d, emb["corpus"])
            stacked_ref = np.vstack([ref_q, ref_d])
            stacked_cand = np.vstack([emb["queries"], emb["corpus"]])
            drift["overall"] = cosine_drift_report(stacked_ref, stacked_cand)
            drift["rank_agreement"] = mean_rank_agreement(
                ranking_store["pytorch"], ranking_store[name], ks=DEFAULT_AGREEMENT_KS
            )
        per_backend[name] = {
            "encode": encode_meta[name],
            "retrieval": result["retrieval"],
            "drift": drift,
        }

    report: dict[str, Any] = {
        **base_metadata(),
        "model": settings.model_name,
        "device": "cpu",
        "dataset": "BeIR/nfcorpus",
        "qrels_split": args.qrels_split,
        "n_queries": len(query_ids),
        "n_docs": len(doc_ids),
        "n_qrels": sum(len(v) for v in qrels.values()),
        "ks": list(ks),
        "reference": "pytorch",
        "backends": per_backend,
    }

    # Cosine drift table
    drift_rows: list[list[str]] = []
    for name, backend in per_backend.items():
        if name == "pytorch":
            continue
        overall = backend["drift"]["overall"]
        queries = backend["drift"]["queries"]
        corpus = backend["drift"]["corpus"]
        agree = backend["drift"]["rank_agreement"]
        drift_rows.append(
            [
                name,
                fmt(queries["cosine_mean"]),
                fmt(corpus["cosine_mean"]),
                fmt(overall["cosine_mean"]),
                fmt(overall["cosine_p05"]),
                fmt(overall["cosine_min"]),
                fmt(overall["angle_deg_mean"], 2),
                fmt(agree["top1_overlap"]),
                fmt(agree["top10_overlap"]),
            ]
        )
    print_table(
        "Cosine drift vs pytorch (higher cosine / overlap is better)",
        [
            "backend",
            "cos queries",
            "cos corpus",
            "cos overall",
            "cos p05",
            "cos min",
            "angle° mean",
            "top1 overlap",
            "top10 overlap",
        ],
        drift_rows,
    )

    retrieval_rows: list[list[str]] = []
    ref_ndcg = per_backend["pytorch"]["retrieval"]["ndcg@10"]
    for name, backend in per_backend.items():
        r = backend["retrieval"]
        delta = r["ndcg@10"] - ref_ndcg
        retrieval_rows.append(
            [
                name,
                fmt(r["ndcg@10"]),
                fmt(r["ndcg@100"]),
                fmt(r["recall@10"]),
                fmt(r["recall@100"]),
                fmt(r["mrr@10"]),
                f"{delta:+.4f}",
            ]
        )
    print_table(
        "nfcorpus test retrieval (cosine ranking, L2-normalized)",
        ["backend", "nDCG@10", "nDCG@100", "recall@10", "recall@100", "MRR@10", "Δ nDCG@10"],
        retrieval_rows,
    )

    write_json(report, args.output)
    print(f"\nWrote {args.output}")

    min_cosine = {
        "onnx-fp32": args.min_cosine_fp32,
        "onnx-int8": args.min_cosine_int8,
    }
    max_drop = {
        "onnx-fp32": args.max_ndcg10_drop_fp32,
        "onnx-int8": args.max_ndcg10_drop_int8,
    }
    failures = qa_failures(report, min_cosine, max_drop)
    if failures:
        print("\nQA gates:")
        for item in failures:
            print(f"  FAIL {item}")
        if args.qa:
            return 1
    else:
        print("\nQA gates: pass")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())