"""Forge model server v2 — free HF ZeroGPU Gradio Space. Design constraints (all verified empirically, not assumed): * Free accounts: creating Docker or cpu-basic Spaces returns HTTP 402. ZeroGPU Gradio Spaces are the only free compute. ZeroGPU REQUIRES a function decorated with @spaces.GPU to exist at startup, or the Space is killed with "No @spaces.GPU function detected during startup" — v1 died exactly there. * Free ZeroGPU quota is 5 min/day on an RTX Pro 6000 Blackwell (48 GB), so the GPU is a fast path, not the workhorse. Unlimited generations run on a CPU engine: llamafile (prebuilt Cosmopolitan binary, no compiler needed) serving a Q4_K_M GGUF on localhost:7861, proxied through this app. * Clients choose the engine with the model name: anything ending in "-gpu" takes the ZeroGPU path; everything else takes the unlimited CPU path. * Gradio 6's default SSR mode runs a Node.js proxy that owns the port; ssr_mode=False serves straight from Python, which is what lets the OpenAI-compatible /v1/* routes below live on the same port as the UI. * `import spaces` must come before any CUDA-touching import: gradio imports spaces lazily in a watcher thread, and if torch has already initialised CUDA that import raises and kills the app. """ import spaces # noqa: F401 — import order is the point; keep this first import errno import functools import json import os import shutil import subprocess import threading import time import urllib.error import urllib.request from pathlib import Path # ---------------------------------------------------------------- config ---- MODEL_ID = os.environ.get("MODEL_ID", "mlabonne/Qwen3-4B-abliterated") GGUF_REPO = os.environ.get("GGUF_REPO", "mradermacher/Qwen3-4B-abliterated-GGUF") GGUF_FILE = os.environ.get("GGUF_FILE", "Qwen3-4B-abliterated.Q4_K_M.gguf") LLAMAFILE_URL = os.environ.get( "LLAMAFILE_URL", "https://github.com/mozilla-ai/llamafile/releases/download/0.10.5/llamafile-0.10.5-thin", ) MODEL_ALIAS = os.environ.get("MODEL_ALIAS", "qwen3-4b-abliterated") GPU_ALIAS = MODEL_ALIAS + "-gpu" ENGINE_DIR = Path(os.environ.get("FORGE_ENGINE_DIR", str(Path.home() / "forge-engine"))) LLAMA_PORT = int(os.environ.get("LLAMA_PORT", "7861")) PORT = int(os.environ.get("PORT", "7860")) ENGINE_BOOT_WAIT = int(os.environ.get("ENGINE_BOOT_WAIT", "600")) # cold download is ~2.6 GB GPU_DURATION = int(os.environ.get("GPU_DURATION", "120")) ENGINE_BIN = ENGINE_DIR / "llamafile" ENGINE_URL = "http://127.0.0.1:%d" % LLAMA_PORT _engine = {"proc": None, "ready": False, "err": None, "since": 0.0} _engine_lock = threading.Lock() _gpu_lock = threading.Lock() _gpu = {"model": None, "tok": None, "err": None} # ---------------------------------------------------------------- gpu path ---- def _gpu_load(): """Lazy-load the transformers model. On ZeroGPU this runs inside a @spaces.GPU call, so real CUDA is present. Elsewhere it falls back to CPU (which is how this file is tested on a machine with no GPU at all). Plain bf16, deliberately: the card has 48 GB so there is no reason to quantize a 4B model, and nf4 was measurably dumber — it copied the system-prompt example verbatim instead of designing the requested level (verified 2026-09-12). The Q4_K_M GGUF on the CPU path does not do this.""" if _gpu["model"] is not None: return _gpu["model"], _gpu["tok"] import torch from transformers import AutoModelForCausalLM, AutoTokenizer tok = AutoTokenizer.from_pretrained(MODEL_ID) device = "cuda" if torch.cuda.is_available() else "cpu" dtype = torch.bfloat16 if device == "cuda" else torch.float32 try: model = AutoModelForCausalLM.from_pretrained( MODEL_ID, dtype=dtype, device_map=device, ) except Exception as exc: # older transformers: torch_dtype spelling print("[gpu] dtype load failed (%r); trying torch_dtype" % exc, flush=True) model = AutoModelForCausalLM.from_pretrained( MODEL_ID, torch_dtype=dtype, device_map=device, ) model.eval() _gpu["model"], _gpu["tok"] = model, tok return model, tok def _gpu_generate_inner(messages, max_tokens, temperature): import torch model, tok = _gpu_load() # Qwen3 hybrid thinking: disable it — GameForge needs bare JSON, no . try: text = tok.apply_chat_template( messages, tokenize=False, add_generation_prompt=True, enable_thinking=False, ) except TypeError: # template without the kwarg text = tok.apply_chat_template(messages, tokenize=False, add_generation_prompt=True) inputs = tok(text, return_tensors="pt").to(model.device) with torch.inference_mode(): out = model.generate( **inputs, max_new_tokens=int(max_tokens), do_sample=float(temperature) > 0, temperature=max(float(temperature), 1e-4), top_p=0.95, pad_token_id=tok.eos_token_id, ) return tok.decode(out[0][inputs["input_ids"].shape[1]:], skip_special_tokens=True) @spaces.GPU(duration=GPU_DURATION) def gpu_generate(messages, max_tokens=1200, temperature=0.7): """ZeroGPU fast path. Quota: ~5 min/day on a free account. If the cached model references a stale CUDA context (GPU detached between calls), drop it and reload once before giving up.""" with _gpu_lock: try: return _gpu_generate_inner(messages, max_tokens, temperature) except Exception as first: print("[gpu] generate failed (%r); reloading once" % first, flush=True) _gpu["model"] = _gpu["err"] = None try: return _gpu_generate_inner(messages, max_tokens, temperature) except Exception as second: _gpu["err"] = repr(second) raise # ---------------------------------------------------------------- cpu engine ---- def _log(tag, msg): print("[%s] %s" % (tag, msg), flush=True) def _download(url, dest: Path): dest.parent.mkdir(parents=True, exist_ok=True) tmp = dest.with_suffix(dest.suffix + ".part") _log("engine", "downloading %s -> %s" % (url, dest)) with urllib.request.urlopen(url, timeout=120) as r, open(tmp, "wb") as fh: shutil.copyfileobj(r, fh, 1 << 20) tmp.rename(dest) def _engine_boot(): """Download llamafile + GGUF (if missing), start llama-server on localhost, wait for /health. Runs in a background thread so the Gradio app binds its port immediately — ZeroGPU validation only needs the decorated function.""" tag = "engine" try: with _engine_lock: if _engine["proc"] is not None and _engine["proc"].poll() is None: return # An engine from a previous app instance may still be bound (it is # daemonised, so it survives the parent). Reuse it if healthy. try: with urllib.request.urlopen(ENGINE_URL + "/health", timeout=2) as r: if r.status == 200: _engine["ready"] = True _log(tag, "reusing engine already running on %s" % ENGINE_URL) return except Exception: pass ENGINE_DIR.mkdir(parents=True, exist_ok=True) if not ENGINE_BIN.exists(): _download(LLAMAFILE_URL, ENGINE_BIN) ENGINE_BIN.chmod(0o755) from huggingface_hub import hf_hub_download gguf = Path(hf_hub_download(repo_id=GGUF_REPO, filename=GGUF_FILE)) threads = max(2, min(8, os.cpu_count() or 2)) cmd = [ str(ENGINE_BIN), "--model", str(gguf), "--server", "--host", "127.0.0.1", "--port", str(LLAMA_PORT), "--ctx-size", "6144", "--threads", str(threads), "--alias", MODEL_ALIAS, "--jinja", "--no-webui", ] _log(tag, "starting: %s" % " ".join(cmd)) logfh = open(ENGINE_DIR / "llamafile.log", "ab") def _popen(argv): # APE binaries are hybrid ELF+shell. Some kernels refuse the # ELF header with ENOEXEC; the documented fallback is /bin/sh, # which the shell-script half of the binary understands. try: return subprocess.Popen( argv, cwd=str(ENGINE_DIR), stdout=logfh, stderr=subprocess.STDOUT, start_new_session=True, ) except OSError as exc: if exc.errno != errno.ENOEXEC: raise return subprocess.Popen( ["/bin/sh", *argv], cwd=str(ENGINE_DIR), stdout=logfh, stderr=subprocess.STDOUT, start_new_session=True, ) proc = _popen(cmd) _engine["proc"] = proc deadline = time.time() + 300 while time.time() < deadline: if proc.poll() is not None: raise RuntimeError( "llamafile exited rc=%s — see llamafile.log" % proc.returncode ) try: with urllib.request.urlopen(ENGINE_URL + "/health", timeout=2) as r: if r.status == 200: _engine["ready"] = True _engine["err"] = None _log(tag, "ready on %s" % ENGINE_URL) return except Exception: time.sleep(1.5) raise RuntimeError("llamafile /health timeout after 300s") except Exception as exc: _engine["err"] = repr(exc) _log(tag, "BOOT FAILED: %r" % exc) def engine_wait(timeout=ENGINE_BOOT_WAIT): """Block until the CPU engine answers, so requests that arrive during a cold start wait instead of failing.""" deadline = time.time() + timeout while time.time() < deadline: if _engine["ready"]: return True if _engine["err"] and _engine["proc"] is None: return False time.sleep(2) return False def engine_chat(body): """POST an OpenAI-shaped body to the local llamafile. Qwen3 soft-switch: /no_think keeps the model from emitting blocks into the JSON. Verified empirically: response_format json_schema is NOT honoured by this build (the grammar is not applied), so clients must not rely on it.""" messages = body.get("messages") or [] messages = list(messages) if messages and isinstance(messages[0], dict) and messages[0].get("role") == "system": if "/no_think" not in messages[0].get("content", ""): messages[0] = dict(messages[0]) messages[0]["content"] = messages[0]["content"].rstrip() + "\n/no_think" else: # No system message: Qwen3's template defaults to thinking mode, which # lands in reasoning_content and leaves content empty. Suppress it. messages.insert(0, {"role": "system", "content": "/no_think"}) # The soft switch is documented for the user turn, and some builds only # check the LAST user message — tag that too. for i in range(len(messages) - 1, -1, -1): if isinstance(messages[i], dict) and messages[i].get("role") == "user": if "/no_think" not in str(messages[i].get("content", "")): messages[i] = dict(messages[i]) messages[i]["content"] = str(messages[i]["content"]).rstrip() + "\n/no_think" break body = dict(body, messages=messages, stream=False) # Harmless if this build ignores it; honoured by newer llama.cpp builds. body.setdefault("chat_template_kwargs", {"enable_thinking": False}) if not body.get("model"): body["model"] = MODEL_ALIAS data = json.dumps(body).encode("utf-8") req = urllib.request.Request( ENGINE_URL + "/v1/chat/completions", data=data, headers={"Content-Type": "application/json"}, method="POST", ) with urllib.request.urlopen(req, timeout=900) as r: return json.loads(r.read().decode("utf-8", "replace")) # ---------------------------------------------------------------- ui ---- def status(): gpu_state = "cached" if _gpu["model"] is not None else ( "error: %s" % _gpu["err"] if _gpu["err"] else "cold" ) if _engine["ready"]: eng = "cpu engine ready (%s)" % MODEL_ALIAS elif _engine["err"]: eng = "cpu engine ERROR: %s" % _engine["err"] else: eng = "cpu engine booting (cold start downloads ~2.6 GB)" return "%s | gpu(%s): %s | quota: ~5 min/day free" % (eng, GPU_ALIAS, gpu_state) def ui_generate(system_prompt, user_prompt, max_tokens, use_gpu): try: if use_gpu: return gpu_generate( [{"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}], int(max_tokens), 0.7, ) if not engine_wait(): return "engine not ready: %s" % (_engine["err"] or "timeout") out = engine_chat({ "model": MODEL_ALIAS, "max_tokens": int(max_tokens), "temperature": 0.7, "messages": [{"role": "system", "content": system_prompt}, {"role": "user", "content": user_prompt}], }) return out["choices"][0]["message"]["content"] except Exception as exc: return "error: %r" % exc import gradio as gr # noqa: E402 with gr.Blocks(title="Forge model server") as demo: gr.Markdown( "## Forge model server\n" "OpenAI-compatible backend for repo GameForge. Unlimited CPU engine " "(`%s`), ZeroGPU fast path (`%s`).\n\n" "Point an OpenAI client at `/v1` on this Space's URL." % (MODEL_ALIAS, GPU_ALIAS) ) with gr.Row(): sys_in = gr.Textbox(label="system", lines=5) usr_in = gr.Textbox(label="user", lines=5) with gr.Row(): mt = gr.Slider(64, 2000, value=900, step=32, label="max tokens") gpu = gr.Checkbox(label="use ZeroGPU fast path (daily quota)") out = gr.Textbox(label="output", lines=16) gr.Button("generate").click(ui_generate, [sys_in, usr_in, mt, gpu], out, api_name="forge") st = gr.Textbox(label="status", value=status(), interactive=False) gr.Button("status").click(status, None, st, api_name="status") # ---------------------------------------------------------------- routes ---- import asyncio # noqa: E402 from fastapi import Request # noqa: E402 from fastapi.responses import JSONResponse # noqa: E402 def _openai_response(content, model): return { "id": "forge-%d" % int(time.time()), "object": "chat.completion", "created": int(time.time()), "model": model, "choices": [{ "index": 0, "message": {"role": "assistant", "content": content}, "finish_reason": "stop", }], "usage": {"prompt_tokens": 0, "completion_tokens": 0, "total_tokens": 0}, } async def _chat_completions(request: Request): try: body = await request.json() except Exception: return JSONResponse({"error": {"message": "invalid JSON body"}}, status_code=400) if body.get("stream"): return JSONResponse( {"error": {"message": "streaming is not supported; send stream=false"}}, status_code=400, ) messages = body.get("messages") if not messages: return JSONResponse({"error": {"message": "messages required"}}, status_code=400) model = body.get("model") or MODEL_ALIAS max_tokens = int(body.get("max_tokens") or 1200) temperature = float(body.get("temperature") or 0.7) loop = asyncio.get_running_loop() if str(model).endswith("-gpu"): try: content = await loop.run_in_executor( None, functools.partial(gpu_generate, messages, max_tokens, temperature) ) return JSONResponse(_openai_response(content, model)) except Exception as exc: # Quota exhausted / GPU unavailable: 429 so clients fall back. return JSONResponse( {"error": {"message": "gpu path failed: %r" % exc}}, status_code=429 ) if not await loop.run_in_executor(None, engine_wait): return JSONResponse( {"error": { "message": "cpu engine not ready: %s" % (_engine["err"] or "cold start timeout") }}, status_code=503, ) try: out = await loop.run_in_executor(None, functools.partial(engine_chat, body)) return JSONResponse(out) except urllib.error.HTTPError as exc: try: detail = json.loads(exc.read().decode("utf-8", "replace")) except Exception: detail = {"error": {"message": "engine HTTP %d" % exc.code}} return JSONResponse(detail, status_code=exc.code) except Exception as exc: return JSONResponse( {"error": {"message": "engine error: %r" % exc}}, status_code=500 ) async def _models(): loop = asyncio.get_running_loop() if not await loop.run_in_executor(None, functools.partial(engine_wait, 30)): return JSONResponse({"object": "list", "data": [ {"id": MODEL_ALIAS, "object": "model"}, {"id": GPU_ALIAS, "object": "model"}, ]}) try: with urllib.request.urlopen(ENGINE_URL + "/v1/models", timeout=20) as r: return JSONResponse(json.loads(r.read().decode("utf-8", "replace"))) except Exception: return JSONResponse( {"object": "list", "data": [{"id": MODEL_ALIAS, "object": "model"}]} ) async def _healthz(): return JSONResponse({"ok": True, "status": status()}) _launch_result = demo.launch( server_name="0.0.0.0", server_port=PORT, ssr_mode=False, prevent_thread_lock=True, quiet=True, show_error=True, ) _fastapi = _launch_result[0] if isinstance(_launch_result, tuple) else None if _fastapi is None: raise RuntimeError("gradio 6 launch() did not return the FastAPI app") # Plain `def` endpoints: FastAPI runs them in its threadpool, so the blocking # engine_wait()/generation calls never stall the event loop. _fastapi.add_api_route("/v1/chat/completions", _chat_completions, methods=["POST"]) _fastapi.add_api_route("/v1/models", _models, methods=["GET"]) _fastapi.add_api_route("/healthz", _healthz, methods=["GET"]) threading.Thread(target=_engine_boot, daemon=True).start() if __name__ == "__main__": _log("app", "serving on :%d (engine on :%d)" % (PORT, LLAMA_PORT)) demo.block_thread()