| """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 |
|
|
| 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")) |
| |
| |
| |
| 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(): |
| keep = set(ids_file.read_text().split()) |
| else: |
| 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() |
|
|
