Blame for live/app.py

"""Side-by-side demo: the same small model answers a math question with no tools and with a
sandboxed run_python tool. Runs llama.cpp's llama-server on the Space's free CPU."""
import os, re, json, time, tarfile, subprocess, urllib.request, threading, glob
import gradio as gr
from huggingface_hub import hf_hub_download
import sandbox
from proxy import chat_with_python

LLAMA_TAG = "b11938"
MODEL_REPO, MODEL_FILE = "LiquidAI/LFM3-1.7B-GGUF", "LFM3-1.7B-QAD-Q4_0.gguf"
PORT = 8080
URL = os.environ.get("LLAMA_URL", f"http://127.0.0.1:{PORT}")
TOOL_REPO = "https://huggingface.co/austindixson/python-sandbox-math-tool"
SYS = "Solve the math problem. Reason step by step, then put the final answer in \\boxed{}."
_ready = threading.Event(); _status = {"msg": "starting"}

def _start_server():
    if os.environ.get("LLAMA_URL"):
        _ready.set(); return
    try:
        d = "/tmp/llama"
        if not glob.glob(f"{d}/**/llama-server", recursive=True):
            _status["msg"] = "downloading llama.cpp"
            os.makedirs(d, exist_ok=True)
            tgz = f"{d}/llama.tgz"
            urllib.request.urlretrieve(f"https://github.com/ggml-org/llama.cpp/releases/download/{LLAMA_TAG}/llama-{LLAMA_TAG}-bin-ubuntu-x64.tar.gz", tgz)
            with tarfile.open(tgz) as t: t.extractall(d)
        srv = glob.glob(f"{d}/**/llama-server", recursive=True)[0]
        _status["msg"] = "downloading model"
        gguf = hf_hub_download(MODEL_REPO, MODEL_FILE)
        _status["msg"] = "loading model"
        env = dict(os.environ, LD_LIBRARY_PATH=os.path.dirname(srv))
        subprocess.Popen([srv, "-m", gguf, "-c", "8192", "-np", "1", "-t", os.environ.get("LLAMA_THREADS", "2"),
                          "--host", "127.0.0.1", "--port", str(PORT), "--jinja"], env=env,
                         stdout=open("/tmp/llama.log", "w"), stderr=subprocess.STDOUT, cwd=os.path.dirname(srv))
        for _ in range(300):
            try:
                if json.load(urllib.request.urlopen(URL + "/health", timeout=2)).get("status") == "ok":
                    _ready.set(); _status["msg"] = "ready"; return
            except Exception: pass
            time.sleep(1)
        _status["msg"] = "model server did not start: " + open("/tmp/llama.log").read()[-500:]
    except Exception as e:
        _status["msg"] = f"startup failed: {e}"
threading.Thread(target=_start_server, daemon=True).start()

EXAMPLES = [
    "What is the sum of the digits of 2^1000?",
    "What is the remainder when 7^2026 is divided by 1000?",
    "What is 987654321 × 123456789?",
    "How many ways can you make change for one dollar using pennies, nickels, dimes and quarters?",
    "How many primes are there below 100,000?",
]

def _boxed(s):
    i = s.rfind("\\boxed{")
    if i < 0: return None
    j, d = i + 7, 1
    for k in range(j, len(s)):
        d += {"{": 1, "}": -1}.get(s[k], 0)
        if d == 0: return s[j:k]
    return None

def _one(q, use_tool, max_tokens):
    body = {"model": "lfm3", "max_tokens": int(max_tokens), "temperature": 0.0, "top_k": 1,
            "chat_template_kwargs": {"enable_thinking": False},
            "messages": [{"role": "system", "content": SYS}, {"role": "user", "content": q}]}
    t = time.time(); r = chat_with_python(URL, body, use_tool=use_tool); dt = time.time() - t
    m = r["choices"][0]["message"]; txt = re.sub(r"<think>.*?</think>", "", m.get("content") or "", flags=re.S).strip()
    x = r["x_python_tool"]; fin = r["choices"][0].get("finish_reason")
    stats = (f"**Final answer:** `{_boxed(txt) or '(no boxed answer)'}`  \n"
             f"**Generated tokens:** {r['usage']['completion_tokens']:,}  ·  **Time:** {dt:.1f} s  ·  "
             f"**Model calls:** {x['llm_calls']}  ·  **Tool calls:** {len(x['calls'])}"
             + ("  \n⚠️ hit the token cap" if fin == "length" else ""))
    calls = "\n\n".join(f"**run_python #{i}**\n```python\n{c['code']}\n```\n```text\n{c['output']}\n```" for i, c in enumerate(x["calls"], 1)) or "_no tool calls_"
    return stats, txt, calls

def compare(q, max_tokens):
    if not q.strip():
        yield "Type a question first.", "", "", "", "", ""; return
    if not _ready.wait(timeout=1):
        yield f"Model is still loading ({_status['msg']}). Try again in a minute.", "", "", "", "", ""; return
    wait = "_running..._"
    yield wait, "", "", "_waiting for the no-tools run to finish..._", "", ""
    a = _one(q, False, max_tokens)
    yield a[0], a[1], "", wait, "", ""
    b = _one(q, True, max_tokens)
    yield a[0], a[1], "", b[0], b[1], b[2]

RECORDED = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "RECORDED.md")).read()

backend = sandbox.pick_backend()
with gr.Blocks(title="Python sandbox tool: with vs without") as demo:
    gr.Markdown(f"""# Small model, with and without a Python tool
The same model (LiquidAI **LFM3-1.7B**, QAD-Q4_0 GGUF, llama.cpp {LLAMA_TAG}, greedy, thinking off) answers your
question twice: once on its own, once with a `run_python` tool it can call up to 6 times. Runs on this Space's
free 2-vCPU CPU, so expect roughly 10–60 s per side. Tool code: [{TOOL_REPO.split('co/')[1]}]({TOOL_REPO}).

**Sandbox here:** `{backend}`: {sandbox.ISOLATION.get(backend)}. On a Mac the tool uses `sandbox-exec`
(network denied by the OS, writes only in a temp dir); on Linux with working user namespaces it uses bubblewrap.""")
    with gr.Tab("Try it"):
        with gr.Row():
            q = gr.Textbox(label="Math question", value=EXAMPLES[0], lines=2, scale=4)
            mt = gr.Slider(256, 2048, value=1024, step=128, label="Token cap per answer (demo only)", scale=1)
        gr.Examples(EXAMPLES, inputs=q)
        go = gr.Button("Run both", variant="primary")
        with gr.Row():
            with gr.Column():
                gr.Markdown("### No tools")
                s1 = gr.Markdown(); a1 = gr.Markdown(); c1 = gr.Markdown(visible=False)
            with gr.Column():
                gr.Markdown("### With run_python")
                s2 = gr.Markdown(); a2 = gr.Markdown()
                with gr.Accordion("Tool calls and outputs", open=True):
                    c2 = gr.Markdown()
        go.click(compare, [q, mt], [s1, a1, c1, s2, a2, c2], concurrency_limit=1)
        gr.Markdown("The token cap here is a demo setting. Every recorded benchmark run used a 25,000-token budget per problem.")
    with gr.Tab("Recorded results"):
        gr.Markdown(RECORDED)
        with gr.Row():
            gr.Image("fig_tokens.png", label="Tokens per problem", show_label=True)
            gr.Image("fig_aime.png", label="Hard math (AIME 2025)", show_label=True)

if __name__ == "__main__":
    demo.queue(default_concurrency_limit=1).launch()