Spaces:
Running on Zero
Running on Zero
| 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'<div class="clef-row {highlight}"><div class="clef-label">{html.escape(str(label))}</div>' | |
| f'<div class="clef-bar"><div class="clef-fill" style="width:{value * 100:.1f}%"></div></div>' | |
| f'<div class="clef-val">{value * 100:.1f}%</div></div>' | |
| ) | |
| extra = "" | |
| if question["type"] == "score": | |
| expected = sum(i * p[str(i)] for i in range(len(question["criteria"]))) | |
| extra = f' · expected score <b>{expected:.2f}</b>' | |
| instr = question.get("instructions") or question_id | |
| cards.append( | |
| f'<div class="clef-card"><div class="clef-head"><span class="clef-id">{html.escape(question_id)}</span>' | |
| f'<span class="clef-type">{question["type"]}</span></div>' | |
| f'<div class="clef-instr">{html.escape(str(instr))}</div>' | |
| f'<div class="clef-answer">→ <b>{html.escape(str(best))}</b>{extra}</div>{"".join(rows)}</div>' | |
| ) | |
| footer = f'<div class="clef-foot">{n_tokens} input tokens · forward pass {latency_ms:.0f} ms</div>' | |
| return '<div class="clef-wrap">' + "".join(cards) + footer + "</div>" | |
| 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)) | |
| 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) | |