| import spaces |
|
|
| 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"<tool_call>\s*<function=(?P<name>[^>]+)>(?P<body>.*?)</function>\s*</tool_call>", re.DOTALL) |
| PARAM_RE = re.compile(r"<parameter=(?P<name>[^>]+)>(?P<value>.*?)</parameter>", re.DOTALL) |
|
|
|
|
| def _parse_tool_calls(text: str): |
| """Extract <tool_call> 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) |
|
|
