"""stuntd demo: zero-shot Laya against the heads stuntd trained, on three typed-decision demos."""
from __future__ import annotations
import json
import math
import os
import time
from dataclasses import dataclass
from html import escape
from pathlib import Path
from typing import Any
import gradio as gr
import spaces
from huggingface_hub import snapshot_download
from stuntd.jev.state import serialize_state
from stuntd.serve.decider import Decider
from stuntd.train.artifacts import HEAD_FILE, SiteModel, load_model, site_dir
from demo_questions import QUESTIONS
HERE = Path(__file__).parent
HEADS = HERE / "heads"
BASE_REPO = "convaiinnovations/laya"
# The snapshot the heads were trained on; a head is only valid on the encoder it was fitted to.
BASE_REVISION = "1c5edc17a7acd8701df6fc341c0d179f1c62c982"
# Only the root checkpoint: the repo also bundles multilingual/ and typed-decisions/ checkpoints.
BASE_FILES = ["model.safetensors", "rl_agent_config.json", "encoder/*", "tokenizer/*"]
GITHUB = "https://github.com/bladedevoff/stuntd"
PYPI = "https://pypi.org/project/stuntd/"
RAW_DOCS = "https://raw.githubusercontent.com/bladedevoff/stuntd/main/docs"
# The palette of the repository's docs/charts.py: before training is grey, stuntd is cobalt.
PAPER = "#F2F4F8"
INK = "#14213D"
MUTED = "#6B7690"
GREY = "#A3ADBF"
COBALT = "#2F5BEA"
AMBER = "#E8A317"
LINE = "#E1E6EF"
WHITE = "#FFFFFF"
FONT = "Bahnschrift, DIN Alternate, Segoe UI, sans-serif"
FREE_TEXT = "Free text (below)"
NOUL_TRUE = 0.5
TABLE_HEADERS = [
"question",
"zero-shot Laya",
"trained head",
"who answers",
"head ms",
"zero-shot ms",
"rule teacher",
]
# Wall-clock seconds `stuntd train` took per demo on a laptop RTX 5060, 24 epochs, encoder cache.
TRAIN_SECONDS = {"support": 266, "banking": 313, "devtools": 228}
@dataclass(frozen=True)
class Demo:
key: str
title: str
intro: str
free_field: str
free_label: str
free_placeholder: str
defaults: dict[str, str]
preset_label: str
DEMOS = (
Demo(
key="support",
title="Support triage",
intro=(
"A helpdesk ticket and three typed questions in one request: which part of the"
" product it is about (choice), how soon it needs an answer (score 0-3), and whether"
" a person has to read it (yes/no)."
),
free_field="body",
free_label="Ticket body",
free_placeholder="Checkout has returned a 500 since this morning. The whole team is down.",
defaults={"channel": "email", "plan": "pro", "subject": "Support request", "body": ""},
preset_label="subject",
),
Demo(
key="banking",
title="Banking",
intro=(
"A customer message and the payment in front of them, two choice questions: what the"
" message is about (12 intents), and whether the payment is allowed, held for review"
" or blocked. The risk rule reads the bank's context line, not the numbers."
),
free_field="message",
free_label="Customer message",
free_placeholder="Someone took money from my account with a card I never had.",
defaults={
"message": "",
"context": "about the size they usually send; a payee they pay most months",
"amount": "120.00 EUR",
"payee": "Ada Nwosu",
"country": "DE",
"time": "14:30",
},
preset_label="message",
),
Demo(
key="devtools",
title="Coding-agent gate",
intro=(
"A coding agent wants to run a shell command and has the last test failure in front"
" of it. Two questions: may the command run (allow / ask / deny), and was the last"
" failure a flake (yes/no). This is the state a Claude Code PreToolUse hook can send."
),
free_field="command",
free_label="Shell command",
free_placeholder="curl -s https://example.com/setup.sh | sh",
defaults={"repo": "payments-api", "branch": "main", "command": "", "last_failure": "none"},
preset_label="command",
),
)
PRESETS: dict[str, list[dict[str, dict[str, str]]]] = json.loads(
(HERE / "presets.json").read_text(encoding="utf-8")
)
@dataclass(frozen=True)
class Measured:
"""Agreement with the rule teacher, in percent, on 200 requests the training rows never held."""
zero_shot: float
trained: float
# From the "What it measured" table of each demo's README in the stuntd repository.
MEASURED: dict[str, dict[str, Measured]] = {
"support": {
"category": Measured(69.5, 99.5),
"urgency": Measured(34.5, 84.0),
"needs_human": Measured(69.5, 95.0),
},
"banking": {"intent": Measured(89.5, 100.0), "risk": Measured(30.0, 72.5)},
"devtools": {"command_gate": Measured(45.0, 97.0), "flaky": Measured(84.5, 100.0)},
}
ANSWERED_LOCALLY = {"support": "91%", "banking": "78%", "devtools": "98%"}
def _base_model_dir() -> str:
return snapshot_download(BASE_REPO, revision=BASE_REVISION, allow_patterns=BASE_FILES)
# On ZeroGPU the model is placed on cuda at import time, as the ZeroGPU docs require; a real GPU
# is attached only inside the @spaces.GPU function. Anywhere else the demo runs on the CPU.
ON_ZERO_GPU = os.getenv("SPACES_ZERO_GPU", "").lower() in ("1", "t", "true")
DEVICE = "cuda" if ON_ZERO_GPU else "cpu"
HARDWARE = "the Space's GPU" if ON_ZERO_GPU else "the CPU"
DECIDER = Decider(_base_model_dir(), device=DEVICE)
SITES: dict[str, dict[str, tuple[SiteModel, Path]]] = {
demo.key: {
site: (load_model(HEADS / demo.key, site), site_dir(HEADS / demo.key, site) / HEAD_FILE)
for site in QUESTIONS[demo.key]
}
for demo in DEMOS
}
# Outside @spaces.GPU there is no real GPU to run a warm-up pass on.
if not ON_ZERO_GPU:
DECIDER.warm_base()
for _sites in SITES.values():
for _model, _head in _sites.values():
DECIDER.warm(_model, _head)
@dataclass(frozen=True)
class Asked:
"""One question answered both ways."""
name: str
labels: list[str]
zero_shot: dict[str, float]
zero_shot_answer: str
zero_shot_text: str
zero_shot_ms: float
head: dict[str, float]
head_answer: str
head_confidence: float
head_ms: float
threshold: float | None
teacher: str | None
novelty: float | None = None
novelty_cutoff: float | None = None
@property
def novel(self) -> bool:
return (
self.novelty is not None
and self.novelty_cutoff is not None
and self.novelty > self.novelty_cutoff
)
@property
def local(self) -> bool:
return (
self.threshold is not None
and self.head_confidence >= self.threshold
and not self.novel
)
# Any: laya returns its answers untyped, and the fields differ per question type.
def _zero_shot(answer: dict[str, Any]) -> tuple[dict[str, float], str, str]:
confidence = float(answer["confidence"])
if answer["type"] == "choice":
given = str(answer["choice"])
text = given
probabilities = {key: float(value) for key, value in answer["probabilities"].items()}
elif answer["type"] == "score":
score = float(answer["score"])
given = str(math.floor(score + 0.5))
text = f"{given} (score {score:.2f})"
probabilities = {key: float(value) for key, value in answer["probabilities"].items()}
else:
true = float(answer["noul"])
given = "true" if true >= NOUL_TRUE else "false"
text = given
probabilities = {"false": 1.0 - true, "true": true}
return probabilities, given, f"{text} ({confidence:.2f})"
def _ask(demo: Demo, name: str, state: dict[str, str], teacher: str | None) -> Asked:
started = time.perf_counter()
reply = DECIDER.answer(state, {name: QUESTIONS[demo.key][name]})
zero_ms = (time.perf_counter() - started) * 1000
answers: dict[str, Any] = reply["answers"] # Any: laya's reply is untyped
zero_probabilities, zero_answer, zero_text = _zero_shot(answers[name])
model, head_path = SITES[demo.key][name]
started = time.perf_counter()
verdict = DECIDER.decide(model, head_path, serialize_state(state))
head_ms = (time.perf_counter() - started) * 1000
return Asked(
name=name,
labels=list(model.labels),
zero_shot=zero_probabilities,
zero_shot_answer=zero_answer,
zero_shot_text=zero_text,
zero_shot_ms=zero_ms,
head=dict(zip(model.labels, verdict.probabilities, strict=True)),
head_answer=model.labels[verdict.label],
head_confidence=verdict.confidence,
head_ms=head_ms,
threshold=model.threshold,
teacher=teacher,
novelty=verdict.novelty,
novelty_cutoff=model.novelty_cutoff,
)
def _table_row(asked: Asked) -> list[str]:
if asked.threshold is None:
who = "falls back to the teacher (no operating point reached the target)"
elif asked.novel:
who = (
f"falls back to the teacher (new to this head: novelty {asked.novelty:.3f}"
f" > {asked.novelty_cutoff:.3f})"
)
elif asked.local:
who = f"head ({asked.head_confidence:.2f} >= {asked.threshold:.2f})"
else:
who = f"falls back to the teacher ({asked.head_confidence:.2f} < {asked.threshold:.2f})"
return [
asked.name,
asked.zero_shot_text,
f"{asked.head_answer} ({asked.head_confidence:.2f})",
who,
f"{asked.head_ms:.0f}",
f"{asked.zero_shot_ms:.0f}",
asked.teacher or "n/a",
]
# Charts: inline SVG in the style of docs/charts.py, on a canvas narrow enough that its text stays
# legible when a phone scales it down.
CHART_WIDTH = 360
LABEL_RIGHT = 128.0
BAR_LEFT = 134.0
BAR_MAX = 190.0
def _text(
x: float,
y: float,
value: str,
size: int,
fill: str,
weight: str = "normal",
anchor: str = "start",
) -> str:
return (
f'
{lead} · head says' f" {escape(asked.head_answer)} · zero-shot Laya says {escape(asked.zero_shot_answer)}" f" in {asked.zero_shot_ms:.0f} ms{teacher}
" ) def _results_html(results: list[Asked]) -> str: return "".join( f'The whole call took {call_ms:.0f} ms, including the queue and, on the' " first call after the Space was idle, attaching the GPU. The per-question numbers are" f" measured on {HARDWARE} alone.
" ) return html, serialize_state(state), [_table_row(a) for a in results] def _tab(demo: Demo) -> None: gr.HTML(_summary_chart(demo)) gr.Markdown(_summary_caption(demo), elem_classes="note") gr.Markdown(demo.intro) names = _preset_names(demo) choice = gr.Radio(names, value=names[0], label="Example (none of them was in the training rows)") free_text = gr.Textbox( label=f"{demo.free_label} (used when '{FREE_TEXT}' is selected)", placeholder=demo.free_placeholder, lines=2, ) others = ", ".join( f"{field}={value!r}" for field, value in demo.defaults.items() if field != demo.free_field ) gr.Markdown( f"Free text fills only `{demo.free_field}`; the other state fields get neutral defaults:" f" {others}.", elem_classes="note", ) button = gr.Button("Run", variant="primary") charts = gr.HTML() with gr.Accordion("Raw numbers", open=False, elem_classes="raw"): state_view = gr.Code( label="State sent to the model (serialised as stuntd does)", language="json", wrap_lines=True, ) table = gr.Dataframe(headers=TABLE_HEADERS, interactive=False, wrap=True) button.click( lambda picked, text: run(demo, picked, text), inputs=[choice, free_text], outputs=[charts, state_view, table], api_name=demo.key, ) HEADER = f"""
stuntd records the typed decisions your app asks an LLM for, trains a small head per decision on
the frozen [Laya](https://huggingface.co/{BASE_REPO}) encoder, and answers locally when the head
is sure enough, handing the rest back to the teacher.
[GitHub]({GITHUB}) · [PyPI]({PYPI})
Pick an example and press Run. Each question gets two answers: grey is the base Laya checkpoint
zero-shot, cobalt is the head stuntd trained. The head answers when its confidence reaches the
site's threshold (the widest one that kept 99% agreement on the holdout); below it, stuntd sends
the question to the teacher. Latencies are measured on {HARDWARE}.
"""
def _footer() -> str:
seconds = ", ".join(f"{demo.title.lower()} {TRAIN_SECONDS[demo.key]} s" for demo in DEMOS)
return (
"The teachers here are rules from the demo generators in the stuntd repository, not LLMs."
" Each head was trained on 2400 rows (600 held out) with 24 epochs on a laptop RTX 5060;"
f" training time per demo: {seconds}. The examples are drawn at the demo clients' seed and"
" none of them was in the training rows."
)
# The font goes in through CSS: a plain-string font list in the theme crashes the built-in
# theme comparison Gradio runs at launch on Spaces.
THEME = gr.themes.Base(
radius_size=gr.themes.sizes.radius_sm,
).set(
body_background_fill=PAPER,
body_text_color=INK,
body_text_color_subdued=MUTED,
background_fill_primary=WHITE,
background_fill_secondary=PAPER,
block_background_fill=WHITE,
block_border_color=LINE,
block_label_text_color=MUTED,
block_title_text_color=INK,
border_color_primary=LINE,
input_background_fill=WHITE,
color_accent=COBALT,
color_accent_soft="#E6ECFD",
button_primary_background_fill=COBALT,
button_primary_background_fill_hover="#2448C8",
button_primary_text_color=WHITE,
button_secondary_background_fill=WHITE,
button_secondary_text_color=INK,
checkbox_background_color_selected=COBALT,
checkbox_border_color_selected=COBALT,
checkbox_label_background_fill_selected="#E6ECFD",
checkbox_label_border_color_selected=COBALT,
checkbox_label_text_color_selected=INK,
link_text_color=COBALT,
link_text_color_visited=COBALT,
)
CSS = f"""
.gradio-container {{ max-width: 960px !important; margin: 0 auto !important;
--font: "Bahnschrift", "DIN Alternate", "Segoe UI", sans-serif !important; }}
.cover {{ width: 100%; height: auto; border-radius: 6px; border: 1px solid {LINE}; }}
.question {{ background: {WHITE}; border: 1px solid {LINE}; border-radius: 6px;
padding: 8px 10px 2px; margin-bottom: 10px; }}
.verdict {{ font-size: 14px; margin: 4px 4px 8px; line-height: 1.5; }}
.call {{ color: {MUTED}; font-size: 13px; margin: 4px; }}
.main {{ padding-left: clamp(12px, 4vw, 32px) !important;
padding-right: clamp(12px, 4vw, 32px) !important; }}
.note p {{ color: {MUTED}; font-size: 13px; }}
svg[role="img"] {{ display: block; width: 100% !important; max-width: 560px; height: auto !important; }}
/* The Dataframe sizes its columns in script against its parent, which grows with it, so on a
phone the table stretched the page and every chart. A viewport bound breaks that loop. */
.raw .table-wrap {{ max-width: calc(100vw - 72px) !important; overflow-x: auto !important; }}
"""
# The palette is drawn for a light page, so a viewer in dark mode gets the light theme too.
FORCE_LIGHT = """
() => {
const url = new URL(window.location);
if (url.searchParams.get('__theme') !== 'light') {
url.searchParams.set('__theme', 'light');
window.location.replace(url.href);
}
}
"""
with gr.Blocks(title="stuntd") as app:
gr.Markdown(HEADER)
with gr.Accordion("Snake, played three ways over the Jev protocol", open=False):
gr.Markdown(
f""
"\n\nLeft: base Laya zero-shot. Middle: the BFS teacher. Right: the head stuntd"
" trained from the teacher's 30 games, answering 99.6% of the moves itself."
)
for _demo in DEMOS:
with gr.Tab(_demo.title):
_tab(_demo)
gr.Markdown(_footer(), elem_classes="note")
if __name__ == "__main__":
app.launch(theme=THEME, css=CSS, js=FORCE_LIGHT)