"""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 @gpu_wrap 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']}%") @gpu_wrap 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 @gpu_wrap 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()