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 PIL import Image MODEL_ID = "Cloudflare/clef" MODEL_PATH = Path(snapshot_download(MODEL_ID)) sys.path.insert(0, str(MODEL_PATH)) from joint_schema_model import ( # noqa: E402 ClefModel, JointSchemaHead, QUESTION_TYPES, collate_records, encode_record, systemone_answer, ) from safetensors.torch import load_file # noqa: E402 from transformers import AutoProcessor, Qwen3_5ForConditionalGeneration # noqa: E402 # Same as joint_schema_model.load_release_model, but loaded to CPU first and moved with # .to("cuda") so the ZeroGPU hijack can pack the weights (device_map would bypass it). backbone = Qwen3_5ForConditionalGeneration.from_pretrained(MODEL_PATH, dtype=torch.bfloat16) 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(dtype=torch.bfloat16) model = ClefModel(backbone, head).eval().to("cuda") processor = AutoProcessor.from_pretrained(MODEL_PATH) MAX_IMAGE_SIDE = 1024 EX_SUPPORT_STATE = "Checkout has been failing for every customer for the last hour." EX_SUPPORT_Q = { "urgent": {"type": "noul", "instructions": "Is this support request urgent?"}, "team": { "type": "choice", "instructions": "Which team should handle this request?", "criteria": { "billing": "Payments, invoices, refunds", "technical": "Bugs, errors, outages", "sales": "Pricing, plans, new purchases", }, }, "severity": { "type": "score", "instructions": "How severe is the customer impact?", "criteria": ["No impact", "Minor", "Major", "Critical"], }, } 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_ROUTING_STATE = "Our checkout started returning errors and orders are blocked." EX_ROUTING_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_FOOD_STATE = "A user uploaded this photo to a restaurant review listing." EX_FOOD_Q = { "category": { "type": "choice", "instructions": "What does the photo mainly show?", "criteria": { "food": "A dish or drink", "interior": "The inside of a venue", "people": "People or portraits", "other": "Anything else", }, }, "appetizing": { "type": "score", "instructions": "How appetizing is the photo?", "criteria": ["Unappetizing", "Neutral", "Appetizing", "Very appetizing"], }, "safe": {"type": "noul", "instructions": "Is the photo safe to publish on the listing?"}, } EX_ROOM_STATE = {"listing": {"type": "apartment rental", "claimed_style": "minimalist"}} EX_ROOM_Q = { "room": { "type": "choice", "instructions": "Which room is shown?", "criteria": { "living_room": "Living room or lounge", "kitchen": "Kitchen", "bedroom": "Bedroom", "bathroom": "Bathroom", }, }, "matches_style": {"type": "noul", "instructions": "Does the photo match the claimed style?"}, "clutter": { "type": "score", "instructions": "How cluttered is the room?", "criteria": ["Spotless", "Tidy", "Some clutter", "Very cluttered"], }, } def _dump(value) -> str: return value if isinstance(value, str) else json.dumps(value, indent=2, ensure_ascii=False) def _parse_state(state: str): text = (state or "").strip() if not text: raise gr.Error("Please provide a state (plain text or JSON).") try: return json.loads(text) except json.JSONDecodeError: return text def _parse_questions(questions: str) -> dict: try: parsed = json.loads(questions or "") except json.JSONDecodeError as error: raise gr.Error(f"Questions must be valid JSON: {error}") if not isinstance(parsed, dict) or not parsed: raise gr.Error("Questions must be a non-empty JSON object keyed by question ID.") for question_id, question in parsed.items(): if not isinstance(question, dict) or question.get("type") not in QUESTION_TYPES: raise gr.Error(f"{question_id}: type must be one of noul, choice, score.") if question["type"] == "choice" and not isinstance(question.get("criteria"), dict): raise gr.Error(f"{question_id}: choice criteria must be an object of option -> description.") if question["type"] == "score" and not isinstance(question.get("criteria"), list): raise gr.Error(f"{question_id}: score criteria must be a list of level descriptions.") if question["type"] != "noul" and not question.get("criteria"): raise gr.Error(f"{question_id}: criteria must not be empty.") return parsed def _prep_image(image: Image.Image) -> Image.Image: image = image.convert("RGB") scale = MAX_IMAGE_SIDE / max(image.size) if scale < 1: image = image.resize((round(image.width * scale), round(image.height * scale)), Image.LANCZOS) return image def _render(questions: dict, probs: dict, latency_ms: float, n_tokens: int) -> str: cards = [] for question_id, question in questions.items(): p = probs[question_id] if question["type"] == "noul": options = [("true", "true", p["true"]), ("false", "false", p["false"])] elif question["type"] == "choice": options = [(k, f"{k} — {v}" if v else k, p[str(k)]) for k, v in question["criteria"].items()] else: options = [(str(i), f"{i} — {v}", p[str(i)]) for i, v in enumerate(question["criteria"])] best = max(options, key=lambda o: o[2])[0] rows = [] for key, label, value in options: highlight = "clef-best" if key == best else "" rows.append( f'
{html.escape(str(label))}
' f'
' f'
{value * 100:.1f}%
' ) extra = "" if question["type"] == "score": expected = sum(i * p[str(i)] for i in range(len(question["criteria"]))) extra = f' · expected score {expected:.2f}' instr = question.get("instructions") or question_id cards.append( f'
{html.escape(question_id)}' f'{question["type"]}
' f'
{html.escape(str(instr))}
' f'
→ {html.escape(str(best))}{extra}
{"".join(rows)}
' ) footer = f'
{n_tokens} input tokens · forward pass {latency_ms:.0f} ms
' return '
' + "".join(cards) + footer + "
" def _gpu_duration(record: dict) -> int: # Measured on xlarge: ~0.4 s for a short schema, ~2-3 s with an image, ~12 s at ~13k tokens. # A 15 s floor is needed so a cold worker (streaming 55 GB of weights) is not aborted. chars = len(json.dumps(record["state"], ensure_ascii=False)) + len(json.dumps(record["questions"])) seconds = 15 + chars / 2000 + (4 if record.get("images") else 0) return int(min(40, seconds)) @spaces.GPU(duration=_gpu_duration, size="xlarge") def _forward(record: dict): encoded = encode_record(processor.tokenizer, record, processor=processor) batch = collate_records([encoded], processor.tokenizer.pad_token_id, torch.device("cuda")) torch.cuda.synchronize() start = time.perf_counter() with torch.inference_mode(): logits = model(batch)[0] torch.cuda.synchronize() latency_ms = (time.perf_counter() - start) * 1000 probs = { q.question_id: dict(zip(q.option_ids, ql.float().softmax(-1).cpu().tolist())) for q, ql in zip(encoded.questions, logits) } return probs, latency_ms, len(encoded.input_ids) def decide(state: str, questions: str, image: Image.Image | None = None): """Answer a typed question schema about a state with Cloudflare Clef. Args: state: The situation to decide on, as plain text or a JSON value. questions: 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). image: Optional image that is part of the state. Returns: An HTML rendering of per-option probabilities and a SystemOne-style JSON response with the answer, confidence and probabilities for every question. """ parsed_questions = _parse_questions(questions) record = {"state": _parse_state(state), "questions": parsed_questions} if image is not None: record["images"] = [_prep_image(image)] probs, latency_ms, n_tokens = _forward(record) response = { "model": "clef", "answers": {qid: systemone_answer(parsed_questions[qid], probs[qid]) for qid in parsed_questions}, "usage": {"input_tokens": n_tokens, "output_tokens": 0}, } return _render(parsed_questions, probs, latency_ms, n_tokens), response CSS = """ #col-container { max-width: 1200px; margin: 0 auto; } .dark .gradio-container { color: var(--body-text-color); } .clef-wrap { display: flex; flex-direction: column; gap: 12px; } .clef-card { border: 1px solid var(--border-color-primary); border-radius: 10px; padding: 12px 14px; } .clef-head { display: flex; gap: 8px; align-items: center; margin-bottom: 2px; } .clef-id { font-weight: 700; font-family: var(--font-mono); } .clef-type { font-size: 0.75em; padding: 1px 8px; border-radius: 999px; background: var(--color-accent-soft); } .clef-instr { opacity: 0.8; font-size: 0.92em; } .clef-answer { margin: 6px 0 8px; } .clef-row { display: grid; grid-template-columns: minmax(120px, 40%) 1fr 56px; gap: 8px; align-items: center; margin: 3px 0; font-size: 0.9em; } .clef-label { overflow: hidden; text-overflow: ellipsis; white-space: nowrap; } .clef-bar { height: 12px; border-radius: 6px; background: var(--block-background-fill); border: 1px solid var(--border-color-primary); overflow: hidden; } .clef-fill { height: 100%; background: var(--neutral-400); } .clef-best .clef-fill { background: var(--color-accent); } .clef-best .clef-label { font-weight: 600; } .clef-val { text-align: right; font-family: var(--font-mono); } .clef-foot { font-size: 0.8em; opacity: 0.7; } """ with gr.Blocks(title="Clef decision model") as demo: with gr.Column(elem_id="col-container"): gr.Markdown( "# Clef: typed decisions from one forward pass\n" "[Cloudflare/clef](https://huggingface.co/Cloudflare/clef) is a 27B multimodal decision model. " "Give it a **state** (text, JSON, optionally an image) and a **schema of typed questions**: " "`noul` (true/false), `choice` (named options) or `score` (ordered levels). " "It returns a probability for every allowed option. No text is generated. " "· [Blog](https://blog.cloudflare.com/clef-decision-models)" ) with gr.Row(): with gr.Column(scale=1): state = gr.Textbox( label="State (plain text or JSON)", lines=5, placeholder="Describe the situation, paste a JSON record, a support ticket, a log line…", ) image = gr.Image(label="Image (optional)", type="pil", height=240) questions = gr.Code( label="Questions (JSON schema)", language="json", value=_dump(EX_SUPPORT_Q), lines=18, ) run = gr.Button("Decide", variant="primary") with gr.Column(scale=1): result_html = gr.HTML(label="Decisions") with gr.Accordion("SystemOne JSON response", open=False): result_json = gr.JSON(label="Response") run.click(decide, inputs=[state, questions, image], outputs=[result_html, result_json]) gr.Examples( label="Text examples", examples=[ [EX_SUPPORT_STATE, _dump(EX_SUPPORT_Q)], [_dump(EX_INVOICE_STATE), _dump(EX_INVOICE_Q)], [EX_ROUTING_STATE, _dump(EX_ROUTING_Q)], ], inputs=[state, questions], outputs=[result_html, result_json], fn=decide, cache_examples=True, cache_mode="lazy", ) gr.Examples( label="Image examples", examples=[ [EX_FOOD_STATE, _dump(EX_FOOD_Q), "examples/gourmet_burger.jpg"], [_dump(EX_ROOM_STATE), _dump(EX_ROOM_Q), "examples/living_room_minimalist.jpg"], ], inputs=[state, questions, image], outputs=[result_html, result_json], fn=decide, cache_examples=True, cache_mode="lazy", ) if __name__ == "__main__": demo.launch(theme=gr.themes.Citrus(), css=CSS, mcp_server=True)