Spaces:
Running
Running
Download backend/src/evaluate.py from VivDubs/eumora-api: direct link, hf CLI and curl.
- Browser
- Download file 24.1 kB
-
https://huggingface.co/spaces/VivDubs/eumora-api/resolve/main/backend/src/evaluate.py
- Command line
-
hf download hf://spaces/VivDubs/eumora-api/backend/src/evaluate.py
-
curl -L -o evaluate.py https://huggingface.co/spaces/VivDubs/eumora-api/resolve/main/backend/src/evaluate.py
24.1 kB
| """Checkpoint evaluation and ablation reporting. | |
| Produces the numbers that go in the paper: in-domain test metrics, per-class | |
| breakdown, calibration error, and β the measurement this work exists for β the | |
| lyric-register gap from the probe set. | |
| # one checkpoint | |
| python -m backend.src.evaluate --model models/eumora_improved_.../final | |
| # ablation table across checkpoints | |
| python -m backend.src.evaluate \\ | |
| --model baseline=models/eumora_baseline_.../final \\ | |
| --model improved=models/eumora_improved_.../final \\ | |
| --model +tapt=models/eumora_tapt_.../final \\ | |
| --markdown results/ablation.md | |
| `--no-test-set` skips the in-domain split and runs the probe only, which needs | |
| no dataset download and takes seconds. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import os | |
| from pathlib import Path | |
| from typing import Dict, List, Optional | |
| import numpy as np | |
| import torch | |
| from sklearn.metrics import ( | |
| accuracy_score, | |
| classification_report, | |
| confusion_matrix, | |
| f1_score, | |
| ) | |
| from transformers import AutoModelForSequenceClassification, AutoTokenizer | |
| from .calibration import expected_calibration_error, negative_log_likelihood | |
| from .data_prep import TARGET_LABELS | |
| from .lyrics_probe import LABELS as PROBE_LABELS, load_probe | |
| def get_device() -> torch.device: | |
| if torch.cuda.is_available(): | |
| return torch.device("cuda") | |
| if torch.backends.mps.is_available(): | |
| return torch.device("mps") | |
| return torch.device("cpu") | |
| def batched_logits( | |
| model, tokenizer, texts: List[str], device, max_length: int = 256, batch_size: int = 32 | |
| ) -> np.ndarray: | |
| model.eval() | |
| out = [] | |
| for i in range(0, len(texts), batch_size): | |
| enc = tokenizer( | |
| texts[i : i + batch_size], | |
| return_tensors="pt", | |
| truncation=True, | |
| max_length=max_length, | |
| padding=True, | |
| ).to(device) | |
| out.append(model(**enc).logits.float().cpu().numpy()) | |
| return np.concatenate(out, axis=0) | |
| def softmax(z: np.ndarray) -> np.ndarray: | |
| z = z - z.max(axis=1, keepdims=True) | |
| e = np.exp(z) | |
| return e / e.sum(axis=1, keepdims=True) | |
| # ββ Uncertainty βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def bootstrap_ci( | |
| values: np.ndarray, n_boot: int = 10000, alpha: float = 0.05, seed: int = 0 | |
| ) -> tuple: | |
| """Percentile bootstrap CI for the mean of a 0/1 outcome vector. | |
| The probe has 40 items per register, so a reported accuracy carries real | |
| sampling uncertainty β 11.1% on 36 items has a wide interval, and a paper | |
| quoting the point estimate alone overstates what was measured. Bootstrapping | |
| avoids a normal approximation, which is poor for proportions near 0 or 1. | |
| """ | |
| if len(values) == 0: | |
| return (None, None) | |
| rng = np.random.default_rng(seed) | |
| idx = rng.integers(0, len(values), size=(n_boot, len(values))) | |
| means = values[idx].mean(axis=1) | |
| lo, hi = np.percentile(means, [100 * alpha / 2, 100 * (1 - alpha / 2)]) | |
| return (round(float(lo), 4), round(float(hi), 4)) | |
| def paired_gap_ci( | |
| literal: np.ndarray, figurative: np.ndarray, | |
| n_boot: int = 10000, alpha: float = 0.05, seed: int = 0 | |
| ) -> tuple: | |
| """CI for the register gap, resampling *pairs* rather than the two arms. | |
| The probe is built from minimal pairs, so the two subsets are not | |
| independent β a pair shares its gold label and its content. Resampling pairs | |
| preserves that structure; resampling the arms separately would understate | |
| the correlation and give a misleadingly wide interval. | |
| """ | |
| if len(literal) == 0 or len(literal) != len(figurative): | |
| return (None, None) | |
| rng = np.random.default_rng(seed) | |
| idx = rng.integers(0, len(literal), size=(n_boot, len(literal))) | |
| gaps = literal[idx].mean(axis=1) - figurative[idx].mean(axis=1) | |
| lo, hi = np.percentile(gaps, [100 * alpha / 2, 100 * (1 - alpha / 2)]) | |
| return (round(float(lo), 4), round(float(hi), 4)) | |
| # ββ Probe evaluation ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def evaluate_probe_detailed(model, tokenizer, device, max_length: int = 256) -> Dict: | |
| """Full probe breakdown: register gap, per-phenomenon, per-pair failures.""" | |
| all_records = load_probe() | |
| # Compare by label *name*, not index: checkpoints may order their classes | |
| # differently, and older variants may not have all eight. Items whose gold | |
| # label the model cannot express are excluded and reported, rather than | |
| # failing the whole run. | |
| model_labels = [ | |
| model.config.id2label[i] for i in sorted(int(k) for k in model.config.id2label) | |
| ] | |
| supported = [l for l in PROBE_LABELS if l in model_labels] | |
| skipped = [l for l in PROBE_LABELS if l not in model_labels] | |
| records = [r for r in all_records if r["gold"] in supported] | |
| if not records: | |
| raise ValueError( | |
| f"Model labels {model_labels} share no classes with the probe {PROBE_LABELS}" | |
| ) | |
| logits = batched_logits(model, tokenizer, [r["text"] for r in records], device, max_length) | |
| probs = softmax(logits) | |
| pred_names = [model_labels[i] for i in probs.argmax(axis=1)] | |
| correct = np.array( | |
| [int(p == r["gold"]) for p, r in zip(pred_names, records)], dtype=int | |
| ) | |
| # Gold indices in the *model's* label space, for calibration metrics. | |
| gold = np.array([model_labels.index(r["gold"]) for r in records]) | |
| def subset_acc(mask) -> Optional[float]: | |
| idx = [i for i, r in enumerate(records) if mask(r)] | |
| return round(float(correct[idx].mean()), 4) if idx else None | |
| literal = subset_acc(lambda r: r["register"] == "literal") | |
| figurative = subset_acc(lambda r: r["register"] == "figurative") | |
| # Pairs where the literal member is right but the figurative one is wrong β | |
| # the cleanest evidence of register sensitivity, controlling for content. | |
| by_pair: Dict[str, Dict[str, int]] = {} | |
| for i, r in enumerate(records): | |
| by_pair.setdefault(r["pair_id"], {})[r["register"]] = int(correct[i]) | |
| broken = sorted( | |
| p for p, v in by_pair.items() | |
| if v.get("literal") == 1 and v.get("figurative") == 0 | |
| ) | |
| phenomena = sorted({r["phenomenon"] for r in records}) | |
| per_class = { | |
| name: subset_acc(lambda r, n=name: r["gold"] == n) for name in supported | |
| } | |
| # Uncertainty. The gap is bootstrapped over pairs, since the probe's two | |
| # halves are matched rather than independent samples. | |
| def subset_vec(mask): | |
| return correct[[i for i, r in enumerate(records) if mask(r)]] | |
| lit_vec = subset_vec(lambda r: r["register"] == "literal") | |
| fig_vec = subset_vec(lambda r: r["register"] == "figurative") | |
| pair_ids = sorted({r["pair_id"] for r in records}) | |
| lit_by_pair, fig_by_pair = [], [] | |
| for pid in pair_ids: | |
| members = {r["register"]: correct[i] for i, r in enumerate(records) | |
| if r["pair_id"] == pid} | |
| if "literal" in members and "figurative" in members: | |
| lit_by_pair.append(members["literal"]) | |
| fig_by_pair.append(members["figurative"]) | |
| # How often the model retreats to `neutral` on figurative input β the | |
| # specific failure mode behind the low lyric scores in production. | |
| figurative_non_neutral = [ | |
| i for i, r in enumerate(records) | |
| if r["register"] == "figurative" and r["gold"] != "neutral" | |
| ] | |
| neutral_retreat = ( | |
| round(float(np.mean([pred_names[i] == "neutral" for i in figurative_non_neutral])), 4) | |
| if figurative_non_neutral and "neutral" in model_labels | |
| else None | |
| ) | |
| # A model that predicts one class for everything has a register gap of 0, | |
| # which reads as "no register sensitivity" when it actually means the model | |
| # never discriminates at all. The baseline arm does exactly this β it | |
| # collapses to `neutral` and scores 7.5% on both registers. Flag it so the | |
| # gap is not reported as a strength. | |
| distinct_predictions = len(set(pred_names)) | |
| from collections import Counter as _Counter | |
| top_share = _Counter(pred_names).most_common(1)[0][1] / len(pred_names) | |
| degenerate = distinct_predictions <= 2 or top_share >= 0.9 | |
| return { | |
| "overall": round(float(correct.mean()), 4), | |
| "overall_ci": bootstrap_ci(correct), | |
| "literal": literal, | |
| "literal_ci": bootstrap_ci(lit_vec), | |
| "figurative": figurative, | |
| "figurative_ci": bootstrap_ci(fig_vec), | |
| "register_gap_ci": paired_gap_ci(np.array(lit_by_pair), np.array(fig_by_pair)), | |
| "n_complete_pairs": len(lit_by_pair), | |
| "degenerate": degenerate, | |
| "distinct_predictions": distinct_predictions, | |
| "majority_prediction_share": round(top_share, 4), | |
| "register_gap": round(literal - figurative, 4) if None not in (literal, figurative) else None, | |
| "by_phenomenon": {p: subset_acc(lambda r, p=p: r["phenomenon"] == p) for p in phenomena}, | |
| "by_class": per_class, | |
| "mean_confidence": round(float(probs.max(axis=1).mean()), 4), | |
| "ece": round(expected_calibration_error(probs, gold), 4), | |
| "broken_pairs": broken, | |
| "n_broken_pairs": len(broken), | |
| "neutral_rate_figurative": neutral_retreat, | |
| "n_evaluated": len(records), | |
| "n_total": len(all_records), | |
| "skipped_labels": skipped, | |
| # Per-item outcomes, retained so two arms can be compared on the *same* | |
| # items rather than by eyeballing whether their intervals overlap. | |
| "_pair_ids": pair_ids, | |
| "_literal_correct": [int(x) for x in lit_by_pair], | |
| "_figurative_correct": [int(x) for x in fig_by_pair], | |
| } | |
| # ββ In-domain evaluation ββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def evaluate_test_set(model, tokenizer, device, max_length: int = 256, **data_kwargs) -> Dict: | |
| from .data_prep import build_dataset | |
| dataset, _ = build_dataset(**data_kwargs) | |
| texts = dataset["test"]["text"] | |
| gold = np.array(dataset["test"]["label"]) | |
| logits = batched_logits(model, tokenizer, texts, device, max_length) | |
| probs = softmax(logits) | |
| preds = probs.argmax(axis=1) | |
| per_class = f1_score( | |
| gold, preds, average=None, zero_division=0, labels=range(len(TARGET_LABELS)) | |
| ) | |
| return { | |
| "accuracy": round(float(accuracy_score(gold, preds)), 4), | |
| "f1_macro": round(float(f1_score(gold, preds, average="macro", zero_division=0)), 4), | |
| "f1_weighted": round(float(f1_score(gold, preds, average="weighted", zero_division=0)), 4), | |
| "per_class_f1": {n: round(float(s), 4) for n, s in zip(TARGET_LABELS, per_class)}, | |
| "ece": round(expected_calibration_error(probs, gold), 4), | |
| "nll": round(negative_log_likelihood(probs, gold), 4), | |
| "mean_confidence": round(float(probs.max(axis=1).mean()), 4), | |
| "n_samples": len(gold), | |
| "report": classification_report( | |
| gold, preds, labels=range(len(TARGET_LABELS)), | |
| target_names=TARGET_LABELS, zero_division=0, digits=3, | |
| ), | |
| "confusion": confusion_matrix( | |
| gold, preds, labels=range(len(TARGET_LABELS)) | |
| ).tolist(), | |
| } | |
| def load_env() -> None: | |
| """Load .env so private Hub repos resolve without exporting HF_TOKEN by hand. | |
| The deployed classifier lives in a private repo, so benchmarking it β the | |
| baseline row of the ablation table β needs a token. Reading .env here means | |
| `--model VivDubs/eumora-emotion-classifier` just works. | |
| """ | |
| for candidate in ( | |
| Path(__file__).resolve().parents[2] / ".env", # repo root | |
| Path(__file__).resolve().parents[1] / ".env", # backend/ | |
| ): | |
| if not candidate.exists(): | |
| continue | |
| for line in candidate.read_text().splitlines(): | |
| line = line.strip() | |
| if not line or line.startswith("#") or "=" not in line: | |
| continue | |
| key, value = line.split("=", 1) | |
| os.environ.setdefault(key.strip(), value.strip().strip('"').strip("'")) | |
| def evaluate_checkpoint( | |
| model_path: str, | |
| run_test_set: bool = True, | |
| max_length: int = 256, | |
| ) -> Dict: | |
| device = get_device() | |
| print(f"\nEvaluating {model_path}") | |
| token = os.environ.get("HF_TOKEN") or os.environ.get("HF_API_TOKEN") | |
| auth = {"token": token} if token and not Path(model_path).exists() else {} | |
| if not Path(model_path).exists() and not token: | |
| print(" (no HF_TOKEN found β private Hub repos will fail to load)") | |
| tokenizer = AutoTokenizer.from_pretrained(model_path, **auth) | |
| model = AutoModelForSequenceClassification.from_pretrained(model_path, **auth).to(device) | |
| result: Dict = {"model_path": str(model_path)} | |
| calib = getattr(model.config, "calibration", None) | |
| if calib: | |
| result["calibration_folded"] = calib | |
| print(f" calibration folded in: T={calib.get('temperature')}") | |
| print(" lyric-register probe...") | |
| result["probe"] = evaluate_probe_detailed(model, tokenizer, device, max_length) | |
| p = result["probe"] | |
| if p["skipped_labels"]: | |
| print(f" note: model cannot predict {p['skipped_labels']} β " | |
| f"scoring {p['n_evaluated']}/{p['n_total']} items") | |
| def ci(key): | |
| lo, hi = p.get(key) or (None, None) | |
| return f" [{lo:.0%}-{hi:.0%}]" if lo is not None else "" | |
| print(f" overall {p['overall']:.1%}{ci('overall_ci')} | " | |
| f"literal {p['literal']:.1%}{ci('literal_ci')} | " | |
| f"figurative {p['figurative']:.1%}{ci('figurative_ci')}") | |
| gci = p.get("register_gap_ci") or (None, None) | |
| gap_txt = (f" [{gci[0]:+.0%} to {gci[1]:+.0%}]" if gci[0] is not None else "") | |
| print(f" register gap {p['register_gap']:+.1%}{gap_txt} " | |
| f"({p.get('n_complete_pairs', 0)} pairs)") | |
| if p.get("degenerate"): | |
| print(f" DEGENERATE: predicts {p['distinct_predictions']} distinct class(es), " | |
| f"{p['majority_prediction_share']:.0%} one class β the gap is meaningless here") | |
| if p["neutral_rate_figurative"] is not None: | |
| print(f" neutral-retreat on figurative: {p['neutral_rate_figurative']:.1%}") | |
| if run_test_set: | |
| print(" in-domain test set...") | |
| result["test"] = evaluate_test_set(model, tokenizer, device, max_length) | |
| t = result["test"] | |
| print(f" accuracy {t['accuracy']:.1%} | macro-F1 {t['f1_macro']:.1%} | ECE {t['ece']:.4f}") | |
| return result | |
| def compare_arms( | |
| a_name: str, a_probe: Dict, b_name: str, b_probe: Dict, | |
| n_boot: int = 10000, seed: int = 0, | |
| ) -> Optional[Dict]: | |
| """Bootstrap the *difference* in register gap between two models. | |
| Two arms evaluated on the same probe are matched data, so the question is | |
| not whether their individual intervals overlap β overlapping intervals are | |
| entirely compatible with a significant paired difference, and reading them | |
| that way is a standard error. Resample the same pair indices for both models | |
| and bootstrap the difference directly. | |
| Returns None when the two runs did not score the same pairs, since the | |
| comparison is only paired if the items match. | |
| """ | |
| if a_probe.get("_pair_ids") != b_probe.get("_pair_ids"): | |
| return None | |
| a_lit = np.array(a_probe["_literal_correct"]) | |
| a_fig = np.array(a_probe["_figurative_correct"]) | |
| b_lit = np.array(b_probe["_literal_correct"]) | |
| b_fig = np.array(b_probe["_figurative_correct"]) | |
| if len(a_lit) == 0: | |
| return None | |
| rng = np.random.default_rng(seed) | |
| idx = rng.integers(0, len(a_lit), size=(n_boot, len(a_lit))) | |
| gap_a = a_lit[idx].mean(axis=1) - a_fig[idx].mean(axis=1) | |
| gap_b = b_lit[idx].mean(axis=1) - b_fig[idx].mean(axis=1) | |
| diff = gap_b - gap_a # negative => b closed the gap | |
| fig_diff = b_fig[idx].mean(axis=1) - a_fig[idx].mean(axis=1) | |
| lit_diff = b_lit[idx].mean(axis=1) - a_lit[idx].mean(axis=1) | |
| def ci(v): | |
| lo, hi = np.percentile(v, [2.5, 97.5]) | |
| return (round(float(lo), 4), round(float(hi), 4)) | |
| gap_lo, gap_hi = ci(diff) | |
| return { | |
| "from": a_name, "to": b_name, "n_pairs": int(len(a_lit)), | |
| "gap_change": round(float(diff.mean()), 4), "gap_change_ci": (gap_lo, gap_hi), | |
| "figurative_change": round(float(fig_diff.mean()), 4), | |
| "figurative_change_ci": ci(fig_diff), | |
| "literal_change": round(float(lit_diff.mean()), 4), | |
| "literal_change_ci": ci(lit_diff), | |
| # Only claim a change when the interval clears zero. | |
| "gap_significantly_closed": bool(gap_hi < 0), | |
| "gap_significantly_widened": bool(gap_lo > 0), | |
| } | |
| # ββ Reporting βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ | |
| def markdown_table(results: Dict[str, Dict]) -> str: | |
| """Ablation table, ready to paste into the paper.""" | |
| has_test = any("test" in r for r in results.values()) | |
| lines = ["# EUMORA ablation results", ""] | |
| header = ["Variant", "Probe overall", "Literal", "Figurative", "Register gap", "Neutral retreat"] | |
| if has_test: | |
| header += ["Test acc", "Test macro-F1", "ECE"] | |
| lines.append("| " + " | ".join(header) + " |") | |
| lines.append("|" + "|".join(["---"] * len(header)) + "|") | |
| def pct(v, signed=False): | |
| if v is None: | |
| return "β" | |
| return f"{v:+.1%}" if signed else f"{v:.1%}" | |
| for name, r in results.items(): | |
| p = r["probe"] | |
| label = name + ("*" if p.get("skipped_labels") else "") + ("β " if p.get("degenerate") else "") | |
| row = [ | |
| label, | |
| pct(p["overall"]), | |
| pct(p["literal"]), | |
| pct(p["figurative"]), | |
| (pct(p["register_gap"], signed=True) + | |
| (f" [{p['register_gap_ci'][0]:+.0%}, {p['register_gap_ci'][1]:+.0%}]" | |
| if p.get("register_gap_ci") and p["register_gap_ci"][0] is not None else "")), | |
| pct(p["neutral_rate_figurative"]), | |
| ] | |
| if has_test: | |
| t = r.get("test") | |
| row += ( | |
| [f"{t['accuracy']:.1%}", f"{t['f1_macro']:.1%}", f"{t['ece']:.4f}"] | |
| if t else ["β", "β", "β"] | |
| ) | |
| lines.append("| " + " | ".join(row) + " |") | |
| lines += [ | |
| "", | |
| "**Register gap** = literal accuracy β figurative accuracy on matched minimal", | |
| "pairs. Lower is better; it isolates sensitivity to lyric phrasing while holding", | |
| "content and label distribution fixed.", | |
| "", | |
| "**Neutral retreat** = share of non-neutral figurative items predicted `neutral`.", | |
| "This is the failure mode behind the low lyric scores in production.", | |
| "", | |
| ] | |
| degen = {n: r["probe"] for n, r in results.items() if r["probe"].get("degenerate")} | |
| for n, pr in degen.items(): | |
| lines.append( | |
| f"β `{n}` predicts only {pr['distinct_predictions']} distinct class(es) " | |
| f"({pr['majority_prediction_share']:.0%} a single class). Its register gap is " | |
| f"an artefact of uniform failure, not evidence of register robustness." | |
| ) | |
| if degen: | |
| lines.append("") | |
| partial = { | |
| n: r["probe"]["skipped_labels"] | |
| for n, r in results.items() if r["probe"].get("skipped_labels") | |
| } | |
| for n, sk in partial.items(): | |
| r = results[n]["probe"] | |
| lines.append( | |
| f"\\* `{n}` cannot predict {', '.join(sk)}; scored on " | |
| f"{r['n_evaluated']}/{r['n_total']} probe items." | |
| ) | |
| if partial: | |
| lines.append("") | |
| names = list(results) | |
| if len(names) > 1: | |
| pairwise = [] | |
| for i in range(len(names) - 1): | |
| c = compare_arms(names[i], results[names[i]]["probe"], | |
| names[i + 1], results[names[i + 1]]["probe"]) | |
| if c: | |
| pairwise.append(c) | |
| if pairwise: | |
| lines += ["## Paired comparison between arms", "", | |
| "Differences are bootstrapped over the *same* probe pairs for both", | |
| "models. Overlapping individual intervals are compatible with a", | |
| "significant paired difference, so this is the comparison to read.", "", | |
| "| change | register gap | figurative | literal |", | |
| "|---|---|---|---|"] | |
| for c in pairwise: | |
| verdict = ("**closed**" if c["gap_significantly_closed"] | |
| else "**widened**" if c["gap_significantly_widened"] | |
| else "not significant") | |
| lines.append( | |
| f"| {c['from']} -> {c['to']} | " | |
| f"{c['gap_change']:+.1%} [{c['gap_change_ci'][0]:+.0%}, {c['gap_change_ci'][1]:+.0%}] {verdict} | " | |
| f"{c['figurative_change']:+.1%} [{c['figurative_change_ci'][0]:+.0%}, {c['figurative_change_ci'][1]:+.0%}] | " | |
| f"{c['literal_change']:+.1%} [{c['literal_change_ci'][0]:+.0%}, {c['literal_change_ci'][1]:+.0%}] |" | |
| ) | |
| lines.append("") | |
| phenomena = sorted({p for r in results.values() for p in r["probe"]["by_phenomenon"]}) | |
| lines += ["## Accuracy by linguistic phenomenon", ""] | |
| lines.append("| Variant | " + " | ".join(phenomena) + " |") | |
| lines.append("|" + "|".join(["---"] * (len(phenomena) + 1)) + "|") | |
| for name, r in results.items(): | |
| vals = r["probe"]["by_phenomenon"] | |
| lines.append( | |
| "| " + name + " | " | |
| + " | ".join(f"{vals[p]:.1%}" if vals.get(p) is not None else "β" for p in phenomena) | |
| + " |" | |
| ) | |
| return "\n".join(lines) + "\n" | |
| def main(): | |
| load_env() | |
| parser = argparse.ArgumentParser(description="Evaluate EUMORA checkpoints") | |
| parser.add_argument( | |
| "--model", action="append", required=True, | |
| help="Path, or name=path to label it in the table. Repeatable.", | |
| ) | |
| parser.add_argument("--no-test-set", action="store_true", | |
| help="Probe only β no dataset download needed") | |
| parser.add_argument("--max-length", type=int, default=256) | |
| parser.add_argument("--json", type=str, default=None, help="Write full results JSON here") | |
| parser.add_argument("--markdown", type=str, default=None, help="Write the ablation table here") | |
| parser.add_argument("--per-class", action="store_true", help="Print the sklearn report") | |
| args = parser.parse_args() | |
| results: Dict[str, Dict] = {} | |
| for spec in args.model: | |
| name, path = spec.split("=", 1) if "=" in spec else (Path(spec).parent.name, spec) | |
| results[name] = evaluate_checkpoint(path, not args.no_test_set, args.max_length) | |
| if args.per_class: | |
| for name, r in results.items(): | |
| if "test" in r: | |
| print(f"\n=== {name} β per-class (in-domain test) ===") | |
| print(r["test"]["report"]) | |
| print("\n" + markdown_table(results)) | |
| if args.json: | |
| out = Path(args.json) | |
| out.parent.mkdir(parents=True, exist_ok=True) | |
| serializable = { | |
| n: {k: v for k, v in r.items() if k != "test"} | ( | |
| {"test": {k: v for k, v in r["test"].items() if k != "report"}} | |
| if "test" in r else {} | |
| ) | |
| for n, r in results.items() | |
| } | |
| out.write_text(json.dumps(serializable, indent=2, default=float)) | |
| print(f"JSON -> {out}") | |
| if args.markdown: | |
| out = Path(args.markdown) | |
| out.parent.mkdir(parents=True, exist_ok=True) | |
| out.write_text(markdown_table(results)) | |
| print(f"Table -> {out}") | |
| if __name__ == "__main__": | |
| main() | |