import spaces # noqa: E402 (must be imported before torch/transformers) import json import re import time import gradio as gr import torch from transformers import AutoModelForCausalLM, AutoTokenizer MODEL_ID = "Qwen/Qwen3-Coder-30B-A3B-Instruct-FP8" tokenizer = AutoTokenizer.from_pretrained(MODEL_ID) model = AutoModelForCausalLM.from_pretrained( MODEL_ID, dtype="auto", ).to("cuda") model.eval() TOOL_CALL_RE = re.compile(r"\s*[^>]+)>(?P.*?)\s*", re.DOTALL) PARAM_RE = re.compile(r"[^>]+)>(?P.*?)", re.DOTALL) def _parse_tool_calls(text: str): """Extract blocks from generated text into OpenAI-style tool_calls.""" tool_calls = [] for match in TOOL_CALL_RE.finditer(text): name = match.group("name").strip() body = match.group("body") arguments = {} for pmatch in PARAM_RE.finditer(body): pname = pmatch.group("name").strip() pvalue = pmatch.group("value").strip("\n") arguments[pname] = pvalue tool_calls.append( { "id": f"call_{len(tool_calls)}_{int(time.time() * 1000)}", "type": "function", "function": {"name": name, "arguments": json.dumps(arguments)}, } ) leftover = TOOL_CALL_RE.sub("", text).strip() return leftover, tool_calls @spaces.GPU(duration=120) def chat_completions(messages_json: str, tools_json: str = "", max_new_tokens: int = 2048) -> str: """Run one chat-completion turn against Qwen3-Coder-30B-A3B-Instruct-FP8. messages_json: JSON-encoded list of {"role", "content"} (and optionally "tool_calls" / "tool_call_id" for prior turns), OpenAI chat format. tools_json: JSON-encoded list of OpenAI-style tool/function definitions, or an empty string for no tools. max_new_tokens: generation budget for this turn. Returns a JSON-encoded object: {"content": str, "tool_calls": [...]}. """ messages = json.loads(messages_json) tools = json.loads(tools_json) if tools_json else None prompt = tokenizer.apply_chat_template( messages, tools=tools, add_generation_prompt=True, tokenize=False, ) inputs = tokenizer(prompt, return_tensors="pt").to("cuda") with torch.no_grad(): output_ids = model.generate( **inputs, max_new_tokens=max_new_tokens, do_sample=False, ) generated = output_ids[0][inputs["input_ids"].shape[1]:] text = tokenizer.decode(generated, skip_special_tokens=True) content, tool_calls = _parse_tool_calls(text) return json.dumps({"content": content, "tool_calls": tool_calls}) demo = gr.Interface( fn=chat_completions, inputs=[ gr.Text(label="messages_json"), gr.Text(label="tools_json", value=""), gr.Number(label="max_new_tokens", value=2048, precision=0), ], outputs=gr.Text(label="completion_json"), title="Qwen3-Coder-30B-A3B-Instruct-FP8 \u2014 chat completions", description=( "Internal inference endpoint for the Coding Model project. " "Called via the local OpenAI-compatible shim, not directly by end users." ), api_name="chat_completions", ) if __name__ == "__main__": demo.launch(mcp_server=True)