"""lev on Hugging Face Spaces: one state, several typed questions, one forward pass on CPU. Local run: LEV_RUN=runs/v4-350m-subset/00-trial-0/checkpoint uv run --with gradio python space/app.py """ import html, json, os, time import gradio as gr import torch from huggingface_hub import snapshot_download from peft import PeftModel from lev.api import SystemOneRequest, to_answers, to_record from lev.model import DecisionModel, load_tokenizer RUN = os.environ.get("LEV_RUN", "franckverrot/lev-350m") MAX_TOKENS = 2048 HERE = os.path.dirname(os.path.abspath(__file__)) FIGURE = next((p for p in (os.path.join(HERE, "lev-architecture.png"), os.path.join(HERE, "..", "docs", "lev-architecture.png")) if os.path.exists(p)), None) def load(run): path = run if os.path.isdir(run) else snapshot_download(run, allow_patterns=["*.json", "*.safetensors", "*.pt", "*.jinja"]) meta = torch.load(os.path.join(path, "head.pt"), map_location="cpu") tok = load_tokenizer(meta["base"], revision=meta.get("base_revision")) model = DecisionModel(meta["base"], tok, "cpu", revision=meta.get("base_revision"), head_dim=meta.get("head_dim", 256), option_isolation=meta.get("option_isolation", False)) model.lm = PeftModel.from_pretrained(model.lm, path) model.head.load_state_dict(meta["head"]); model.eval() calibration = os.path.join(path, "calibration.json") temperature = json.load(open(calibration))["temperature"] if os.path.exists(calibration) else 1.0 return tok, model, float(temperature), meta["base"] TOK, MODEL, TEMPERATURE, BASE = load(RUN) EXAMPLES = [ ["Shoes arrived two weeks late and in the wrong size. Also I see two charges on my card.", {"department": {"type": "choice", "instructions": "Which team should handle this?", "criteria": {"returns": "Exchanges, refunds, wrong or damaged items", "shipping": "Delivery status, delays, lost packages", "billing": "Charges, invoices, payment problems"}}, "escalate": {"type": "noul", "instructions": "Does this need urgent human attention?"}, "frustration": {"type": "score", "instructions": "How frustrated is the customer?", "criteria": ["Calm", "Frustrated", "Very angry"]}}], ["The central bank held rates steady on Thursday and signalled two cuts next year as inflation cooled to 2.4 percent.", {"topic": {"type": "choice", "instructions": "What is the topic of this article?", "criteria": {"world": None, "sports": None, "business": None, "scitech": None}}, "is_finance": {"type": "noul", "instructions": "Is this article about finance?"}}], ["The room was clean and the staff were friendly, but the wifi dropped every ten minutes and breakfast ran out by eight.", {"stars": {"type": "score", "instructions": "How many stars did this reviewer give?", "criteria": ["1 star", "2 stars", "3 stars", "4 stars", "5 stars"]}, "recommend": {"type": "noul", "instructions": "Would this reviewer recommend the hotel?"}}], ] def bars(answers): rows = [] for qid, a in answers.items(): if a["type"] == "noul": dist, top = {"yes": a["noul"], "no": round(1 - a["noul"], 3)}, "yes" if a["noul"] >= 0.5 else "no" elif a["type"] == "choice": dist, top = a["probabilities"], a["choice"] else: dist = {a["legend"][k]: v for k, v in a["probabilities"].items()}; top = max(dist, key=dist.get) head = f"
{html.escape(qid)} {a['type']}" + (f", expected level {a['score']}" if a["type"] == "score" else "") + "
" lines = "".join(f"
{html.escape(str(k))}
" f"
" f"
{v:.3f}
" for k, v in dist.items()) rows.append(head + lines) return "".join(rows) def answer(state_text, questions): try: text = state_text.strip() state = json.loads(text) if text[:1] in "{[" else state_text except json.JSONDecodeError: state = state_text try: if isinstance(questions, str): questions = json.loads(questions) request = SystemOneRequest.model_validate({"state": state, "model": "lev-latest", "questions": questions}) record, meta = to_record(request) enc = MODEL.encode(TOK, record, max_state=MAX_TOKENS, max_branch=MAX_TOKENS) except Exception as error: raise gr.Error(f"Invalid request: {error}") if len(enc["ids"]) > MAX_TOKENS: raise gr.Error(f"This Space accepts up to {MAX_TOKENS} packed tokens; the request has {len(enc['ids'])}.") started = time.perf_counter() with torch.no_grad(): probs = [torch.softmax(z / TEMPERATURE, -1).tolist() for z in MODEL.forward(enc)] answers = to_answers(probs, meta) return bars(answers), {"model": RUN, "answers": answers, "usage": {"input_tokens": len(enc["ids"])}, "latency_ms": round(1000 * (time.perf_counter() - started), 1)} with gr.Blocks(title="lev") as demo: gr.Markdown("# lev\nA clone of [kev](https://github.com/jaredpalmer/kev), a Jev-style decision model, on [LiquidAI LFM2.5-350M](https://huggingface.co/LiquidAI/LFM2.5-350M). " "Write a state, define typed questions, and get a probability for every option from one forward pass. " "Code: [github.com/franckverrot/lev](https://github.com/franckverrot/lev). Weights: " + (f"`{RUN}`" if os.path.isdir(RUN) else f"[{RUN}](https://huggingface.co/{RUN})") + ". Running on CPU.") with gr.Row(): with gr.Column(): state = gr.Textbox(label="State (text or JSON)", lines=6, value=EXAMPLES[0][0]) questions = gr.Code(label="Questions (JSON): noul = yes/no, choice = options with descriptions, score = ordered levels", language="json", lines=16, value=json.dumps(EXAMPLES[0][1], indent=2)) run = gr.Button("Answer", variant="primary") with gr.Column(): chart = gr.HTML(label="Probabilities") raw = gr.JSON(label="Response, same shape as POST /v1/systemone") gr.Examples([[s, json.dumps(q, indent=2)] for s, q in EXAMPLES], inputs=[state, questions], label="Examples") if FIGURE: with gr.Accordion("How it works", open=False): gr.Image(FIGURE, show_label=False, interactive=False) run.click(answer, [state, questions], [chart, raw]) if __name__ == "__main__": demo.launch()