app.py

"""Glyph v2 Space — the REAL finetuned model (ZeroGPU).

Pipeline per message: English → speaker (LoRA) → glyph message → builder (LoRA)
→ Python → executed. Glyph input goes straight to builder + translator.
"""
import os
import subprocess
import sys
import tempfile

import gradio as gr
import spaces
import torch
from peft import PeftModel
from transformers import AutoModelForCausalLM, AutoTokenizer

BASE = "Qwen/Qwen2.5-Coder-3B-Instruct"
ADAPTER = "robertkeus/glyph-adapters"
DEMO = {
    "IL": [3, -1, 2, 2, -5],
    "SL": ["hello", "World", "", "ada"],
    "RL": [{"id": 1, "name": "Ada", "age": 36, "email": "ada@x.io", "password": "s3cretpw"},
           {"id": 2, "name": "Bo", "age": 12, "email": "", "password": "pw"},
           {"id": 3, "name": "Cy", "age": 65, "email": "cy@y.com", "password": "hunter2"}],
}


def demo_for(code):
    """Pick the demo input by sniffing the generated code's loop variable/field access."""
    if "d[" in code:
        return DEMO["RL"]
    if " s in r" in code or "(s)" in code or ".join(r)" in code or "key=len" in code:
        return DEMO["SL"]
    return DEMO["IL"]

PROMPT = {
    "speaker": "Encode this task as glyph symbols.\nTask: {x}\nSymbols:",
    "builder": "Write Python for this glyph message.\nSymbols: {x}\nCode:",
    "translator": "Translate this glyph message into English.\nSymbols: {x}\nEnglish:",
}

tok = AutoTokenizer.from_pretrained(BASE)
model = AutoModelForCausalLM.from_pretrained(BASE, torch_dtype=torch.float32)  # CPU-safe dtype
model = PeftModel.from_pretrained(model, ADAPTER, subfolder=os.environ.get("GLYPH_V", "v4"),
                                  torch_device="cpu")
model.eval()
_dev = {"d": "cpu"}  # moved to cuda lazily inside the GPU context if available

# wire glyphs are CJK (0x4E00 block); display remap to syllabics (0x1400) = alien look
_A, _C = 0x1400, 0x4E00
alien = lambda s: "".join(chr(_A + ord(c) - _C) if 0x4E00 <= ord(c) <= 0x9FFF else c for c in s)
unalien = lambda s: "".join(chr(_C + ord(c) - _A) if 0x1400 <= ord(c) <= 0x167F else c for c in s)
is_glyphs = lambda s: all(0x4E00 <= ord(c) <= 0x9FFF for c in s)


def _gen(prompt, max_new):
    enc = tok(prompt, return_tensors="pt", add_special_tokens=False).to(_dev["d"])
    with torch.no_grad():
        out = model.generate(**enc, max_new_tokens=max_new, do_sample=False,
                             pad_token_id=tok.eos_token_id)
    return tok.decode(out[0][enc.input_ids.shape[1]:], skip_special_tokens=True).strip()


@spaces.GPU(duration=40)
def pipeline(msg, raw_glyphs):
    """ONE GPU context per message (speaker+translator+builder) — 1 quota charge, not 3."""
    global model
    if _dev["d"] == "cpu" and torch.cuda.is_available():
        model = model.to("cuda", dtype=torch.bfloat16)
        _dev["d"] = "cuda"
    glyphs = raw_glyphs or _gen(PROMPT["speaker"].format(x=msg), 24)
    if not glyphs or not is_glyphs(glyphs):
        return {"error": "speaker produced no valid glyphs", "got": (glyphs or "")[:40]}
    code = _gen(PROMPT["builder"].format(x=glyphs), 200)
    return {"glyphs": glyphs, "alien": alien(glyphs), "bytes": 2 * len(glyphs),
            "english": _gen(PROMPT["translator"].format(x=glyphs), 60), "code": code}


def run_code(code):
    demo = demo_for(code)
    src = f"{code}\n\nprint(solve({demo!r}))\n"
    with tempfile.NamedTemporaryFile("w", suffix=".py", delete=False) as f:
        f.write(src); path = f.name
    try:
        p = subprocess.run([sys.executable, "-I", path], capture_output=True,
                           text=True, timeout=5)
        out = p.stdout.strip() if p.returncode == 0 else "error: " + p.stderr.strip()[-120:]
    except Exception as e:
        out = f"error: {type(e).__name__}"
    finally:
        os.unlink(path)
    return demo, out


def respond(msg, _history):
    msg = (msg or "").strip()[:300]
    if not msg:
        return "Ask a list/string/record task — I answer in my glyph language, then code."
    d = api_json(msg)
    if "error" in d:
        return f"⚠️ {d['error']}"
    return (f"**glyph message:** {d['alien']}  ·  {d['bytes']} bytes\n\n"
            f"**model reads it as:** {d['english']}\n\n"
            f"```python\n{d['code']}\n```\n\n`solve({d['input']})` → `{d['result']}`")


def api_json(msg):
    """Structured endpoint for custom clients: {glyphs, alien, english, code, result}."""
    msg = (msg or "").strip()[:300]
    raw = unalien(msg)
    try:
        d = pipeline(msg, raw if is_glyphs(raw) else None)
        if "code" in d:
            d["input"], d["result"] = run_code(d["code"])
        return d
    except Exception as e:
        return {"error": f"{type(e).__name__}: {str(e)[:150]}"}


EX = ["keep the positive numbers, square each, then return their sum",
      "keep values greater than 7, then count them",
      "drop records where email is empty, then extract the name of each record",
      "delete the record with id 2, then count the records",
      "uppercase each string, then render them as an HTML unordered list"]

with gr.Blocks(title="Glyph v2") as demo:
    gr.ChatInterface(respond, examples=EX, title="Glyph v2 — live finetuned model",
                     description="Every reply is the real trained pipeline: speaker → glyphs "
                                 "→ builder → executed Python. (v4: 99 prims incl. operands, CRUD, UI; builder 99.5% held-out)")
    _i = gr.Textbox(visible=False)
    _o = gr.JSON(visible=False)
    _b = gr.Button(visible=False)
    _b.click(api_json, _i, _o, api_name="glyph")   # POST /gradio_api/call/glyph
demo.launch()