Buckets:
| """WO38 — error anatomy of the urd-g2p byt5-base checkpoint. | |
| The data-quality arm closed (6 conflicts in 610K pairs — the | |
| dictionary is clean), so the 15.65% CER is model difficulty. This | |
| eval classifies every error char by type: stress/dots (convention | |
| cosmetics), vowels+diacritics, consonants (retroflex/aspirate | |
| inventory), punctuation — telling us whether a future arm has | |
| phoneme-level headroom or is chasing notation. | |
| Usage (cpu-performance): python /ckpt/run-059-urd-g2p-base/anatomy.py | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import time | |
| from pathlib import Path | |
| from collections import Counter | |
| import torch | |
| DATA = Path("/data-ru/data") | |
| CKPT = Path("/ckpt") / os.environ.get("U_RUN", "run-059-urd-g2p-base") | |
| STRESS = "ˈˌ." | |
| def norm(s: str) -> str: | |
| for ch in STRESS: | |
| s = s.replace(ch, "") | |
| return s.replace(" ", "") | |
| def lev_align(a: str, b: str): | |
| """Returns list of (op, ref_char, pred_char) for the optimal path.""" | |
| n, m = len(a), len(b) | |
| dp = [[0] * (m + 1) for _ in range(n + 1)] | |
| for i in range(n + 1): | |
| dp[i][0] = i | |
| for j in range(m + 1): | |
| dp[0][j] = j | |
| for i in range(1, n + 1): | |
| for j in range(1, m + 1): | |
| dp[i][j] = min(dp[i-1][j] + 1, dp[i][j-1] + 1, | |
| dp[i-1][j-1] + (a[i-1] != b[j-1])) | |
| ops = [] | |
| i, j = n, m | |
| while i > 0 or j > 0: | |
| if i > 0 and j > 0 and dp[i][j] == dp[i-1][j-1] + (a[i-1] != b[j-1]): | |
| ops.append(("match" if a[i-1] == b[j-1] else "sub", a[i-1], b[j-1])) | |
| i, j = i - 1, j - 1 | |
| elif i > 0 and dp[i][j] == dp[i-1][j] + 1: | |
| ops.append(("del", a[i-1], "")) | |
| i -= 1 | |
| else: | |
| ops.append(("ins", "", b[j-1])) | |
| j -= 1 | |
| return ops[::-1] | |
| def classify(ref_char: str) -> str: | |
| if ref_char in STRESS or ref_char == ".": | |
| return "stress/dots" | |
| if ref_char in "əɪʊæɐeɛoɔaːiːuːeːoːãẽʊ̃ɐ̃": | |
| return "vowels/diacritics" | |
| if ref_char in "ː": | |
| return "length" | |
| return "consonants" | |
| def main() -> None: | |
| from transformers import AutoModelForSeq2SeqLM, AutoTokenizer | |
| tok = AutoTokenizer.from_pretrained(CKPT / "best") | |
| model = AutoModelForSeq2SeqLM.from_pretrained(CKPT / "best").cuda().eval() | |
| rows = [json.loads(l) for l in (DATA / "test.jsonl").read_text( | |
| encoding="utf-8").splitlines() if l.strip()] | |
| print(f"[test] {len(rows)} rows", flush=True) | |
| preds = [] | |
| with torch.no_grad(): | |
| for i in range(0, len(rows), 16): | |
| batch = [r["src"].strip() for r in rows[i:i+16]] | |
| enc = tok(batch, return_tensors="pt", padding=True, | |
| truncation=True, max_length=256).to("cuda") | |
| with torch.autocast("cuda", torch.bfloat16): | |
| gen = model.generate(**enc, max_new_tokens=256, num_beams=1) | |
| preds.extend(tok.batch_decode(gen, skip_special_tokens=True)) | |
| if (i // 16) % 100 == 0: | |
| print(f"[gen] {i + 16}/{len(rows)}", flush=True) | |
| cats = {} | |
| raw_cers, norm_cers = [], [] | |
| for r, pred in zip(rows, preds): | |
| src, gold = r["src"].strip(), r["tgt"].strip() | |
| def _lev(a, b): | |
| prev = list(range(len(b) + 1)) | |
| for i, ca in enumerate(a, 1): | |
| cur = [i] | |
| for j, cb in enumerate(b, 1): | |
| cur.append(min(prev[j] + 1, cur[j-1] + 1, | |
| prev[j-1] + (ca != cb))) | |
| prev = cur | |
| return prev[-1] | |
| raw_cers.append(_lev(pred, gold) / max(1, len(gold))) | |
| ng, npd = norm(gold), norm(pred) | |
| norm_cers.append(_lev(npd, ng) / max(1, len(ng))) | |
| ops = lev_align(norm(gold), norm(pred)) | |
| for op, rc, pc in ops: | |
| if op == "match": | |
| continue | |
| cat = classify(rc) if op in ("sub", "del") else "insertion" | |
| c = cats.setdefault(cat, Counter()) | |
| c[op] += 1 | |
| import numpy as np # noqa: PLC0415 | |
| res = { | |
| "n": len(rows), | |
| "cer_raw": round(sum(raw_cers) / len(raw_cers), 4), | |
| "cer_normalized": round(sum(norm_cers) / len(norm_cers), 4), | |
| "error_ops_by_class": {k: dict(v) for k, v in sorted( | |
| cats.items(), key=lambda kv: -sum(kv[1].values()))}, | |
| } | |
| total_err = sum(sum(v.values()) for v in cats.values()) | |
| res["error_share_by_class"] = { | |
| k: round(sum(v.values()) / total_err, 3) | |
| for k, v in res["error_ops_by_class"].items()} | |
| (CKPT / "anatomy.json").write_text(json.dumps(res, indent=1)) | |
| print(json.dumps(res, indent=1), flush=True) | |
| (CKPT / "ANATOMY_DONE").write_text("ok\n") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 4.74 kB
- Xet hash:
- 45ef7e5463c6c8cf21cf0a8b95088fd6a0879d55128bb7d2575a6c4694c2f8d1
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.