Upload app.py

app.py

@@ -8,7 +8,8 @@ from typing import AsyncIterator, Optional
 import gradio as gr
 import spaces
 import torch
-from fastapi import HTTPException
+import uvicorn
+from fastapi import FastAPI, HTTPException
 from fastapi.responses import JSONResponse, StreamingResponse
 from pydantic import BaseModel, Field
 from transformers import AutoModelForCausalLM, AutoTokenizer, TextIteratorStreamer
@@ -168,79 +169,79 @@ You can also chat directly below.
     gr.ChatInterface(fn=gradio_chat)
 
 # ---------------------------------------------------------------------------
-# Register FastAPI-style routes onto Gradio's own FastAPI app via app_kwargs,
-# BEFORE launch() — so ZeroGPU's launch hook still does its registration scan.
+# FastAPI app — built ourselves so the routes are guaranteed to exist before
+# Gradio is mounted into it. (demo.app does not exist until demo.launch()
+# runs, so routes can never be attached to it beforehand.)
 # ---------------------------------------------------------------------------
 
+app = FastAPI(title="Qwen3-30B-A3B OpenAI-compatible API")
 
-def add_custom_routes(fastapi_app):
-    @fastapi_app.get("/v1/models")
-    async def list_models():
-        return {
-            "object": "list",
-            "data": [{"id": MODEL_ALIAS, "object": "model", "created": int(time.time()), "owned_by": "qwen"}],
-        }
 
-    @fastapi_app.post("/v1/chat/completions")
-    async def chat_completions(request: ChatCompletionRequest):
-        try:
-            prompt = build_prompt(request.messages, request.enable_thinking or False)
-            gen_kwargs = make_generation_kwargs(request)
-        except Exception as exc:
-            raise HTTPException(status_code=422, detail=str(exc))
+@app.get("/v1/models")
+async def list_models():
+    return {
+        "object": "list",
+        "data": [{"id": MODEL_ALIAS, "object": "model", "created": int(time.time()), "owned_by": "qwen"}],
+    }
+
+
+@app.post("/v1/chat/completions")
+async def chat_completions(request: ChatCompletionRequest):
+    try:
+        prompt = build_prompt(request.messages, request.enable_thinking or False)
+        gen_kwargs = make_generation_kwargs(request)
+    except Exception as exc:
+        raise HTTPException(status_code=422, detail=str(exc))
 
-        if request.stream:
-            completion_id = f"chatcmpl-{uuid.uuid4().hex}"
+    if request.stream:
+        completion_id = f"chatcmpl-{uuid.uuid4().hex}"
 
-            async def token_generator() -> AsyncIterator[str]:
-                role_chunk = {
-                    "id": completion_id, "object": "chat.completion.chunk",
-                    "created": int(time.time()), "model": request.model,
-                    "choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": None}],
-                }
-                yield f"data: {json.dumps(role_chunk)}\n\n"
+        async def token_generator() -> AsyncIterator[str]:
+            role_chunk = {
+                "id": completion_id, "object": "chat.completion.chunk",
+                "created": int(time.time()), "model": request.model,
+                "choices": [{"index": 0, "delta": {"role": "assistant"}, "finish_reason": None}],
+            }
+            yield f"data: {json.dumps(role_chunk)}\n\n"
 
-                streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True)
-                thread = Thread(target=_generate_streaming, args=(prompt, gen_kwargs, streamer), daemon=True)
-                thread.start()
+            streamer = TextIteratorStreamer(tokenizer, skip_prompt=True, skip_special_tokens=True)
+            thread = Thread(target=_generate_streaming, args=(prompt, gen_kwargs, streamer), daemon=True)
+            thread.start()
 
-                try:
-                    for token_text in streamer:
-                        if token_text:
-                            yield stream_chunk(token_text, request.model, completion_id)
-                        await asyncio.sleep(0)
-                finally:
-                    thread.join()
+            try:
+                for token_text in streamer:
+                    if token_text:
+                        yield stream_chunk(token_text, request.model, completion_id)
+                    await asyncio.sleep(0)
+            finally:
+                thread.join()
 
-                yield stream_chunk("", request.model, completion_id, finish_reason="stop")
-                yield "data: [DONE]\n\n"
+            yield stream_chunk("", request.model, completion_id, finish_reason="stop")
+            yield "data: [DONE]\n\n"
 
-            return StreamingResponse(token_generator(), media_type="text/event-stream",
-                                     headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})
+        return StreamingResponse(token_generator(), media_type="text/event-stream",
+                                 headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"})
 
-        try:
-            content = _generate_response(prompt, gen_kwargs)
-        except Exception as exc:
-            raise HTTPException(status_code=500, detail=f"Generation failed: {exc}")
+    try:
+        content = _generate_response(prompt, gen_kwargs)
+    except Exception as exc:
+        raise HTTPException(status_code=500, detail=f"Generation failed: {exc}")
 
-        return JSONResponse(chat_completion_object(content, request.model))
+    return JSONResponse(chat_completion_object(content, request.model))
 
-    @fastapi_app.get("/health")
-    async def health():
-        return {"status": "ok", "model": MODEL_ID}
+
+@app.get("/health")
+async def health():
+    return {"status": "ok", "model": MODEL_ID}
 
 
 # ---------------------------------------------------------------------------
-# Entry-point — demo.launch() is what triggers spaces.one_launch's patched
-# hook, which is what registers @spaces.GPU functions with the HF platform.
+# Mount Gradio into our FastAPI app at the ROOT path. ZeroGPU's scanner
+# inspects the module for @spaces.GPU usage — it does not require
+# demo.launch() to be called, so this mount-and-uvicorn pattern is safe.
 # ---------------------------------------------------------------------------
 
+app = gr.mount_gradio_app(app, demo, path="/")
+
 if __name__ == "__main__":
-    demo.queue()
-    add_custom_routes(demo.app)
-    demo.launch(
-        server_name="0.0.0.0",
-        server_port=7860,
-        app_kwargs={"docs_url": None},
-    )
-    
+    uvicorn.run(app, host="0.0.0.0", port=7860)