Buckets:
| """Procedural testbed (E6): tasksource/procedural-typed-decisions -> S1 canonical jsonl (shared_yesno_data_v1). | |
| One question per state (seeded random choice), stratified over 18 task families x 5 levels. | |
| train: from the train split; dev: from validation; test: from test (held out; never trained on). | |
| """ | |
| import glob | |
| import json | |
| import os | |
| import random | |
| from collections import Counter | |
| import pandas as pd | |
| SRC = "/workspace/data/procedural/all" | |
| OUT = "/workspace/data/procedural/canonical" | |
| def parse_state(s): | |
| if not isinstance(s, str): | |
| return s | |
| try: | |
| return json.loads(s) | |
| except json.JSONDecodeError: | |
| return s # plain-text states stay text (S1's serialize_state handles both) | |
| def to_row(r, qid, q, ans, split): | |
| t = q["type"] | |
| crit = q.get("criteria") | |
| if t == "noul": | |
| p = float(ans["noul"]) if "noul" in ans else float(ans["probabilities"]["true"]) | |
| probs, keys = {"false": 1 - p, "true": p}, ["false", "true"] | |
| elif t == "choice": | |
| keys = list(crit) | |
| probs = {k: float((ans.get("probabilities") or {}).get(k, 0.0)) for k in keys} | |
| if sum(probs.values()) == 0 and "choice" in ans: | |
| probs[ans["choice"]] = 1.0 | |
| else: | |
| keys = [str(i) for i in range(len(crit))] | |
| pr = ans.get("probabilities") | |
| probs = {k: float(pr.get(k, 0.0)) for k in keys} if pr else {k: float(int(round(ans["score"])) == i) | |
| for i, k in enumerate(keys)} | |
| tot = sum(probs.values()) | |
| if tot <= 0: | |
| return None | |
| probs = {k: v / tot for k, v in probs.items()} | |
| hard = sorted(probs.values())[-1] == 1.0 | |
| qd = {k: q[k] for k in ("type", "instructions", "criteria") if k in q} | |
| return dict(schema_version="shared_yesno_data_v1", split=split, case_id="%s:%s" % (r["id"], qid), | |
| group_id=str(r["id"]), view_id="0", source=dict(name="proc/%s" % r["task"]), | |
| level=int(r["level"]), state=parse_state(r["state"]), | |
| questions={qid: qd}, gold={qid: dict(probabilities=probs, kind="hard" if hard else "known_distribution")}, | |
| option_ids={qid: keys}) | |
| def build(split, per_cell, seed): | |
| df = pd.concat([pd.read_parquet(p) for p in sorted(glob.glob("%s/%s-*.parquet" % (SRC, split)))], ignore_index=True) | |
| rng = random.Random(seed) | |
| rows, skipped = [], Counter() | |
| for (task, level), g in df.groupby(["task", "level"], sort=True): | |
| idx = list(g.index) | |
| rng.shuffle(idx) | |
| n = 0 | |
| for i in idx: | |
| if n >= per_cell: | |
| break | |
| r = df.loc[i] | |
| qs, ans = json.loads(r["questions"]), json.loads(r["answers"]) | |
| names = sorted(qs) | |
| rng.shuffle(names) | |
| for qid in names: | |
| row = to_row(r, qid, qs[qid], ans[qid], "train" if split == "train" else ("dev" if split == "validation" else "test")) | |
| if row is not None: | |
| rows.append(row) | |
| n += 1 | |
| break | |
| skipped["no_target"] += 1 | |
| rng.shuffle(rows) | |
| return rows, skipped | |
| def main(): | |
| os.makedirs(OUT, exist_ok=True) | |
| for split, per_cell, name in (("train", 889, "train"), ("validation", 20, "dev"), ("test", 40, "test")): | |
| rows, skipped = build(split, per_cell, 42) | |
| with open("%s/%s.jsonl" % (OUT, name), "w") as f: | |
| for r in rows: | |
| f.write(json.dumps(r) + "\n") | |
| print(name, len(rows), dict(skipped), Counter(r["questions"][next(iter(r["questions"]))]["type"] for r in rows)) | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 3.67 kB
- Xet hash:
- d51269fe1ac2d17f87615d3a9fe77bea5c0fa8b0486fe410ef8d238356a801c6
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.