Spaces:
Running on Zero
Running on Zero
| """MLOL — MultiDomain LLM Optimisation Lab (Gradio Space). | |
| Control plane only (spec §1): renders with no model loaded and no GPU acquired. | |
| GPU work happens exclusively inside gpu_wrap-decorated handlers (P7). | |
| """ | |
| import json | |
| import os | |
| import pathlib | |
| import threading | |
| import time | |
| # ZeroGPU: `spaces` must be imported before torch. | |
| try: | |
| import spaces | |
| _HAS_SPACES = True | |
| except ImportError: | |
| _HAS_SPACES = False | |
| import gradio as gr | |
| import yaml | |
| from src.config_loader import get_configs | |
| from src.schemas import ExperimentManifest, config_hash, new_run_id | |
| from src.services import assistant as assistant_svc | |
| from src.services import dataset_prep, evaluation, reporting, routing | |
| from src.services.persistence import get_store | |
| from src.services.training_backends import ColabExporter, HFJobsBackend, MockBackend, ZeroGPUDemoBackend | |
| CFG = get_configs() | |
| STORE = get_store() | |
| IS_ZEROGPU = bool(os.environ.get("SPACES_ZERO_GPU")) | |
| PREMIUM_CODES = {c.strip() for c in os.environ.get("PREMIUM_ACCESS_CODES", "").split(",") if c.strip()} | |
| CONTACT = "finpy07@gmail.com" | |
| if _HAS_SPACES: | |
| gpu_wrap = spaces.GPU(duration=240) | |
| else: | |
| def gpu_wrap(fn): | |
| return fn | |
| threading.Thread(target=STORE.recover_from_hub, daemon=True).start() | |
| def _has_cuda(): | |
| try: | |
| import torch # lazy (P7/§12) | |
| return torch.cuda.is_available() or IS_ZEROGPU | |
| except ImportError: | |
| return IS_ZEROGPU | |
| def _accelerator(): | |
| return "zerogpu" if IS_ZEROGPU else ("cuda" if _has_cuda() else "cpu") | |
| # ---------------------------------------------------------------- helpers | |
| def run_choices(): | |
| return [f"{m.run_id} · {m.title or m.model_repo} · {m.state}" for m in STORE.list_runs()] | |
| def rid_of(choice): | |
| return choice.split(" · ")[0] if choice else None | |
| def model_info_md(name): | |
| m = CFG.model_by_name(name) | |
| if not m: | |
| return "" | |
| return (f"**{m.name}** — `{m.repo}`\n\n" | |
| f"| Parameters | Context | License | Quantization | Gated |\n|--|--|--|--|--|\n" | |
| f"| {m.params_b} B | {m.context:,} | {m.license} | {', '.join(m.quant)} | " | |
| f"{'yes — HF_TOKEN required' if m.gated else 'no'} |\n" | |
| + (f"\n{m.notes}" if m.notes else "")) | |
| def _generate_fn(base_repo, adapter_dir, max_new_tokens): | |
| from src.inference.engine import InferenceEngine | |
| eng = InferenceEngine() | |
| eng.load(base_repo, adapter_dir) | |
| def gen(prompt_msgs): | |
| return eng.chat(prompt_msgs, max_new_tokens=max_new_tokens, temperature=0.0) | |
| return gen | |
| def _train_cfg(model_name, epochs, batch, lr, lora_r, lora_alpha, dropout, | |
| scheduler, grad_accum, warmup, decay, seed, seq_len): | |
| m = CFG.model_by_name(model_name) | |
| return {"model_name": model_name, "model_repo": m.repo, "epochs": int(epochs), | |
| "batch_size": int(batch), "learning_rate": float(lr), "lora_r": int(lora_r), | |
| "lora_alpha": int(lora_alpha), "lora_dropout": float(dropout), | |
| "scheduler": scheduler, "grad_accum": int(grad_accum), | |
| "warmup_ratio": float(warmup), "weight_decay": float(decay), | |
| "seed": int(seed), "seq_len": int(seq_len)} | |
| # ---------------------------------------------------------------- tier 1 handlers | |
| def prepare_dataset(file, title, domain_id, model_name): | |
| if file is None: | |
| return "Upload a file first.", gr.update(), "" | |
| dom = CFG.domain_by_id(domain_id) | |
| try: | |
| records, summary = dataset_prep.prepare(file, dom.system_prompt if dom else "") | |
| except dataset_prep.DatasetError as e: | |
| return f"❌ Rejected: {e}", gr.update(), "" | |
| m = ExperimentManifest(run_id=new_run_id(), title=title or pathlib.Path(file).stem, | |
| domain=domain_id or "general", | |
| model_repo=CFG.model_by_name(model_name).repo if CFG.model_by_name(model_name) else "", | |
| dataset_fingerprint=summary["fingerprint"]) | |
| m.go("data-ready", note=f"{summary['samples']} samples") | |
| STORE.save_manifest(m) | |
| dataset_prep.save_jsonl(records, STORE.artifact_path(m.run_id, "dataset.jsonl")) | |
| STORE.save_artifact(m.run_id, "dataset_summary.json", summary) | |
| hrs, _ = routing.estimate(CFG.model_by_name(model_name).params_b if CFG.model_by_name(model_name) else 1, | |
| summary["samples"], summary["avg_tokens_per_sample"], 2, 512) | |
| md = (f"✅ **Experiment `{m.run_id}` created** — state `{m.state}`\n\n" | |
| f"| Samples | Tokens (est) | Avg tokens | Duplicates removed | References |\n|--|--|--|--|--|\n" | |
| f"| {summary['samples']} | {summary['est_tokens']:,} | {summary['avg_tokens_per_sample']} " | |
| f"| {summary['duplicates_removed']} | {'yes' if summary['has_reference_answers'] else 'no'} |\n\n" | |
| f"Estimated training time: ~{hrs:.2f} GPU-hours (2 epochs).") | |
| return md, gr.update(choices=run_choices(), value=None), m.run_id | |
| def check_routing(run_choice, model_name, epochs, seq_len): | |
| rid = rid_of(run_choice) | |
| if not rid: | |
| return "Select an experiment first." | |
| ds = STORE.load_artifact(rid, "dataset_summary.json") or {} | |
| jobs_ok, jobs_why = HFJobsBackend().eligible() | |
| d = routing.decide(model_name, ds.get("samples", 0), ds.get("avg_tokens_per_sample", 200), | |
| int(epochs), int(seq_len), "qlora", | |
| user_authed=jobs_ok, jobs_eligible=jobs_ok, | |
| zerogpu_present=IS_ZEROGPU or _has_cuda()) | |
| lines = [f"**Recommended backend: `{d.backend}`** · est {d.est_gpu_hours:.2f} GPU-h · est {d.est_vram_gb} GB VRAM\n"] | |
| for b, (ok, why) in d.eligible.items(): | |
| lines.append(f"- {'✅' if ok else '🚫'} `{b}` — {why}") | |
| if not jobs_ok: | |
| lines.append(f"\n_HF Jobs: {jobs_why}_") | |
| return "\n".join(lines) | |
| def _eval_pass(rid, model_name, which, level, progress=gr.Progress()): | |
| """Shared baseline/post evaluation. which: 'baseline'|'post_training'.""" | |
| m = STORE.load_manifest(rid) | |
| ds = dataset_prep.load_jsonl(STORE.artifact_path(rid, "dataset.jsonl")) | |
| lim = CFG.limits.get("evaluation", {}) | |
| n = lim.get("quick_items", 25) if level == "Quick" else lim.get("standard_items", 100) | |
| seed = m.train_config.get("seed", 42) if m.train_config else 42 | |
| items = evaluation.sample_items(ds, n, seed) | |
| existing = STORE.load_artifact(rid, f"{which}_items.json") or {} | |
| adapter = None | |
| if which == "post_training": | |
| ad = STORE.artifact_path(rid, "adapter") | |
| adapter = str(ad) if ad.exists() else None | |
| model = CFG.model_by_name(model_name) | |
| gen = _generate_fn(model.repo, adapter, lim.get("max_new_tokens", 192)) | |
| state_name = "baseline-running" if which == "baseline" else "post-evaluation-running" | |
| if m.can_go(state_name): | |
| m.go(state_name) | |
| STORE.save_manifest(m) | |
| results = evaluation.evaluate_items(gen, items, existing, | |
| lim.get("max_new_tokens", 192), | |
| progress=lambda f: progress(f, desc=f"{which} eval")) | |
| STORE.save_artifact(rid, f"{which}_items.json", results) | |
| summary = evaluation.summarize(results, seed, level) | |
| STORE.save_artifact(rid, f"{which}.json", summary) | |
| m = STORE.load_manifest(rid) | |
| nxt = "baseline-complete" if which == "baseline" else "complete" | |
| if m.can_go(nxt): | |
| m.go(nxt) | |
| STORE.save_manifest(m) | |
| return summary | |
| def run_baseline(run_choice, model_name, level, progress=gr.Progress()): | |
| rid = rid_of(run_choice) | |
| if not rid: | |
| return "Select an experiment first." | |
| try: | |
| s = _eval_pass(rid, model_name, "baseline", level, progress) | |
| except Exception as e: # noqa: BLE001 | |
| import traceback; traceback.print_exc() | |
| return f"❌ Baseline evaluation failed: {type(e).__name__}: {e}" | |
| mm = s["metrics"] | |
| return (f"✅ Baseline saved (`baseline.json`) — n={s['n_items']}, seed={s['seed']}\n\n" | |
| f"accuracy {mm['accuracy']['mean']} · ROUGE-L {mm['rougeL']['mean']} · " | |
| f"BLEU {mm['bleu']['mean']} · latency {mm['latency_s']['mean']}s · " | |
| f"hallucination est. {s['hallucination_estimate']['composite_pct']}%") | |
| def run_training(run_choice, backend_choice, model_name, epochs, batch, lr, lora_r, | |
| lora_alpha, dropout, scheduler, grad_accum, warmup, decay, seed, | |
| seq_len, progress=gr.Progress()): | |
| rid = rid_of(run_choice) | |
| if not rid: | |
| return "Select an experiment first.", None | |
| m = STORE.load_manifest(rid) | |
| cfg = _train_cfg(model_name, epochs, batch, lr, lora_r, lora_alpha, dropout, | |
| scheduler, grad_accum, warmup, decay, seed, seq_len) | |
| m.train_config, m.config_hash, m.model_repo = cfg, config_hash(cfg), cfg["model_repo"] | |
| ds = STORE.load_artifact(rid, "dataset_summary.json") or {} | |
| d = routing.decide(model_name, ds.get("samples", 0), ds.get("avg_tokens_per_sample", 200), | |
| int(epochs), int(seq_len), "qlora", | |
| user_authed=True, jobs_eligible=False, | |
| zerogpu_present=IS_ZEROGPU or _has_cuda()) | |
| if backend_choice == "Colab export": | |
| m.backend = "colab_export" | |
| if m.can_go("training-submitted"): | |
| m.go("training-submitted", note="colab package generated") | |
| STORE.save_manifest(m) | |
| path = ColabExporter().build(m, cfg) | |
| return (f"📦 Pinned Colab package ready — run it in Colab; the dashboard " | |
| f"updates when the completion metadata comes back."), str(path) | |
| if backend_choice == "Mock (test pipeline)": | |
| backend, note = MockBackend(), "mock" | |
| else: | |
| ok, why = d.eligible["zerogpu_demo"] | |
| if not ok: | |
| return f"🚫 ZeroGPU demo not eligible: {why}. Use Colab export or HF Jobs.", None | |
| backend, note = ZeroGPUDemoBackend(), "zerogpu demo" | |
| records = dataset_prep.load_jsonl(STORE.artifact_path(rid, "dataset.jsonl")) | |
| try: | |
| for st in ("training-submitted", "training-running"): | |
| if m.can_go(st): | |
| m.go(st, note=note) | |
| m.backend = backend.name | |
| STORE.save_manifest(m) | |
| progress(0.05, desc="training") | |
| out = backend.train(m, records, cfg) | |
| m = STORE.load_manifest(rid) | |
| if m.can_go("training-complete"): | |
| m.go("training-complete") | |
| STORE.save_manifest(m) | |
| except Exception as e: # noqa: BLE001 | |
| import traceback; traceback.print_exc() | |
| m = STORE.load_manifest(rid) | |
| m.error = f"{type(e).__name__}: {e}" | |
| if m.can_go("failed"): | |
| m.go("failed", note=m.error) | |
| STORE.save_manifest(m) | |
| return f"❌ Training failed: {m.error}", None | |
| loss_txt = f"loss {out['losses'][0]:.3f} → {out['losses'][-1]:.3f}" if out.get("losses") else "no loss logged" | |
| return (f"✅ Training complete ({note}) in {out['train_seconds']}s — {loss_txt}. " | |
| f"State: `training-complete`. Now run the post-training evaluation."), None | |
| def run_post_and_compare(run_choice, model_name, level, progress=gr.Progress()): | |
| rid = rid_of(run_choice) | |
| if not rid: | |
| return "Select an experiment first.", None | |
| try: | |
| post = _eval_pass(rid, model_name, "post_training", level, progress) | |
| except Exception as e: # noqa: BLE001 | |
| import traceback; traceback.print_exc() | |
| return f"❌ Post-training evaluation failed: {type(e).__name__}: {e}", None | |
| base = STORE.load_artifact(rid, "baseline.json") | |
| if not base: | |
| return "⚠️ Post eval saved, but no baseline exists — run the baseline first.", None | |
| cmp_ = evaluation.compare(base, post, | |
| STORE.load_artifact(rid, "baseline_items.json"), | |
| STORE.load_artifact(rid, "post_training_items.json")) | |
| STORE.save_artifact(rid, "comparison.json", cmp_) | |
| tl = STORE.load_artifact(rid, "training_log.json") or {} | |
| ds = STORE.load_artifact(rid, "dataset_summary.json") or {} | |
| diags = evaluation.diagnostics(ds, tl, cmp_) | |
| STORE.save_artifact(rid, "diagnostics.json", {"items": diags}) | |
| rows = [[r["metric"], r["baseline"], r["finetuned"], r["change"], | |
| r["p_value"], "✔" if r["significant"] else "", r["direction"]] | |
| for r in cmp_["rows"]] | |
| md = (f"### Overall: **{cmp_['overall']}** ({cmp_['n_paired_items']} paired items; {cmp_['method']})\n\n" | |
| + "\n".join(f"- **{d['reason']}** — {d['evidence']}" for d in diags)) | |
| return md, rows | |
| def generate_reports(run_choice): | |
| rid = rid_of(run_choice) | |
| if not rid: | |
| return "Select an experiment.", None, None, None | |
| m = STORE.load_manifest(rid) | |
| needed = {n: STORE.load_artifact(rid, f"{n}.json") for n in | |
| ("dataset_summary", "training_log", "baseline", "post_training", "comparison", "diagnostics")} | |
| missing = [k for k, v in needed.items() if v is None] | |
| if missing: | |
| return f"⚠️ Missing artifacts: {', '.join(missing)}. Complete the pipeline first.", None, None, None | |
| env = reporting.environment_block(m, m.backend or "unknown", _accelerator(), | |
| "bf16/4bit", needed["baseline"].get("seed", 42), | |
| needed["post_training"].get("n_items", 0), | |
| demo_run=m.backend in ("mock", "zerogpu_demo")) | |
| hw = routing.hardware_recommendations( | |
| routing.estimate(CFG.model_by_name(next((x.name for x in CFG.models if x.repo == m.model_repo), CFG.models[0].name)).params_b | |
| if any(x.repo == m.model_repo for x in CFG.models) else 7.0, | |
| needed["dataset_summary"].get("samples", 0), | |
| needed["dataset_summary"].get("avg_tokens_per_sample", 200), 2, 512)[1]) | |
| cert = reporting.build_certificate(m, needed["dataset_summary"], needed["training_log"], | |
| needed["baseline"], needed["post_training"], | |
| needed["comparison"], needed["diagnostics"]["items"], env, hw) | |
| STORE.save_artifact(rid, "certificate.json", cert) | |
| pdf = reporting.certificate_pdf(cert) | |
| pdf_path = STORE.save_binary(rid, "report.pdf", pdf) | |
| csv_path = STORE.artifact_path(rid, "certificate.csv") | |
| csv_path.write_text(reporting.certificate_csv(cert)) | |
| json_path = STORE.artifact_path(rid, "certificate.json") | |
| md = (f"## {cert['section_3_overall']} · Confidence {cert['section_4_confidence']['stars']}\n" | |
| f"**Deployment: {cert['section_7_deployment']}**\n\n" | |
| + "\n".join(f"- {s}" for s in cert["section_9_research_summary"])) | |
| return md, str(pdf_path), str(csv_path), str(json_path) | |
| # ---------------------------------------------------------------- assistant | |
| def assistant_chat(message, history, mode, provider_name, user_key, run_choice): | |
| if not message.strip(): | |
| return history, "" | |
| providers = {p.name: p.id for p in CFG.providers} | |
| pid = providers.get(provider_name, CFG.default_provider) | |
| ctx = "" | |
| rid = rid_of(run_choice) | |
| if mode in ("Experiment", "Report") and rid: | |
| parts = {} | |
| for n in ("dataset_summary", "training_log", "baseline", "post_training", | |
| "comparison", "diagnostics", "certificate"): | |
| a = STORE.load_artifact(rid, f"{n}.json") | |
| if a: | |
| parts[n] = a | |
| m = STORE.load_manifest(rid) | |
| if m: | |
| parts["config"] = m.train_config | |
| parts["state"] = m.state | |
| ctx = json.dumps(parts, default=str)[:24000] | |
| elif mode == "Hardware": | |
| ctx = json.dumps({"profiles": [p.model_dump() for p in CFG.hardware], | |
| "zerogpu_demo_limits": CFG.limits.get("zerogpu_demo", {})}) | |
| msgs = assistant_svc.build_messages(mode, message, history, ctx) | |
| reply = assistant_svc.chat(pid, msgs, user_key) | |
| reply, tool = assistant_svc.parse_tool_call(reply) | |
| if tool: | |
| if tool["action"] == "suggest_hyperparameters" and rid: | |
| ds = STORE.load_artifact(rid, "dataset_summary.json") or {} | |
| sug = assistant_svc.suggest_hyperparameters(ds.get("samples", 0), 1.5) | |
| reply += f"\n\n🔧 Suggested config: `{json.dumps(sug)}` — copy into the training form." | |
| else: | |
| reply += f"\n\n🔧 Requested action `{tool['action']}` — open the relevant tab to apply it." | |
| history = history + [{"role": "user", "content": message}, | |
| {"role": "assistant", "content": reply}] | |
| return history, "" | |
| # ---------------------------------------------------------------- UI | |
| CSS = """ | |
| #assistant-panel {position: fixed; bottom: 12px; right: 12px; width: 400px; max-height: 75vh; | |
| z-index: 1000; background: var(--background-fill-primary); | |
| border: 1px solid var(--border-color-primary); border-radius: 12px; | |
| box-shadow: 0 4px 18px rgba(0,0,0,.25); overflow-y: auto;} | |
| """ | |
| with gr.Blocks(title="MLOL — MultiDomain LLM Optimisation Lab", css=CSS) as demo: | |
| gr.Markdown("# 🧪 MLOL — MultiDomain LLM Optimisation Lab\n" | |
| "Choose a model → upload data → baseline → fine-tune → evaluate → certify → deploy. " | |
| "_Research platform; outputs are not professional advice._") | |
| if CFG.errors: | |
| gr.Markdown("⚠️ **Config validation:** " + " · ".join(CFG.errors)) | |
| with gr.Tab("🏠 Home"): | |
| gr.Markdown("### Pipeline\n" | |
| "`Model → Dataset → Validation → Config → Hardware → Baseline → " | |
| "Fine-tune → Post-eval → Comparison → Certificate → Report → Deploy`\n\n" | |
| f"Accelerator: **{_accelerator()}** · Persistence: " | |
| f"**{'Hub-mirrored' if STORE.hub_available() else 'local only (no write token)'}**") | |
| home_tbl = gr.Dataframe(headers=["run", "title", "state", "domain", "model"], | |
| interactive=False, label="Recent experiments") | |
| home_refresh = gr.Button("Refresh") | |
| def _home(): | |
| return [[m.run_id, m.title, m.state, m.domain, m.model_repo] for m in STORE.list_runs()[:20]] | |
| home_refresh.click(_home, None, home_tbl) | |
| with gr.Tab("🔬 Tier 1 — Fine-Tuning Lab"): | |
| with gr.Row(): | |
| with gr.Column(scale=1): | |
| model_dd = gr.Dropdown([m.name for m in CFG.models], value=CFG.models[0].name, | |
| label="Base model (configs/models.yaml)") | |
| model_md = gr.Markdown(model_info_md(CFG.models[0].name)) | |
| model_dd.change(model_info_md, model_dd, model_md) | |
| exp_title = gr.Textbox(label="Experiment title", placeholder="my-first-run") | |
| domain_dd = gr.Dropdown(["general"] + [d.id for d in CFG.domains], value="general", | |
| label="Domain (Tier 2 pre-fills this)") | |
| data_file = gr.File(label="Dataset (CSV/JSON/JSONL/TXT/PDF/DOCX)", type="filepath") | |
| prep_btn = gr.Button("1️⃣ Prepare dataset", variant="primary") | |
| with gr.Column(scale=2): | |
| prep_md = gr.Markdown() | |
| run_dd = gr.Dropdown(choices=run_choices(), label="Active experiment", interactive=True, allow_custom_value=True) | |
| run_refresh = gr.Button("↻ refresh experiments", size="sm") | |
| run_refresh.click(lambda: gr.update(choices=run_choices()), None, run_dd) | |
| with gr.Accordion("Training configuration", open=True): | |
| with gr.Row(): | |
| epochs = gr.Slider(1, 5, 2, step=1, label="Epochs") | |
| batch = gr.Slider(1, 8, 2, step=1, label="Batch size") | |
| lr = gr.Textbox("2e-4", label="Learning rate") | |
| with gr.Row(): | |
| lora_r = gr.Slider(4, 64, 16, step=4, label="LoRA rank") | |
| lora_alpha = gr.Slider(8, 128, 32, step=8, label="Alpha") | |
| dropout = gr.Slider(0.0, 0.3, 0.05, step=0.01, label="Dropout") | |
| with gr.Accordion("Advanced", open=False): | |
| with gr.Row(): | |
| scheduler = gr.Dropdown(["cosine", "linear", "constant"], value="cosine", label="Scheduler") | |
| grad_accum = gr.Slider(1, 16, 4, step=1, label="Grad accumulation") | |
| warmup = gr.Slider(0.0, 0.2, 0.03, step=0.01, label="Warmup ratio") | |
| with gr.Row(): | |
| decay = gr.Textbox("0.001", label="Weight decay") | |
| seed = gr.Number(42, label="Seed", precision=0) | |
| seq_len = gr.Slider(128, 2048, 512, step=128, label="Max seq length") | |
| route_btn = gr.Button("2️⃣ Check routing") | |
| route_md = gr.Markdown() | |
| with gr.Row(): | |
| level_dd = gr.Dropdown(["Quick", "Standard"], value="Quick", label="Eval level") | |
| baseline_btn = gr.Button("3️⃣ Baseline eval") | |
| baseline_md = gr.Markdown() | |
| backend_dd = gr.Radio(["ZeroGPU demo", "Colab export", "Mock (test pipeline)"], | |
| value="ZeroGPU demo", label="Training backend") | |
| train_btn = gr.Button("4️⃣ Fine-tune", variant="primary") | |
| train_md = gr.Markdown() | |
| colab_file = gr.File(label="Colab package", visible=True) | |
| post_btn = gr.Button("5️⃣ Post-training eval + compare", variant="primary") | |
| post_md = gr.Markdown() | |
| cmp_tbl = gr.Dataframe(headers=["metric", "baseline", "finetuned", "Δ", "p", "sig", "direction"], | |
| interactive=False) | |
| new_run_state = gr.State("") | |
| prep_btn.click(prepare_dataset, [data_file, exp_title, domain_dd, model_dd], | |
| [prep_md, run_dd, new_run_state]) | |
| route_btn.click(check_routing, [run_dd, model_dd, epochs, seq_len], route_md) | |
| baseline_btn.click(run_baseline, [run_dd, model_dd, level_dd], baseline_md) | |
| train_btn.click(run_training, | |
| [run_dd, backend_dd, model_dd, epochs, batch, lr, lora_r, lora_alpha, | |
| dropout, scheduler, grad_accum, warmup, decay, seed, seq_len], | |
| [train_md, colab_file]) | |
| post_btn.click(run_post_and_compare, [run_dd, model_dd, level_dd], [post_md, cmp_tbl]) | |
| with gr.Tab("🏛 Tier 2 — Domain Foundry"): | |
| t2_unlocked = gr.State(not PREMIUM_CODES) | |
| gr.Markdown("Domain-specific configurations: curated datasets, templates, benchmarks, " | |
| f"recommended hyperparameters. **Premium tier** — request access: **{CONTACT}**.") | |
| t2_code = gr.Textbox(label="Premium access code", type="password", | |
| visible=bool(PREMIUM_CODES)) | |
| t2_unlock = gr.Button("Unlock", visible=bool(PREMIUM_CODES)) | |
| t2_msg = gr.Markdown("" if PREMIUM_CODES else "✅ Premium open (dev mode — no codes set).") | |
| t2_dom = gr.Dropdown([d.name for d in CFG.domains], value=CFG.domains[0].name, label="Domain") | |
| t2_md = gr.Markdown() | |
| def t2_show(name, unlocked): | |
| d = next((x for x in CFG.domains if x.name == name), None) | |
| if not unlocked: | |
| return f"🔒 Premium locked — request a code via **{CONTACT}**." | |
| bm = "\n".join(f"- **{b.name}** (`{b.source}`, metric: {b.metric})" for b in d.benchmarks) | |
| dsl = "\n".join(f"- `{x}`" for x in d.datasets) or "- (bring your own via Tier 1 upload)" | |
| return (f"## {d.name}\n**System prompt:** {d.system_prompt}\n\n" | |
| f"**Curated datasets:**\n{dsl}\n\n**Benchmarks (sampled in-Space):**\n{bm}\n\n" | |
| f"**Recommended hyperparameters:** `{json.dumps(d.hyperparameters)}`\n\n" | |
| f"Use Tier 1 with Domain = `{d.id}` — the system prompt and settings apply " | |
| f"automatically.\n\n_{d.disclaimer}_") | |
| def t2_try_unlock(code): | |
| ok = code.strip() in PREMIUM_CODES if PREMIUM_CODES else True | |
| return ok, ("✅ Unlocked." if ok else f"❌ Invalid code — request one via **{CONTACT}**.") | |
| t2_unlock.click(t2_try_unlock, t2_code, [t2_unlocked, t2_msg]) | |
| t2_dom.change(t2_show, [t2_dom, t2_unlocked], t2_md) | |
| t2_unlocked.change(t2_show, [t2_dom, t2_unlocked], t2_md) | |
| with gr.Tab("📊 Evaluation Lab"): | |
| gr.Markdown("Baseline vs fine-tuned with **identical items and seeds**, bootstrap CIs, and " | |
| "paired permutation significance (α=0.05). Sampled evaluation — the certificate " | |
| "always discloses n, seed, and that the full benchmark was not executed.") | |
| ev_run = gr.Dropdown(choices=run_choices(), label="Experiment", allow_custom_value=True) | |
| gr.Button("↻ refresh", size="sm").click(lambda: gr.update(choices=run_choices()), None, ev_run) | |
| ev_view = gr.Button("Show stored evaluations") | |
| ev_md = gr.Markdown() | |
| def show_evals(choice): | |
| rid = rid_of(choice) | |
| if not rid: | |
| return "Select an experiment." | |
| out = [] | |
| for n in ("baseline", "post_training"): | |
| a = STORE.load_artifact(rid, f"{n}.json") | |
| if a: | |
| mm = a["metrics"] | |
| out.append(f"**{n}** (n={a['n_items']}, seed={a['seed']}): " | |
| f"acc {mm['accuracy']['mean']} [{mm['accuracy']['ci_low']}–{mm['accuracy']['ci_high']}] · " | |
| f"ROUGE-L {mm['rougeL']['mean']} · BLEU {mm['bleu']['mean']} · " | |
| f"halluc. est {a['hallucination_estimate']['composite_pct']}%") | |
| c = STORE.load_artifact(rid, "comparison.json") | |
| if c: | |
| out.append(f"**Comparison:** {c['overall']} ({c['n_paired_items']} paired items)") | |
| d = STORE.load_artifact(rid, "diagnostics.json") | |
| if d: | |
| out += [f"- {x['reason']}: {x['evidence']}" for x in d["items"]] | |
| return "\n\n".join(out) or "No evaluations stored yet — run them in Tier 1." | |
| ev_view.click(show_evals, ev_run, ev_md) | |
| with gr.Tab("📄 Reports"): | |
| rp_run = gr.Dropdown(choices=run_choices(), label="Experiment", allow_custom_value=True) | |
| gr.Button("↻ refresh", size="sm").click(lambda: gr.update(choices=run_choices()), None, rp_run) | |
| rp_btn = gr.Button("Generate optimisation report + certificate", variant="primary") | |
| rp_md = gr.Markdown() | |
| with gr.Row(): | |
| rp_pdf = gr.File(label="Certificate PDF") | |
| rp_csv = gr.File(label="CSV") | |
| rp_json = gr.File(label="JSON") | |
| rp_btn.click(generate_reports, rp_run, [rp_md, rp_pdf, rp_csv, rp_json]) | |
| gr.Markdown("### Research dashboard — compare any two experiments") | |
| with gr.Row(): | |
| cmp_a = gr.Dropdown(choices=run_choices(), label="Experiment A", allow_custom_value=True) | |
| cmp_b = gr.Dropdown(choices=run_choices(), label="Experiment B", allow_custom_value=True) | |
| cmp_btn = gr.Button("Compare") | |
| cmp2_tbl = gr.Dataframe(interactive=False) | |
| def compare_two(a, b): | |
| rows = [] | |
| for label, ch in (("A", a), ("B", b)): | |
| rid = rid_of(ch) | |
| post = STORE.load_artifact(rid, "post_training.json") if rid else None | |
| if post: | |
| mm = post["metrics"] | |
| rows.append([label, rid, mm["accuracy"]["mean"], mm["bleu"]["mean"], | |
| mm["rougeL"]["mean"], mm["latency_s"]["mean"], | |
| post["hallucination_estimate"]["composite_pct"]]) | |
| return gr.update(value=rows, | |
| headers=["exp", "run", "accuracy", "bleu", "rougeL", "latency", "halluc%"]) | |
| cmp_btn.click(compare_two, [cmp_a, cmp_b], cmp2_tbl) | |
| with gr.Tab("📚 Adapter Library"): | |
| lib_btn = gr.Button("↻ refresh") | |
| lib_tbl = gr.Dataframe(headers=["run", "title", "domain", "base model", "state", "adapter", "date"], | |
| interactive=False) | |
| def lib(): | |
| rows = [] | |
| for m in STORE.list_runs(): | |
| ad = STORE.artifact_path(m.run_id, "adapter") | |
| rows.append([m.run_id, m.title, m.domain, m.model_repo, m.state, | |
| "✅ local" if ad.exists() else "—", | |
| time.strftime("%Y-%m-%d", time.localtime(m.created_at))]) | |
| return rows | |
| lib_btn.click(lib, None, lib_tbl) | |
| with gr.Tab("🖥 Hardware Advisor"): | |
| hw_model = gr.Dropdown([m.name for m in CFG.models], value=CFG.models[0].name, label="Model") | |
| with gr.Row(): | |
| hw_n = gr.Number(1000, label="Samples", precision=0) | |
| hw_ep = gr.Slider(1, 5, 2, step=1, label="Epochs") | |
| hw_seq = gr.Slider(128, 4096, 512, step=128, label="Seq length") | |
| hw_btn = gr.Button("Estimate") | |
| hw_md = gr.Markdown() | |
| hw_tbl = gr.Dataframe(headers=["hardware", "VRAM", "verdict", "note"], interactive=False) | |
| def hw_go(name, n, ep, seq): | |
| m = CFG.model_by_name(name) | |
| hrs, vram = routing.estimate(m.params_b, int(n), 300, int(ep), int(seq)) | |
| rows = [[r["hardware"], r["vram_gb"], r["verdict"], r["note"]] | |
| for r in routing.hardware_recommendations(vram)] | |
| return (f"**{name}** — estimated **{vram} GB VRAM**, **{hrs:.2f} GPU-hours** " | |
| f"(order-of-magnitude estimates; certificates report actuals)"), rows | |
| hw_btn.click(hw_go, [hw_model, hw_n, hw_ep, hw_seq], [hw_md, hw_tbl]) | |
| with gr.Tab("📖 Documentation"): | |
| guide = pathlib.Path("docs/USER_GUIDE.md") | |
| gr.Markdown(guide.read_text() if guide.exists() else "See docs/MASTER_SPEC.md") | |
| with gr.Column(elem_id="assistant-panel"): | |
| with gr.Accordion("🤖 AI Research Assistant", open=False): | |
| as_mode = gr.Radio(["General", "Experiment", "Hardware", "Report"], | |
| value="General", label="Mode") | |
| as_provider = gr.Dropdown([p.name for p in CFG.providers], | |
| value=next((p.name for p in CFG.providers | |
| if p.id == CFG.default_provider), None), | |
| label="LLM provider") | |
| as_key = gr.Textbox(label="API key (only for Claude/GPT; never stored)", type="password") | |
| as_run = gr.Dropdown(choices=run_choices(), label="Experiment context (optional)", allow_custom_value=True) | |
| gr.Button("↻", size="sm").click(lambda: gr.update(choices=run_choices()), None, as_run) | |
| as_chat = gr.Chatbot(type="messages", height=280, label="Assistant") | |
| as_msg = gr.Textbox(placeholder="Why did my model perform worse?", label="Ask") | |
| as_msg.submit(assistant_chat, [as_msg, as_chat, as_mode, as_provider, as_key, as_run], | |
| [as_chat, as_msg]) | |
| if __name__ == "__main__": | |
| demo.launch() | |