Upload mine_hard_negs.py with huggingface_hub

mine_hard_negs.py

@@ -0,0 +1,82 @@
+"""Mine hard negatives from a V9+ eval dump for next-iteration training.
+
+For each failure (intent_em or slots_em wrong on canonical 253):
+  - Output the row with corrected gold (from canonical scenario)
+  - Tag with failure type: wrong_intent | missing_slot | extra_slot | wrong_segment_count
+  - Optionally synthesize a near-duplicate by light text mutation
+
+Output: poc/llm-finetune/training/data_canonical_v2/hard_negatives.jsonl
+"""
+from __future__ import annotations
+import argparse
+import json
+import re
+from collections import Counter
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[3]
+
+
+def diff_segments(pred_segs: list, gold_segs: list) -> dict:
+    out = {"seg_count_match": len(pred_segs) == len(gold_segs)}
+    out["pred_n"] = len(pred_segs)
+    out["gold_n"] = len(gold_segs)
+    pred_intents = [s.get("intent") for s in pred_segs]
+    gold_intents = [s.get("intent") for s in gold_segs]
+    out["intent_match_ordered"] = pred_intents == gold_intents
+    out["intent_overlap"] = list(Counter(gold_intents) & Counter(pred_intents))
+    return out
+
+
+def main():
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("dump")
+    p.add_argument("--out", default="poc/llm-finetune/training/data_canonical_v2/hard_negatives.jsonl")
+    args = p.parse_args()
+
+    dump = json.loads(Path(args.dump).read_text())
+    rows = dump["rows"]
+
+    failures = []
+    type_counter = Counter()
+    for r in rows:
+        gold = r.get("gold") or {}
+        pred = r.get("parsed") or {}
+        gold_segs = gold.get("segments", [])
+        pred_segs = pred.get("segments") or []
+        diff = diff_segments(pred_segs, gold_segs)
+
+        if diff["intent_match_ordered"]:
+            continue  # not a failure on intent
+        # tag failure
+        if not diff["seg_count_match"]:
+            failure_type = f"seg_count_{diff['pred_n']}_vs_{diff['gold_n']}"
+        elif not diff["intent_overlap"]:
+            failure_type = "intent_disjoint"
+        else:
+            failure_type = "intent_order_or_partial_overlap"
+        type_counter[failure_type] += 1
+        failures.append({
+            "text": r["text"],
+            "compound": r.get("compound", False),
+            "gold_segments": gold_segs,
+            "pred_segments_for_audit": pred_segs,
+            "failure_type": failure_type,
+        })
+
+    print(f"loaded {len(rows)} rows; failures: {len(failures)}")
+    print("\nfailure type distribution:")
+    for ft, n in type_counter.most_common():
+        print(f"  {ft:35s} {n}")
+
+    out_path = Path(args.out)
+    out_path.parent.mkdir(parents=True, exist_ok=True)
+    with out_path.open("w") as f:
+        for fail in failures:
+            f.write(json.dumps(fail, ensure_ascii=False) + "\n")
+    print(f"\nwrote {len(failures)} hard negatives -> {out_path}")
+    print(f"\nnext: feed back into build_canonical_v1.py via mine_v9_failures() helper, retrain V10")
+
+
+if __name__ == "__main__":
+    main()