File size: 6,636 Bytes
ec24fb0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | """Normalise raw eval data into one MCQ schema under data/processed/.
Schema (one JSON object per line):
uid, dataset, q_no, subject, category, level, question, options[list[str]], answer (1-based int), meta{}
Outputs:
sinhalammlu_dev.jsonl public Easy tier (dev)
gmmlu_si_val.jsonl fixed held-out sample of Global-MMLU si test (harder internal validation);
ids in configs/splits/ so they are excluded from any later training mix
sinhalammlu_dev_5shot.jsonl fixed 5 few-shot examples drawn from the dev pool (ids frozen in configs/splits)
gmmlu_si_dev.jsonl Global-MMLU si dev split (5/subject), candidate pool for fixed few-shot examples
belebele_si.jsonl Belebele sin_Sinh, EVALUATION ONLY
"""
from __future__ import annotations
import argparse
import json
import random
import sys
from collections import Counter, defaultdict
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from eval.common import ROOT, read_jsonl, write_jsonl # noqa: E402
RAW = ROOT / "data" / "raw"
OUT = ROOT / "data" / "processed"
SPLITS = ROOT / "configs" / "splits"
LETTERS = "ABCD"
def prep_sinhalammlu() -> list[dict]:
rows, dropped = [], []
for f in sorted((RAW / "sinhalammlu" / "TEST").rglob("*.json")):
recs = json.loads(f.read_text(encoding="utf-8"))
# A few records spell their subject differently from the rest of the file
# ("Eastern_music" vs "Eastern music", trailing spaces): group by the file's majority spelling,
# keep the original in subject_raw because Codabench predictions must echo it.
canonical = Counter(r["subject"].strip() for r in recs).most_common(1)[0][0]
for r in recs:
level = r["metadata"].get("difficulty", f.parent.name)
uid = f"smmlu/{level}/{r['subject']}/{r['q_no']}"
if not (isinstance(r["answer"], int) and 1 <= r["answer"] <= len(r["choices"])):
dropped.append((uid, r["answer"]))
continue
rows.append(dict(
uid=uid, dataset="sinhalammlu", q_no=r["q_no"], subject=canonical, subject_raw=r["subject"],
category=r["category"], level=level, question=r["question"].strip(),
options=[str(c).strip() for c in r["choices"]], answer=r["answer"],
meta={k: r["metadata"].get(k) for k in ("grade", "year", "type", "subject_original")},
))
for uid, a in dropped:
print(f" dropped {uid}: invalid answer {a!r}")
return rows
def select_fewshot(rows: list[dict], k: int, seed: int) -> list[dict]:
"""Fixed few-shot set from the dev pool: one question per domain, gold answers spread over 1..4.
Chosen at random (not by inspection) and frozen in configs/splits; excluded from evaluation by the harness."""
ids_file = SPLITS / f"sinhalammlu_dev_{k}shot_ids.txt"
by_uid = {r["uid"]: r for r in rows}
if ids_file.exists():
return [by_uid[u] for u in ids_file.read_text().split()]
rng = random.Random(seed)
pool = sorted(rows, key=lambda r: r["uid"])
rng.shuffle(pool)
picked, answers = [], []
for cat in sorted({r["category"] for r in rows}):
r = next(r for r in pool if r["category"] == cat and r["answer"] not in answers)
picked.append(r)
answers.append(r["answer"])
for r in pool:
if len(picked) >= k:
break
if r not in picked and (len(set(answers)) == 4 or r["answer"] not in answers):
picked.append(r)
answers.append(r["answer"])
rng.shuffle(picked)
ids_file.write_text("\n".join(r["uid"] for r in picked) + "\n")
return picked
def _gmmlu_rows(df, split: str) -> list[dict]:
rows = []
for r in df.itertuples():
rows.append(dict(
uid=f"gmmlu_si/{split}/{r.sample_id}", dataset="gmmlu_si", q_no=r.sample_id,
subject=r.subject, category=r.subject_category, level="gmmlu",
question=r.question.strip(),
options=[str(getattr(r, f"option_{c}")).strip() for c in "abcd"],
answer=LETTERS.index(r.answer) + 1,
meta={"cultural_sensitivity": r.cultural_sensitivity_label},
))
return rows
def prep_gmmlu(val_size: int, seed: int) -> tuple[list[dict], list[dict]]:
import pandas as pd
d = RAW / "global_mmlu" / "si"
test = _gmmlu_rows(pd.read_parquet(d / "test-00000-of-00001.parquet"), "test")
dev = _gmmlu_rows(pd.read_parquet(d / "dev-00000-of-00001.parquet"), "dev")
ids_file = SPLITS / "gmmlu_si_val_ids.txt"
if ids_file.exists(): # split is frozen once created
keep = set(ids_file.read_text().split())
else: # proportional stratified sample by subject
by_subj = defaultdict(list)
for r in test:
by_subj[r["subject"]].append(r["uid"])
rng = random.Random(seed)
keep = set()
for subj in sorted(by_subj):
uids = sorted(by_subj[subj])
k = max(1, round(val_size * len(uids) / len(test)))
keep.update(rng.sample(uids, min(k, len(uids))))
SPLITS.mkdir(parents=True, exist_ok=True)
ids_file.write_text("\n".join(sorted(keep)) + "\n")
return [r for r in test if r["uid"] in keep], dev
def prep_belebele() -> list[dict]:
rows = []
for r in read_jsonl(RAW / "belebele" / "data" / "sin_Sinh.jsonl"):
rows.append(dict(
uid=f"belebele_si/{r['link']}#{r['question_number']}", dataset="belebele_si",
q_no=r["question_number"], subject="reading_comprehension", category="reading",
level="belebele", question=f"{r['flores_passage'].strip()}\n\n{r['question'].strip()}",
options=[r[f"mc_answer{i}"].strip() for i in range(1, 5)],
answer=int(r["correct_answer_num"]), meta={},
))
return rows
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--gmmlu-val-size", type=int, default=2000)
ap.add_argument("--seed", type=int, default=1234)
args = ap.parse_args()
outputs = {"sinhalammlu_dev": prep_sinhalammlu()}
outputs["sinhalammlu_dev_5shot"] = select_fewshot(outputs["sinhalammlu_dev"], 5, args.seed)
outputs["gmmlu_si_val"], outputs["gmmlu_si_dev"] = prep_gmmlu(args.gmmlu_val_size, args.seed)
outputs["belebele_si"] = prep_belebele()
for name, rows in outputs.items():
write_jsonl(OUT / f"{name}.jsonl", rows)
ans = Counter(r["answer"] for r in rows)
print(f"{name:18s} n={len(rows):5d} answers={dict(sorted(ans.items()))}")
if __name__ == "__main__":
main()
|