Spaces:
Sleeping
Sleeping
| """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"<div style='margin:14px 0 6px;font-weight:600'>{html.escape(qid)} <span style='font-weight:400;opacity:.65'>{a['type']}" + (f", expected level {a['score']}" if a["type"] == "score" else "") + "</span></div>" | |
| lines = "".join(f"<div style='display:flex;align-items:center;gap:8px;margin:3px 0;font-size:13px'><div style='width:150px;text-align:right;overflow:hidden;text-overflow:ellipsis;white-space:nowrap'>{html.escape(str(k))}</div>" | |
| f"<div style='flex:1;height:12px;background:rgba(128,128,128,.18);border-radius:4px'><div style='width:{100 * v:.1f}%;height:100%;border-radius:4px;background:{'#4f46e5' if k == top else '#a5b4fc'}'></div></div>" | |
| f"<div style='width:48px;font-family:monospace'>{v:.3f}</div></div>" 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() | |