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 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, | |
| ) | |
| # Authors' load_release_model uses device_map, which initializes CUDA before the | |
| # ZeroGPU patch can intercept it. Load eagerly with .to("cuda") instead. | |
| 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 | |
| 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) | |
| def render_html(questions: dict, answers: dict, probs: dict, latency_ms: float, n_tokens: int) -> str: | |
| parts = ['<div class="clef-out">'] | |
| 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(str(ans['choice']))} · {ans['confidence']:.3f}" | |
| labels = {k: html.escape(str(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'<div class="clef-q"><div class="clef-qh"><span class="clef-id">{html.escape(str(qid))}</span>' | |
| f'<span class="clef-type">{html.escape(qtype)}</span></div>' | |
| f'<div class="clef-instr">{instr}</div><div class="clef-head">{headline}</div>' | |
| ) | |
| 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'<div class="clef-row"><div class="clef-lbl">{labels.get(opt, html.escape(str(opt)))}</div>' | |
| f'<div class="clef-track"><div class="{cls}" style="width:{pct:.1f}%"></div></div>' | |
| f'<div class="clef-pct">{pct:.1f}%</div></div>' | |
| ) | |
| parts.append("</div>") | |
| parts.append( | |
| f'<div class="clef-meta">{n_tokens} input tokens · forward pass {latency_ms:.0f} ms</div></div>' | |
| ) | |
| return "".join(parts) | |
| def _parse_state(state: str): | |
| text = (state or "").strip() | |
| if text[:1] in "{[": | |
| try: | |
| return json.loads(text) | |
| except json.JSONDecodeError: | |
| pass | |
| return text | |
| 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, question in questions.items(): | |
| if not isinstance(question, dict) or question.get("type") not in QUESTION_TYPES: | |
| raise gr.Error(f"{qid}: type must be noul, choice, or score.") | |
| if question["type"] == "choice" and not ( | |
| isinstance(question.get("criteria"), dict) and question["criteria"] | |
| ): | |
| raise gr.Error(f"{qid}: a choice question needs criteria as option id → description.") | |
| if question["type"] == "score" and not ( | |
| isinstance(question.get("criteria"), list) and question["criteria"] | |
| ): | |
| raise gr.Error(f"{qid}: a score question needs criteria as a list of level descriptions.") | |
| 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` for true/false, `choice` for named options, `score` for ordered | |
| levels), optional instructions, and criteria. | |
| image: Optional image that is part of the state. | |
| Returns: | |
| An HTML view of per-option probabilities and a SystemOne-style JSON response. | |
| """ | |
| try: | |
| questions = json.loads(questions_json) | |
| except json.JSONDecodeError as exc: | |
| raise gr.Error(f"Questions JSON is invalid: {exc}") from exc | |
| _validate(questions) | |
| if not (state or "").strip() and image is None: | |
| raise gr.Error("Provide a state, an image, or both.") | |
| 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 exc: | |
| raise gr.Error(f"Could not encode the request: {exc}") from exc | |
| batch = collate_records([encoded], processor.tokenizer.pad_token_id, torch.device("cuda")) | |
| torch.cuda.synchronize() | |
| started = time.perf_counter() | |
| with torch.inference_mode(): | |
| logits = model(batch)[0] | |
| torch.cuda.synchronize() | |
| latency_ms = (time.perf_counter() - started) * 1000 | |
| probs, answers = {}, {} | |
| for encoded_question, question_logits in zip(encoded.questions, logits): | |
| probabilities = dict( | |
| zip(encoded_question.option_ids, question_logits.float().softmax(-1).tolist()) | |
| ) | |
| probs[encoded_question.question_id] = probabilities | |
| answers[encoded_question.question_id] = systemone_answer( | |
| questions[encoded_question.question_id], probabilities | |
| ) | |
| response = { | |
| "model": "clef-flash", | |
| "answers": answers, | |
| "usage": {"input_tokens": len(encoded.input_ids), "output_tokens": 0}, | |
| } | |
| print(f"clef-flash forward {latency_ms:.1f} ms tokens={len(encoded.input_ids)}", flush=True) | |
| return render_html(questions, answers, probs, latency_ms, len(encoded.input_ids)), response | |
| CSS = """ | |
| #col-container { max-width: 1200px; margin: 0 auto; } | |
| .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); } | |
| .clef-type { font-size: 0.75em; padding: 1px 8px; border-radius: 99px; background: var(--color-accent-soft); } | |
| .clef-instr { font-size: 0.9em; opacity: 0.8; margin: 4px 0 6px; } | |
| .clef-head { font-weight: 600; margin-bottom: 8px; } | |
| .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; } | |
| .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); } | |
| .clef-meta { font-size: 0.8em; opacity: 0.7; } | |
| """ | |
| with gr.Blocks(title="Clef-Flash") as demo: | |
| with gr.Column(elem_id="col-container"): | |
| gr.Markdown( | |
| "# Clef-Flash\n" | |
| "Give [Cloudflare/clef-flash](https://huggingface.co/Cloudflare/clef-flash) a **state** " | |
| "(text, JSON, or an image) and a **schema of typed questions**. It returns a probability " | |
| "for every allowed option of every question in one forward pass — no text generation and " | |
| "no output parsing.\n\n" | |
| "`noul` is true/false, `choice` is named options, `score` is ordered levels. " | |
| "[Announcement](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, 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 or JSON") | |
| 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_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) | |