Upload score_baseline.py with huggingface_hub

score_baseline.py

@@ -0,0 +1,164 @@
+"""Offline scorer for eval_baseline.py dumps.
+
+Reads a dump JSON (from eval_baseline.py) and computes metrics under
+multiple slot-matching strategies without re-running inference.
+
+Strategies:
+  exact:      pred dict == gold dict (case-sensitive)
+  ci:         case-insensitive keys/values
+  subset:     every gold key+value present in pred (tolerates extra pred slots)
+  ci-subset:  case-insensitive subset (the V32-baseline-equivalent semantic)
+  intent_snap: also snap pred intent to nearest 50-enum (post-hoc constraint)
+
+Usage:
+  uv run python poc/llm-finetune/training/score_baseline.py \\
+      poc/llm-finetune/training/eval_dump_q3b_adapter_v7_r16.json \\
+      --strategy ci-subset --by intent
+
+  uv run python poc/llm-finetune/training/score_baseline.py <dump> --strategy all
+"""
+from __future__ import annotations
+import argparse
+import json
+import re
+from collections import defaultdict
+from difflib import get_close_matches
+from pathlib import Path
+
+INTENTS_50 = sorted(json.loads(Path("/Users/jean-patricksmith/digital/kingly/apps/production/naac/poc/deberta_intent/checkpoints-base/label_mapping.json").read_text())["intent2id"].keys())
+INTENT_SET = set(INTENTS_50)
+
+
+def normalize(d: dict, ci: bool) -> dict:
+    if not isinstance(d, dict):
+        return {}
+    if ci:
+        return {str(k).upper(): str(v).strip().lower() for k, v in d.items() if v is not None and str(v) != ""}
+    return {str(k): str(v) for k, v in d.items() if v is not None}
+
+
+def slot_match(pred_slots: dict, gold_slots: dict, strategy: str) -> bool:
+    if strategy == "exact":
+        return pred_slots == gold_slots
+    if strategy == "ci":
+        return normalize(pred_slots, ci=True) == normalize(gold_slots, ci=True)
+    if strategy == "subset":
+        return all(pred_slots.get(k) == v for k, v in gold_slots.items())
+    if strategy == "ci-subset":
+        p = normalize(pred_slots, ci=True)
+        g = normalize(gold_slots, ci=True)
+        return all(p.get(k) == v for k, v in g.items())
+    raise ValueError(f"unknown strategy: {strategy}")
+
+
+def snap_intent(intent_str: str | None) -> str | None:
+    if not intent_str or intent_str in INTENT_SET:
+        return intent_str
+    matches = get_close_matches(intent_str, INTENTS_50, n=1, cutoff=0.4)
+    return matches[0] if matches else intent_str
+
+
+def score(rows: list[dict], strategy: str, intent_snap: bool = False) -> dict:
+    schema_ok = intent_em = slots_em = 0
+    invented_intents = 0
+    by_intent = defaultdict(lambda: [0, 0, 0])
+    by_scenario = defaultdict(lambda: [0, 0, 0])
+    failures = []
+
+    for r in rows:
+        gold_intent = r["gold_intent"]
+        gold_params = r["gold_parameters"] or {}
+        parsed = r["parsed"]
+
+        if parsed:
+            schema_ok += 1
+
+        pred_intent = (parsed or {}).get("intent")
+        if pred_intent and pred_intent not in INTENT_SET:
+            invented_intents += 1
+            if intent_snap:
+                pred_intent = snap_intent(pred_intent)
+
+        is_intent = bool(parsed and pred_intent == gold_intent)
+        is_slots = bool(parsed and slot_match((parsed or {}).get("slots") or {}, gold_params, strategy))
+
+        intent_em += int(is_intent); slots_em += int(is_slots)
+        by_intent[gold_intent][0] += 1
+        by_intent[gold_intent][1] += int(is_intent)
+        by_intent[gold_intent][2] += int(is_slots)
+        by_scenario[r["scenario_id"]][0] += 1
+        by_scenario[r["scenario_id"]][1] += int(is_intent)
+        by_scenario[r["scenario_id"]][2] += int(is_slots)
+
+        if not (is_intent and is_slots) and len(failures) < 20:
+            failures.append({
+                "scenario": r["scenario_id"],
+                "turn": r["turn_index"],
+                "text": r["text"][:120],
+                "gold": {"intent": gold_intent, "slots": gold_params},
+                "pred": parsed,
+                "intent_ok": is_intent, "slots_ok": is_slots,
+            })
+
+    n = len(rows)
+    return {
+        "n": n, "strategy": strategy, "intent_snap": intent_snap,
+        "schema_valid": schema_ok, "intent_em": intent_em, "slots_em": slots_em,
+        "schema_pct": schema_ok / max(n, 1), "intent_pct": intent_em / max(n, 1), "slots_pct": slots_em / max(n, 1),
+        "invented_intents": invented_intents,
+        "by_intent": dict(by_intent), "by_scenario": dict(by_scenario),
+        "failures": failures,
+    }
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("dump", help="JSON dump from eval_baseline.py")
+    p.add_argument("--strategy", choices=["exact", "ci", "subset", "ci-subset", "all"], default="ci-subset")
+    p.add_argument("--snap-intent", action="store_true", help="post-hoc snap pred intent to nearest 50-enum")
+    p.add_argument("--by", choices=["intent", "scenario", "both", "none"], default="intent")
+    p.add_argument("--top", type=int, default=20)
+    p.add_argument("--show-fails", action="store_true")
+    p.add_argument("--out", default=None)
+    args = p.parse_args()
+
+    dump = json.loads(Path(args.dump).read_text())
+    rows = dump["rows"]
+    print(f"loaded {len(rows)} rows from {args.dump}")
+    print(f"adapter: {dump.get('adapter')}\n")
+
+    strategies = ["exact", "ci", "subset", "ci-subset"] if args.strategy == "all" else [args.strategy]
+    results = []
+    for strat in strategies:
+        r = score(rows, strat, intent_snap=args.snap_intent)
+        results.append(r)
+        print(f"=== strategy={strat}{'+snap' if args.snap_intent else ''} ===")
+        print(f"  schema_valid: {r['schema_valid']}/{r['n']} = {r['schema_pct']:.1%}")
+        print(f"  intent_em:    {r['intent_em']}/{r['n']} = {r['intent_pct']:.1%}    (V32 baseline: 87.4%)")
+        print(f"  slots_em:     {r['slots_em']}/{r['n']} = {r['slots_pct']:.1%}    (V32 baseline: 80.6%)")
+        print(f"  invented_intents: {r['invented_intents']}")
+
+        if args.by in ("intent", "both"):
+            print(f"  --- per-intent (top {args.top}) ---")
+            for intent, (n_i, i_em, s_em) in sorted(r["by_intent"].items(), key=lambda kv: -kv[1][0])[:args.top]:
+                print(f"    {intent:30s} n={n_i:4d}  intent={i_em}/{n_i} ({i_em/n_i:.0%})  slots={s_em}/{n_i} ({s_em/n_i:.0%})")
+        if args.by in ("scenario", "both"):
+            print(f"  --- per-scenario (top {args.top}) ---")
+            for scn, (n_s, i_em, s_em) in sorted(r["by_scenario"].items(), key=lambda kv: -kv[1][0])[:args.top]:
+                print(f"    {scn:30s} n={n_s:4d}  intent={i_em}/{n_s} ({i_em/n_s:.0%})  slots={s_em}/{n_s} ({s_em/n_s:.0%})")
+        if args.show_fails:
+            print(f"  --- first {len(r['failures'])} failures ---")
+            for f in r["failures"]:
+                print(f"    [{f['scenario']}/{f['turn']}] intent_ok={f['intent_ok']} slots_ok={f['slots_ok']}")
+                print(f"      text: {f['text']}")
+                print(f"      gold: {f['gold']}")
+                print(f"      pred: {f['pred']}")
+        print()
+
+    if args.out:
+        Path(args.out).write_text(json.dumps({"adapter": dump.get("adapter"), "results": results}, indent=2))
+        print(f"saved scoring report to {args.out}")
+
+
+if __name__ == "__main__":
+    main()