Qwen3.5-35B-A3B-Instruct-NVFP4A16

import argparse
import atexit
import json
import os
import shutil
import subprocess
import sys
import time
import urllib.error
import urllib.request


# ---------------------------
# User-facing configuration
# ---------------------------
DEFAULTS = {
    "model": "YCWTG/Qwen3.5-35B-A3B-Instruct-NVFP4A16",
    "served_model_name": "YCWTG/Qwen3.5-35B-A3B-Instruct-NVFP4A16",
    "host": "localhost",
    "port": 8000,
    "max_model_len": 143616,
    "enable_auto_tool_choice": True,
    "tool_call_parser": "qwen3_coder",
    "attention_backend": "FLASHINFER",
    "max_num_seqs": 1,
    "allowed_local_media_path": None,
    "trust_remote_code": True,
    "language_model_only": False,
}

RUNTIME = {
    "gpu_memory_utilization": 0.97,
    "startup_timeout_sec": 180,
    "healthcheck_timeout_sec": 3,
    "healthcheck_interval_sec": 1,
    "chat_timeout_sec": 600,
}

SERVE_VALUE_OPTIONS = (
    ("--served-model-name", "served_model_name"),
    ("--host", "host"),
    ("--port", "port"),
    ("--max-model-len", "max_model_len"),
    ("--tool-call-parser", "tool_call_parser"),
    ("--attention-backend", "attention_backend"),
    ("--max-num-seqs", "max_num_seqs"),
)

CLIENT_VALUE_OPTIONS = (
    ("--model", "model"),
    *SERVE_VALUE_OPTIONS,
)

SERVE_BOOL_OPTIONS = (
    ("--enable-auto-tool-choice", "enable_auto_tool_choice"),
    ("--trust-remote-code", "trust_remote_code"),
    ("--language-model-only", "language_model_only"),
)

CLIENT_BOOL_OPTIONS = (
    ("--language-model-only", "--no-language-model-only", "language_model_only"),
    ("--enable-auto-tool-choice", "--no-enable-auto-tool-choice", "enable_auto_tool_choice"),
    ("--trust-remote-code", "--no-trust-remote-code", "trust_remote_code"),
)


def append_value_options(cmd, args, options):
    for flag, attr in options:
        cmd.extend([flag, str(getattr(args, attr))])


def append_true_bool_options(cmd, args, options):
    for flag, attr in options:
        if getattr(args, attr):
            cmd.append(flag)


def append_boolean_optional_options(cmd, args, options):
    for positive_flag, negative_flag, attr in options:
        cmd.append(positive_flag if getattr(args, attr) else negative_flag)


def append_optional_value_option(cmd, args, flag, attr):
    value = getattr(args, attr)
    if value is None:
        return
    if isinstance(value, str) and not value.strip():
        return
    cmd.extend([flag, str(value)])


def multiline_input():
    print('User (type "END" on a single line to send, "exit" to quit):')
    lines = []
    while True:
        line = input()
        text = line.strip()
        if text.lower() in {"exit", "quit"}:
            return None
        if text == "END":
            break
        lines.append(line)
    return "\n".join(lines)


def resolve_client_host(host):
    return "127.0.0.1" if host in {"0.0.0.0", "::"} else host


def launch_vllm(args):
    cmd = ["vllm", "serve", args.model]
    append_value_options(cmd, args, SERVE_VALUE_OPTIONS)
    append_optional_value_option(cmd, args, "--allowed-local-media-path", "allowed_local_media_path")
    cmd.extend(
        [
            "--gpu-memory-utilization",
            str(RUNTIME["gpu_memory_utilization"]),
        ]
    )
    append_true_bool_options(cmd, args, SERVE_BOOL_OPTIONS)

    print("Launching vLLM:")
    print(" ".join(cmd))
    try:
        return subprocess.Popen(cmd)
    except FileNotFoundError as e:
        raise RuntimeError("vllm command not found. Activate an environment that has vllm installed.") from e


def stop_vllm(proc):
    if proc and proc.poll() is None:
        proc.terminate()
        try:
            proc.wait(timeout=10)
        except subprocess.TimeoutExpired:
            proc.kill()


def wait_vllm_ready(base_url, timeout_sec=RUNTIME["startup_timeout_sec"]):
    deadline = time.time() + timeout_sec
    url = f"{base_url}/v1/models"
    req = urllib.request.Request(url=url)
    while time.time() < deadline:
        try:
            with urllib.request.urlopen(req, timeout=RUNTIME["healthcheck_timeout_sec"]) as resp:
                if resp.status == 200:
                    return True
        except urllib.error.URLError:
            pass
        time.sleep(RUNTIME["healthcheck_interval_sec"])
    return False


def chat_once(base_url, model_name, messages):
    payload = {"model": model_name, "messages": messages}
    req = urllib.request.Request(
        url=f"{base_url}/v1/chat/completions",
        data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
        headers={"Content-Type": "application/json"},
        method="POST",
    )
    with urllib.request.urlopen(req, timeout=RUNTIME["chat_timeout_sec"]) as resp:
        data = json.loads(resp.read().decode("utf-8"))
    return data["choices"][0]["message"]


