diff --git a/tests/edge/test_qnn_export.py b/tests/edge/test_qnn_export.py new file mode 100644 index 000000000..2f7d641b2 --- /dev/null +++ b/tests/edge/test_qnn_export.py @@ -0,0 +1,144 @@ +"""WP4/mobile: the decode-step graph exported for phone accelerators (QNN / TFLite / ORT). + +Checks the maths of the grouped-query attention layout against a straightforward reference, the KV-cache contract, +and that the exported ONNX is free of the three constructs the Qualcomm toolchains reject. +""" + +from __future__ import annotations + +import importlib.util +import math + +import pytest +import torch + +from vllm_omni.edge import qnn_export as qx + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + +CFG = qx.DecodeStepConfig( + hidden_size=32, + num_hidden_layers=2, + num_attention_heads=4, + num_key_value_heads=2, + head_dim=8, + intermediate_size=48, + vocab_size=16, + rope_theta=10000.0, +) +L = 6 + + +def _module(seed: int = 0) -> qx.DecodeStep: + torch.manual_seed(seed) + module = qx.DecodeStep(CFG).eval() + for p in module.parameters(): + torch.nn.init.normal_(p, std=0.15) + return module + + +def test_config_rejects_head_counts_that_do_not_group(): + with pytest.raises(ValueError): + qx.DecodeStepConfig(hidden_size=32, num_hidden_layers=1, num_attention_heads=6, num_key_value_heads=4, head_dim=8, intermediate_size=48, vocab_size=16) + + +def test_names_and_example_input_shapes_line_up(): + names, outs = qx.input_names(CFG), qx.output_names(CFG) + args = qx.example_inputs(CFG, L) + assert len(names) == len(args) == 4 + 2 * CFG.num_hidden_layers + assert names[:4] == ["x", "cos", "sin", "mask"] and names[4:6] == ["k_cache_0", "v_cache_0"] + assert outs[0] == "logits" and outs[1:3] == ["k_new_0", "v_new_0"] + x, cos, sin, mask, *caches = args + assert x.shape == (1, 1, CFG.hidden_size) and cos.shape == sin.shape == (1, 1, 1, CFG.head_dim) + assert mask.shape == (1, 1, 1, L + 1) + assert all(c.shape == (1, CFG.num_key_value_heads, L, CFG.head_dim) for c in caches) + + +def test_output_contract(): + module = _module() + out = module(*qx.example_inputs(CFG, L)) + assert len(out) == 1 + 2 * CFG.num_hidden_layers + assert out[0].shape == (1, CFG.vocab_size) + assert all(t.shape == (1, CFG.num_key_value_heads, 1, CFG.head_dim) for t in out[1:]) + + +def test_wrong_number_of_caches_is_rejected(): + module = _module() + x, cos, sin, mask, *caches = qx.example_inputs(CFG, L) + with pytest.raises(ValueError): + module(x, cos, sin, mask, *caches[:-1]) + + +def test_grouped_query_layout_matches_expanded_reference(): + """The exported layout reshapes q; the reference expands k/v. They must agree.""" + module = _module(1) + x, cos, sin, mask, *caches = qx.example_inputs(CFG, L) + layer = module.layers[0] + got, k_new, v_new = layer(x, cos, sin, caches[0], caches[1], mask) + + d, kv, g = CFG.head_dim, CFG.num_key_value_heads, CFG.group_size + h = qx._rms(x, layer.input_layernorm, CFG.rms_norm_eps) + q = qx._rope(qx._rms(layer.q_proj(h).view(1, CFG.num_attention_heads, 1, d), layer.q_norm, CFG.rms_norm_eps), cos, sin) + k = qx._rope(qx._rms(layer.k_proj(h).view(1, kv, 1, d), layer.k_norm, CFG.rms_norm_eps), cos, sin) + v = layer.v_proj(h).view(1, kv, 1, d) + keys = torch.cat([caches[0], k], 2).repeat_interleave(g, dim=1) + vals = torch.cat([caches[1], v], 2).repeat_interleave(g, dim=1) + scores = torch.matmul(q, keys.transpose(2, 3)) / math.sqrt(d) + mask + attn = torch.matmul(torch.softmax(scores, -1), vals).transpose(1, 2).reshape(1, 1, CFG.num_attention_heads * d) + expected = x + layer.o_proj(attn) + hh = qx._rms(expected, layer.post_attention_layernorm, CFG.rms_norm_eps) + expected = expected + layer.down_proj(torch.nn.functional.silu(layer.gate_proj(hh)) * layer.up_proj(hh)) + + assert torch.allclose(got, expected, atol=1e-5) + assert torch.allclose(k_new, k) and torch.allclose(v_new, v) + + +def test_masked_cache_slots_do_not_change_the_result(): + """Slots the mask disables may hold anything: that is what lets the host keep one fixed-size cache.""" + module = _module(2) + args = list(qx.example_inputs(CFG, L, position=2)) + first = module(*args)[0] + for i in range(4, len(args)): + cache = args[i].clone() + cache[:, :, 2:L] = 7.0 # masked region + args[i] = cache + assert torch.allclose(first, module(*args)[0], atol=1e-5) + + +def test_placement_table_prefers_npu_for_the_step_and_gpu_for_the_vocoder(): + p = qx.PLACEMENT_MS_PER_FRAME + assert p["talker_step"]["npu"] < p["talker_step"]["gpu"] + assert p["code2wav_chunk"]["gpu"] < p["code2wav_chunk"]["npu"] + + +@pytest.mark.skipif(importlib.util.find_spec("onnx") is None, reason="onnx not installed") +def test_export_onnx_is_clean_for_the_qualcomm_toolchains(tmp_path): + import onnx + + module = _module(3) + manifest = qx.export_onnx(module, L, tmp_path / "step.onnx") + assert manifest["cache_len"] == L and manifest["config"]["hidden_size"] == CFG.hidden_size + model = onnx.load(str(tmp_path / "step.onnx")) + onnx.checker.check_model(model, full_check=True) + assert [i.name for i in model.graph.input] == qx.input_names(CFG) + assert [o.name for o in model.graph.output] == qx.output_names(CFG) + assert not [n for n in model.graph.node if n.op_type == "IsNaN"] + assert not [n for n in model.graph.node for a in n.attribute if n.op_type == "Reshape" and a.name == "allowzero" and a.i] + inits = {i.name: i for i in model.graph.initializer} + assert not [n for n in model.graph.node if n.op_type in ("Unsqueeze", "Squeeze") and len(n.input) > 1 and n.input[1] in inits and not inits[n.input[1]].dims] + + +@pytest.mark.skipif(importlib.util.find_spec("onnxruntime") is None or importlib.util.find_spec("onnx") is None, reason="onnx/onnxruntime not installed") +def test_exported_graph_reproduces_eager(tmp_path): + import onnxruntime as ort + + module = _module(4) + args = qx.example_inputs(CFG, L) + with torch.inference_mode(): + expected = module(*args) + qx.export_onnx(module, L, tmp_path / "step.onnx") + sess = ort.InferenceSession(str(tmp_path / "step.onnx"), providers=["CPUExecutionProvider"]) + feed = {name: arg.numpy() for name, arg in zip(qx.input_names(CFG), args)} + got = sess.run(None, feed) + for name, want, have in zip(qx.output_names(CFG), expected, got): + assert torch.allclose(want, torch.from_numpy(have), atol=1e-4), name diff --git a/vllm_omni/edge/qnn_export.py b/vllm_omni/edge/qnn_export.py new file mode 100644 index 000000000..4e1b7fc4a --- /dev/null +++ b/vllm_omni/edge/qnn_export.py @@ -0,0 +1,364 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""Export the Qwen3-TTS talker / code-predictor **decode step** for phone accelerators (QNN, TFLite, ONNX Runtime). + +Companion to :mod:`vllm_omni.edge.decoder_export`, which exports the Code2Wav vocoder. Together the two cover the +whole on-device path served by the codec-token stream (WP5): the talker and its code predictor produce codec tokens, +the vocoder turns them into audio. + +Layout +------ +One token in, one token out, with an explicit static KV cache:: + + inputs x [1, 1, H], cos [1, 1, 1, D], sin [1, 1, 1, D], mask [1, 1, 1, L+1], + k_cache_i / v_cache_i [1, KV, L, D] for each layer i + outputs logits [1, vocab], k_new_i / v_new_i [1, KV, 1, D] for each layer i + +Everything that is cheap on a CPU and awkward on an accelerator stays on the host: embedding lookups, the rotary +angles (passed in as ``cos``/``sin``, which also keeps M-RoPE's interleaved sections out of the graph), sampling, and +appending ``k_new``/``v_new`` into the caches. + +Two details are what make this fast, and both were measured on real devices (Snapdragon 8 Gen 3 / 8 Elite via +Qualcomm AI Hub, 2026-09-10; see ``analysis/mobile_htp_simulation.md`` §5): + +* **per-layer cache tensors**, not one stacked ``[num_layers, ...]`` tensor that each layer slices. The stacked form + spent ~70 % of its NPU cycles in cache ``select``/``Reshape`` and fp32→fp16 conversion of the whole cache; +* **grouped-query attention by reshaping the query** to ``[1, KV, G, D]`` instead of expanding keys and values to + ``[1, H, L+1, D]``. Expanding multiplies the attention tensors by ``G`` before the matmul. + +On a Snapdragon 8 Elite this took the 0.6B talker step from 36.9 ms to 15.8 ms (fp16, 256-token cache); the code +predictor step is 2.25 ms, so a full 80 ms frame (talker + 15 predictor sub-steps) costs ≈50 ms on the NPU. + +Placement +--------- +Measured, same phone, same graphs (per 80 ms frame, fp16): + +=================== =========== ========== ========= +component Hexagon NPU Adreno GPU phone CPU +=================== =========== ========== ========= +talker step 15.8 ms 29.9 ms 29.8 ms +predictor step x15 33.9 ms 65.0 ms 72.3 ms +Code2Wav vocoder 182 ms 21.7 ms 109 ms +=================== =========== ========== ========= + +So run **this** graph on the NPU and the **vocoder on the GPU**: 8.4x for the vocoder, ~2x for the transformer steps, +and the two stages pipeline the way they already do in the server deployment. Sending everything to the NPU is 3.3x +slower than the split. Integer quantization is *not* the lever it appears to be: every automatic int8 build measured +was numerically broken (predictor 3.1 dB, vocoder -4.3 dB vs the fp32 reference), while fp16 is faithful +(39-48 dB on the NPU). + +ONNX for the Qualcomm toolchains +-------------------------------- +:func:`sanitize_onnx` fixes the three things the dynamo exporter emits that ``qairt-converter`` and AI Hub reject: +``IsNaN``/``Where`` guards, ``Reshape`` with ``allowzero=1``, and rank-0 ``axes`` initializers on +``Unsqueeze``/``Squeeze`` (which crash AI Hub's shape inference). + +CLI:: + + python -m vllm_omni.edge.qnn_export --model Qwen/Qwen3-TTS-12Hz-0.6B-CustomVoice \\ + --component talker --cache-len 256 --out-dir ./qnn_export +""" + +from __future__ import annotations + +import argparse +import json +import math +import sys +from dataclasses import asdict, dataclass +from pathlib import Path +from typing import Any + +import torch +from torch import nn + +__all__ = [ + "DecodeStepConfig", + "DecodeStep", + "config_from_qwen3_tts", + "load_qwen3_tts_weights", + "example_inputs", + "input_names", + "output_names", + "export_onnx", + "sanitize_onnx", + "PLACEMENT_MS_PER_FRAME", +] + +#: Measured on Samsung Galaxy S25 (Snapdragon 8 Elite), fp16, per 80 ms audio frame. See the module docstring. +PLACEMENT_MS_PER_FRAME = { + "talker_step": {"npu": 15.8, "gpu": 29.9, "cpu": 29.8}, + "predictor_step_x15": {"npu": 33.9, "gpu": 65.0, "cpu": 72.3}, + "code2wav_chunk": {"npu": 182.0, "gpu": 21.7, "cpu": 109.0}, +} + + +@dataclass(frozen=True) +class DecodeStepConfig: + """Shape of one autoregressive step. Mirrors the fields of a Qwen3 decoder config that the graph needs.""" + + hidden_size: int + num_hidden_layers: int + num_attention_heads: int + num_key_value_heads: int + head_dim: int + intermediate_size: int + vocab_size: int + rms_norm_eps: float = 1e-6 + rope_theta: float = 1e6 + + def __post_init__(self) -> None: + if self.num_attention_heads % self.num_key_value_heads: + raise ValueError(f"num_attention_heads {self.num_attention_heads} is not a multiple of num_key_value_heads {self.num_key_value_heads}") + + @property + def group_size(self) -> int: + return self.num_attention_heads // self.num_key_value_heads + + +def _rms(x: torch.Tensor, weight: torch.Tensor, eps: float) -> torch.Tensor: + return weight * (x * torch.rsqrt(x.pow(2).mean(-1, keepdim=True) + eps)) + + +def _rope(x: torch.Tensor, cos: torch.Tensor, sin: torch.Tensor) -> torch.Tensor: + half = x.shape[-1] // 2 + x1, x2 = x[..., :half], x[..., half:] + return x * cos + torch.cat([-x2, x1], dim=-1) * sin + + +class _Layer(nn.Module): + def __init__(self, cfg: DecodeStepConfig): + super().__init__() + self.cfg = cfg + h, d = cfg.hidden_size, cfg.head_dim + self.input_layernorm = nn.Parameter(torch.ones(h)) + self.post_attention_layernorm = nn.Parameter(torch.ones(h)) + self.q_norm = nn.Parameter(torch.ones(d)) + self.k_norm = nn.Parameter(torch.ones(d)) + self.q_proj = nn.Linear(h, cfg.num_attention_heads * d, bias=False) + self.k_proj = nn.Linear(h, cfg.num_key_value_heads * d, bias=False) + self.v_proj = nn.Linear(h, cfg.num_key_value_heads * d, bias=False) + self.o_proj = nn.Linear(cfg.num_attention_heads * d, h, bias=False) + self.gate_proj = nn.Linear(h, cfg.intermediate_size, bias=False) + self.up_proj = nn.Linear(h, cfg.intermediate_size, bias=False) + self.down_proj = nn.Linear(cfg.intermediate_size, h, bias=False) + + def forward(self, x, cos, sin, k_cache, v_cache, mask): + cfg = self.cfg + kv, d, g = cfg.num_key_value_heads, cfg.head_dim, cfg.group_size + h = _rms(x, self.input_layernorm, cfg.rms_norm_eps) + q = _rms(self.q_proj(h).view(1, cfg.num_attention_heads, 1, d), self.q_norm, cfg.rms_norm_eps) + k = _rms(self.k_proj(h).view(1, kv, 1, d), self.k_norm, cfg.rms_norm_eps) + v = self.v_proj(h).view(1, kv, 1, d) + q, k = _rope(q, cos, sin), _rope(k, cos, sin) + keys = torch.cat([k_cache, k], dim=2) + vals = torch.cat([v_cache, v], dim=2) + # GQA without expanding keys/values: group the query heads that share a kv head. + scores = torch.matmul(q.view(1, kv, g, d), keys.transpose(2, 3)) * (1.0 / math.sqrt(d)) + mask + attn = torch.matmul(torch.softmax(scores, dim=-1), vals).view(1, 1, cfg.num_attention_heads * d) + x = x + self.o_proj(attn) + h = _rms(x, self.post_attention_layernorm, cfg.rms_norm_eps) + x = x + self.down_proj(torch.nn.functional.silu(self.gate_proj(h)) * self.up_proj(h)) + return x, k, v + + +class DecodeStep(nn.Module): + """One decode step with an explicit per-layer KV cache; see the module docstring for the tensor contract.""" + + def __init__(self, cfg: DecodeStepConfig): + super().__init__() + self.cfg = cfg + self.layers = nn.ModuleList(_Layer(cfg) for _ in range(cfg.num_hidden_layers)) + self.norm = nn.Parameter(torch.ones(cfg.hidden_size)) + self.lm_head = nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False) + + def forward(self, x, cos, sin, mask, *caches): + if len(caches) != 2 * self.cfg.num_hidden_layers: + raise ValueError(f"expected {2 * self.cfg.num_hidden_layers} cache tensors, got {len(caches)}") + new: list[torch.Tensor] = [] + for i, layer in enumerate(self.layers): + x, k, v = layer(x, cos, sin, caches[2 * i], caches[2 * i + 1], mask) + new += [k, v] + logits = self.lm_head(_rms(x, self.norm, self.cfg.rms_norm_eps))[:, 0] + return (logits, *new) + + +def input_names(cfg: DecodeStepConfig) -> list[str]: + return ["x", "cos", "sin", "mask"] + [f"{kv}_cache_{i}" for i in range(cfg.num_hidden_layers) for kv in ("k", "v")] + + +def output_names(cfg: DecodeStepConfig) -> list[str]: + return ["logits"] + [f"{kv}_new_{i}" for i in range(cfg.num_hidden_layers) for kv in ("k", "v")] + + +def example_inputs(cfg: DecodeStepConfig, cache_len: int, position: int | None = None, dtype: torch.dtype = torch.float32) -> tuple[torch.Tensor, ...]: + """Concrete inputs for tracing: a random hidden state, the rotary angles for ``position`` and a causal mask.""" + pos = cache_len // 2 if position is None else position + d = cfg.head_dim + inv = 1.0 / (cfg.rope_theta ** (torch.arange(0, d, 2, dtype=torch.float32) / d)) + ang = pos * inv + cos = torch.cat([ang.cos(), ang.cos()]).view(1, 1, 1, d).to(dtype) + sin = torch.cat([ang.sin(), ang.sin()]).view(1, 1, 1, d).to(dtype) + mask = torch.zeros(1, 1, 1, cache_len + 1, dtype=dtype) + mask[..., pos:cache_len] = torch.finfo(torch.float16).min / 4 # positions not yet written + x = (torch.randn(1, 1, cfg.hidden_size) * 0.1).to(dtype) + caches = [torch.randn(1, cfg.num_key_value_heads, cache_len, d, dtype=dtype) * 0.5 for _ in range(2 * cfg.num_hidden_layers)] + return (x, cos, sin, mask, *caches) + + +def config_from_qwen3_tts(hf_config: dict[str, Any], component: str) -> DecodeStepConfig: + """Build a :class:`DecodeStepConfig` from a Qwen3-TTS ``config.json`` dict. ``component``: talker | predictor.""" + talker = hf_config["talker_config"] if "talker_config" in hf_config else hf_config + cfg = talker if component == "talker" else talker["code_predictor_config"] + return DecodeStepConfig( + hidden_size=cfg["hidden_size"], + num_hidden_layers=cfg["num_hidden_layers"], + num_attention_heads=cfg["num_attention_heads"], + num_key_value_heads=cfg["num_key_value_heads"], + head_dim=cfg.get("head_dim", cfg["hidden_size"] // cfg["num_attention_heads"]), + intermediate_size=cfg["intermediate_size"], + vocab_size=cfg["vocab_size"], + rms_norm_eps=cfg.get("rms_norm_eps", 1e-6), + rope_theta=cfg.get("rope_theta", 1e6), + ) + + +def load_qwen3_tts_weights(module: DecodeStep, safetensors_path: str | Path, component: str, lm_head_index: int = 0) -> int: + """Copy real weights into ``module``. ``component``: ``talker`` (codec head) or ``predictor`` (``lm_head.``).""" + from safetensors import safe_open + + prefix = "talker." if component == "talker" else "talker.code_predictor." + head_key = "codec_head.weight" if component == "talker" else f"lm_head.{lm_head_index}.weight" + loaded = 0 + with safe_open(str(safetensors_path), "pt") as st: + get = lambda key: st.get_tensor(prefix + key).float() # noqa: E731 + for i, layer in enumerate(module.layers): + p = f"model.layers.{i}." + for name, param in (("input_layernorm", layer.input_layernorm), ("post_attention_layernorm", layer.post_attention_layernorm)): + param.data.copy_(get(p + name + ".weight")) + loaded += 1 + for name, param in (("q_norm", layer.q_norm), ("k_norm", layer.k_norm)): + param.data.copy_(get(p + "self_attn." + name + ".weight")) + loaded += 1 + for name in ("q_proj", "k_proj", "v_proj", "o_proj"): + getattr(layer, name).weight.data.copy_(get(p + "self_attn." + name + ".weight")) + loaded += 1 + for name in ("gate_proj", "up_proj", "down_proj"): + getattr(layer, name).weight.data.copy_(get(p + "mlp." + name + ".weight")) + loaded += 1 + module.norm.data.copy_(get("model.norm.weight")) + module.lm_head.weight.data.copy_(get(head_key)) + loaded += 2 + return loaded + + +def sanitize_onnx(path: str | Path) -> dict[str, int]: + """Make a dynamo-exported ONNX file digestible by ``qairt-converter`` and Qualcomm AI Hub, in place. + + Removes ``IsNaN``/``Where`` guard pairs (the converter has no ``IsNaN``), clears ``Reshape``'s ``allowzero`` + attribute (rejected), and promotes rank-0 ``axes`` initializers on ``Unsqueeze``/``Squeeze`` to rank 1 (AI Hub's + shape inference raises "Inferred shape and existing shape differ in rank" on them). Returns the counts. + """ + import onnx + + model = onnx.load(str(path)) + graph = model.graph + produced = {out: node for node in graph.node for out in node.output} + renamed: dict[str, str] = {} + for node in list(graph.node): + if node.op_type != "Where" or node.input[0] not in produced: + continue + guard = produced[node.input[0]] + if guard.op_type != "IsNaN": + continue + source = guard.input[0] + keep = source if source in (node.input[1], node.input[2]) else None + if keep is None: + continue + renamed[node.output[0]] = keep + graph.node.remove(node) + graph.node.remove(guard) + initializers = {init.name: init for init in graph.initializer} + constants = {node.output[0]: node for node in graph.node if node.op_type == "Constant"} + stats = {"nan_guards_removed": len(renamed), "allowzero_cleared": 0, "scalar_axes_fixed": 0} + scalar_axes: list[str] = [] + for node in graph.node: + for i, name in enumerate(node.input): + if name in renamed: + node.input[i] = renamed[name] + if node.op_type == "Reshape": + for attr in node.attribute: + if attr.name == "allowzero" and attr.i: + attr.i = 0 + stats["allowzero_cleared"] += 1 + if node.op_type in ("Unsqueeze", "Squeeze") and len(node.input) > 1: + axes = node.input[1] + if axes in initializers and not initializers[axes].dims: + initializers[axes].dims.extend([1]) + scalar_axes.append(axes) + elif axes in constants: + for attr in constants[axes].attribute: + if attr.name == "value" and not attr.t.dims: + attr.t.dims.extend([1]) + scalar_axes.append(axes) + stats["scalar_axes_fixed"] = len(scalar_axes) + for collection in (graph.value_info, graph.input): + for info in list(collection): + if info.name in scalar_axes: + collection.remove(info) # stale rank-0 shape annotation + for out in graph.output: + if out.name in renamed: + out.name = renamed[out.name] + onnx.save(model, str(path)) + return stats + + +def export_onnx(module: DecodeStep, cache_len: int, path: str | Path, opset: int = 18) -> dict[str, Any]: + """Export ``module`` to ONNX at ``path`` (dynamo exporter) and sanitize it for the Qualcomm toolchains.""" + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + cfg = module.cfg + args = example_inputs(cfg, cache_len) + program = torch.onnx.export( + module, + args, + input_names=input_names(cfg), + output_names=output_names(cfg), + opset_version=opset, + dynamo=True, + optimize=True, + ) + program.save(str(path)) + stats = sanitize_onnx(path) + return {"path": str(path), "cache_len": cache_len, "config": asdict(cfg), "sanitize": stats} + + +def _main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--model", default="Qwen/Qwen3-TTS-12Hz-0.6B-CustomVoice") + parser.add_argument("--component", choices=("talker", "predictor"), default="talker") + parser.add_argument("--cache-len", type=int, default=256) + parser.add_argument("--lm-head-index", type=int, default=0, help="predictor only: which residual head to attach") + parser.add_argument("--out-dir", default="./qnn_export") + parser.add_argument("--no-weights", action="store_true", help="export the graph shape only (random weights)") + args = parser.parse_args(argv) + + from huggingface_hub import snapshot_download + + snapshot = Path(snapshot_download(args.model, allow_patterns=["config.json", "model.safetensors"])) + cfg = config_from_qwen3_tts(json.loads((snapshot / "config.json").read_text()), args.component) + module = DecodeStep(cfg).eval() + if not args.no_weights: + loaded = load_qwen3_tts_weights(module, snapshot / "model.safetensors", args.component, args.lm_head_index) + print(f"loaded {loaded} tensors ({sum(p.numel() for p in module.parameters()) / 1e6:.0f}M params)", flush=True) + out = Path(args.out_dir) / f"{args.component}_step_L{args.cache_len}.onnx" + manifest = export_onnx(module, args.cache_len, out) + manifest["model"] = args.model + manifest["component"] = args.component + (Path(args.out_dir) / f"{args.component}_step_L{args.cache_len}.json").write_text(json.dumps(manifest, indent=2)) + print(json.dumps(manifest, indent=2), flush=True) + return 0 + + +if __name__ == "__main__": + sys.exit(_main())