SGLang command to run Qwen 3.8 27B (full precision, DSpark)

sglang-qwen-3.8-27b.sh

"$HOME/sglang/bin/sglang" serve Qwen/Qwen3.8-27B \
    --served-model-name Qwen3.8-27B \
    --host 0.0.0.0 \
    --port 8080 \
    --trust-remote-code \
    --tp 1 \
    --dtype bfloat16 \
    --context-length 262144 \
    --kv-cache-dtype bfloat16 \
    --mem-fraction-static 0.95 \
    --max-running-requests 1 \
    --max-total-tokens 262144 \
    --chunked-prefill-size 2048 \
    --attention-backend flashinfer \
    --linear-attn-prefill-backend flashinfer \
    --linear-attn-decode-backend triton \
    --mamba-radix-cache-strategy extra_buffer_lazy \
    --max-mamba-cache-size 12 \
    --speculative-algorithm DSPARK \
    --speculative-draft-model-path \
      "$HOME/models/huggingface/hub/models--RadixArk--Qwen3.8-27B-DSpark/snapshots/923ed3a8572615643f0137e424e4ce4edd7f1cda" \
    --speculative-draft-model-quantization unquant \
    --speculative-dspark-block-size 7 \
    --reasoning-parser qwen3 \
    --tool-call-parser qwen3_coder \
    --default-chat-template-kwargs '{"preserve_thinking":true}'

Benchmark script:

#!/usr/bin/env python3

import json
import statistics
import time

import requests
from transformers import AutoTokenizer


BASE_URL = "http://127.0.0.1:8080"
MODEL_PATH = (
    "/home/behnam/models/huggingface/hub/"
    "models--Qwen--Qwen3.8-27B/snapshots/"
    "1d4bf0f2ff6012fd82039f2fa52739d0dd7c60c0"
)
CONTEXTS = (10_000, 60_000)
OUTPUT_TOKENS = 256
WARMUPS = 1
MEASURED_RUNS = 2


def make_prompt(tokenizer, target_tokens):
    prefix = tokenizer.encode(
        "You are auditing a distributed observatory. Review these deterministic records.\n",
        add_special_tokens=False,
    )
    suffix = tokenizer.encode(
        "\nFinal task: Write a correct Python function named summarize_observations(records). "
        "It must parse pipe-delimited records, return the highest temperature, the average "
        "humidity, and the sorted station IDs whose status is ALERT. Explain malformed-input "
        "handling and include focused tests.",
        add_special_tokens=False,
    )
    available = target_tokens - len(prefix) - len(suffix)
    if available <= 0:
        raise ValueError("target token count is too small")

    body = []
    index = 0
    while len(body) < available:
        record = (
            f"station={index % 97:02d}|minute={index:06d}|temperature={12 + (index % 31)}."
            f"{index % 10}|humidity={35 + (index % 61)}|status="
            f"{'ALERT' if index % 23 == 0 else 'OK'}|note=calibrated sensor reading\n"
        )
        body.extend(tokenizer.encode(record, add_special_tokens=False))
        index += 1

    prompt = prefix + body[:available] + suffix
    if len(prompt) != target_tokens:
        raise RuntimeError(f"constructed {len(prompt)} tokens, expected {target_tokens}")
    return prompt


def flush_cache(session):
    response = session.post(f"{BASE_URL}/flush_cache", timeout=30)
    response.raise_for_status()


def run_once(session, input_ids):
    flush_cache(session)
    payload = {
        "input_ids": input_ids,
        "sampling_params": {
            "temperature": 0.0,
            "max_new_tokens": OUTPUT_TOKENS,
            "ignore_eos": True,
        },
        "stream": True,
    }

    started = time.perf_counter()
    first_token_at = None
    final_event = None
    with session.post(f"{BASE_URL}/generate", json=payload, stream=True, timeout=300) as response:
        response.raise_for_status()
        for raw_line in response.iter_lines():
            if not raw_line.startswith(b"data:"):
                continue
            raw_data = raw_line[5:].strip()
            if not raw_data or raw_data == b"[DONE]":
                continue
            event = json.loads(raw_data)
            final_event = event
            if first_token_at is None and event.get("text"):
                first_token_at = time.perf_counter()
    finished = time.perf_counter()

    if first_token_at is None or final_event is None:
        raise RuntimeError("stream completed without a generated token")
    meta = final_event.get("meta_info", {})
    completion_tokens = meta.get("completion_tokens", OUTPUT_TOKENS)
    if completion_tokens != OUTPUT_TOKENS:
        raise RuntimeError(
            f"generated {completion_tokens} tokens, expected exactly {OUTPUT_TOKENS}"
        )

    ttft = first_token_at - started
    end_to_end = finished - started
    decode_seconds = end_to_end - ttft
    decode_tps = (completion_tokens - 1) / decode_seconds
    return {
        "ttft": ttft,
        "decode_tps": decode_tps,
        "end_to_end": end_to_end,
        "completion_tokens": completion_tokens,
    }


def main():
    tokenizer = AutoTokenizer.from_pretrained(MODEL_PATH, local_files_only=True)
    prompts = {context: make_prompt(tokenizer, context) for context in CONTEXTS}
    session = requests.Session()
    all_results = {}

    for context in CONTEXTS:
        for warmup_index in range(WARMUPS):
            result = run_once(session, prompts[context])
            print(
                json.dumps(
                    {"context": context, "kind": "warmup", "run": warmup_index + 1, **result}
                ),
                flush=True,
            )

        measured = []
        for run_index in range(MEASURED_RUNS):
            result = run_once(session, prompts[context])
            measured.append(result)
            print(
                json.dumps(
                    {"context": context, "kind": "measured", "run": run_index + 1, **result}
                ),
                flush=True,
            )

        summary = {
            "ttft": statistics.mean(item["ttft"] for item in measured),
            "decode_tps": statistics.mean(item["decode_tps"] for item in measured),
            "end_to_end": statistics.mean(item["end_to_end"] for item in measured),
        }
        all_results[context] = summary
        print(json.dumps({"context": context, "kind": "summary", **summary}), flush=True)

    print(json.dumps({"kind": "final", "results": all_results}), flush=True)


if __name__ == "__main__":
    main()