model-3/scripts/bench_models.py

"""A/B any embedding model on REAL data before committing hours to it.

    python scripts/bench_models.py --gpu 1 --models e5-small e5-large bge-m3

What this measures, exactly
---------------------------
This scores the BLOCKING/RETRIEVAL stage, not the final F0.5. That is deliberate:
blocking recall is the hard ceiling on F0.5 (you cannot match what you never
retrieve), and it is the only thing the embedding model influences. The final
F0.5 is produced by src/run_train_eval.py on a held-out split.

Data
  queries : --n-india / --n-us non-singleton TRAIN Source-1 entities, seed 0.
            Train only, because test has no labels. Countries are read from the
            data, never hard-coded.
  pool    : every Source-2/Source-3 record that is a TRUE match of a sampled query,
            PLUS `--negs` random same-country distractors per query. The negatives
            come from the same country so the test reflects the country-gated
            search the pipeline actually performs.
  truth   : dataset/train/train_ground_truth.tsv

Metric
  recall@k -- the share of true pairs whose match is in the query's top-k, overall
  and on the two slices that decide this problem: non-Latin matched names and
  empty matched addresses. Plus throughput (texts/s) and peak VRAM.

Honest limits
  * The pool is ~`negs`/query, not the full ~4M per country, so ABSOLUTE recall is
    optimistic. Use it to RANK models, not to predict production recall.
  * France is test-only and unlabelled, so no unseen-country signal here.
"""
from __future__ import annotations

import argparse
import json
import os
import sys
import time
import unicodedata as ud

import numpy as np
import pandas as pd

sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "..", "src"))
from config import EMB, PATHS, model_spec        # noqa: E402

KS = (10, 20, 50)
T0 = time.time()


def log(m):
    print(f"[{time.time() - T0:6.0f}s] {m}", flush=True)


def script_of(name):
    for c in name:
        if c.isalpha() and not ud.name(c, "").startswith("LATIN"):
            return "nonlatin"
    return "latin"


