Upload build_canonical_v1.py with huggingface_hub

build_canonical_v1.py

@@ -0,0 +1,201 @@
+"""Build canonical-v1 corpus: multi-segment JSON schema + unknown class + hard negatives.
+
+Schema (always emit `segments` list, even for single-intent):
+    {"segments": [{"intent": "...", "slots": {...}, "text": "..."}], "abstain_reason": null}
+
+Sources:
+  - data/src/partner_graph/output/scenarios/*.json
+    - 442 single-intent ATC turns -> 1-segment list
+    - 76 compound turns with expected_segments[] -> N-segment list
+  - V8 failures from eval_dump_v8_canonical.json -> hard-negative gold-corrected
+  - Synthetic `unknown` adversarials (garbled / corrupted transcripts) -> abstain training
+
+Output: poc/llm-finetune/training/data_canonical_v1/{train,valid,test}.jsonl
+"""
+from __future__ import annotations
+import json
+import random
+import re
+from collections import defaultdict
+from pathlib import Path
+
+ROOT = Path("/Users/jean-patricksmith/digital/kingly/apps/production/naac")
+INTENTS_50 = sorted(json.loads((ROOT / "poc/deberta_intent/checkpoints-base/label_mapping.json").read_text())["intent2id"].keys())
+INTENTS_51 = INTENTS_50 + ["unknown"]
+SLOTS_LOWER = ["altimeter_setting", "altitude", "approach_type", "call_sign", "clock_position",
+               "direction", "distance", "facility", "fix", "frequency", "heading", "pattern_leg",
+               "route", "runway", "speed", "taxiway", "time", "transponder_code", "turn_direction",
+               "sequence"]
+
+SYSTEM_PROMPT = (
+    "You are an ATC parser. Parse the air traffic control transmission into a JSON object.\n\n"
+    "OUTPUT SCHEMA (always emit `segments` list, even single-intent turns):\n"
+    '{\n  "segments": [\n    {"intent": <one-of-enum>, "slots": {<lowercase_key>: <value>}, "text": <segment substring>}\n  ],\n  "abstain_reason": null | <short reason if unsure>\n}\n\n'
+    "Intent enum (51 values, includes `unknown` for ambiguous/garbled):\n"
+    + ", ".join(INTENTS_51) + "\n\n"
+    "Slot keys (lowercase only): " + ", ".join(SLOTS_LOWER) + "\n\n"
+    "Rules:\n"
+    "1. Compound transmissions get multiple segments — split by comma, period, or new clause.\n"
+    "2. Single-intent transmissions still emit one segment in the list.\n"
+    "3. Use `unknown` intent + `abstain_reason` when transcript is garbled, partial, or doesn't match any enum.\n"
+    "4. Slots is FMM-relevant subset only — do NOT include callsign or facility unless required for the action.\n"
+    "5. Output ONLY the JSON object."
+)
+
+
+def to_chat(text: str, segments: list[dict], abstain_reason: str | None = None) -> dict:
+    payload = {"segments": segments}
+    if abstain_reason:
+        payload["abstain_reason"] = abstain_reason
+    else:
+        payload["abstain_reason"] = None
+    gold = json.dumps(payload, ensure_ascii=False)
+    return {
+        "messages": [
+            {"role": "system", "content": SYSTEM_PROMPT},
+            {"role": "user", "content": text.strip()},
+            {"role": "assistant", "content": gold},
+        ]
+    }
+
+
+def normalize_segment(seg: dict) -> dict:
+    """Convert a raw expected_segment dict to canonical (lowercase keys, FMM subset)."""
+    intent = seg.get("intent")
+    if intent not in INTENTS_50:
+        return None
+    slots_raw = seg.get("slots") or {}
+    slots = {k.lower(): str(v) for k, v in slots_raw.items() if v is not None and str(v) != ""}
+    text = (seg.get("text") or "").strip()
+    return {"intent": intent, "slots": slots, "text": text}
+
+
+def walk_canonical_turns():
+    """Yield (text, segments_list, source) for each ATC-spoken turn."""
+    scenario_dir = ROOT / "data/src/partner_graph/output/scenarios"
+    for path in sorted(scenario_dir.glob("*.json")):
+        try:
+            scn = json.loads(path.read_text())
+        except Exception:
+            continue
+        for turn in scn.get("turns", []):
+            if turn.get("speaker", "").lower() != "atc":
+                continue
+            text = (turn.get("expected_transcript") or "").strip()
+            if not text:
+                continue
+
+            # Compound turn → use expected_segments
+            if turn.get("compound") and turn.get("expected_segments"):
+                segs = []
+                for s in turn["expected_segments"]:
+                    norm = normalize_segment(s)
+                    if norm:
+                        segs.append(norm)
+                if not segs:
+                    continue
+                yield text, segs, "compound"
+            else:
+                # Single intent
+                intent = turn.get("expected_intent")
+                if intent not in INTENTS_50:
+                    continue
+                params = turn.get("expected_parameters") or {}
+                slots = {k.lower(): str(v) for k, v in params.items() if v is not None and str(v) != ""}
+                yield text, [{"intent": intent, "slots": slots, "text": text}], "single"
+
+
+def mine_v8_failures():
+    """Yield (text, segments_list, source='hard_negative') from V8 failures with corrected gold."""
+    dump = json.loads((ROOT / "poc/llm-finetune/training/eval_dump_v8_canonical.json").read_text())
+    for r in dump["rows"]:
+        gold_intent = r["gold_intent"]
+        gold_params = r["gold_parameters"] or {}
+        slots = {k.lower(): str(v) for k, v in gold_params.items() if v is not None and str(v) != ""}
+        # Re-emit gold as single-segment hard-negative (not yet compound-aware in canonical-v0)
+        yield r["text"], [{"intent": gold_intent, "slots": slots, "text": r["text"]}], "hard_negative"
+
+
+def synthesize_unknowns():
+    """Synthesize `unknown`-intent adversarial examples for abstention training."""
+    examples = [
+        ("zzzkkrrr static partial garbled transmission unintelligible", "garbled_transcript"),
+        ("station calling please say again", "incomplete_transmission"),
+        ("uhh what was that", "non_atc_speech"),
+        ("hello world this is a test", "out_of_domain"),
+        ("the cat sat on the mat", "out_of_domain"),
+        ("November One Seven X-ray X-ray, kkkrr static break sssh", "partial_transmission"),
+        ("press the red button to launch missiles", "non_atc_command"),
+        ("Cleared for landing on the moon", "implausible_atc"),
+        ("Altitude pizza at runway sandwich", "nonsense_slots"),
+        ("November One Seven X-ray X-ray, blah blah blah unknown command", "unrecognizable_intent"),
+        ("static noise crackle hiss", "noise_only"),
+        ("eee aaa ooo speaking gibberish", "non_words"),
+        ("ATC robot voice synthesis test 12345", "test_transmission"),
+        ("Going to the store to buy bread", "non_atc"),
+        ("Tower this is invisible plane requesting cloud taxi", "implausible_atc"),
+    ]
+    for text, reason in examples:
+        yield text, [{"intent": "unknown", "slots": {}, "text": text}], "unknown_synth"
+
+
+def main():
+    out = ROOT / "poc/llm-finetune/training/data_canonical_v1"
+    out.mkdir(parents=True, exist_ok=True)
+
+    rows = []
+    counts = defaultdict(int)
+
+    print("walking canonical scenarios (single + compound)...")
+    for text, segs, source in walk_canonical_turns():
+        rows.append({"text": text, "segments": segs, "source": source})
+        counts[source] += 1
+
+    print("mining V8 hard negatives...")
+    for text, segs, source in mine_v8_failures():
+        rows.append({"text": text, "segments": segs, "source": source})
+        counts[source] += 1
+
+    print("synthesizing unknown adversarials...")
+    for text, segs, source in synthesize_unknowns():
+        rows.append({"text": text, "segments": segs, "source": source})
+        counts[source] += 1
+
+    print(f"\nraw counts: {dict(counts)}")
+    print(f"total raw: {len(rows)}")
+
+    # Dedupe by text
+    seen = set()
+    dedup = []
+    for r in rows:
+        if r["text"] in seen:
+            continue
+        seen.add(r["text"])
+        dedup.append(r)
+    print(f"deduped: {len(dedup)}")
+
+    rng = random.Random(42)
+    rng.shuffle(dedup)
+    n = len(dedup)
+    n_valid = max(20, int(n * 0.05))
+    n_test = max(20, int(n * 0.05))
+    n_train = n - n_valid - n_test
+    splits = {"train": dedup[:n_train], "valid": dedup[n_train:n_train+n_valid], "test": dedup[n_train+n_valid:]}
+
+    for name, items in splits.items():
+        path = out / f"{name}.jsonl"
+        with path.open("w") as f:
+            for r in items:
+                f.write(json.dumps(to_chat(r["text"], r["segments"]), ensure_ascii=False) + "\n")
+        print(f"{name}: {len(items)} -> {path}")
+
+    # Stats
+    n_compound = sum(1 for r in dedup if len(r["segments"]) > 1)
+    n_unknown = sum(1 for r in dedup if r["segments"][0]["intent"] == "unknown")
+    print(f"\ncompound rows: {n_compound}")
+    print(f"unknown rows:  {n_unknown}")
+    print(f"single rows:   {len(dedup) - n_compound - n_unknown}")
+
+
+if __name__ == "__main__":
+    main()