Buckets:
| """Build data/train_pool.jsonl, eval.jsonl, calib.jsonl and summary.json from the Hub. | |
| Every exclusion is a whole item, counted in summary.json. Nothing is truncated. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import collections | |
| import hashlib | |
| import json | |
| import os | |
| import random | |
| import pandas as pd | |
| from huggingface_hub import hf_hub_download, list_repo_files | |
| from transformers import AutoTokenizer | |
| from common import CONTEXT, MAX_OPTIONS, MAX_PROMPT_TOKENS, render, write_jsonl | |
| PROC = "tasksource/procedural-typed-decisions" | |
| TS = "tasksource/tasksource-jev-typed-decisions" | |
| OD = "jgeuter/OpenDecisions" | |
| LL = "LocalLLaMA/typed-decisions" | |
| SEED = 17 | |
| def fetch(repo: str, pattern_prefix: str, raw: str) -> list[str]: | |
| files = [f for f in list_repo_files(repo, repo_type="dataset") if f.startswith(pattern_prefix) and f.endswith(".parquet")] | |
| return [hf_hub_download(repo, f, repo_type="dataset", local_dir=os.path.join(raw, repo.replace("/", "__"))) | |
| for f in sorted(files)] | |
| def finish(item: dict, counts: collections.Counter, tag: str) -> dict | None: | |
| """Validate option count and gold; None (and a count) for an excluded item.""" | |
| k = len(item["options"]) | |
| if k < 2 or k > MAX_OPTIONS: | |
| counts[f"{tag}:options_out_of_range"] += 1 | |
| return None | |
| t = [float(x) for x in item["target"]] | |
| s = sum(t) | |
| if s <= 0: | |
| counts[f"{tag}:empty_target"] += 1 | |
| return None | |
| t = [x / s for x in t] | |
| top = sorted(t, reverse=True) | |
| if abs(top[0] - top[1]) < 1e-9: | |
| counts[f"{tag}:gold_tie"] += 1 | |
| return None | |
| item["target"] = t | |
| item["gold"] = max(range(k), key=lambda i: t[i]) | |
| return item | |
| def noul_options(criteria) -> list[dict]: | |
| crit = criteria if isinstance(criteria, dict) else {} | |
| return [{"id": "true", "desc": crit.get("true") or "The proposition is true."}, | |
| {"id": "false", "desc": crit.get("false") or "The proposition is false."}] | |
| def typed_question(qid: str, q: dict, ans: dict) -> tuple[list[dict], list[float]]: | |
| """Options and target of one System One question (procedural / LocalLLaMA shape).""" | |
| qt, crit = q["type"], q.get("criteria") | |
| if qt == "noul": | |
| p = float(ans["noul"]) if "noul" in ans else float(ans["probabilities"]["true"]) | |
| return noul_options(crit), [p, 1 - p] | |
| probs = ans.get("probabilities") or {} | |
| if qt == "choice": | |
| if isinstance(crit, dict): | |
| opts = [{"id": k, "desc": v} for k, v in crit.items()] | |
| else: | |
| opts = [{"id": str(k), "desc": None} for k in crit] | |
| return opts, [float(probs.get(o["id"], 0.0)) for o in opts] | |
| # score: ordered levels, ids 0..K-1 | |
| if isinstance(crit, dict): | |
| texts = list(crit.values()) | |
| else: | |
| texts = list(crit) | |
| opts = [{"id": str(i), "desc": t} for i, t in enumerate(texts)] | |
| return opts, [float(probs.get(str(i), 0.0)) for i in range(len(opts))] | |
| def procedural(paths: list[str], set_name: str, counts, per_family: int | None, per_level: int | None, rng) -> list[dict]: | |
| df = pd.concat([pd.read_parquet(p) for p in paths], ignore_index=True) | |
| out = [] | |
| for task, g in df.groupby("task", sort=True): | |
| g = g.sample(frac=1.0, random_state=rng.randrange(1 << 30)) | |
| picked, by_level = [], collections.Counter() | |
| for _, row in g.iterrows(): | |
| qs, ans = json.loads(row["questions"]), json.loads(row["answers"]) | |
| names = list(qs) | |
| rng.shuffle(names) | |
| for name in names: # first valid question of this state | |
| opts, target = typed_question(name, qs[name], ans[name]) | |
| item = finish({"id": f"proc:{row['id']}:{name}", "set": set_name, "family": task, | |
| "level": int(row["level"]), "tier": None, "qtype": qs[name]["type"], | |
| "question_name": name, "evidence": row["state"], | |
| "criterion": qs[name]["instructions"], "options": opts, "target": target}, | |
| counts, set_name) | |
| if item is None: | |
| continue | |
| if per_level and by_level[item["level"]] >= per_level: | |
| break | |
| picked.append(item) | |
| by_level[item["level"]] += 1 | |
| break | |
| if per_family and len(picked) >= per_family: | |
| break | |
| out += picked | |
| return out | |
| def tasksource(paths: list[str], set_name: str, counts, n: int, per_source: int, rng) -> list[dict]: | |
| df = pd.concat([pd.read_parquet(p, columns=["id", "state", "kind", "options", "target", "question", "source"]) | |
| for p in paths], ignore_index=True) | |
| proc = df["source"].str.startswith("procedural-typed-decisions/") | |
| counts[f"{set_name}:procedural_copies_removed"] += int(proc.sum()) | |
| df = df[~proc].sample(frac=1.0, random_state=rng.randrange(1 << 30)) | |
| out, per = [], collections.Counter() | |
| for row in df.itertuples(index=False): | |
| if per[row.source] >= per_source: | |
| continue | |
| if row.kind == "noul": | |
| p = float(row.target[0]) | |
| opts, target = noul_options(None), [p, 1 - p] | |
| else: | |
| opts = ([{"id": str(i), "desc": o} for i, o in enumerate(row.options)] if row.kind == "score" | |
| else [{"id": None, "desc": o} for o in row.options]) | |
| target = list(row.target) | |
| item = finish({"id": f"ts:{row.id}", "set": set_name, "family": row.source, "level": None, "tier": None, | |
| "qtype": row.kind, "evidence": row.state, "criterion": row.question, | |
| "options": opts, "target": target}, counts, set_name) | |
| if item is None: | |
| continue | |
| out.append(item) | |
| per[row.source] += 1 | |
| if len(out) >= n: | |
| break | |
| return out | |
| def opendecisions(path: str, counts) -> list[dict]: | |
| out = [] | |
| with open(path, encoding="utf-8") as f: | |
| for line in f: | |
| it = json.loads(line) | |
| q, labels = it["question"], it["labels"] | |
| state = it["state"] if isinstance(it["state"], str) else json.dumps(it["state"], ensure_ascii=False) | |
| tp = it.get("target_probs") or None | |
| if q["type"] == "noul": | |
| opts = noul_options(q.get("criteria")) | |
| labmap = ["yes", "no"] # option 0 = true = yes | |
| target = [float(tp["yes"]), float(tp["no"])] if tp else [float(it["expected"] == "yes"), float(it["expected"] == "no")] | |
| elif q["type"] == "choice": | |
| crit = q["criteria"] | |
| opts = ([{"id": k, "desc": v or None} for k, v in crit.items()] if isinstance(crit, dict) | |
| else [{"id": str(k), "desc": None} for k in crit]) | |
| labmap = [o["id"] for o in opts] | |
| target = [float(tp.get(l, 0.0)) for l in labmap] if tp else [float(l == it["expected"]) for l in labmap] | |
| else: | |
| opts = [{"id": str(i), "desc": t} for i, t in enumerate(q["criteria"])] | |
| labmap = [str(i) for i in range(len(opts))] | |
| target = [float(tp.get(l, 0.0)) for l in labmap] if tp else [float(l == str(it["expected"])) for l in labmap] | |
| item = finish({"id": f"od:{it['id']}", "set": "opendecisions", "family": it["family"], "level": None, | |
| "tier": it["tier"], "qtype": q["type"], "evidence": state, "criterion": q["instructions"], | |
| "options": opts, "target": target, "od_id": it["id"], "od_labels": labmap, | |
| "language": it.get("language")}, counts, "opendecisions") | |
| if item is not None and tp: # gold is the dataset's expected label | |
| exp = it["expected"] if q["type"] != "noul" else ("yes" if it["expected"] == "yes" else "no") | |
| item["gold"] = labmap.index(str(exp)) | |
| if item is not None: | |
| out.append(item) | |
| return out | |
| def localllama(path: str, counts) -> list[dict]: | |
| df = pd.read_parquet(path) | |
| out = [] | |
| for row in df.itertuples(index=False): | |
| qs, gold, agree = json.loads(row.questions), json.loads(row.gold), json.loads(row.label_agreement) | |
| for name, q in qs.items(): | |
| opts, target = typed_question(name, q, gold[name]) | |
| item = finish({"id": f"ll:{row.id}:{name}", "set": "localllama", "family": row.workflow, "level": None, | |
| "tier": None, "qtype": q["type"], "question_name": name, "evidence": row.state, | |
| "criterion": q["instructions"], "options": opts, "target": target, | |
| "panel_agrees": bool(agree.get(name, {}).get("argmax_agree", False))}, | |
| counts, "localllama") | |
| if item is not None: | |
| out.append(item) | |
| return out | |
| def key(item: dict) -> str: | |
| return hashlib.sha1((item["evidence"] + "\x00" + item["criterion"]).encode()).hexdigest() | |
| def main() -> None: | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--model", default="Qwen/Qwen3.5-4B") | |
| ap.add_argument("--out", default="data") | |
| ap.add_argument("--raw", default="data/raw") | |
| ap.add_argument("--proc-train-per-family", type=int, default=500) | |
| ap.add_argument("--ts-train", type=int, default=9000) | |
| ap.add_argument("--ts-train-per-source", type=int, default=30) | |
| ap.add_argument("--small", action="store_true", help="tiny sets for the preflight") | |
| args = ap.parse_args() | |
| os.makedirs(args.out, exist_ok=True) | |
| rng = random.Random(SEED) | |
| counts = collections.Counter() | |
| small = args.small | |
| proc_train = fetch(PROC, "all/train", args.raw) | |
| proc_val = fetch(PROC, "all/validation", args.raw) | |
| proc_test = fetch(PROC, "all/test", args.raw) | |
| ts_train = fetch(TS, "data/train", args.raw) | |
| ts_val = fetch(TS, "data/validation", args.raw) | |
| ts_test = fetch(TS, "data/test", args.raw) | |
| od_path = hf_hub_download(OD, "opendecisions.jsonl", repo_type="dataset", local_dir=os.path.join(args.raw, OD.replace("/", "__"))) | |
| ll_path = fetch(LL, "all/test", args.raw)[0] | |
| pool = procedural(proc_train, "train_proc", counts, 8 if small else args.proc_train_per_family, None, rng) | |
| pool += tasksource(ts_train, "train_ts", counts, 64 if small else args.ts_train, args.ts_train_per_source, rng) | |
| ev = procedural(proc_test, "eval_proc", counts, 2 if small else 30, 1 if small else 6, rng) | |
| ev += tasksource(ts_test, "eval_ts", counts, 16 if small else 600, 4, rng) | |
| od = opendecisions(od_path, counts) | |
| ll = localllama(ll_path, counts) | |
| ev += od[:16] if small else od | |
| ev += ll[:10] if small else ll | |
| calib = procedural(proc_val, "calib_proc", counts, 2 if small else 12, None, rng) | |
| calib += tasksource(ts_val, "calib_ts", counts, 16 if small else 300, 2, rng) | |
| # leakage: drop train items whose evidence+criterion appears in any eval or calibration item; unique ids | |
| held = {key(x) for x in ev + calib} | |
| before = len(pool) | |
| pool = [x for x in pool if key(x) not in held] | |
| counts["train:overlap_with_eval_removed"] += before - len(pool) | |
| seen, dedup = set(), [] | |
| for name, rows in (("eval", ev), ("calib", calib), ("train", pool)): | |
| kept = [] | |
| for x in rows: | |
| if x["id"] in seen: | |
| counts[f"{name}:duplicate_id"] += 1 | |
| continue | |
| seen.add(x["id"]) | |
| kept.append(x) | |
| dedup.append(kept) | |
| ev, calib, pool = dedup | |
| # whole-item length exclusion on the thinking prompt (the longer of the two renders) | |
| tok = AutoTokenizer.from_pretrained(args.model) | |
| def fits(rows, name): | |
| texts = [render(tok, x, think=False) for x in rows] | |
| lens = [len(t) for t in tok(texts, add_special_tokens=False)["input_ids"]] | |
| kept = [] | |
| for x, n in zip(rows, lens): | |
| if n > MAX_PROMPT_TOKENS: | |
| counts[f"{name}:prompt_over_{MAX_PROMPT_TOKENS}_tokens"] += 1 | |
| continue | |
| x["prompt_tokens"] = n | |
| kept.append(x) | |
| return kept | |
| pool, ev, calib = fits(pool, "train"), fits(ev, "eval"), fits(calib, "calib") | |
| rng.shuffle(pool) # the fixed order generation walks | |
| write_jsonl(os.path.join(args.out, "train_pool.jsonl"), pool) | |
| write_jsonl(os.path.join(args.out, "eval.jsonl"), ev) | |
| write_jsonl(os.path.join(args.out, "calib.jsonl"), calib) | |
| by = lambda rows, f: dict(collections.Counter(f(x) for x in rows)) | |
| summary = {"model_tokenizer": args.model, "context": CONTEXT, "max_prompt_tokens": MAX_PROMPT_TOKENS, | |
| "train_pool": len(pool), "eval": len(ev), "calib": len(calib), | |
| "train_by_set": by(pool, lambda x: x["set"]), "eval_by_set": by(ev, lambda x: x["set"]), | |
| "calib_by_set": by(calib, lambda x: x["set"]), | |
| "train_by_type": by(pool, lambda x: x["qtype"]), "eval_by_type": by(ev, lambda x: x["qtype"]), | |
| "exclusions": dict(sorted(counts.items())), | |
| "prompt_tokens_max": {"train": max(x["prompt_tokens"] for x in pool), | |
| "eval": max(x["prompt_tokens"] for x in ev)}} | |
| with open(os.path.join(args.out, "summary.json"), "w") as f: | |
| json.dump(summary, f, indent=1) | |
| print(json.dumps(summary, indent=1)) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 13.5 kB
- Xet hash:
- 72e685d3e95a08828a01d7a6447b7f1398ad72961fe091668bc9bbfb06ee142b
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.