import spaces # must be imported before torch import html import json import sys import time from pathlib import Path import gradio as gr import torch from huggingface_hub import snapshot_download from safetensors.torch import load_file from transformers import AutoProcessor, Qwen3_5ForConditionalGeneration MODEL_ID = "Cloudflare/clef-flash" MODEL_PATH = Path(snapshot_download(MODEL_ID)) sys.path.insert(0, str(MODEL_PATH)) from joint_schema_model import ( # noqa: E402 QUESTION_TYPES, ClefModel, JointSchemaHead, collate_records, encode_record, systemone_answer, ) # Same as the authors' load_release_model, but with an eager .to("cuda") (ZeroGPU-friendly) backbone = Qwen3_5ForConditionalGeneration.from_pretrained( MODEL_PATH, dtype=torch.bfloat16, attn_implementation="sdpa" ).to("cuda") backbone.config.use_cache = False head = JointSchemaHead(**json.loads((MODEL_PATH / "joint_head_config.json").read_text())) head.load_state_dict(load_file(MODEL_PATH / "joint_head.safetensors"), strict=True) head = head.to(device="cuda", dtype=torch.bfloat16) model = ClefModel(backbone, head).eval() processor = AutoProcessor.from_pretrained(MODEL_PATH) MAX_IMAGE_SIDE = 1024 # ---------------------------------------------------------------- examples EX_SUPPORT_STATE = "Our checkout started returning errors and orders are blocked." EX_SUPPORT_Q = { "department": { "type": "choice", "instructions": "Which team should handle the message?", "criteria": {"billing": "Payments or invoices", "technical": "Bugs or outages"}, }, "urgency": {"type": "score", "criteria": ["Can wait", "This week", "Today"]}, "outage": {"type": "noul", "instructions": "Is a service down?"}, } EX_INVOICE_STATE = {"invoice": {"vendor": "Acme", "total": 1250.0, "currency": "USD", "status": "overdue"}} EX_INVOICE_Q = { "status": { "type": "choice", "instructions": "What is the invoice status?", "criteria": {"paid": "Invoice is paid.", "overdue": "Invoice is past due.", "draft": "Not sent."}, }, "large": {"type": "noul", "instructions": "Is the total above 1000 USD?"}, } EX_REVIEW_STATE = ( "Ordered the noise-cancelling headphones two weeks ago. Sound is great but the left ear cup " "started crackling yesterday and the support chat keeps disconnecting. I want a replacement, not a refund." ) EX_REVIEW_Q = { "sentiment": { "type": "score", "instructions": "Overall customer sentiment", "criteria": ["Very negative", "Negative", "Mixed", "Positive", "Very positive"], }, "action": { "type": "choice", "instructions": "What should the agent do next?", "criteria": { "replace": "Ship a replacement unit", "refund": "Issue a refund", "troubleshoot": "Send troubleshooting steps", "escalate": "Escalate to a supervisor", }, }, "defect": {"type": "noul", "instructions": "Does the customer report a hardware defect?"}, } EX_FOOD_STATE = {"task": "Moderate a photo uploaded to a restaurant listing.", "listing": "Vegan Garden Bistro"} EX_FOOD_Q = { "category": { "type": "choice", "instructions": "What does the photo mainly show?", "criteria": {"food": "A dish or meal", "interior": "Restaurant interior", "menu": "A menu or text", "people": "People"}, }, "matches_listing": {"type": "noul", "instructions": "Is the dish plausibly vegan, consistent with the listing?"}, "appetizing": {"type": "score", "instructions": "How appetizing is the photo?", "criteria": ["Not at all", "Somewhat", "Very"]}, } EX_ROOM_STATE = "A guest submitted this photo with a listing for a short-term rental apartment." EX_ROOM_Q = { "room": { "type": "choice", "instructions": "Which room is shown?", "criteria": {"bedroom": "Bedroom", "kitchen": "Kitchen", "living_room": "Living room", "bathroom": "Bathroom"}, }, "has_tv": {"type": "noul", "instructions": "Is there a television in the room?"}, "tidiness": {"type": "score", "instructions": "How tidy is the room?", "criteria": ["Messy", "Average", "Spotless"]}, } def _j(obj) -> str: return json.dumps(obj, indent=2, ensure_ascii=False) # ---------------------------------------------------------------- rendering def render_html(questions: dict, answers: dict, probs: dict, latency_ms: float, n_tokens: int) -> str: parts = ['
'] for qid, q in questions.items(): qtype = q["type"] p = probs[qid] ans = answers[qid] if qtype == "noul": headline = f"{'true' if p['true'] >= 0.5 else 'false'} · P(true) = {p['true']:.3f}" labels = {"true": "true", "false": "false"} elif qtype == "choice": headline = f"{html.escape(ans['choice'])} · {ans['confidence']:.3f}" labels = {k: k for k in p} else: headline = f"expected score {ans['score']:.2f} / {len(q['criteria']) - 1}" labels = {str(i): f"{i} · {html.escape(str(c))}" for i, c in enumerate(q["criteria"])} best = max(p, key=p.get) instr = html.escape(str(q.get("instructions") or qid)) parts.append( f'
{html.escape(qid)}' f'{qtype}
' f'
{instr}
{headline}
' ) order = list(p.keys()) if qtype in ("score", "noul") else sorted(p, key=p.get, reverse=True) for opt in order: pct = 100 * p[opt] cls = "clef-bar best" if opt == best else "clef-bar" parts.append( f'
{labels.get(opt, html.escape(opt))}
' f'
' f'
{pct:.1f}%
' ) parts.append("
") parts.append(f'
{n_tokens} input tokens · forward pass {latency_ms:.0f} ms
') return "".join(parts) def _parse_state(state: str): s = (state or "").strip() if s[:1] in "{[": try: return json.loads(s) except json.JSONDecodeError: pass return s def _validate(questions): if not isinstance(questions, dict) or not questions: raise gr.Error("Questions must be a non-empty JSON object mapping question IDs to questions.") for qid, q in questions.items(): if not isinstance(q, dict) or q.get("type") not in QUESTION_TYPES: raise gr.Error(f"{qid}: 'type' must be one of noul, choice, score.") if q["type"] == "choice" and not (isinstance(q.get("criteria"), dict) and q["criteria"]): raise gr.Error(f"{qid}: a choice question needs 'criteria' as an object of option_id -> description.") if q["type"] == "score" and not (isinstance(q.get("criteria"), list) and q["criteria"]): raise gr.Error(f"{qid}: a score question needs 'criteria' as a list of level descriptions.") @spaces.GPU(duration=20) def decide(state: str, questions_json: str, image=None): """Score every allowed option of every typed question about a state with Clef-Flash. Args: state: the situation to decide on, as free text or a JSON value. questions_json: JSON object mapping question IDs to questions. Each question has a `type` (`noul` = true/false, `choice` = named options, `score` = ordered levels), optional `instructions`, and `criteria` (object of option -> description for choice, list of level descriptions for score, optional true/false descriptions for noul). image: optional image that is part of the state. Returns: An HTML visualisation of per-option probabilities and a SystemOne-style JSON response. """ try: questions = json.loads(questions_json) except json.JSONDecodeError as e: raise gr.Error(f"Questions JSON is invalid: {e}") _validate(questions) record = {"model": "clef-flash", "state": _parse_state(state), "questions": questions} if image is not None: image = image.convert("RGB") image.thumbnail((MAX_IMAGE_SIDE, MAX_IMAGE_SIDE)) record["images"] = [image] try: encoded = encode_record(processor.tokenizer, record, max_length=16384, processor=processor) except Exception as e: raise gr.Error(f"Could not encode request: {e}") batch = collate_records([encoded], processor.tokenizer.pad_token_id, torch.device("cuda")) torch.cuda.synchronize() t0 = time.perf_counter() with torch.inference_mode(): logits = model(batch)[0] torch.cuda.synchronize() latency_ms = (time.perf_counter() - t0) * 1000 probs, answers = {}, {} for eq, ql in zip(encoded.questions, logits): p = dict(zip(eq.option_ids, ql.float().softmax(-1).tolist())) probs[eq.question_id] = p answers[eq.question_id] = systemone_answer(questions[eq.question_id], p) response = { "model": "clef-flash", "answers": answers, "usage": {"input_tokens": len(encoded.input_ids), "output_tokens": 0}, } return render_html(questions, answers, probs, latency_ms, len(encoded.input_ids)), response CSS = """ #col-container { max-width: 1200px; margin: 0 auto; } .dark .gradio-container { color: var(--body-text-color); } .clef-out { display: flex; flex-direction: column; gap: 12px; } .clef-q { border: 1px solid var(--border-color-primary); border-radius: 10px; padding: 12px 14px; background: var(--block-background-fill); } .clef-qh { display: flex; gap: 8px; align-items: center; } .clef-id { font-weight: 700; font-family: var(--font-mono); color: var(--body-text-color); } .clef-type { font-size: 0.75em; padding: 1px 8px; border-radius: 99px; background: var(--color-accent-soft); color: var(--body-text-color); } .clef-instr { font-size: 0.9em; opacity: 0.8; margin: 4px 0 6px; color: var(--body-text-color); } .clef-head { font-weight: 600; margin-bottom: 8px; color: var(--body-text-color); } .clef-row { display: grid; grid-template-columns: minmax(90px, 38%) 1fr 56px; gap: 8px; align-items: center; margin: 3px 0; } .clef-lbl { font-size: 0.85em; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; color: var(--body-text-color); } .clef-track { height: 12px; border-radius: 6px; background: var(--background-fill-secondary); overflow: hidden; } .clef-bar { height: 100%; background: var(--neutral-400); border-radius: 6px; } .clef-bar.best { background: var(--color-accent); } .clef-pct { font-size: 0.85em; text-align: right; font-family: var(--font-mono); color: var(--body-text-color); } .clef-meta { font-size: 0.8em; opacity: 0.7; color: var(--body-text-color); } """ with gr.Blocks(title="Clef-Flash") as demo: with gr.Column(elem_id="col-container"): gr.Markdown( "# Clef-Flash: typed decisions in one forward pass\n" "Give [Cloudflare/clef-flash](https://huggingface.co/Cloudflare/clef-flash) a **state** (text, JSON, optionally an image) " "and a **schema of typed questions**. It returns a probability for every allowed option of every question " "— no text generation, no output parsing. Question types: `noul` (true/false), `choice` (named options), " "`score` (ordered levels). [Blog post](https://blog.cloudflare.com/clef-decision-models)" ) with gr.Row(): with gr.Column(scale=1): state = gr.Textbox( label="State (free text or JSON)", lines=5, value=EX_SUPPORT_STATE, placeholder="Describe the situation to decide on, or paste a JSON object…", ) questions = gr.Code(label="Questions schema (JSON)", language="json", value=_j(EX_SUPPORT_Q), lines=14) image = gr.Image(label="Image in the state (optional)", type="pil", height=240) run = gr.Button("Decide", variant="primary") with gr.Column(scale=1): out_html = gr.HTML(label="Decisions") with gr.Accordion("SystemOne JSON response", open=False): out_json = gr.JSON(label="Response") gr.Markdown("### Examples — text / JSON state") text_examples = gr.Examples( examples=[ [EX_SUPPORT_STATE, _j(EX_SUPPORT_Q)], [_j(EX_INVOICE_STATE), _j(EX_INVOICE_Q)], [EX_REVIEW_STATE, _j(EX_REVIEW_Q)], ], inputs=[state, questions], outputs=[out_html, out_json], fn=decide, cache_examples=True, cache_mode="lazy", ) gr.Markdown("### Examples — with an image") gr.Examples( examples=[ [_j(EX_FOOD_STATE), _j(EX_FOOD_Q), "examples/burger.jpg"], [EX_ROOM_STATE, _j(EX_ROOM_Q), "examples/living_room.jpg"], ], inputs=[state, questions, image], outputs=[out_html, out_json], fn=decide, cache_examples=True, cache_mode="lazy", ) # Text-only examples should not keep a previously loaded image around. text_examples.dataset.click(lambda: None, outputs=image, api_visibility="private") run.click(decide, inputs=[state, questions, image], outputs=[out_html, out_json], api_name="decide") if __name__ == "__main__": demo.launch(theme=gr.themes.Citrus(), css=CSS, mcp_server=True)