| """Mentor pipeline: hinted re-solves and the training set built from the student's own work.
|
|
|
| hinted writes the curriculum tasks with my advice for the tasks where the student failed or was weak:
|
| mentor_build.py hinted --hints data/visual/mentor_hints.json --out data/visual/mentor_hinted.jsonl
|
| sft builds token-level examples (prompt WITHOUT the advice -> the student's own thinking + answer):
|
| mentor_build.py sft --hinted-tag mtr_hint_dm --base-tag mtr_dm --out data/mentor/sft_mentor.jsonl
|
| --rft-only keeps only the student's unhinted good attempts (the control arm: self-training without me).
|
|
|
| A hinted attempt is kept only if it has an answer (ended by itself or read out at the budget), renders cleanly, does not mention the advice,
|
| scores higher with the vision model (VQAScore of the task text) than the student's best unhinted attempt, and I
|
| approved its screenshot (data/visual/mentor_approved.json, written after looking at the review sheets).
|
| """
|
| import argparse
|
| import collections
|
| import json
|
| import os
|
| import re
|
|
|
| ADVICE = "Advice from an experienced developer: "
|
| MENTION = re.compile(r"\b(advice|advised|advises|mentor|hint|experienced developer)\b", re.I)
|
| EOS, THINK_END = 124900, 124902
|
|
|
| ap = argparse.ArgumentParser()
|
| ap.add_argument("cmd", choices=["hinted", "sft"])
|
| ap.add_argument("--tasks", default="data/visual/mentor_tasks.jsonl")
|
| ap.add_argument("--hints", default="data/visual/mentor_hints.json")
|
| ap.add_argument("--approved", default="data/visual/mentor_approved.json")
|
| ap.add_argument("--hinted-tag", default="mtr_hint_dm")
|
| ap.add_argument("--base-tag", default="mtr_dm", help="the student's unhinted attempts (after the read-out module)")
|
| ap.add_argument("--per-task", type=int, default=2)
|
| ap.add_argument("--rft-only", action="store_true")
|
| ap.add_argument("--min-vqa", type=float, default=0.5, help="unhinted attempts: minimum VQAScore to count as good")
|
| ap.add_argument("--max-len", type=int, default=14000)
|
| ap.add_argument("--out", required=True)
|
| a = ap.parse_args()
|
| tasks = {json.loads(l)["id"]: json.loads(l) for l in open(a.tasks, encoding="utf-8")}
|
|
|
|
|
| def split_prompt(p):
|
| i = p.index("\n\nAnswer with")
|
| return p[:i], p[i:]
|
|
|
|
|
| if a.cmd == "hinted":
|
| hints = json.load(open(a.hints, encoding="utf-8"))
|
| os.makedirs(os.path.dirname(a.out), exist_ok=True)
|
| with open(a.out, "w", encoding="utf-8") as f:
|
| for tid, t in tasks.items():
|
| h = hints.get(tid.split("/")[-1])
|
| if h:
|
| body, tail = split_prompt(t["prompt"])
|
| f.write(json.dumps({**t, "prompt": body + "\n\n" + ADVICE + h + tail}, ensure_ascii=False) + "\n")
|
| print(f"{sum(1 for _ in open(a.out, encoding='utf-8'))} hinted tasks -> {a.out}")
|
| raise SystemExit
|
|
|
| from transformers import AutoTokenizer
|
|
|
| tok = AutoTokenizer.from_pretrained("models/LFM3-1.7B")
|
| approved = set(json.load(open(a.approved))) if os.path.exists(a.approved) else None
|
|
|
|
|
| def load(tag):
|
| recs = {}
|
| for p in os.listdir(os.path.join("results", tag)):
|
| if p.endswith(".jsonl"):
|
| for l in open(os.path.join("results", tag, p), encoding="utf-8"):
|
| r = json.loads(l)
|
| recs[(r["id"].split("/")[-1], r["sample"])] = r
|
| rep = {(r["id"].split("/")[-1], r["sample"]): r for r in json.load(open(os.path.join("results", tag, "visual.json")))}
|
| vqa_p = os.path.join("results", tag, "vqa.json")
|
| vqa = json.load(open(vqa_p)) if os.path.exists(vqa_p) else {}
|
| return recs, rep, vqa
|
|
|
|
|
| def clean(r, task):
|
| return (r is not None and not r.get("parse_error") and not r["errors"] and not r["failed"] and not r["blank"]
|
| and (r["animated"] or not task["animated"]))
|
|
|
|
|
| def completion_ids(rec):
|
| """Generated token ids (the prompt ends with <think>); records without saved tokens are re-tokenized."""
|
| if rec.get("tokens") and rec.get("module", "none") == "none":
|
| return [x for x in rec["tokens"] if x != EOS]
|
| return tok(rec["completion"][len("<think>"):], add_special_tokens=False)["input_ids"]
|
|
|
|
|
| def usable(rec):
|
| """Ended by itself (the thinking closed and the answer finished) and does not mention the advice."""
|
| return ("</think>" in rec["completion"] and not rec.get("truncated")
|
| and "render-repaired" not in rec["completion"] and not MENTION.search(rec["completion"]))
|
|
|
|
|
| base, base_rep, base_vqa = load(a.base_tag)
|
| hint, hint_rep, hint_vqa = ({}, {}, {}) if a.rft_only else load(a.hinted_tag)
|
| out, stats = [], collections.Counter()
|
| for tid, t in tasks.items():
|
| name = tid.split("/")[-1]
|
| p_ids = tok.apply_chat_template([{"role": "user", "content": t["prompt"]}], tokenize=True,
|
| add_generation_prompt=True, return_dict=False)
|
| own = [(s, base_vqa.get(f"{name}_{s}", 0.0)) for (n, s) in base if n == name]
|
| own_clean = [(s, v) for s, v in own if clean(base_rep.get((name, s)), t)]
|
| best_own = max([v for _, v in own_clean], default=0.0)
|
| picks = []
|
| if not a.rft_only:
|
| cands = [(s, hint_vqa.get(f"{name}_{s}", 0.0)) for (n, s) in hint if n == name]
|
| for s, v in sorted(cands, key=lambda x: -x[1]):
|
| rec = hint[(name, s)]
|
| if not clean(hint_rep.get((name, s)), t):
|
| stats["hinted: not clean"] += 1
|
| elif not usable(rec):
|
| stats["hinted: cut or mentions advice"] += 1
|
| elif v <= best_own:
|
| stats["hinted: not better than own best"] += 1
|
| elif approved is not None and f"hint:{name}_{s}" not in approved:
|
| stats["hinted: not approved"] += 1
|
| else:
|
| picks.append(("hinted", rec))
|
| if not picks:
|
| for s, v in sorted(own_clean, key=lambda x: -x[1]):
|
| rec = base[(name, s)]
|
| if v >= a.min_vqa and usable(rec) and (approved is None or f"own:{name}_{s}" in approved):
|
| picks.append(("own", rec))
|
| for kind, rec in picks[: a.per_task]:
|
| comp = completion_ids(rec) + [EOS]
|
| if len(p_ids) + len(comp) > a.max_len:
|
| stats["too long"] += 1
|
| continue
|
| out.append({"id": tid, "task": "visual", "kind": kind, "sample": rec["sample"], "prompt_ids": p_ids,
|
| "completion_ids": comp})
|
| stats[f"examples {kind} ({t['cat']})"] += 1
|
| os.makedirs(os.path.dirname(a.out), exist_ok=True)
|
| with open(a.out, "w", encoding="utf-8") as f:
|
| for r in out:
|
| f.write(json.dumps(r) + "\n")
|
| print(f"{len(out)} examples from {len({r['id'] for r in out})} tasks -> {a.out}")
|
| for k, v in sorted(stats.items()):
|
| print(f" {k}: {v}")
|
|
|
