app.py

import os
import json
from concurrent.futures import ThreadPoolExecutor

import gradio as gr
import spaces
from huggingface_hub import InferenceClient

@spaces.GPU
def _ensure_gpu_functions_present_for_zerogpu():
    """ZeroGPU startup probe: required for zero-a10g spaces detection."""
    return None


MODEL_MATRIX = [
    {
        "key": "gemma4-31b",
        "label": "Gemma 4 31B-IT",
        "model_id": "unsloth/gemma-4-31B-it-unsloth-bnb-4bit",
    },
    {
        "key": "gemma4-26b",
        "label": "Gemma 4 26B-IT",
        "model_id": "unsloth/gemma-4-26B-A4B-it",
    },
    {
        "key": "qwen3.6-35b",
        "label": "Qwen 3.6 35B-IT",
        "model_id": "unsloth/Qwen3.6-35B-A3B-GGUF",
    },
    {
        "key": "qwen3.6-27b",
        "label": "Qwen 3.6 27B",
        "model_id": "unsloth/Qwen3.6-27B-GGUF",
    },
]

HF_TOKEN = os.getenv("HF_API_TOKEN") or os.getenv("HUGGINGFACE_HUB_TOKEN") or os.getenv("HF_TOKEN")
HF_TIMEOUT = int(os.getenv("HF_TIMEOUT", "120"))

_SYSTEM_PROMPT = (
    "You are a useful assistant. Answer briefly and directly unless the user asks for "
    "long-form detail. Keep responses focused and practical."
)


def _init_client() -> InferenceClient:
    return InferenceClient(token=HF_TOKEN)


def _extract_chat_text(response):
    if response is None:
        return "No response from model."

    if isinstance(response, dict):
        if "choices" in response and response["choices"]:
            choice = response["choices"][0]
            if isinstance(choice, dict) and "message" in choice and isinstance(choice["message"], dict):
                return (choice["message"].get("content") or "").strip()
            if isinstance(choice, dict) and "text" in choice:
                return str(choice["text"]).strip()
        if "generated_text" in response:
            return str(response["generated_text"]).strip()
        if "text" in response:
            return str(response["text"]).strip()

    if hasattr(response, "choices"):
        choices = getattr(response, "choices")
        if choices:
            first = choices[0]
            if hasattr(first, "message") and getattr(first.message, "content", None) is not None:
                return str(first.message.content).strip()
            if hasattr(first, "text"):
                return str(first.text).strip()

    text = str(response)
    if text and text != "{}":
        return text.strip()
    return "Model returned an empty response."


def _build_messages(history, user_message):
    messages = [{"role": "system", "content": _SYSTEM_PROMPT}]
    for pair in history:
        if not pair:
            continue
        user_turn, assistant_turn = pair
        if user_turn:
            messages.append({"role": "user", "content": str(user_turn)})
        if assistant_turn:
            messages.append({"role": "assistant", "content": str(assistant_turn)})
    messages.append({"role": "user", "content": user_message})
    return messages


def _chat_completion(model_id, messages, max_new_tokens, temperature, top_p):
    client = _init_client()
    try:
        response = client.chat_completion(
            model=model_id,
            messages=messages,
            max_tokens=max_new_tokens,
            temperature=temperature,
            top_p=top_p,
            timeout=HF_TIMEOUT,
        )
        return _extract_chat_text(response)
    except Exception as chat_error:
        fallback_prompt = "\n".join([f"{m['role']}: {m['content']}" for m in messages]) + "\nassistant:"
        try:
            response = client.text_generation(
                prompt=fallback_prompt,
                max_new_tokens=max_new_tokens,
                temperature=temperature,
                top_p=top_p,
                timeout=HF_TIMEOUT,
            )
            return str(response).strip()
        except Exception as text_error:
            return (
                f"Could not query model '{model_id}'."
                f" Chat error: {chat_error.__class__.__name__}"
                f"\nFallback error: {text_error.__class__.__name__}"
            )


def _safe_append(history, user_message, assistant_message):
    updated = list(history or [])
    updated.append((user_message, assistant_message))
    return updated


def _query_model(model_id, messages, max_new_tokens, temperature, top_p):
    return _chat_completion(model_id, messages, max_new_tokens, temperature, top_p)


