app.py

"""CoderInstruc Gradio Space — UI + agent API (remote HF inference, no local model)."""

from __future__ import annotations

import os
from typing import Any, Callable, TypeVar

import gradio as gr
import httpx

DEFAULT_MODEL = "Qwen/Qwen3-Coder-30B-A3B-Instruct"
DEFAULT_ENDPOINT = "https://router.huggingface.co/v1"
SYSTEM_PROMPT = (
    "You are CoderInstruc, a coding assistant specialized in Python, TypeScript, Go, and SQL. "
    "Prefer correct, minimal, production-ready code. Use tools/schema style answers when asked "
    "to edit files or run tests. Refuse requests for malware or secret exfiltration."
)

F = TypeVar("F", bound=Callable[..., Any])


def gpu_if_zerogpu(fn: F) -> F:
    """
    ZeroGPU Spaces require at least one @spaces.GPU at startup.
    We only call remote Inference API — decorate when the real HF `spaces` lib is present.
    """
    try:
        import spaces  # type: ignore

        gpu = getattr(spaces, "GPU", None)
        if gpu is None:
            return fn
        return gpu(duration=120)(fn)  # type: ignore[return-value]
    except Exception:  # noqa: BLE001 — local / CPU / wrong PyPI package
        return fn


def _settings() -> dict[str, str]:
    token = os.getenv("HF_TOKEN") or os.getenv("CODERINSTRUC_HF_TOKEN") or ""
    model = os.getenv("CODERINSTRUC_HF_INFERENCE_MODEL") or DEFAULT_MODEL
    endpoint = (
        os.getenv("CODERINSTRUC_HF_INFERENCE_ENDPOINT") or DEFAULT_ENDPOINT
    ).rstrip("/")
    return {"token": token, "model": model, "endpoint": endpoint}


def _is_unsafe(text: str) -> bool:
    lowered = text.lower()
    blocked = (
        "ignore previous instructions",
        "bypass safety",
        "write ransomware",
        "create malware to steal",
    )
    return any(b in lowered for b in blocked)


@gpu_if_zerogpu
def complete(message: str, history: list[dict[str, str]] | None = None) -> str:
    """Core completion used by UI and agent API."""
    cfg = _settings()
    if not message.strip():
        return "Envie uma pergunta de código."
    if _is_unsafe(message):
        return "Request blocked by safety filter."
    if not cfg["token"]:
        return (
            "Configure o secret `HF_TOKEN` no Space "
            "(Settings → Variables and secrets) para habilitar a inferência."
        )

    messages: list[dict[str, Any]] = [{"role": "system", "content": SYSTEM_PROMPT}]
    for turn in history or []:
        role = turn.get("role")
        content = turn.get("content")
        if role in {"user", "assistant"} and content:
            messages.append({"role": role, "content": content})
    messages.append({"role": "user", "content": message})

    payload = {
        "model": cfg["model"],
        "messages": messages,
        "max_tokens": 2048,
        "temperature": 0.7,
        "top_p": 0.8,
        "stream": False,
    }
    headers = {
        "Authorization": f"Bearer {cfg['token']}",
        "Content-Type": "application/json",
    }
    url = f"{cfg['endpoint']}/chat/completions"
    try:
        with httpx.Client(timeout=180.0) as client:
            resp = client.post(url, json=payload, headers=headers)
            if resp.status_code >= 400:
                return f"Erro HF ({resp.status_code}): {resp.text[:600]}"
            data = resp.json()
        return data["choices"][0]["message"]["content"]
    except Exception as exc:  # noqa: BLE001
        return f"Falha ao chamar inference: {exc}"


def ask(prompt: str) -> str:
    """Stateless agent endpoint — api_name='ask' for /agents.md clients."""
    return complete(prompt, history=None)


def build_ui() -> gr.Blocks:
    cfg = _settings()
    with gr.Blocks(title="CoderInstruc") as demo:
        gr.Markdown(
            f"""
# CoderInstruc
Assistente de código (base Qwen3-Coder → post-train CoderInstruc).

**Modelo:** `{cfg["model"]}` · Inferência remota (pesos fora deste Space)

**Agents:** `curl https://huggingface.co/spaces/amarorn/CoderInstruc/agents.md`

> Preferência de hardware: **CPU Basic** (não precisa ZeroGPU). Se estiver em ZeroGPU, o app
> usa `@spaces.GPU` só para satisfazer o runtime — a carga real é a Inference API.
"""
        )
        with gr.Tab("Chat"):
            chatbot = gr.Chatbot(height=480)
            msg = gr.Textbox(
                placeholder="Ex.: Write a Python binary search with tests",
                label="Mensagem",
                lines=3,
            )
            with gr.Row():
                send = gr.Button("Enviar", variant="primary")
                clear = gr.Button("Limpar")

            def respond(user_msg: str, history: list[dict[str, str]] | None):
                reply = complete(user_msg, history)
                history = list(history or [])
                history.append({"role": "user", "content": user_msg})
                history.append({"role": "assistant", "content": reply})
                return history, ""

            send.click(respond, inputs=[msg, chatbot], outputs=[chatbot, msg])
            msg.submit(respond, inputs=[msg, chatbot], outputs=[chatbot, msg])
            clear.click(lambda: ([], ""), outputs=[chatbot, msg])

            gr.Examples(
                examples=[
                    "Write a Python function that reverses a string",
                    "TypeScript debounce utility with generics",
                    "SQL: top 5 users by order count",
                ],
                inputs=msg,
            )

        with gr.Tab("Agent API"):
            gr.Markdown(
                "Endpoint estático para agents (`api_name=ask`). "
                "Também disponível via Gradio API / `agents.md`."
            )
            prompt = gr.Textbox(label="prompt", lines=4)
            answer = gr.Textbox(label="answer", lines=12)
            ask_btn = gr.Button("Ask", variant="primary")
            ask_btn.click(ask, inputs=prompt, outputs=answer, api_name="ask")

    return demo


demo = build_ui()

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