Spaces:
Running
Running
| """Which DeBERTa reads the listener's own text better? (HANDOFF §4 item 4) | |
| The same weights read the user's text (Vercel `/api/emotion`) and the songs' | |
| lyrics. A candidate text model has to win on lyrics through the whole system | |
| (`train_fusion compare`) *and* not lose on user text. This measures the second | |
| half, on text neither model trained on: | |
| tweet_eval SemEval-2018 emotion tweets, test split (1,421). anger, joy, | |
| optimism and sadness; optimism counts as joy. Not part of | |
| dair-ai/emotion, GoEmotions or tweet_eval/irony. | |
| literal the register probe's 55 literal statements, the closest thing | |
| to what people type into the box ("I feel so sad and empty...") | |
| Both models are read the way the route reads them: plain softmax, argmax over | |
| all 8 classes, so a prediction outside the gold set (e.g. `neutral`) is wrong. | |
| python -m backend.src.user_text_eval --live models/<live>/final --candidate models/<cand>/final --tag full | |
| Ship rule, fixed before the first run: the candidate passes the user-text side | |
| if, on *both* sets, the paired 95% bootstrap interval of its accuracy minus the | |
| live model's has an upper bound >= 0 (not significantly worse). Decided on | |
| unrounded values. Macro-F1 and VAD error are reported, not ruled on. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| from pathlib import Path | |
| from typing import Dict, List | |
| import numpy as np | |
| import torch | |
| from sklearn.metrics import f1_score | |
| from transformers import AutoModelForSequenceClassification, AutoTokenizer | |
| from .acoustic import EMOTION_VAD | |
| from .fusion import user_vad | |
| from .lyrics_probe import load_probe | |
| ROOT = Path(__file__).resolve().parents[2] | |
| RESULTS = ROOT / "results" | |
| TWEET_EVAL_MAP = {0: "anger", 1: "joy", 2: "joy", 3: "sadness"} # optimism -> joy | |
| N_BOOT = 10000 | |
| SEED = 0 | |
| def user_sets() -> Dict[str, List[Dict]]: | |
| from datasets import load_dataset | |
| te = load_dataset("tweet_eval", "emotion", split="test") | |
| return { | |
| "tweet_eval": [{"text": t, "gold": TWEET_EVAL_MAP[int(l)]} for t, l in zip(te["text"], te["label"])], | |
| "literal": [{"text": r["text"], "gold": r["gold"]} for r in load_probe() if r["register"] == "literal"], | |
| } | |
| def softmax_probs(path: str, texts: List[str], device: torch.device, batch: int = 32) -> List[Dict[str, float]]: | |
| tok = AutoTokenizer.from_pretrained(path) | |
| model = AutoModelForSequenceClassification.from_pretrained(path).to(device).eval() | |
| labels = [model.config.id2label[i] for i in range(model.config.num_labels)] | |
| out = [] | |
| for i in range(0, len(texts), batch): | |
| enc = tok(texts[i:i + batch], return_tensors="pt", truncation=True, max_length=256, padding=True).to(device) | |
| p = torch.softmax(model(**enc).logits.float(), dim=-1).cpu().numpy() | |
| out.extend({l: float(v) for l, v in zip(labels, row)} for row in p) | |
| return out | |
| def vad_error(probs: Dict[str, float], gold: str) -> float: | |
| """Distance from the user's VAD point to the gold emotion's; an all-neutral | |
| reading sits at neutral, as the app has nothing else to place it by.""" | |
| v = user_vad(probs) | |
| v = np.asarray(EMOTION_VAD["neutral"]) if v is None else v | |
| return float(np.linalg.norm(v - np.asarray(EMOTION_VAD[gold]))) | |
| def score(probs: List[Dict[str, float]], items: List[Dict]) -> Dict: | |
| pred = [max(p, key=p.get) for p in probs] | |
| gold = [it["gold"] for it in items] | |
| present = sorted(set(gold)) | |
| return { | |
| "correct": np.array([int(a == b) for a, b in zip(pred, gold)]), | |
| "accuracy": float(np.mean([a == b for a, b in zip(pred, gold)])), | |
| "macro_f1": float(f1_score(gold, pred, labels=present, average="macro", zero_division=0)), | |
| "vad_error": float(np.mean([vad_error(p, g) for p, g in zip(probs, gold)])), | |
| "neutral_rate": float(np.mean([x == "neutral" for x in pred])), | |
| } | |
| def paired_ci(a: np.ndarray, b: np.ndarray) -> tuple: | |
| """95% percentile interval of mean(b) - mean(a), resampling items.""" | |
| rng = np.random.default_rng(SEED) | |
| idx = rng.integers(0, len(a), size=(N_BOOT, len(a))) | |
| d = b[idx].mean(axis=1) - a[idx].mean(axis=1) | |
| return float(np.percentile(d, 2.5)), float(np.percentile(d, 97.5)) | |
| def compare(live: str, candidate: str, tag: str) -> Dict: | |
| device = torch.device("cuda" if torch.cuda.is_available() else "cpu") | |
| report: Dict = {"live": live, "candidate": candidate, "sets": {}} | |
| for name, items in user_sets().items(): | |
| texts = [it["text"] for it in items] | |
| a, b = score(softmax_probs(live, texts, device), items), score(softmax_probs(candidate, texts, device), items) | |
| lo, hi = paired_ci(a["correct"], b["correct"]) | |
| report["sets"][name] = { | |
| "n": len(items), | |
| "live": {k: round(v, 4) for k, v in a.items() if k != "correct"}, | |
| tag: {k: round(v, 4) for k, v in b.items() if k != "correct"}, | |
| "accuracy_diff": round(b["accuracy"] - a["accuracy"], 4), | |
| "ci": [round(lo, 4), round(hi, 4)], | |
| "not_worse": bool(hi >= 0), | |
| } | |
| report["ship_user_text"] = all(s["not_worse"] for s in report["sets"].values()) | |
| for name, s in report["sets"].items(): | |
| print(f" {name:10s} n={s['n']:5d} live {s['live']['accuracy']:.3f} {tag} {s[tag]['accuracy']:.3f} " | |
| f"diff {s['accuracy_diff']:+.3f} [{s['ci'][0]:+.3f}, {s['ci'][1]:+.3f}] " | |
| f"macro-F1 {s['live']['macro_f1']:.3f} -> {s[tag]['macro_f1']:.3f} " | |
| f"VAD err {s['live']['vad_error']:.2f} -> {s[tag]['vad_error']:.2f}") | |
| print(f" user-text side passes: {report['ship_user_text']}") | |
| out = RESULTS / f"user_text_{tag}.json" | |
| out.write_text(json.dumps(report, indent=2)) | |
| print(f"wrote {out}") | |
| return report | |
| def main() -> int: | |
| parser = argparse.ArgumentParser(description="Live vs candidate DeBERTa on the listener's own text") | |
| parser.add_argument("--live", required=True, help="local copy of the deployed weights") | |
| parser.add_argument("--candidate", required=True) | |
| parser.add_argument("--tag", required=True) | |
| args = parser.parse_args() | |
| compare(args.live, args.candidate, args.tag) | |
| return 0 | |
| if __name__ == "__main__": | |
| raise SystemExit(main()) | |