File size: 3,395 Bytes
1ec3957 d4280a1 1ec3957 d4280a1 1ec3957 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | 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"<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)
|