def run_compare(
    user_message,
    hist1,
    hist2,
    hist3,
    hist4,
    temperature,
    top_p,
    max_tokens,
):
    if not user_message or not str(user_message).strip():
        return hist1, hist2, hist3, hist4, "", "", "", ""

    user_message = str(user_message).strip()
    m1, m2, m3, m4 = MODEL_MATRIX
    model_ids = [m1["model_id"], m2["model_id"], m3["model_id"], m4["model_id"]]

    histories = [
        list(hist1 or []),
        list(hist2 or []),
        list(hist3 or []),
        list(hist4 or []),
    ]
    messages = [
        _build_messages(histories[i], user_message)
        for i in range(4)
    ]

    with ThreadPoolExecutor(max_workers=4) as executor:
        futures = [
            executor.submit(_query_model, model_ids[idx], messages[idx], max_tokens, temperature, top_p)
            for idx in range(4)
        ]
        answers = [f.result() for f in futures]

    new_hist1 = _safe_append(histories[0], user_message, answers[0])
    new_hist2 = _safe_append(histories[1], user_message, answers[1])
    new_hist3 = _safe_append(histories[2], user_message, answers[2])
    new_hist4 = _safe_append(histories[3], user_message, answers[3])

    return (
        new_hist1,
        new_hist2,
        new_hist3,
        new_hist4,
        m1["label"],
        m2["label"],
        m3["label"],
        m4["label"],
    )


def run_clear():
    return [], [], [], [], "", "", "", ""


css = """
.gradio-container { max-width: 100%; }
.column {
    border: 1px solid #d7dee8;
    border-radius: 12px;
    padding: 10px;
    background: linear-gradient(160deg, #f8fbff, #f2f4ff);
}
"""

with gr.Blocks(css=css) as demo:
    gr.Markdown(
        "# Side-by-side Unsloth model comparison\n"
        "All generations run through Hugging Face hosted inference (zero local GPU)."
    )

    with gr.Row():
        status1 = gr.Textbox(label="Gemma4 31B-IT", value="", interactive=False)
        status2 = gr.Textbox(label="Gemma4 26B-IT", value="", interactive=False)
        status3 = gr.Textbox(label="Qwen 3.6 35B-IT", value="", interactive=False)
        status4 = gr.Textbox(label="Qwen 3.6 27B", value="", interactive=False)

    with gr.Row(equal_height=True):
        with gr.Column(elem_classes=["column"]):
            gr.Markdown("### Gemma4 31B-IT")
            chat1 = gr.Chatbot(height=470, label="Gemma4 31B-IT")

        with gr.Column(elem_classes=["column"]):
            gr.Markdown("### Gemma4 26B-IT")
            chat2 = gr.Chatbot(height=470, label="Gemma4 26B-IT")

        with gr.Column(elem_classes=["column"]):
            gr.Markdown("### Qwen 3.6 35B-IT")
            chat3 = gr.Chatbot(height=470, label="Qwen 3.6 35B-IT")

        with gr.Column(elem_classes=["column"]):
            gr.Markdown("### Qwen 3.6 27B")
            chat4 = gr.Chatbot(height=470, label="Qwen 3.6 27B")

    with gr.Accordion("Advanced generation settings", open=False):
        temperature = gr.Slider(0.1, 1.2, value=0.7, step=0.05, label="Temperature")
        top_p = gr.Slider(0.2, 1.0, value=0.9, step=0.05, label="Top-p")
        max_tokens = gr.Slider(64, 2048, value=512, step=16, label="Max new tokens")

    with gr.Row():
        prompt = gr.Textbox(
            label="User prompt",
            placeholder="Ask the same question to all models...",
            lines=2,
            scale=3,
        )
        send = gr.Button("Compare", variant="primary")
        clear = gr.Button("Clear")

    send.click(
        run_compare,
        inputs=[
            prompt,
            chat1,
            chat2,
            chat3,
            chat4,
            temperature,
            top_p,
            max_tokens,
        ],
        outputs=[chat1, chat2, chat3, chat4, status1, status2, status3, status4],
    )

    prompt.submit(
        run_compare,
        inputs=[
            prompt,
            chat1,
            chat2,
            chat3,
            chat4,
            temperature,
            top_p,
            max_tokens,
        ],
        outputs=[chat1, chat2, chat3, chat4, status1, status2, status3, status4],
    )

    clear.click(
        run_clear,
        outputs=[chat1, chat2, chat3, chat4, status1, status2, status3, status4],
    )


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