| 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""" |
| <div style="margin-bottom: 15px; display: flex; gap: 10px; align-items: center;"> |
| <a href="{tunnel_url}" target="_blank" style=" |
| display: inline-block; |
| padding: 10px 20px; |
| background-color: #2563EB; |
| color: white; |
| text-decoration: none; |
| border-radius: 6px; |
| font-weight: 600; |
| font-size: 14px; |
| transition: background-color 0.2s; |
| " onmouseover="this.style.backgroundColor='#1D4ED8'" onmouseout="this.style.backgroundColor='#2563EB'"> |
| 🗪 Open Full Chat (llama.cpp UI) |
| </a> |
| <span style="color: #6B7280; font-size: 12px;">Opens the backend interface in a new window</span> |
| </div> |
| """) |
| |
| 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() |
