""" export_distillation.py — turn captured teacher examples into student-ready JSONL. Offline/admin tool. Reads the `distillation_examples` Firestore collection (written by distill.py) and emits one JSONL file per task under an output directory: intent_parse.jsonl / translate.jsonl / codegen.jsonl / coach.jsonl / insight.jsonl → chat-format SFT {"messages":[{system},{user},{assistant}]} vision.jsonl → {"image_uri", "instruction", "output"} asr.jsonl → {"audio_uri", "transcript"} rejected.jsonl → hard negatives (verdict=rejected), for DPO / filtering This is sequence-level (response-based) distillation: targets are the teacher's text / JSON outputs and the human-verified gold label — Gemini exposes no logits. Usage: python scripts/export_distillation.py --out ./dataset \ --verdicts confirmed corrected pending (defaults: SFT uses confirmed+corrected; pass --include-pending to add unlabeled) Env: FIREBASE (service-account JSON), as used by the app. """ import argparse import json import os from collections import Counter, defaultdict from typing import Any, Dict, List, Optional def _init_db(): import firebase_admin from firebase_admin import credentials, firestore if not firebase_admin._apps: sa = json.loads(os.environ["FIREBASE"]) cred = credentials.Certificate(sa) opts = {} bucket = os.environ.get("FIREBASE_STORAGE_BUCKET") if bucket: opts["storageBucket"] = bucket firebase_admin.initialize_app(cred, opts) return firestore.client() # Tasks that train as text chat SFT. _TEXT_TASKS = {"intent_parse", "translate", "codegen", "coach", "insight"} def _gold_or_teacher(ex: Dict[str, Any]) -> Optional[str]: """Prefer the human-corrected gold output; fall back to the teacher output.""" return ex.get("gold_output") or ex.get("output_raw") def _to_chat(ex: Dict[str, Any]) -> Optional[Dict[str, Any]]: inp = ex.get("input") or {} user = inp.get("text") or inp.get("transcript") or "" system = inp.get("system_prompt") or "" target = _gold_or_teacher(ex) if not user or not target: return None messages = [] if system: messages.append({"role": "system", "content": system}) messages.append({"role": "user", "content": user}) messages.append({"role": "assistant", "content": target}) return {"messages": messages, "task": ex.get("task"), "teacher_model": ex.get("teacher_model"), "prompt_version": ex.get("prompt_version"), "verdict": ex.get("verdict")} def _to_vision(ex: Dict[str, Any]) -> Optional[Dict[str, Any]]: inp = ex.get("input") or {} uri = inp.get("media_uri") target = _gold_or_teacher(ex) if not uri or not target: return None return {"image_uri": uri, "instruction": inp.get("caption") or "Extract the bookkeeping transactions as JSON.", "output": target, "verdict": ex.get("verdict"), "teacher_model": ex.get("teacher_model")} def _to_asr(ex: Dict[str, Any]) -> Optional[Dict[str, Any]]: inp = ex.get("input") or {} uri = inp.get("media_uri") transcript = ex.get("output_raw") if not uri or not transcript: return None return {"audio_uri": uri, "transcript": transcript, "teacher_model": ex.get("teacher_model")} def main(): ap = argparse.ArgumentParser() ap.add_argument("--out", default="./dataset", help="output directory") ap.add_argument("--verdicts", nargs="*", default=["confirmed", "corrected"], help="verdicts to include for SFT (default: confirmed corrected)") ap.add_argument("--include-pending", action="store_true", help="also include unlabeled (pending) examples in SFT") ap.add_argument("--limit", type=int, default=0, help="max docs (0 = all)") args = ap.parse_args() sft_verdicts = set(args.verdicts) if args.include_pending: sft_verdicts.add("pending") db = _init_db() os.makedirs(args.out, exist_ok=True) writers: Dict[str, Any] = {} def _w(name: str): if name not in writers: writers[name] = open(os.path.join(args.out, f"{name}.jsonl"), "w", encoding="utf-8") return writers[name] counts: Counter = Counter() # Dedup identical (user,target) pairs per task. seen: Dict[str, set] = defaultdict(set) q = db.collection("distillation_examples") docs = q.limit(args.limit).stream() if args.limit else q.stream() for doc in docs: ex = doc.to_dict() or {} task = ex.get("task") or "other" verdict = ex.get("verdict") or "pending" if verdict == "rejected": rec = _to_chat(ex) or {"input": ex.get("input"), "output": ex.get("output_raw")} _w("rejected").write(json.dumps(rec, ensure_ascii=False, default=str) + "\n") counts["rejected"] += 1 continue if task == "asr": rec = _to_asr(ex) if rec: _w("asr").write(json.dumps(rec, ensure_ascii=False, default=str) + "\n") counts["asr"] += 1 continue if verdict not in sft_verdicts: continue if task == "vision": rec = _to_vision(ex) if rec: _w("vision").write(json.dumps(rec, ensure_ascii=False, default=str) + "\n") counts["vision"] += 1 continue if task in _TEXT_TASKS: rec = _to_chat(ex) if not rec: continue key = (rec["messages"][-2]["content"], rec["messages"][-1]["content"]) if key in seen[task]: continue seen[task].add(key) _w(task).write(json.dumps(rec, ensure_ascii=False, default=str) + "\n") counts[task] += 1 for f in writers.values(): f.close() print("Exported examples per file:") for name, n in sorted(counts.items()): print(f" {name}.jsonl: {n}") print(f"Output: {os.path.abspath(args.out)}") if __name__ == "__main__": main()