import os import subprocess import urllib.request import zipfile import tarfile import json import time import re import httpx import requests import spaces from fastapi import Request from fastapi.responses import StreamingResponse import gradio as gr @spaces.GPU def dummy_gpu_check(): pass def get_latest_llama_tag(): try: url = "https://api.github.com/repos/ggml-org/llama.cpp/releases/latest" req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'}) with urllib.request.urlopen(req) as response: data = json.loads(response.read().decode()) return data["tag_name"] except Exception: return "b4610" def download_and_extract(tag): zip_url = f"https://github.com/ggml-org/llama.cpp/releases/download/{tag}/llama-{tag}-bin-ubuntu-x64.zip" tar_url = f"https://github.com/ggml-org/llama.cpp/releases/download/{tag}/llama-{tag}-bin-ubuntu-x64.tar.gz" try: urllib.request.urlretrieve(zip_url, "llama.zip") with zipfile.ZipFile("llama.zip", 'r') as zip_ref: zip_ref.extractall(".") return True except Exception: try: urllib.request.urlretrieve(tar_url, "llama.tar.gz") with tarfile.open("llama.tar.gz", "r:gz") as tar_ref: tar_ref.extractall(".") return True except Exception: return False def find_binary(name): for root, dirs, files in os.walk("."): if name in files: full_path = os.path.join(root, name) os.chmod(full_path, 0o755) return full_path return None def setup_cloudflared(): cf_path = os.path.abspath("cloudflared") if not os.path.exists(cf_path): print("Downloading cloudflared...") url = "https://github.com/cloudflare/cloudflared/releases/latest/download/cloudflared-linux-amd64" try: urllib.request.urlretrieve(url, cf_path) os.chmod(cf_path, 0o755) except Exception as e: print(f"Failed to download cloudflared: {e}") return None return cf_path def start_tunnel(port): cf_path = setup_cloudflared() if not cf_path: return None, None cmd = [cf_path, "tunnel", "--url", f"http://127.0.0.1:{port}"] process = subprocess.Popen( cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1 ) tunnel_url = None start_time = time.time() while time.time() - start_time < 30: line = process.stdout.readline() if not line: break print(f"[Cloudflared] {line.strip()}") match = re.search(r"https://[a-zA-Z0-9-]+\.trycloudflare\.com", line) if match: tunnel_url = match.group(0) print(f"Cloudflare Tunnel URL: {tunnel_url}") break return process, tunnel_url client = httpx.AsyncClient(base_url="http://127.0.0.1:8000") def make_proxy_handler(target_prefix): async def handler(request: Request, path: str = None): full_path = target_prefix if path: full_path = f"{target_prefix}/{path}" url = f"http://127.0.0.1:8000/{full_path}" headers = dict(request.headers) headers.pop("host", None) body = await request.body() req = client.build_request( method=request.method, url=url, headers=headers, content=body, params=request.query_params ) r = await client.send(req, stream=True) return StreamingResponse( r.aiter_raw(), status_code=r.status_code, headers=dict(r.headers) ) return handler def respond(message, history): messages = [] for h in history: if isinstance(h, dict): messages.append({"role": h["role"], "content": h["content"]}) else: messages.append({"role": "user", "content": h[0]}) messages.append({"role": "assistant", "content": h[1]}) messages.append({"role": "user", "content": message}) payload = { "messages": messages, "stream": True, "temperature": 0.7, "max_tokens": 10240 } response = "" try: with requests.post("http://127.0.0.1:8000/v1/chat/completions", json=payload, stream=True) as r: for line in r.iter_lines(): if line: line_str = line.decode('utf-8').strip() if line_str.startswith("data: "): data_str = line_str[6:] if data_str == "[DONE]": break try: data = json.loads(data_str) delta = data["choices"][0]["delta"] if "content" in delta: response += delta["content"] yield response except Exception: pass except Exception as e: yield f"Connection error: {e}" def main(): tag = get_latest_llama_tag() binary_path = find_binary("llama-server") if not binary_path: if download_and_extract(tag): binary_path = find_binary("llama-server") if not binary_path: raise RuntimeError("llama-server binary not found!") print("Downloading GGUF files...") from huggingface_hub import hf_hub_download model_path = hf_hub_download( repo_id="unsloth/gemma-4-26B-A4B-it-qat-GGUF", filename="gemma-4-26B-A4B-it-qat-UD-Q4_K_XL.gguf" ) draft_path = hf_hub_download( repo_id="unsloth/gemma-4-26B-A4B-it-qat-GGUF", filename="mtp-gemma-4-26B-A4B-it.gguf" ) binary_dir = os.path.dirname(os.path.abspath(binary_path)) env = os.environ.copy() env["LD_LIBRARY_PATH"] = f"{binary_dir}:{env.get('LD_LIBRARY_PATH', '')}" cmd = [ binary_path, "-m", model_path, "--port", "8000", "--host", "0.0.0.0", "-t", "16", "-tb", "16", "--spec-draft-model", draft_path, "--spec-type", "draft-mtp", "--spec-draft-n-max", "3" ] print("Launching llama-server on port 8000...") llama_process = subprocess.Popen(cmd, env=env) time.sleep(5) print("Starting Cloudflare tunnel...") cf_process, tunnel_url = start_tunnel(8000) with gr.Blocks() as demo: gr.Markdown("# Gemma 4 26B") if tunnel_url: gr.HTML(f"""
🗪 Open Full Chat (llama.cpp UI) Opens the backend interface in a new window
""") gr.ChatInterface(respond) print("Launching Gradio on port 7860...") app, _, _ = demo.launch( server_name="0.0.0.0", server_port=7860, prevent_thread_lock=True ) app.api_route("/v1/{path:path}", methods=["GET", "POST", "PUT", "DELETE", "OPTIONS"])(make_proxy_handler("v1")) app.api_route("/completion", methods=["POST", "OPTIONS"])(make_proxy_handler("completion")) app.api_route("/tokenize", methods=["POST", "OPTIONS"])(make_proxy_handler("tokenize")) app.api_route("/detokenize", methods=["POST", "OPTIONS"])(make_proxy_handler("detokenize")) app.api_route("/embedding", methods=["POST", "OPTIONS"])(make_proxy_handler("embedding")) app.api_route("/slots", methods=["GET", "OPTIONS"])(make_proxy_handler("slots")) app.api_route("/props", methods=["GET", "OPTIONS"])(make_proxy_handler("props")) app.api_route("/health", methods=["GET", "OPTIONS"])(make_proxy_handler("health")) print("Proxy endpoints registered. Waiting for processes...") try: llama_process.wait() finally: if cf_process: cf_process.terminate() if __name__ == "__main__": main()