def chat_loop(base_url, model_name):
    print("\n===== Chat Started =====\n")
    messages = []

    while True:
        user_text = multiline_input()
        if user_text is None:
            break

        messages.append({"role": "user", "content": user_text})
        try:
            assistant_msg = chat_once(base_url, model_name, messages)
        except Exception as e:
            print(f"\nRequest failed: {e}\n")
            messages.pop()
            continue

        content = assistant_msg.get("content")
        tool_calls = assistant_msg.get("tool_calls")

        if content:
            print(f"\nAssistant:\n{content}\n")
        elif tool_calls:
            print("\nAssistant(tool_calls):")
            print(json.dumps(tool_calls, ensure_ascii=False, indent=2))
            print()
        else:
            print("\nAssistant:\n(empty response)\n")

        normalized_msg = {"role": "assistant", "content": content or ""}
        if tool_calls:
            normalized_msg["tool_calls"] = tool_calls
        messages.append(normalized_msg)


def build_client_command(args):
    cmd = [sys.executable, os.path.abspath(__file__), "--_client"]
    append_value_options(cmd, args, CLIENT_VALUE_OPTIONS)
    append_boolean_optional_options(cmd, args, CLIENT_BOOL_OPTIONS)
    return cmd


def spawn_chat_terminal(args):
    client_cmd = build_client_command(args)

    terminal_cmd = None
    if os.name == "nt":
        # Open a new cmd window on Windows and keep it alive for interactive chat.
        terminal_cmd = [
            "cmd",
            "/c",
            "start",
            "",
            "cmd",
            "/k",
            subprocess.list2cmdline(client_cmd),
        ]
    elif shutil.which("gnome-terminal"):
        terminal_cmd = ["gnome-terminal", "--", *client_cmd]
    elif shutil.which("x-terminal-emulator"):
        terminal_cmd = ["x-terminal-emulator", "-e", *client_cmd]

    if not terminal_cmd:
        return False

    try:
        subprocess.Popen(terminal_cmd)
        return True
    except Exception as e:
        print(f"Failed to open a new terminal automatically: {e}")
        return False


def parse_args():
    parser = argparse.ArgumentParser(description="Minimal local vLLM chat script")
    parser.add_argument("--_client", action="store_true", help=argparse.SUPPRESS)
    parser.add_argument("--model", default=DEFAULTS["model"])
    parser.add_argument(
        "--served-model-name",
        default=DEFAULTS["served_model_name"],
    )
    parser.add_argument("--host", default=DEFAULTS["host"])
    parser.add_argument("--port", type=int, default=DEFAULTS["port"])
    parser.add_argument("--max-model-len", type=int, default=DEFAULTS["max_model_len"])
    parser.add_argument(
        "--max-num-seqs",
        "--max_num_seqs",
        dest="max_num_seqs",
        type=int,
        default=DEFAULTS["max_num_seqs"],
    )
    parser.add_argument(
        "--enable-auto-tool-choice",
        action=argparse.BooleanOptionalAction,
        default=DEFAULTS["enable_auto_tool_choice"],
    )
    parser.add_argument(
        "--allowed-local-media-path",
        default=DEFAULTS["allowed_local_media_path"],
        help="Optional local media path. Leave empty to disable.",
    )
    parser.add_argument("--tool-call-parser", default=DEFAULTS["tool_call_parser"])
    parser.add_argument("--attention-backend", default=DEFAULTS["attention_backend"])
    parser.add_argument(
        "--trust-remote-code",
        action=argparse.BooleanOptionalAction,
        default=DEFAULTS["trust_remote_code"],
    )
    parser.add_argument(
        "--language-model-only",
        action=argparse.BooleanOptionalAction,
        default=DEFAULTS["language_model_only"],
    )
    return parser.parse_args()


def main():
    args = parse_args()
    base_url = f"http://{resolve_client_host(args.host)}:{args.port}"
    if args._client:
        chat_loop(base_url, args.served_model_name)
        return

    proc = launch_vllm(args)
    atexit.register(stop_vllm, proc)

    print(f"Waiting for service to become ready: {base_url}")
    if not wait_vllm_ready(base_url):
        print("vLLM startup timed out. Check server logs above.")
        stop_vllm(proc)
        sys.exit(1)

    if spawn_chat_terminal(args):
        print("Model is ready. Opened a new terminal for chat; this terminal keeps server logs.")
        print("Press Ctrl+C here to stop vLLM.")
        try:
            proc.wait()
        except KeyboardInterrupt:
            print("\nInterrupted. Stopping vLLM...")
    else:
        print("No supported terminal found. Falling back to chat in this terminal.")
        chat_loop(base_url, args.served_model_name)


if __name__ == "__main__":
    main()