def build_sample(args):
    cols = ["entity_id", "business_name", "business_address", "country"]
    gt = pd.read_csv(PATHS["train"]["ground_truth"], sep="\t", dtype=str, keep_default_na=False)
    gt["ids"] = gt["matched_entity_ids"].str.split(",")
    gt = gt[gt["matched_entity_ids"] != ""]
    s1 = pd.read_parquet(PATHS["train"]["source1"], columns=cols)
    s1 = s1[s1.entity_id.isin(set(gt.source1_entity_id))]
    rng = np.random.default_rng(0)
    picks = []
    for c, n in args.per_country:
        sub = s1[s1.country == c]
        if len(sub) == 0:
            log(f"WARNING: no train records with country={c!r}; skipping")
            continue
        picks.append(sub.iloc[rng.choice(len(sub), min(n, len(sub)), replace=False)])
    Q = pd.concat(picks).reset_index(drop=True)
    truth = gt[gt.source1_entity_id.isin(set(Q.entity_id))]
    need = {m for l in truth.ids for m in l}
    log(f"queries {len(Q):,}  true pairs {sum(len(l) for l in truth.ids):,}")

    # full read of the pool: 10.3M rows x 4 short string cols fits comfortably
    P = pd.concat([pd.read_parquet(PATHS["train"]["source2"], columns=cols),
                   pd.read_parquet(PATHS["train"]["source3"], columns=cols)], ignore_index=True)
    pos = P[P.entity_id.isin(need)]
    negs = []
    for c in Q.country.unique():
        pc = P[(P.country == c) & (~P.entity_id.isin(need))]
        k = min(len(pc), args.negs * int((Q.country == c).sum()))
        negs.append(pc.iloc[rng.choice(len(pc), k, replace=False)])
    pool = pd.concat([pos] + negs, ignore_index=True).drop_duplicates("entity_id").reset_index(drop=True)
    log(f"pool {len(pool):,} ({len(pos):,} true matches + random same-country distractors)")
    return Q, pool, truth


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--gpu", default=os.environ.get("M3_GPU"))
    ap.add_argument("--models", nargs="+", default=["e5-small"])
    ap.add_argument("--n-india", type=int, default=2000)
    ap.add_argument("--n-us", type=int, default=1000)
    ap.add_argument("--negs", type=int, default=250, help="random distractors per query")
    ap.add_argument("--out", default=None)
    a = ap.parse_args()
    if a.gpu is None:
        sys.exit("pass --gpu 0 / --gpu 1 / --gpu cpu (or set M3_GPU)")
    dev = "cpu" if str(a.gpu) == "cpu" else f"cuda:{int(a.gpu)}"
    # country labels are data, not code: discover them, then map the two CLI sizes on
    a.per_country = []
    s1c = pd.read_parquet(PATHS["train"]["source1"], columns=["country"])["country"].value_counts()
    for i, c in enumerate(s1c.index.tolist()):
        a.per_country.append((c, a.n_india if i == 0 else a.n_us))
    log(f"sampling {a.per_country} (countries read from the data)")

    Q, pool, truth = build_sample(a)
    qtext = [f"{n} | {ad} | {c}" for n, ad, c in
             zip(Q.business_name, Q.business_address, Q.country)]
    ptext = [f"{n} | {ad} | {c}" for n, ad, c in
             zip(pool.business_name, pool.business_address, pool.country)]
    qpos = {e: i for i, e in enumerate(Q.entity_id)}
    ppos = {e: i for i, e in enumerate(pool.entity_id)}
    pairs = [(qpos[s], ppos[m]) for s, l in zip(truth.source1_entity_id, truth.ids)
             for m in l if m in ppos]
    tq = np.array([x[0] for x in pairs]); tp = np.array([x[1] for x in pairs])
    pinfo = pool.set_index("entity_id")
    mids = [m for s, l in zip(truth.source1_entity_id, truth.ids) for m in l if m in ppos]
    nonlatin = np.array([script_of(pinfo.business_name[m]) == "nonlatin" for m in mids])
    addrempty = np.array([str(pinfo.business_address[m]).strip() == "" for m in mids])
    qc, pc = Q.country.to_numpy(), pool.country.to_numpy()

    from encoder import Encoder
    rows = {}
    for name in a.models:
        hf, qpre, dpre, _d, lic = model_spec(name)
        log(f"--- {name} ({hf}, {lic})")
        enc = Encoder(hf, dev, EMB["max_len"], EMB["batch"])
        enc.encode(ptext[:2000], dpre)                   # warm-up
        t = time.time()
        PE = enc.encode(ptext, dpre).astype(np.float32)
        rate = len(ptext) / (time.time() - t)
        QE = enc.encode(qtext, qpre).astype(np.float32)
        rank = np.full(len(tq), np.inf)
        for c in np.unique(qc):
            qi, pi = np.flatnonzero(qc == c), np.flatnonzero(pc == c)
            for s in range(0, len(qi), 512):
                qq = qi[s:s + 512]
                S = QE[qq] @ PE[pi].T
                kk = min(max(KS), S.shape[1])
                part = np.argpartition(-S, kk - 1, axis=1)[:, :kk]
                order = np.take_along_axis(S, part, 1).argsort(1)[:, ::-1]
                top = np.take_along_axis(part, order, 1)
                for r, q in enumerate(qq):
                    loc = {pi[j]: z + 1 for z, j in enumerate(top[r])}
                    for t_ in np.flatnonzero(tq == q):
                        rank[t_] = loc.get(tp[t_], np.inf)
        row = {"params": f"{sum(p.numel() for p in enc.model.parameters())/1e6:.0f}M",
               "dim": enc.dim, "texts/s": round(rate), "vram_gb": round(enc.peak_vram_gb(), 2)}
        for k in KS:
            h = rank <= k
            row[f"R@{k}"] = round(float(h.mean()), 4)
            row[f"R@{k}_nonLatin"] = round(float(h[nonlatin].mean()), 4) if nonlatin.any() else None
            row[f"R@{k}_addrEmpty"] = round(float(h[addrempty].mean()), 4) if addrempty.any() else None
        rows[name] = row
        log(f"{name}: {json.dumps(row)}")
        del enc, PE, QE
        import gc; gc.collect()
        if dev != "cpu":
            import torch; torch.cuda.empty_cache()

    df = pd.DataFrame(rows).T
    print("\n" + df.to_string())
    out = a.out or os.path.join(PATHS["work_dir"], "bench_models.tsv")
    os.makedirs(os.path.dirname(out), exist_ok=True)
    df.to_csv(out, sep="\t")
    log(f"wrote {out}")


if __name__ == "__main__":
    main()