diff --git a/.github/actions/gpu-smoke-prepare/action.yml b/.github/actions/gpu-smoke-prepare/action.yml
index b3b416a28..4e610f60a 100644
--- a/.github/actions/gpu-smoke-prepare/action.yml
+++ b/.github/actions/gpu-smoke-prepare/action.yml
@@ -20,7 +20,7 @@ runs:
export UV_CACHE_DIR="${UV_CACHE_DIR:-${HOME}/.cache/uv}"
git config --global http.postBuffer 524288000 || true
# Base image PyPI mirror may lag pypi.org (e.g. kernels, fa3-fwd).
- uv pip install --system --break-system-packages ".[gpu,dev,audio]"
+ uv pip install --system --break-system-packages ".[gpu,dev,audio,omni]"
uv pip install --system --break-system-packages "vllm-omni @ git+https://github.com/vllm-project/vllm-omni.git@$(cat .github/vllm_omni_pin.txt)"
uv pip install --system --break-system-packages --no-deps --reinstall "verl @ git+https://github.com/verl-project/verl.git@$(cat .github/verl_pin.txt)"
uv pip install --system --break-system-packages TransferQueue==0.1.8
@@ -30,7 +30,10 @@ runs:
# TODO: rm --no-deps when VeOmni supports the vLLM torch pin (torch 2.13 as of vLLM 0.28)
uv pip install --system --break-system-packages veomni==0.1.11 --no-deps
uv pip install --system --break-system-packages torchcodec librosa soundfile av audioread
- uv pip install --system --break-system-packages "transformers[mistral-common]==5.14.1"
+ # The released qwen-tts package targets Transformers 4.57.3. Install the
+ # tested upstream Transformers 5 source without changing this repo's stack.
+ uv pip install --system --break-system-packages --no-deps --reinstall \
+ "qwen-tts @ git+https://github.com/QwenLM/Qwen3-TTS.git@$(cat .github/qwen_tts_pin.txt)"
# NCCL checkpoint engine (diffusion v1 separate_async weight sync).
uv pip install --system --break-system-packages pyzmq
uv pip install --system --break-system-packages cupy-cuda12x || uv pip install --system --break-system-packages cupy-cuda13x
diff --git a/.github/qwen_tts_pin.txt b/.github/qwen_tts_pin.txt
new file mode 100644
index 000000000..c0af4ea17
--- /dev/null
+++ b/.github/qwen_tts_pin.txt
@@ -0,0 +1 @@
+00969daa8064e23adc9e5f52cdf20cf247f94159
diff --git a/.github/workflows/cpu_unit_tests.yml b/.github/workflows/cpu_unit_tests.yml
index 4dd2b7485..8ee56e0ae 100644
--- a/.github/workflows/cpu_unit_tests.yml
+++ b/.github/workflows/cpu_unit_tests.yml
@@ -10,6 +10,7 @@ on:
- "tests/**/*_on_cpu.py"
- "pyproject.toml"
- .github/workflows/cpu_unit_tests.yml
+ - .github/qwen_tts_pin.txt
- .github/vllm_omni_pin.txt
- .github/verl_pin.txt
pull_request:
@@ -22,6 +23,7 @@ on:
- "tests/**/*_on_cpu.py"
- "pyproject.toml"
- .github/workflows/cpu_unit_tests.yml
+ - .github/qwen_tts_pin.txt
- .github/vllm_omni_pin.txt
- .github/verl_pin.txt
@@ -56,6 +58,7 @@ jobs:
cache: pip
cache-dependency-path: |
pyproject.toml
+ .github/qwen_tts_pin.txt
.github/vllm_omni_pin.txt
.github/verl_pin.txt
- name: Install dependencies
@@ -64,7 +67,9 @@ jobs:
pip install "vllm-omni @ git+https://github.com/vllm-project/vllm-omni.git@$(cat .github/vllm_omni_pin.txt)"
pip install TransferQueue==0.1.8
pip install --no-deps "verl @ git+https://github.com/verl-project/verl.git@$(cat .github/verl_pin.txt)"
- pip install ".[dev]"
+ pip install ".[omni,dev]"
+ pip install --no-deps \
+ "qwen-tts @ git+https://github.com/QwenLM/Qwen3-TTS.git@$(cat .github/qwen_tts_pin.txt)"
pip install --no-deps -e .
- name: Verify the documented train install resolves
run: |
diff --git a/.github/workflows/gpu_smoke.yml b/.github/workflows/gpu_smoke.yml
index 06888ce8d..a386adeb3 100644
--- a/.github/workflows/gpu_smoke.yml
+++ b/.github/workflows/gpu_smoke.yml
@@ -14,6 +14,7 @@ on:
- "tests/special_e2e/**"
- "pyproject.toml"
- .github/workflows/gpu_smoke.yml
+ - .github/qwen_tts_pin.txt
- .github/vllm_omni_pin.txt
- .github/verl_pin.txt
- .github/actions/gpu-smoke-prepare/**
@@ -32,6 +33,7 @@ on:
- "tests/special_e2e/**"
- "pyproject.toml"
- .github/workflows/gpu_smoke.yml
+ - .github/qwen_tts_pin.txt
- .github/vllm_omni_pin.txt
- .github/verl_pin.txt
- .github/actions/gpu-smoke-prepare/**
diff --git a/README.md b/README.md
index 6c944e3aa..74491be69 100644
--- a/README.md
+++ b/README.md
@@ -199,16 +199,20 @@ Visit our [documentation](https://verl-omni.readthedocs.io/en/latest/index.html)
| ✅ |
- | Qwen3-TTS |
- Audio-modality |
- Text → Audio |
+ Qwen3-TTS |
+ Audio-modality |
+ Text → Audio |
DPO |
WIP |
-
+
| GSPO |
WIP |
+
+ | GRPO |
+ ✅ |
+
diff --git a/docs/examples/qwen3_tts/grpo_trainer_qwen3_tts.md b/docs/examples/qwen3_tts/grpo_trainer_qwen3_tts.md
new file mode 120000
index 000000000..3bffcd1a3
--- /dev/null
+++ b/docs/examples/qwen3_tts/grpo_trainer_qwen3_tts.md
@@ -0,0 +1 @@
+../../../examples/grpo_trainer/qwen3_tts/README.md
\ No newline at end of file
diff --git a/docs/index.md b/docs/index.md
index ebb122c13..f9f360b22 100644
--- a/docs/index.md
+++ b/docs/index.md
@@ -84,6 +84,7 @@ examples/mixgrpo_trainer.md
examples/diffusionopd_trainer.md
examples/flowgrpo_trainer_sd35_drm.md
examples/bagel/flowgrpo_trainer_bagel.md
+examples/qwen3_tts/grpo_trainer_qwen3_tts.md
examples/qwen_image_edit/flowgrpo_trainer_qwen_image_edit.md
examples/ltx2/flowgrpo_trainer_ltx2.md
examples/minimax_h3/diffusionnft_trainer_minimax_h3.md
diff --git a/docs/start/models.md b/docs/start/models.md
index 277329538..7b7ba11f5 100644
--- a/docs/start/models.md
+++ b/docs/start/models.md
@@ -235,6 +235,21 @@ parquet pairs and does not start rollout or reward workers.
---
+### Qwen3-TTS-12Hz-0.6B Base
+
+| Property | Detail |
+|----------|--------|
+| **Hugging Face ID** | `Qwen/Qwen3-TTS-12Hz-0.6B-Base` |
+| **Trainable component** | Talker codec-0 policy, full-parameter example |
+| **Rollout** | Two-stage vLLM-Omni Talker + code2wav pipeline |
+| **Algorithm** | Stock GRPO, vanilla PPO loss, optional direct KL |
+| **Reward** | Generic decoded-audio reward; SpeechJudge-BTRM external scorer example |
+
+The example uses two training GPUs and an independently deployed audio scorer.
+See [Qwen3-TTS GRPO with an audio reward](../../examples/grpo_trainer/qwen3_tts/README.md).
+
+---
+
## Model Architecture Summary
| Model | Architecture | Text encoder |
@@ -248,6 +263,7 @@ parquet pairs and does not start rollout or reward workers.
| MiniMax-H3 | MiniMax H3 transformer | H3 text encoder |
| BAGEL | Unified MM | — |
| Qwen3-Omni-30B | Omni MoE | Qwen3 |
+| Qwen3-TTS-12Hz-0.6B | Talker + code2wav | Qwen3 |
---
@@ -262,7 +278,8 @@ parquet pairs and does not start rollout or reward workers.
| CLAP | `laion/larger_clap_general` | Audio | LTX-2.3 (Flow-GRPO), MiniMax-H3 (DiffusionNFT) | Local transformers load |
| ImageBind | Local `.pth` | Audio + Video | LTX-2.3 (Flow-GRPO), MiniMax-H3 (DiffusionNFT) | Local ImageBind package (CC-BY-NC-SA 4.0) |
| DiNa-LRM | HTTP latent scorer | Diffusion latents | SD3.5 (Flow-GRPO DRM) | Separate `diffusion-rm` process, safetensors HTTP |
-| HTTP scorer | External HTTP service | Any | Any model | Gunicorn/Flask, pickle protocol |
+| HTTP scorer | External HTTP service | Image/audio | Any model | Pickle image or JSON audio protocol |
+| SpeechJudge-BTRM | `RMSnow/SpeechJudge-BTRM` | Audio quality | Qwen3-TTS example | External service; CC-BY-NC-4.0 |
| JPEG incompressibility | Rule-based | Image stats | Any diffusion model | No model process needed |
For end-to-end instructions on setting up each reward, see the respective
@@ -272,18 +289,20 @@ trainer's README in `examples/`.
## Which Trainer for Which Model?
-| Algorithm | Qwen-Image | Qwen-Image-Edit | SD3.5 | Wan2.2 | LTX-2.3 | MiniMax-H3 | BAGEL | Qwen3-Omni |
-|-----------|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:|
-| Flow-GRPO | ✅ | ✅ | ✅ | — | ✅ | WIP | ✅ | — |
-| Flow-DPPO | ✅ | — | — | — | — | — | — | — |
-| GRPO-Guard | ✅ | — | — | — | — | — | — | — |
-| Mix-GRPO | ✅ | — | — | — | — | — | — | — |
-| DanceGRPO | — | — | — | ✅ | — | — | — | — |
-| DPO | ✅ | — | ✅ | — | — | — | — | ✅ |
-| DiffusionNFT | ✅ | — | — | — | — | ✅ | — | — |
-| [DiffusionOPD](../algo/diffusion_opd.md) (incl. MOPD) | — | — | ✅ | — | — | — | — | — |
-| GSPO (incl. OPD) | — | — | — | — | — | — | — | ✅ |
-
-HunyuanImage-3.0 (MixGRPO / SRPO) and Qwen3-TTS (DPO / GSPO) appear on the
-project README as Planned or WIP and do not yet have a ready-to-run recipe, so
-they are omitted from the catalogue above.
+| Algorithm | Qwen-Image | Qwen-Image-Edit | SD3.5 | Wan2.2 | LTX-2.3 | MiniMax-H3 | BAGEL | Qwen3-Omni | Qwen3-TTS |
+|-----------|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:|:---:|
+| GRPO | — | — | — | — | — | — | — | — | ✅ |
+| Flow-GRPO | ✅ | ✅ | ✅ | — | ✅ | WIP | ✅ | — | — |
+| Flow-DPPO | ✅ | — | — | — | — | — | — | — | — |
+| GRPO-Guard | ✅ | — | — | — | — | — | — | — | — |
+| Mix-GRPO | ✅ | — | — | — | — | — | — | — | — |
+| DanceGRPO | — | — | — | ✅ | — | — | — | — | — |
+| DPO | ✅ | — | ✅ | — | — | — | — | ✅ | WIP |
+| DiffusionNFT | ✅ | — | — | — | — | ✅ | — | — | — |
+| [DiffusionOPD](../algo/diffusion_opd.md) (incl. MOPD) | — | — | ✅ | — | — | — | — | — | — |
+| GSPO (incl. OPD) | — | — | — | — | — | — | — | ✅ | WIP |
+
+HunyuanImage-3.0 (MixGRPO / SRPO) appears on the project README as Planned or
+WIP and does not yet have a ready-to-run recipe, so it is omitted from the
+catalogue above. Qwen3-TTS DPO and GSPO remain WIP; its ready-to-run GRPO recipe
+is listed above.
diff --git a/examples/grpo_trainer/qwen3_tts/README.md b/examples/grpo_trainer/qwen3_tts/README.md
new file mode 100644
index 000000000..57ef3b4be
--- /dev/null
+++ b/examples/grpo_trainer/qwen3_tts/README.md
@@ -0,0 +1,160 @@
+# Qwen3-TTS GRPO with an audio reward
+
+Last updated: 09/04/2026.
+
+This example full-parameter tunes the codec-0 policy of
+`Qwen/Qwen3-TTS-12Hz-0.6B-Base`. It uses verl's stock GRPO advantage,
+vanilla PPO policy loss, and optional direct reference-model KL. The other
+15 codec codebooks and code2wav stage remain frozen but are retained so every
+candidate can be decoded and scored as audio.
+
+The launcher follows the V1 omni-model integration guide: it calls
+`verl_omni.trainer.main_omni` and expresses the recipe as CLI overrides on the
+standard `omni_trainer` config, without a model-specific Trainer or config tree.
+
+SpeechJudge-BTRM is one possible pointwise scorer. SpeechJudge's published vLLM
+entry point targets the pairwise generative GRM, while BTRM uses a scalar reward
+head with Transformers. This example therefore keeps reward inference behind the
+generic audio HTTP protocol instead of adding a SpeechJudge-specific Trainer
+path. The scorer may run in a separate environment from the Transformers 5.x
+vLLM training stack.
+
+## Algorithm background
+
+This recipe applies the paper's TTS GRPO flow to Qwen3-TTS: grouped codec-token
+rollouts are decoded, scored, converted to group-relative advantages, and
+replayed with optional reference KL. It optimizes codec-0, the autoregressive
+policy sequence described by the Qwen3-TTS architecture, while retaining all 16
+codebooks for replay and waveform decoding. The HTTP scorer is configurable, so
+this is not an exact reproduction of the paper's CER-and-NLL reward. See the
+references below for the algorithm and multi-codebook design details.
+
+## Install
+
+Install the engine before the training stack:
+
+```bash
+uv pip install -e ".[gpu]" --torch-backend=auto
+uv pip install "vllm-omni @ git+https://github.com/vllm-project/vllm-omni.git@$(cat .github/vllm_omni_pin.txt)"
+uv pip install -e ".[omni,train,dev]"
+uv pip install --no-deps --reinstall \
+ "qwen-tts @ git+https://github.com/QwenLM/Qwen3-TTS.git@$(cat .github/qwen_tts_pin.txt)"
+```
+
+The pinned Qwen3-TTS revision is the upstream Transformers 5 support change
+from Qwen3-TTS PR #360. Its package metadata requires Transformers 5.15.1 or
+newer, while this repository intentionally caps Transformers at 5.14.1. The
+`--no-deps` flag preserves that repository-wide cap; the `omni` extra owns the
+runtime dependencies, including `torchaudio==2.11.0` to match vLLM's Torch pin,
+and CI tests the exact Qwen3-TTS revision from `.github/qwen_tts_pin.txt` on this
+stack. The released `qwen-tts==0.1.1` source targets Transformers 4.57.3 and
+cannot be imported unchanged here. The adapter registers the upstream config
+and model with `AutoConfig` and `AutoModelForTextToWaveform`; it does not carry
+a local Transformers compatibility layer. The system `sox` executable is also
+required by qwen-tts.
+
+## Data
+
+Training and validation parquet rows use the normal verl format:
+
+```python
+{
+ "data_source": "tts",
+ "prompt": [{"role": "user", "content": "Text to synthesize"}],
+ "reward_model": {"style": "model", "ground_truth": "Text to synthesize"},
+ "extra_info": {"id": "stable-id", "split": "train"},
+}
+```
+
+Use disjoint prompts. The default recipe evaluates the same complete 100-row
+validation parquet at step 0 and every 20 updates. It uses the rollout engine's
+global seed; model-specific per-request seed derivation is intentionally outside
+this integration.
+
+The concatenated replay layout also requires one fixed speaker embedding JSON.
+Generate it once with the official Qwen3-TTS Base model's
+`extract_speaker_embedding` API from a 24 kHz reference recording, then reuse
+the same file for the entire run.
+
+## Audio scorer protocol
+
+The configured endpoint receives one JSON request per candidate:
+
+```json
+{
+ "protocol_version": "1",
+ "waveform_f32_base64": "...",
+ "num_samples": 24000,
+ "sample_rate": 24000,
+ "prompt": "Text to synthesize",
+ "metadata": {"id": "stable-id"}
+}
+```
+
+It must return `{"score": 1.25}` and may include additional scalar metrics.
+The client retries only transient network, timeout, HTTP 408/429, and 5xx
+failures. Missing, malformed, or non-finite results stop the run instead of
+being converted to a valid zero reward.
+
+For SpeechJudge-BTRM, deploy the official
+[`AmphionTeam/SpeechJudge`](https://github.com/AmphionTeam/SpeechJudge) code and
+[`RMSnow/SpeechJudge-BTRM`](https://huggingface.co/RMSnow/SpeechJudge-BTRM)
+checkpoint in a separate environment, then expose its pointwise score through
+this protocol. The official [`main_grm_vllm.py`](https://github.com/AmphionTeam/SpeechJudge/blob/master/infer/main_grm_vllm.py)
+runs a different, pairwise generative GRM path; the BTRM entry point is
+[`main_btrm.py`](https://github.com/AmphionTeam/SpeechJudge/blob/master/infer/main_btrm.py).
+Pin the SpeechJudge source revision and runtime versions in the service
+deployment. SpeechJudge-BTRM is licensed CC-BY-NC-4.0.
+
+## Train
+
+```bash
+MODEL_PATH=/path/to/Qwen3-TTS-12Hz-0.6B-Base \
+TRAIN_FILE=/path/to/train.parquet \
+VAL_FILE=/path/to/fixed-validation-100.parquet \
+SPK_EMBED_PATH=/path/to/speaker.json \
+SCORER_URL=http://scorer-host:18080/score \
+OUTPUT_DIR=/path/to/output \
+bash examples/grpo_trainer/qwen3_tts/run_qwen3_tts_grpo.sh
+```
+
+The example defaults are `B=4`, `G=8`, `lr=1e-6` with 10 warmup steps and a
+constant schedule, direct `low_var_kl` with coefficient `0.12`, two GPUs, and
+500 updates. These are recipe values, not algorithm requirements.
+`norm_adv_by_std_in_grpo` remains at the upstream default. The actor and
+reference keep persistent parameters in FP32, while FSDP uses BF16 parameters
+for forward and backward computation with FP32 gradient reduction and buffers.
+The actor's AdamW state therefore remains FP32, and rollout inference remains
+BF16.
+
+For a two-update implementation smoke test:
+
+```bash
+TOTAL_TRAINING_STEPS=2 TEST_FREQ=-1 SAVE_FREQ=-1 RESUME_MODE=disable \
+OUTPUT_DIR=outputs/qwen3_tts_grpo_smoke \
+bash examples/grpo_trainer/qwen3_tts/run_qwen3_tts_grpo.sh \
+ trainer.val_before_train=false trainer.log_val_generations=0
+```
+
+This smoke proves rollout, finite audio reward, optimizer update, and
+post-update weight sync only. It is not evidence that GRPO improves held-out
+speech quality; that requires the complete fixed-validation curve and paired
+human listening evaluation.
+
+The CI-oriented wrapper at
+[`tests/special_e2e/run_qwen3_tts_grpo_smoke.sh`](../../../tests/special_e2e/run_qwen3_tts_grpo_smoke.sh)
+creates deterministic fixtures, uses an in-process CPU duration reward, and runs
+two updates with a pinned tiny-random checkpoint rebuilt for the 16-codebook
+actor contract.
+
+## References
+
+- Chang Liu, Ya-Jun Hu, Ying-Ying Gao, Shi-Lei Zhang, and Zhen-Hua Ling.
+ [Group Relative Policy Optimization for Text-to-Speech with Large Language
+ Models](https://arxiv.org/abs/2509.18798), 2025.
+- Hangrui Hu et al. [Qwen3-TTS Technical
+ Report](https://arxiv.org/abs/2601.15621), 2026.
+- QwenLM. [Qwen3-TTS PR #360: Support Transformers
+ 5](https://github.com/QwenLM/Qwen3-TTS/pull/360), 2026.
+- Dong Zhang et al. [SpeechAlign: Aligning Speech Generation to Human
+ Preferences](https://arxiv.org/abs/2404.05600), 2024.
diff --git a/examples/grpo_trainer/qwen3_tts/run_qwen3_tts_grpo.sh b/examples/grpo_trainer/qwen3_tts/run_qwen3_tts_grpo.sh
new file mode 100755
index 000000000..1a42a6b1d
--- /dev/null
+++ b/examples/grpo_trainer/qwen3_tts/run_qwen3_tts_grpo.sh
@@ -0,0 +1,155 @@
+#!/usr/bin/env bash
+# Copyright 2026 Bytedance Ltd. and/or its affiliates
+# Licensed under the Apache License, Version 2.0
+
+set -euo pipefail
+
+MODEL_PATH="${MODEL_PATH:?Set MODEL_PATH to a local Qwen3-TTS Base directory}"
+TRAIN_FILE="${TRAIN_FILE:?Set TRAIN_FILE to the training parquet}"
+VAL_FILE="${VAL_FILE:?Set VAL_FILE to the fixed validation parquet}"
+SPK_EMBED_PATH="${SPK_EMBED_PATH:?Set SPK_EMBED_PATH to a speaker embedding JSON file}"
+SCORER_URL="${SCORER_URL:?Set SCORER_URL to an audio scorer /score endpoint}"
+OUTPUT_DIR="${OUTPUT_DIR:-outputs/qwen3_tts_grpo}"
+NUM_GPUS="${NUM_GPUS:-2}"
+SEED="${SEED:-42}"
+TOTAL_TRAINING_STEPS="${TOTAL_TRAINING_STEPS:-500}"
+TEST_FREQ="${TEST_FREQ:-20}"
+SAVE_FREQ="${SAVE_FREQ:-20}"
+RESUME_MODE="${RESUME_MODE:-auto}"
+PYTHON_BIN="${PYTHON_BIN:-python3}"
+for path in "${MODEL_PATH}" "${TRAIN_FILE}" "${VAL_FILE}" "${SPK_EMBED_PATH}"; do
+ [[ -e "${path}" ]] || { printf 'Missing required path: %s\n' "${path}" >&2; exit 2; }
+done
+
+mkdir -p "${OUTPUT_DIR}"
+export PYTHONHASHSEED="${SEED}"
+export TOKENIZERS_PARALLELISM=false
+export VERL_USE_EXTERNAL_MODULES=verl_omni
+export VLLM_USE_FLASHINFER_SAMPLER=0
+
+"${PYTHON_BIN}" -m verl_omni.trainer.main_omni \
+ "data.train_files=${TRAIN_FILE}" \
+ "data.val_files=${VAL_FILE}" \
+ "data.seed=${SEED}" \
+ data.train_batch_size=4 \
+ data.dataloader_num_workers=0 \
+ data.max_prompt_length=512 \
+ data.max_response_length=360 \
+ data.val_max_samples=100 \
+ data.validation_shuffle=false \
+ data.filter_overlong_prompts=true \
+ data.truncation=error \
+ "actor_rollout_ref.model.path=${MODEL_PATH}" \
+ actor_rollout_ref.model.model_stage=talker \
+ actor_rollout_ref.model.hf_config_name=talker_config \
+ actor_rollout_ref.model.trust_remote_code=false \
+ actor_rollout_ref.model.use_remove_padding=false \
+ actor_rollout_ref.model.enable_gradient_checkpointing=true \
+ +actor_rollout_ref.model.override_config.attn_implementation=eager \
+ +actor_rollout_ref.model.override_config.tts_language=Auto \
+ "+actor_rollout_ref.model.override_config.tts_spk_embed_path=${SPK_EMBED_PATH}" \
+ actor_rollout_ref.actor.strategy=fsdp \
+ actor_rollout_ref.actor.ppo_mini_batch_size=4 \
+ actor_rollout_ref.actor.ppo_micro_batch_size_per_gpu=1 \
+ actor_rollout_ref.actor.use_dynamic_bsz=false \
+ actor_rollout_ref.actor.use_kl_loss=true \
+ actor_rollout_ref.actor.kl_loss_coef=0.12 \
+ actor_rollout_ref.actor.kl_loss_type=low_var_kl \
+ actor_rollout_ref.actor.ppo_epochs=1 \
+ actor_rollout_ref.actor.shuffle=false \
+ actor_rollout_ref.actor.clip_ratio_low=0.2 \
+ actor_rollout_ref.actor.clip_ratio_high=0.2 \
+ actor_rollout_ref.actor.entropy_coeff=0.0 \
+ actor_rollout_ref.actor.loss_agg_mode=seq-mean-token-mean \
+ actor_rollout_ref.actor.use_torch_compile=false \
+ actor_rollout_ref.actor.policy_loss.loss_mode=vanilla \
+ actor_rollout_ref.actor.optim.lr=1.0e-6 \
+ actor_rollout_ref.actor.optim.lr_warmup_steps=10 \
+ actor_rollout_ref.actor.optim.lr_scheduler_type=constant \
+ actor_rollout_ref.actor.optim.weight_decay=0.0 \
+ actor_rollout_ref.actor.optim.clip_grad=1.0 \
+ "actor_rollout_ref.actor.data_loader_seed=${SEED}" \
+ "actor_rollout_ref.actor.fsdp_config.fsdp_size=${NUM_GPUS}" \
+ "actor_rollout_ref.actor.fsdp_config.seed=${SEED}" \
+ actor_rollout_ref.actor.fsdp_config.model_dtype=float32 \
+ actor_rollout_ref.actor.fsdp_config.dtype=bfloat16 \
+ '+actor_rollout_ref.actor.fsdp_config.mixed_precision={param_dtype:bf16,reduce_dtype:fp32,buffer_dtype:fp32}' \
+ actor_rollout_ref.actor.fsdp_config.param_offload=false \
+ actor_rollout_ref.actor.fsdp_config.optimizer_offload=false \
+ actor_rollout_ref.actor.fsdp_config.use_orig_params=true \
+ actor_rollout_ref.actor.fsdp_config.use_torch_compile=false \
+ actor_rollout_ref.actor.fsdp_config.wrap_policy.min_num_params=0 \
+ actor_rollout_ref.rollout.name=vllm_omni \
+ actor_rollout_ref.rollout.mode=async \
+ actor_rollout_ref.rollout.n=8 \
+ actor_rollout_ref.rollout.temperature=1.0 \
+ actor_rollout_ref.rollout.top_p=1.0 \
+ actor_rollout_ref.rollout.top_k=-1 \
+ actor_rollout_ref.rollout.dtype=bfloat16 \
+ actor_rollout_ref.rollout.tensor_model_parallel_size=1 \
+ actor_rollout_ref.rollout.gpu_memory_utilization=0.20 \
+ actor_rollout_ref.rollout.max_num_seqs=8 \
+ actor_rollout_ref.rollout.max_num_batched_tokens=1024 \
+ actor_rollout_ref.rollout.free_cache_engine=false \
+ actor_rollout_ref.rollout.calculate_log_probs=true \
+ actor_rollout_ref.rollout.logprobs_mode=processed_logprobs \
+ actor_rollout_ref.rollout.load_format=safetensors \
+ actor_rollout_ref.rollout.layered_summon=true \
+ actor_rollout_ref.rollout.enable_prefix_caching=false \
+ actor_rollout_ref.rollout.enforce_eager=true \
+ actor_rollout_ref.rollout.log_prob_micro_batch_size_per_gpu=1 \
+ actor_rollout_ref.rollout.agent.num_workers=8 \
+ actor_rollout_ref.rollout.agent.default_agent_loop=omni_single_turn_agent \
+ "+actor_rollout_ref.rollout.engine_kwargs.vllm_omni.seed=${SEED}" \
+ +actor_rollout_ref.rollout.engine_kwargs.vllm_omni.output_mode=ar \
+ +actor_rollout_ref.rollout.engine_kwargs.vllm_omni.pipeline_name=qwen3_tts_rl \
+ +actor_rollout_ref.rollout.engine_kwargs.vllm_omni.pipeline_mode=full \
+ +actor_rollout_ref.rollout.engine_kwargs.vllm_omni.async_chunk=false \
+ +actor_rollout_ref.rollout.engine_kwargs.vllm_omni.attention_config.backend=TRITON_ATTN \
+ actor_rollout_ref.rollout.val_kwargs.n=1 \
+ actor_rollout_ref.rollout.val_kwargs.do_sample=true \
+ actor_rollout_ref.rollout.val_kwargs.temperature=1.0 \
+ actor_rollout_ref.rollout.val_kwargs.top_p=1.0 \
+ actor_rollout_ref.rollout.val_kwargs.top_k=-1 \
+ actor_rollout_ref.ref.log_prob_micro_batch_size_per_gpu=1 \
+ actor_rollout_ref.ref.strategy=fsdp \
+ "actor_rollout_ref.ref.fsdp_config.fsdp_size=${NUM_GPUS}" \
+ "actor_rollout_ref.ref.fsdp_config.seed=${SEED}" \
+ actor_rollout_ref.ref.fsdp_config.model_dtype=float32 \
+ actor_rollout_ref.ref.fsdp_config.dtype=bfloat16 \
+ '+actor_rollout_ref.ref.fsdp_config.mixed_precision={param_dtype:bf16,reduce_dtype:fp32,buffer_dtype:fp32}' \
+ actor_rollout_ref.ref.fsdp_config.param_offload=false \
+ actor_rollout_ref.ref.fsdp_config.use_orig_params=true \
+ actor_rollout_ref.ref.fsdp_config.wrap_policy.min_num_params=0 \
+ algorithm.adv_estimator=grpo \
+ algorithm.norm_adv_by_std_in_grpo=true \
+ algorithm.use_kl_in_reward=false \
+ reward.num_workers=1 \
+ reward.custom_reward_function.path=pkg://verl_omni.utils.reward_score.audio_http_scorer_client \
+ reward.custom_reward_function.name=compute_score \
+ "+reward.custom_reward_function.reward_kwargs.server_url=${SCORER_URL}" \
+ +reward.custom_reward_function.reward_kwargs.timeout=120.0 \
+ +reward.custom_reward_function.reward_kwargs.max_retries=2 \
+ +reward.custom_reward_function.reward_kwargs.retry_backoff=0.5 \
+ reward.reward_manager.source=importlib \
+ reward.reward_manager.name=AudioRewardManager \
+ reward.reward_manager.module.path=pkg://verl_omni.reward_loop.reward_manager \
+ trainer.val_before_train=true \
+ trainer.balance_batch=true \
+ trainer.critic_warmup=0 \
+ trainer.logger='["console"]' \
+ trainer.project_name=qwen3_tts_grpo \
+ trainer.experiment_name=qwen3_tts_0_6b_audio_reward \
+ "trainer.n_gpus_per_node=${NUM_GPUS}" \
+ trainer.nnodes=1 \
+ trainer.total_epochs=100 \
+ trainer.log_val_generations=100 \
+ "trainer.total_training_steps=${TOTAL_TRAINING_STEPS}" \
+ "trainer.test_freq=${TEST_FREQ}" \
+ "trainer.save_freq=${SAVE_FREQ}" \
+ "trainer.resume_mode=${RESUME_MODE}" \
+ trainer.max_actor_ckpt_to_keep=26 \
+ "trainer.default_local_dir=${OUTPUT_DIR}/checkpoints" \
+ "trainer.validation_data_dir=${OUTPUT_DIR}/validation" \
+ "trainer.rollout_data_dir=${OUTPUT_DIR}/rollout" \
+ "$@" 2>&1 | tee "${OUTPUT_DIR}/train.log"
diff --git a/pyproject.toml b/pyproject.toml
index 39af3a4f1..08fbe060f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -59,6 +59,16 @@ audio = [
"qwen-omni-utils>=0.0.9",
"audioread",
]
+# Install qwen-tts separately from the pinned upstream Transformers 5 support
+# revision. The released 0.1.1 source and metadata target Transformers 4.57.3.
+omni = [
+ "einops>=0.8.0",
+ "librosa>=0.10.2",
+ "onnxruntime>=1.20.0",
+ "soundfile>=0.12.1",
+ "sox>=1.5.0",
+ "torchaudio==2.11.0",
+]
# CUDA rollout backend (vllm) + actor FA3 (kernels) + liger-kernel. Install in step 1 on GPU.
gpu = [
"vllm==0.28.0",
diff --git a/tests/gpu_smoke/run_gpu_smoke_omni_e2e.sh b/tests/gpu_smoke/run_gpu_smoke_omni_e2e.sh
index 979213742..dd8d5431d 100644
--- a/tests/gpu_smoke/run_gpu_smoke_omni_e2e.sh
+++ b/tests/gpu_smoke/run_gpu_smoke_omni_e2e.sh
@@ -21,4 +21,8 @@ run_test 1 "Qwen3-Omni multimodal offline MLLM DPO LoRA e2e" \
env CUDA_VISIBLE_DEVICES="${CUDA_DEVICE_LIST}" NUM_GPUS=2 \
bash tests/special_e2e/run_qwen3_omni_multimodal_offline_mllm_dpo_lora_smoke.sh "${omni_trainer_args[@]}"
+run_test 2 "Qwen3-TTS Talker full-parameter GRPO e2e" \
+ env CUDA_VISIBLE_DEVICES="${CUDA_DEVICE_LIST}" NUM_GPUS=2 \
+ bash tests/special_e2e/run_qwen3_tts_grpo_smoke.sh "${omni_trainer_args[@]}"
+
gpu_smoke_summary
diff --git a/tests/gpu_smoke/select_gpu_smoke_groups.py b/tests/gpu_smoke/select_gpu_smoke_groups.py
index 17c980338..0e5de914f 100644
--- a/tests/gpu_smoke/select_gpu_smoke_groups.py
+++ b/tests/gpu_smoke/select_gpu_smoke_groups.py
@@ -66,9 +66,14 @@ class SmokeGroup:
"verl_omni/workers/**",
),
"ci-e2e-omni": (
+ ".github/qwen_tts_pin.txt",
+ "examples/grpo_trainer/qwen3_tts/**",
"tests/gpu_smoke/run_gpu_smoke_omni_e2e.sh",
+ "tests/pipelines/test_qwen3_tts*",
"tests/special_e2e/*omni*",
+ "tests/special_e2e/*qwen3_tts*",
"verl_omni/models/transformers/qwen3_omni_thinker.py",
+ "verl_omni/pipelines/qwen3_tts/**",
"verl_omni/trainer/config/omni/**",
"verl_omni/trainer/omni/**",
),
diff --git a/tests/pipelines/test_qwen3_tts_on_cpu.py b/tests/pipelines/test_qwen3_tts_on_cpu.py
new file mode 100644
index 000000000..fc00b3488
--- /dev/null
+++ b/tests/pipelines/test_qwen3_tts_on_cpu.py
@@ -0,0 +1,167 @@
+# Copyright 2026 Bytedance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Dependency-light Qwen3-TTS actor and rollout contract tests."""
+
+import importlib.util
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+import torch
+
+
+def _load(name: str, relative_path: str):
+ spec = importlib.util.spec_from_file_location(name, Path(__file__).parents[2] / relative_path)
+ module = importlib.util.module_from_spec(spec)
+ assert spec.loader is not None
+ sys.modules[name] = module
+ spec.loader.exec_module(module)
+ return module
+
+
+forward = _load("qwen3_tts_forward_test", "verl_omni/pipelines/qwen3_tts/talker_forward.py")
+rollout = _load("qwen3_tts_rollout_test", "verl_omni/pipelines/qwen3_tts/rollout_utils.py")
+
+TOKENS = forward.TalkerTokens(900, 901, 902, 4196, 4197, 4198, 4203, 4204, 4205)
+
+
+def test_talker_batch_matches_auto_language_teacher_forcing_layout():
+ text_ids = torch.tensor([1, 2, 3, 4, 5, 6])
+ codes = torch.arange(3 * 16, dtype=torch.long).reshape(3, 16) % 128
+ codes[:, 0] = torch.tensor([10, 11, 12])
+
+ batch = forward.build_talker_batch([text_ids], [codes], TOKENS, sub_codebook_vocab=2048)
+
+ speaker_slot = 6
+ codec_start = 8 + text_ids.numel() - 1
+ assert batch.input_ids[0, 3:speaker_slot, 1].tolist() == [4203, 4204, 4205]
+ assert not batch.codec_embedding_mask[0, speaker_slot]
+ assert batch.input_ids[0, speaker_slot + 1, 1] == TOKENS.codec_pad
+ torch.testing.assert_close(batch.codec_ids[0, codec_start : codec_start + 3], codes)
+ assert batch.logit_start == [codec_start - 1]
+ assert batch.codec_lens == [3]
+
+
+def test_codec0_mask_matches_rollout_vocabulary():
+ masked = forward.mask_codec0_logits(torch.zeros((1, 2, 4300)), 2048, TOKENS.codec_eos)
+
+ assert (masked[..., 0] < -1e3).all()
+ assert torch.isfinite(masked[..., 1:2048]).all()
+ assert (masked[..., 2048 : TOKENS.codec_eos] < -1e3).all()
+ assert torch.isfinite(masked[..., TOKENS.codec_eos]).all()
+ assert (masked[..., TOKENS.codec_eos + 1 :] < -1e3).all()
+
+
+def test_only_validated_auto_language_layout_is_accepted():
+ assert forward.require_auto_language("auto") == "Auto"
+ with pytest.raises(ValueError, match="supports only tts_language=Auto"):
+ forward.require_auto_language("Chinese")
+ with pytest.raises(ValueError, match="supports only tts_language=Auto"):
+ forward.require_auto_language(None)
+
+
+def test_actor_logits_align_to_effective_codec0_response(monkeypatch):
+ class CodePredictor:
+ @staticmethod
+ def get_input_embeddings():
+ return [torch.nn.Embedding(2048, 4)]
+
+ talker = SimpleNamespace(code_predictor=CodePredictor())
+ model = SimpleNamespace(
+ talker=talker,
+ config=SimpleNamespace(
+ tts_pad_token_id=TOKENS.tts_pad,
+ tts_bos_token_id=TOKENS.tts_bos,
+ tts_eos_token_id=TOKENS.tts_eos,
+ talker_config=SimpleNamespace(
+ codec_pad_id=TOKENS.codec_pad,
+ codec_bos_id=TOKENS.codec_bos,
+ codec_eos_token_id=TOKENS.codec_eos,
+ codec_nothink_id=TOKENS.codec_nothink,
+ codec_think_bos_id=TOKENS.codec_think_bos,
+ codec_think_eos_id=TOKENS.codec_think_eos,
+ ),
+ ),
+ )
+
+ def fake_logits(_talker, batch, _speaker):
+ vocab = torch.arange(1, 4301, dtype=torch.float32).reshape(1, 1, -1)
+ return vocab.expand(1, batch.input_ids.shape[1] - 1, -1)
+
+ monkeypatch.setattr(forward, "codec0_logits", fake_logits)
+ input_ids = torch.zeros((1, 9), dtype=torch.long)
+ input_ids[0, -3:] = torch.tensor([10, 11, 12])
+ codes = torch.zeros((1, 6, 16), dtype=torch.long)
+ codes[0, :3, 0] = torch.tensor([10, 11, 12])
+
+ logits = forward.tts_actor_logits(
+ model,
+ input_ids,
+ torch.ones_like(input_ids),
+ torch.tensor([[1, 2, 3, 4, 5, 6]]),
+ codes,
+ torch.tensor([3]),
+ torch.tensor([6]),
+ torch.zeros((1, 4)),
+ )
+
+ assert torch.nonzero(logits.abs().sum(dim=-1)[0], as_tuple=False).reshape(-1).tolist() == [5, 6, 7]
+ assert logits[0, 5, 0] == -1e4
+ assert logits[0, 5, 1] == 2
+ assert logits[0, 5, 2047] == 2048
+ assert logits[0, 5, 2048] == -1e4
+ assert logits[0, 5, TOKENS.codec_eos] == TOKENS.codec_eos + 1
+
+
+@pytest.mark.parametrize("response_length", [2, 14, 15, 16, 17, 32])
+def test_codec_alignment_recovers_exact_prefix_without_final_residual_row(response_length):
+ token_ids = list(range(100, 100 + response_length - 1)) + [2150]
+ generated = torch.arange((response_length - 1) * 16, dtype=torch.long).reshape(response_length - 1, 16) + 1
+ generated[:, 0] = torch.tensor(token_ids[:-1])
+ raw = torch.cat((torch.zeros(12, 16, dtype=torch.long), generated))
+
+ aligned = rollout.align_audio_codes(raw, token_ids)
+
+ assert aligned[:, 0].tolist() == token_ids
+ torch.testing.assert_close(aligned[:-1, 1:], generated[:, 1:])
+ assert not aligned[-1, 1:].any()
+
+
+def test_codec_alignment_preserves_final_row_and_rejects_heuristic_match():
+ token_ids = [101, 102, 103, 2150]
+ generated = torch.arange(4 * 16, dtype=torch.long).reshape(4, 16) + 1
+ generated[:, 0] = torch.tensor(token_ids)
+ aligned = rollout.align_audio_codes(torch.cat((torch.zeros(12, 16, dtype=torch.long), generated)), token_ids)
+ torch.testing.assert_close(aligned, generated)
+
+ malformed = torch.zeros(15, 16, dtype=torch.long)
+ malformed[12:, 0] = torch.tensor([101, 999, 103])
+ with pytest.raises(RuntimeError, match="Could not exactly align"):
+ rollout.align_audio_codes(malformed, token_ids)
+
+
+def test_codec_alignment_rejects_ambiguous_exact_matches():
+ raw = torch.zeros(2, 16, dtype=torch.long)
+ raw[:, 0] = 101
+
+ with pytest.raises(RuntimeError, match="Ambiguous Qwen3-TTS codec alignment"):
+ rollout.align_audio_codes(raw, [101, 2150])
+
+
+def test_talker_batch_and_logit_mask_reject_contract_mismatches():
+ with pytest.raises(ValueError, match="matching non-empty"):
+ forward.build_talker_batch([], [], TOKENS, sub_codebook_vocab=2048)
+ with pytest.raises(ValueError, match="codebook_vocab"):
+ forward.mask_codec0_logits(torch.zeros((1, 2, 100)), 101, TOKENS.codec_eos)
diff --git a/tests/pipelines/test_qwen3_tts_package_on_cpu.py b/tests/pipelines/test_qwen3_tts_package_on_cpu.py
new file mode 100644
index 000000000..a6c6fa069
--- /dev/null
+++ b/tests/pipelines/test_qwen3_tts_package_on_cpu.py
@@ -0,0 +1,185 @@
+# Copyright 2026 Bytedance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Check the pinned upstream qwen-tts TF5 source on the repository stack."""
+
+import importlib.util
+import json
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+import torch
+
+
+def _load_smoke_helper(filename: str):
+ path = Path(__file__).parents[1] / f"special_e2e/{filename}.py"
+ spec = importlib.util.spec_from_file_location(f"{filename}_under_test", path)
+ module = importlib.util.module_from_spec(spec)
+ assert spec.loader is not None
+ spec.loader.exec_module(module)
+ return module
+
+
+def test_smoke_speaker_dimension_matches_model_config(tmp_path):
+ builder = _load_smoke_helper("create_dummy_qwen3_tts_grpo_data")
+ model_config_path = tmp_path / "config.json"
+ model_config_path.write_text(
+ json.dumps(
+ {
+ "talker_config": {"hidden_size": 128},
+ "speaker_encoder_config": {"enc_dim": 128},
+ }
+ ),
+ encoding="utf-8",
+ )
+
+ assert builder._speaker_dimension(model_config_path) == 128
+
+ model_config_path.write_text(
+ json.dumps(
+ {
+ "talker_config": {"hidden_size": 128},
+ "speaker_encoder_config": {"enc_dim": 1024},
+ }
+ ),
+ encoding="utf-8",
+ )
+ with pytest.raises(ValueError, match="speaker encoder output must match"):
+ builder._speaker_dimension(model_config_path)
+
+
+def test_smoke_tiny_model_mrope_section_matches_head_dimension():
+ builder = _load_smoke_helper("build_qwen3_tts_tiny_random")
+
+ section = builder._scaled_mrope_section([24, 20, 20], head_dim=64)
+
+ assert section == [12, 10, 10]
+ assert sum(section) == 64 // 2
+
+
+def test_qwen_tts_registers_and_runs_without_a_transformers_compatibility_layer(tmp_path, monkeypatch):
+ transformers = pytest.importorskip("transformers")
+ if importlib.util.find_spec("qwen_tts") is None:
+ pytest.skip("qwen-tts is an optional dependency")
+
+ from qwen_tts.core.models.configuration_qwen3_tts import Qwen3TTSConfig
+ from qwen_tts.core.models.modeling_qwen3_tts import Qwen3TTSForConditionalGeneration
+ from transformers import AutoConfig, AutoModelForTextToWaveform
+
+ assert int(transformers.__version__.split(".", maxsplit=1)[0]) >= 5
+ AutoConfig.register("qwen3_tts", Qwen3TTSConfig, exist_ok=True)
+ AutoModelForTextToWaveform.register(
+ Qwen3TTSConfig,
+ Qwen3TTSForConditionalGeneration,
+ exist_ok=True,
+ )
+ assert AutoConfig.for_model("qwen3_tts").__class__ is Qwen3TTSConfig
+ assert AutoModelForTextToWaveform._model_mapping[Qwen3TTSConfig] is Qwen3TTSForConditionalGeneration
+
+ predictor = {
+ "vocab_size": 32,
+ "hidden_size": 8,
+ "intermediate_size": 16,
+ "num_hidden_layers": 1,
+ "num_attention_heads": 2,
+ "num_key_value_heads": 1,
+ "head_dim": 4,
+ "max_position_embeddings": 64,
+ "num_code_groups": 16,
+ "layer_types": ["full_attention"],
+ "pad_token_id": None,
+ }
+ talker = {
+ "code_predictor_config": predictor,
+ "vocab_size": 64,
+ "hidden_size": 8,
+ "intermediate_size": 16,
+ "num_hidden_layers": 1,
+ "num_attention_heads": 2,
+ "num_key_value_heads": 1,
+ "max_position_embeddings": 64,
+ "num_code_groups": 16,
+ "text_hidden_size": 8,
+ "text_vocab_size": 80,
+ "codec_eos_token_id": 50,
+ "codec_nothink_id": 51,
+ "codec_think_bos_id": 52,
+ "codec_think_eos_id": 53,
+ "codec_pad_id": 54,
+ "codec_bos_id": 55,
+ "spk_id": {},
+ "codec_language_id": {},
+ "rope_scaling": {
+ "rope_type": "default",
+ "type": "default",
+ "mrope_section": [1, 1, 0],
+ "interleaved": True,
+ },
+ }
+ config = Qwen3TTSConfig(
+ talker_config=talker,
+ speaker_encoder_config={},
+ tts_model_type="custom",
+ tokenizer_type="12hz",
+ tts_pad_token_id=60,
+ tts_bos_token_id=61,
+ tts_eos_token_id=62,
+ )
+ model = Qwen3TTSForConditionalGeneration(config)
+ output = model.talker(
+ inputs_embeds=torch.randn(2, 5, 8),
+ attention_mask=torch.ones(2, 5, dtype=torch.long),
+ use_cache=False,
+ output_hidden_states=False,
+ )
+
+ assert output.logits.shape == (2, 5, 64)
+
+ import vllm_omni.platforms as platforms
+ from vllm_omni.platforms.interface import UnspecifiedOmniPlatform
+
+ monkeypatch.setattr(platforms, "_current_omni_platform", UnspecifiedOmniPlatform())
+ from verl_omni.pipelines.qwen3_tts.talker_forward import (
+ TalkerTokens,
+ build_talker_batch,
+ codec0_input_embeddings,
+ )
+
+ codes = torch.randint(1, 31, (3, 16), dtype=torch.long)
+ batch = build_talker_batch(
+ [torch.tensor([1, 2, 3])],
+ [codes],
+ TalkerTokens.from_config(model.config),
+ sub_codebook_vocab=32,
+ )
+ embeddings = codec0_input_embeddings(model.talker, batch, torch.zeros(1, 8))
+ assert embeddings.shape == (*batch.input_ids.shape[:2], 8)
+ assert torch.isfinite(embeddings).all()
+
+ from verl_omni.pipelines.qwen3_tts.talker_training_adapter import Qwen3TTSTalkerAdapter
+
+ speaker_path = tmp_path / "speaker.json"
+ speaker_path.write_text(json.dumps([0.0] * 8), encoding="utf-8")
+ configured = Qwen3TTSTalkerAdapter.configure_model(
+ model,
+ SimpleNamespace(
+ use_remove_padding=False,
+ override_config={"tts_spk_embed_path": str(speaker_path), "tts_language": "Auto"},
+ ),
+ )
+ trainable_names = {name for name, parameter in configured.named_parameters() if parameter.requires_grad}
+ assert trainable_names
+ assert all(name.startswith(("talker.model.", "talker.codec_head.")) for name in trainable_names)
+ assert any(not parameter.requires_grad for parameter in configured.parameters())
+ assert configured.get_input_embeddings() is configured.talker.model.codec_embedding
diff --git a/tests/special_e2e/build_qwen3_tts_tiny_random.py b/tests/special_e2e/build_qwen3_tts_tiny_random.py
new file mode 100644
index 000000000..74c3eddd7
--- /dev/null
+++ b/tests/special_e2e/build_qwen3_tts_tiny_random.py
@@ -0,0 +1,124 @@
+#!/usr/bin/env python3
+# Copyright 2026 Bytedance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Build a tiny random 16-codebook Qwen3-TTS checkpoint for GPU smoke tests."""
+
+import argparse
+import json
+import shutil
+from pathlib import Path
+
+import torch
+from safetensors.torch import save_file
+
+
+def _copy_files(source: Path, destination: Path, names: tuple[str, ...]) -> None:
+ for name in names:
+ source_file = source / name
+ if source_file.exists():
+ shutil.copy2(source_file, destination / name)
+
+
+def _save_model(model, output_dir: Path) -> None:
+ state_dict = {name: tensor.detach().cpu().contiguous().clone() for name, tensor in model.state_dict().items()}
+ save_file(state_dict, output_dir / "model.safetensors", metadata={"format": "pt"})
+
+
+def _write_config(source: Path, destination: Path, updates) -> None:
+ config = json.loads(source.read_text(encoding="utf-8"))
+ updates(config)
+ destination.write_text(json.dumps(config, indent=2) + "\n", encoding="utf-8")
+
+
+def _scaled_mrope_section(section: list[int], head_dim: int) -> list[int]:
+ target = head_dim // 2
+ total = sum(section)
+ scaled = [value * target // total for value in section]
+ scaled[-1] += target - sum(scaled)
+ return scaled
+
+
+def _set_model_codebooks(config: dict) -> None:
+ talker_config = config["talker_config"]
+ talker_config["num_code_groups"] = 16
+ talker_config["code_predictor_config"]["num_code_groups"] = 16
+ rope_scaling = talker_config["rope_scaling"]
+ rope_scaling["mrope_section"] = _scaled_mrope_section(rope_scaling["mrope_section"], talker_config["head_dim"])
+
+
+def _set_tokenizer_quantizers(config: dict) -> None:
+ config["dtype"] = "bfloat16"
+ config["encoder_valid_num_quantizers"] = 16
+ config["encoder_config"]["num_quantizers"] = 16
+ config["decoder_config"]["num_quantizers"] = 16
+
+
+def build(source_model_path: Path, output_dir: Path, seed: int = 42) -> Path:
+ from qwen_tts.core.models.configuration_qwen3_tts import Qwen3TTSConfig
+ from qwen_tts.core.models.modeling_qwen3_tts import Qwen3TTSForConditionalGeneration
+ from qwen_tts.core.tokenizer_12hz.configuration_qwen3_tts_tokenizer_v2 import Qwen3TTSTokenizerV2Config
+ from qwen_tts.core.tokenizer_12hz.modeling_qwen3_tts_tokenizer_v2 import Qwen3TTSTokenizerV2Model
+
+ torch.manual_seed(seed)
+ output_dir.mkdir(parents=True, exist_ok=True)
+
+ model_config = Qwen3TTSConfig.from_pretrained(source_model_path)
+ model_config.talker_config.num_code_groups = 16
+ model_config.talker_config.code_predictor_config.num_code_groups = 16
+ rope_scaling = model_config.talker_config.rope_scaling
+ rope_scaling["mrope_section"] = _scaled_mrope_section(
+ rope_scaling["mrope_section"], model_config.talker_config.head_dim
+ )
+ model = Qwen3TTSForConditionalGeneration(model_config).to(torch.bfloat16)
+ _save_model(model, output_dir)
+ _write_config(
+ source_model_path / "config.json",
+ output_dir / "config.json",
+ _set_model_codebooks,
+ )
+ _copy_files(
+ source_model_path,
+ output_dir,
+ ("generation_config.json", "merges.txt", "preprocessor_config.json", "tokenizer_config.json", "vocab.json"),
+ )
+
+ tokenizer_source = source_model_path / "speech_tokenizer"
+ tokenizer_output = output_dir / "speech_tokenizer"
+ tokenizer_output.mkdir(exist_ok=True)
+ tokenizer_config = Qwen3TTSTokenizerV2Config.from_pretrained(tokenizer_source)
+ tokenizer_config.encoder_valid_num_quantizers = 16
+ tokenizer_config.encoder_config.num_quantizers = 16
+ tokenizer_config.decoder_config.num_quantizers = 16
+ tokenizer = Qwen3TTSTokenizerV2Model(tokenizer_config).to(torch.bfloat16)
+ _save_model(tokenizer, tokenizer_output)
+ _write_config(
+ tokenizer_source / "config.json",
+ tokenizer_output / "config.json",
+ _set_tokenizer_quantizers,
+ )
+ _copy_files(tokenizer_source, tokenizer_output, ("configuration.json", "preprocessor_config.json"))
+ return output_dir
+
+
+def main() -> None:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--source-model-path", type=Path, required=True)
+ parser.add_argument("--output-dir", type=Path, required=True)
+ parser.add_argument("--seed", type=int, default=42)
+ args = parser.parse_args()
+ print(build(args.source_model_path, args.output_dir, args.seed))
+
+
+if __name__ == "__main__":
+ main()
diff --git a/tests/special_e2e/create_dummy_qwen3_tts_grpo_data.py b/tests/special_e2e/create_dummy_qwen3_tts_grpo_data.py
new file mode 100644
index 000000000..91d1a8a07
--- /dev/null
+++ b/tests/special_e2e/create_dummy_qwen3_tts_grpo_data.py
@@ -0,0 +1,82 @@
+#!/usr/bin/env python3
+# Copyright 2026 Bytedance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Create deterministic Qwen3-TTS GRPO smoke data and speaker conditioning."""
+
+import argparse
+import json
+import math
+from pathlib import Path
+
+import pandas as pd
+
+
+def _speaker_dimension(model_config_path: Path) -> int:
+ model_config = json.loads(model_config_path.read_text(encoding="utf-8"))
+ talker_dimension = model_config["talker_config"]["hidden_size"]
+ speaker_dimension = model_config["speaker_encoder_config"]["enc_dim"]
+ if talker_dimension != speaker_dimension:
+ raise ValueError(
+ f"speaker encoder output must match the talker hidden size: {speaker_dimension} != {talker_dimension}"
+ )
+ if not isinstance(speaker_dimension, int) or speaker_dimension <= 0:
+ raise ValueError(f"speaker dimension must be a positive integer, got {speaker_dimension!r}")
+ return speaker_dimension
+
+
+def _row(text: str, sample_id: str, split: str) -> dict:
+ extra_info = {"id": sample_id, "split": split}
+ return {
+ "data_source": "tts",
+ "prompt": [{"role": "user", "content": text}],
+ "reward_model": {"style": "model", "ground_truth": text},
+ "extra_info": extra_info,
+ }
+
+
+def main() -> None:
+ parser = argparse.ArgumentParser()
+ parser.add_argument("--output-dir", type=Path, required=True)
+ parser.add_argument("--model-config", type=Path, required=True)
+ args = parser.parse_args()
+ args.output_dir.mkdir(parents=True, exist_ok=True)
+
+ train_texts = (
+ "Please read this sentence at a calm and steady pace.",
+ "A short speech sample checks the complete training path.",
+ "Clear pronunciation makes this audio easy to inspect.",
+ "The weather is pleasant and the morning train is on time.",
+ "Four simple prompts are enough for one smoke-test batch.",
+ "This second batch verifies another optimizer update.",
+ "Generated speech is decoded before the reward is computed.",
+ "The final checkpoint confirms that training completed.",
+ )
+ validation_texts = (
+ "This is fixed validation sample one.",
+ "This is fixed validation sample two.",
+ "This is fixed validation sample three.",
+ "This is fixed validation sample four.",
+ )
+ train_rows = [_row(text, f"train-{index}", "train") for index, text in enumerate(train_texts)]
+ validation_rows = [_row(text, f"validation-{index}", "validation") for index, text in enumerate(validation_texts)]
+ pd.DataFrame(train_rows).to_parquet(args.output_dir / "train.parquet", index=False)
+ pd.DataFrame(validation_rows).to_parquet(args.output_dir / "validation.parquet", index=False)
+
+ speaker_dimension = _speaker_dimension(args.model_config)
+ speaker = [1.0 / math.sqrt(speaker_dimension)] * speaker_dimension
+ (args.output_dir / "speaker.json").write_text(json.dumps(speaker), encoding="utf-8")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/tests/special_e2e/qwen3_tts_dummy_reward.py b/tests/special_e2e/qwen3_tts_dummy_reward.py
new file mode 100644
index 000000000..3cda6b744
--- /dev/null
+++ b/tests/special_e2e/qwen3_tts_dummy_reward.py
@@ -0,0 +1,23 @@
+# Copyright 2026 Bytedance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""CPU-only reward used by the Qwen3-TTS execution smoke test."""
+
+import numpy as np
+
+
+def compute_score(solution_audio, **kwargs):
+ del kwargs
+ waveform, sample_rate = solution_audio
+ duration_s = np.asarray(waveform).size / sample_rate
+ return {"score": float(duration_s), "duration_s": float(duration_s)}
diff --git a/tests/special_e2e/run_qwen3_tts_grpo_smoke.sh b/tests/special_e2e/run_qwen3_tts_grpo_smoke.sh
new file mode 100755
index 000000000..80fa691b0
--- /dev/null
+++ b/tests/special_e2e/run_qwen3_tts_grpo_smoke.sh
@@ -0,0 +1,80 @@
+#!/usr/bin/env bash
+# Qwen3-TTS full-parameter GRPO e2e smoke: tiny model, two updates.
+# This validates execution only; the in-process CPU reward is not a quality reward.
+
+set -xeuo pipefail
+
+export NCCL_IB_DISABLE=1
+export CPATH=/usr/include${CPATH:+:${CPATH}}
+export RAY_ACCEL_ENV_VAR_OVERRIDE_ON_ZERO=0
+
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)"
+cd "${REPO_ROOT}"
+
+NUM_GPUS="${NUM_GPUS:-2}"
+[[ "${NUM_GPUS}" =~ ^[0-9]+$ && "${NUM_GPUS}" -ge 2 ]] || {
+ echo "Qwen3-TTS smoke requires at least two GPUs" >&2
+ exit 2
+}
+PYTHON_BIN="${PYTHON_BIN:-python3}"
+MODEL_REPO="${MODEL_REPO:-optimum-intel-internal-testing/tiny-random-qwen3-tts}"
+MODEL_REVISION="${MODEL_REVISION:-6374d605b31381cac6f9577f5e742af2f76ba79c}"
+WORK_DIR="${WORK_DIR:-${TMPDIR:-/tmp}/qwen3_tts_grpo_smoke_${USER:-user}_$$}"
+DATA_DIR="${WORK_DIR}/data"
+OUTPUT_DIR="${OUTPUT_DIR:-${WORK_DIR}/output}"
+mkdir -p "${WORK_DIR}" "${OUTPUT_DIR}"
+
+"${PYTHON_BIN}" -c \
+ 'from qwen_tts.core.models.modeling_qwen3_tts import Qwen3TTSForConditionalGeneration; import onnxruntime, soundfile, librosa, sox' \
+ || { echo "Qwen3-TTS smoke dependencies must be installed by gpu-smoke-prepare" >&2; exit 2; }
+
+MODEL_PATH="${MODEL_PATH:-}"
+if [[ -z "${MODEL_PATH}" ]]; then
+ SOURCE_MODEL_PATH="$("${PYTHON_BIN}" -c \
+ 'import sys; from huggingface_hub import snapshot_download; print(snapshot_download(sys.argv[1], revision=sys.argv[2]))' \
+ "${MODEL_REPO}" "${MODEL_REVISION}")"
+ # The published tiny checkpoint has four codebooks; the production actor
+ # contract has 16, so rebuild its small architecture with random 16-codebook weights.
+ MODEL_PATH="${WORK_DIR}/tiny-random-qwen3-tts-16-codebooks"
+ "${PYTHON_BIN}" tests/special_e2e/build_qwen3_tts_tiny_random.py \
+ --source-model-path "${SOURCE_MODEL_PATH}" \
+ --output-dir "${MODEL_PATH}"
+fi
+[[ -f "${MODEL_PATH}/config.json" ]] || { echo "Invalid MODEL_PATH: ${MODEL_PATH}" >&2; exit 2; }
+
+"${PYTHON_BIN}" tests/special_e2e/create_dummy_qwen3_tts_grpo_data.py \
+ --output-dir "${DATA_DIR}" \
+ --model-config "${MODEL_PATH}/config.json"
+
+MODEL_PATH="${MODEL_PATH}" \
+TRAIN_FILE="${DATA_DIR}/train.parquet" \
+VAL_FILE="${DATA_DIR}/validation.parquet" \
+SPK_EMBED_PATH="${DATA_DIR}/speaker.json" \
+SCORER_URL="http://unused.invalid/score" \
+OUTPUT_DIR="${OUTPUT_DIR}" \
+NUM_GPUS="${NUM_GPUS}" \
+TOTAL_TRAINING_STEPS=2 \
+TEST_FREQ=-1 \
+SAVE_FREQ=-1 \
+RESUME_MODE=disable \
+PYTHON_BIN="${PYTHON_BIN}" \
+bash examples/grpo_trainer/qwen3_tts/run_qwen3_tts_grpo.sh \
+ actor_rollout_ref.rollout.n=2 \
+ actor_rollout_ref.rollout.agent.num_workers=2 \
+ actor_rollout_ref.rollout.max_num_seqs=4 \
+ "reward.custom_reward_function.path=${REPO_ROOT}/tests/special_e2e/qwen3_tts_dummy_reward.py" \
+ reward.custom_reward_function.name=compute_score \
+ trainer.val_before_train=false \
+ trainer.log_val_generations=0 \
+ "$@"
+
+"${PYTHON_BIN}" - "${OUTPUT_DIR}/train.log" <<'PY'
+import re
+import sys
+from pathlib import Path
+
+steps = [int(step) for step in re.findall(r"training/global_step:(\d+)(?!\d)", Path(sys.argv[1]).read_text())]
+if not steps or max(steps) != 2:
+ raise SystemExit(f"Expected training/global_step to reach exactly 2, observed {steps!r}")
+PY
+echo "Qwen3-TTS GRPO e2e smoke passed; artifacts: ${WORK_DIR}"
diff --git a/tests/special_sanity/test_gpu_smoke_selector_on_cpu.py b/tests/special_sanity/test_gpu_smoke_selector_on_cpu.py
new file mode 100644
index 000000000..cce4a4a77
--- /dev/null
+++ b/tests/special_sanity/test_gpu_smoke_selector_on_cpu.py
@@ -0,0 +1,44 @@
+# Copyright 2026 Bytedance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import importlib.util
+import sys
+from pathlib import Path
+
+
+def _load_selector():
+ path = Path(__file__).parents[1] / "gpu_smoke/select_gpu_smoke_groups.py"
+ spec = importlib.util.spec_from_file_location("gpu_smoke_selector_under_test", path)
+ module = importlib.util.module_from_spec(spec)
+ assert spec.loader is not None
+ sys.modules[spec.name] = module
+ spec.loader.exec_module(module)
+ return module
+
+
+def test_qwen3_tts_smoke_files_select_only_omni_e2e_group():
+ selector = _load_selector()
+
+ selected = selector.select_group_names(
+ [
+ ".github/qwen_tts_pin.txt",
+ "examples/grpo_trainer/qwen3_tts/run_qwen3_tts_grpo.sh",
+ "tests/special_e2e/build_qwen3_tts_tiny_random.py",
+ "tests/special_e2e/run_qwen3_tts_grpo_smoke.sh",
+ "tests/special_e2e/create_dummy_qwen3_tts_grpo_data.py",
+ "tests/special_e2e/qwen3_tts_dummy_reward.py",
+ ]
+ )
+
+ assert selected == ["ci-e2e-omni"]
diff --git a/tests/workers/rollout/rollout_vllm/test_qwen3_tts_rollout_on_cpu.py b/tests/workers/rollout/rollout_vllm/test_qwen3_tts_rollout_on_cpu.py
new file mode 100644
index 000000000..b412bffba
--- /dev/null
+++ b/tests/workers/rollout/rollout_vllm/test_qwen3_tts_rollout_on_cpu.py
@@ -0,0 +1,454 @@
+# Copyright 2026 Bytedance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""CPU contracts for Qwen3-TTS's multi-stage rollout integration."""
+
+import subprocess
+import sys
+from types import SimpleNamespace
+
+import pytest
+import torch
+from tensordict import TensorDict
+
+pytest.importorskip("verl")
+pytest.importorskip("vllm_omni")
+
+from verl.experimental.agent_loop.agent_loop import AgentLoopMetrics, AgentLoopOutput
+from verl.utils.tensordict_utils import list_of_dict_to_tensordict
+from vllm import SamplingParams
+
+from verl_omni.agent_loop.single_turn_agent_loop import OmniSingleTurnAgentLoop
+from verl_omni.pipelines.model_base import OmniRolloutPipelineBase
+from verl_omni.pipelines.qwen3_tts import omni_rollout_adapter
+from verl_omni.pipelines.qwen3_tts.omni_rollout_adapter import (
+ Qwen3TTSRolloutAdapter,
+ prepare_code2wav_input_for_policy_replay,
+)
+from verl_omni.pipelines.qwen3_tts.rollout_utils import QWEN3_TTS_REPLAY_KEY
+from verl_omni.pipelines.qwen3_tts.talker_training_adapter import Qwen3TTSTalkerAdapter
+from verl_omni.workers.rollout.vllm_rollout.vllm_omni_ar_strategy import ARStrategy
+from verl_omni.workers.rollout.vllm_rollout.vllm_omni_async_server import vLLMOmniHttpServer
+
+
+class _Tokenizer:
+ def decode(self, token_ids, **kwargs):
+ return "first text" if token_ids == [1] else "other text"
+
+ def __call__(self, text, **kwargs):
+ return {"input_ids": list(range(len(text)))}
+
+
+def test_external_module_import_registers_omni_agent_loop():
+ code = """
+import vllm.utils.import_utils as import_utils
+
+NoGPU = type("NoGPU", (), {
+ "nvmlInit": staticmethod(lambda: None),
+ "nvmlDeviceGetCount": staticmethod(lambda: 0),
+ "nvmlShutdown": staticmethod(lambda: None),
+})
+import_utils.import_pynvml = lambda: NoGPU
+
+import vllm_omni.platforms as platforms
+from vllm_omni.platforms.interface import UnspecifiedOmniPlatform
+
+platforms._current_omni_platform = UnspecifiedOmniPlatform()
+
+import verl_omni
+from verl.experimental.agent_loop.agent_loop import _agent_loop_registry
+
+target = _agent_loop_registry[\"omni_single_turn_agent\"][\"_target_\"]
+assert target == \"verl_omni.agent_loop.single_turn_agent_loop.OmniSingleTurnAgentLoop\"
+"""
+ subprocess.run([sys.executable, "-c", code], check=True)
+
+
+def test_optional_rollout_hooks_preserve_existing_ar_defaults():
+ first, final = object(), object()
+
+ assert OmniRolloutPipelineBase.supports_async_chunk is True
+ assert OmniRolloutPipelineBase.weight_sync_stage_ids() is None
+ assert OmniRolloutPipelineBase.prepare_engine_prompt([], None, {}) is None
+ assert (
+ OmniRolloutPipelineBase.postprocess_agent_loop_output(
+ final,
+ tokenizer=None,
+ response_length=8,
+ )
+ is final
+ )
+ assert OmniRolloutPipelineBase.combine_engine_outputs([final], {}) == (final, {})
+ with pytest.raises(NotImplementedError, match="multiple final outputs"):
+ OmniRolloutPipelineBase.combine_engine_outputs([first, final], {})
+ with pytest.raises(RuntimeError, match="no outputs"):
+ OmniRolloutPipelineBase.combine_engine_outputs([], {})
+
+
+def test_omni_single_turn_agent_resolves_registered_pipeline_adapter():
+ rollout_config = SimpleNamespace(engine_kwargs={"vllm_omni": {"pipeline_name": "qwen3_tts_rl"}})
+
+ assert OmniSingleTurnAgentLoop._resolve_rollout_adapter(rollout_config) is Qwen3TTSRolloutAdapter
+
+ missing_config = SimpleNamespace(engine_kwargs={"vllm_omni": {"pipeline_name": "missing"}})
+ with pytest.raises(ValueError, match="requires a registered"):
+ OmniSingleTurnAgentLoop._resolve_rollout_adapter(missing_config)
+
+
+def test_rollout_pipeline_registers_upstream_talker(monkeypatch):
+ registered_pipelines = []
+ monkeypatch.setattr(
+ omni_rollout_adapter,
+ "register_pipeline",
+ lambda pipeline: registered_pipelines.append(pipeline),
+ )
+
+ Qwen3TTSRolloutAdapter.ensure_pipeline_registered()
+
+ assert registered_pipelines == [omni_rollout_adapter.QWEN3_TTS_RL_PIPELINE]
+ assert registered_pipelines[0].model_arch == omni_rollout_adapter.QWEN3_TTS_PIPELINE.model_arch
+
+
+def test_rollout_adapter_builds_unique_prompt_and_scopes_weight_sync(tmp_path):
+ speaker = tmp_path / "speaker.json"
+ speaker.write_text("[0.0, 1.0]")
+ model_config = SimpleNamespace(
+ tokenizer=_Tokenizer(),
+ override_config={"tts_spk_embed_path": str(speaker), "tts_language": "Auto"},
+ hf_config=SimpleNamespace(talker_config=SimpleNamespace(codec_eos_token_id=2150)),
+ )
+
+ first = Qwen3TTSRolloutAdapter.prepare_engine_prompt([1], model_config, {})
+ second = Qwen3TTSRolloutAdapter.prepare_engine_prompt([2], model_config, {})
+
+ assert first["additional_information"]["text"] == ["first text"]
+ assert first["cache_salt"] != second["cache_salt"]
+ assert Qwen3TTSRolloutAdapter.weight_sync_stage_ids("full") == [0]
+ stages = Qwen3TTSRolloutAdapter.build_stage_configs("full")
+ assert [stage.final_output_type for stage in stages if stage.final_output] == ["latent", "audio"]
+ assert stages[0].sampling_constraints["min_tokens"] == 2
+ assert stages[1].sync_process_input_func.endswith(".prepare_code2wav_input_for_policy_replay")
+
+
+def test_code2wav_placeholder_uses_retained_policy_token_count_without_mutation():
+ completion = SimpleNamespace(
+ cumulative_token_ids=[101, 201, 202, 2150],
+ multimodal_output=None,
+ )
+ source_output = SimpleNamespace(finished=True, outputs=[completion])
+
+ prepared = prepare_code2wav_input_for_policy_replay([source_output])
+
+ assert len(prepared) == 1
+ assert len(prepared[0]["prompt_token_ids"]) == 3 * 16
+ assert completion.multimodal_output is None
+
+
+def test_ar_strategy_resolves_qwen3_tts_adapter(monkeypatch):
+ server = SimpleNamespace(_rollout_flags={})
+ strategy = ARStrategy(server)
+ deploy_calls = []
+ monkeypatch.setattr(
+ strategy,
+ "_write_deploy_config",
+ lambda engine_kwargs, pipeline_name, adapter_cls, pipeline_mode: deploy_calls.append(
+ (pipeline_name, adapter_cls, pipeline_mode)
+ ),
+ )
+ engine_kwargs = {
+ "output_mode": "ar",
+ "pipeline_name": "qwen3_tts_rl",
+ "pipeline_mode": "full",
+ "async_chunk": False,
+ }
+
+ strategy.preprocess_engine_kwargs(engine_kwargs)
+
+ assert deploy_calls == [("qwen3_tts_rl", Qwen3TTSRolloutAdapter, "full")]
+ assert strategy._rollout_adapter is Qwen3TTSRolloutAdapter
+ assert engine_kwargs == {"async-chunk": False}
+
+
+@pytest.mark.parametrize("async_chunk", [None, True])
+def test_ar_strategy_requires_non_chunked_qwen3_tts_rollout(async_chunk):
+ strategy = ARStrategy(SimpleNamespace(_rollout_flags={}))
+ engine_kwargs = {
+ "pipeline_name": "qwen3_tts_rl",
+ "pipeline_mode": "full",
+ }
+ if async_chunk is not None:
+ engine_kwargs["async_chunk"] = async_chunk
+
+ with pytest.raises(ValueError, match="requires async_chunk=false"):
+ strategy.preprocess_engine_kwargs(engine_kwargs)
+
+
+def test_rollout_adapter_requires_speaker_embedding():
+ model_config = SimpleNamespace(
+ tokenizer=_Tokenizer(),
+ override_config={"tts_language": "Auto"},
+ hf_config=SimpleNamespace(talker_config=SimpleNamespace(codec_eos_token_id=2150)),
+ )
+
+ with pytest.raises(ValueError, match="requires tts_spk_embed_path"):
+ Qwen3TTSRolloutAdapter.prepare_engine_prompt([1], model_config, {})
+
+
+def test_qwen3_tts_adapters_require_explicit_language(tmp_path):
+ speaker = tmp_path / "speaker.json"
+ speaker.write_text("[0.0, 1.0]")
+ model_config = SimpleNamespace(
+ tokenizer=_Tokenizer(),
+ override_config={"tts_spk_embed_path": str(speaker)},
+ hf_config=SimpleNamespace(talker_config=SimpleNamespace(codec_eos_token_id=2150)),
+ )
+
+ with pytest.raises(ValueError, match="supports only tts_language=Auto"):
+ Qwen3TTSRolloutAdapter.prepare_engine_prompt([1], model_config, {})
+ with pytest.raises(ValueError, match="supports only tts_language=Auto"):
+ Qwen3TTSTalkerAdapter.configure_model(SimpleNamespace(config=SimpleNamespace()), model_config)
+
+
+def test_talker_adapter_rejects_remove_padding_before_model_configuration():
+ model_config = SimpleNamespace(use_remove_padding=True)
+
+ with pytest.raises(ValueError, match="use_remove_padding=false"):
+ Qwen3TTSTalkerAdapter.configure_model(SimpleNamespace(), model_config)
+
+
+def test_talker_adapter_pads_exact_rollout_fields_for_actor_forward():
+ model_inputs = {"input_ids": torch.zeros(2, 6, dtype=torch.long)}
+ payloads = [
+ {"text_ids": [1, 2, 6], "audio_codes": torch.ones(3, 16, dtype=torch.long)},
+ {"text_ids": [3, 4, 5], "audio_codes": torch.full((2, 16), 2, dtype=torch.long)},
+ ]
+ micro_batch = list_of_dict_to_tensordict(
+ [
+ AgentLoopOutput(
+ prompt_ids=[1],
+ response_ids=[2],
+ response_mask=[1],
+ metrics=AgentLoopMetrics(),
+ extra_fields={QWEN3_TTS_REPLAY_KEY: item},
+ ).as_dict()
+ for item in payloads
+ ]
+ )
+
+ prepared = Qwen3TTSTalkerAdapter.prepare_model_inputs(model_inputs, micro_batch, None)
+
+ assert prepared["tts_text_ids"].shape == (2, 3)
+ assert prepared["tts_audio_codes"].shape == (2, 3, 16)
+ assert prepared["text_len"].tolist() == [3, 3]
+ assert prepared["response_len"].tolist() == [3, 2]
+ assert not prepared["tts_audio_codes"][1, 2].any()
+
+
+def test_talker_adapter_requires_namespaced_replay_payload():
+ model_inputs = {"input_ids": torch.zeros(1, 4, dtype=torch.long)}
+ micro_batch = TensorDict({}, batch_size=[1])
+
+ with pytest.raises(RuntimeError, match=QWEN3_TTS_REPLAY_KEY):
+ Qwen3TTSTalkerAdapter.prepare_model_inputs(model_inputs, micro_batch, None)
+
+
+def test_rollout_adapter_combines_policy_codes_and_waveform():
+ token_ids = [101, 102, 2150]
+ generated = torch.arange(3 * 16, dtype=torch.long).reshape(3, 16) + 1
+ generated[:, 0] = torch.tensor(token_ids)
+ policy = SimpleNamespace(
+ stage_id=0,
+ outputs=[SimpleNamespace(token_ids=token_ids)],
+ multimodal_output={"codes": {"audio": torch.cat((torch.zeros(12, 16), generated))}},
+ )
+ decoder = SimpleNamespace(
+ stage_id=1,
+ outputs=[],
+ multimodal_output={"audio": torch.ones(2400), "sr": 24_000},
+ )
+ prompt = {"additional_information": {"text": ["first text"]}}
+
+ selected, fields = Qwen3TTSRolloutAdapter.combine_engine_outputs([policy, decoder], prompt)
+
+ assert selected is policy
+ torch.testing.assert_close(fields["tts_audio_codes"], generated.long())
+ torch.testing.assert_close(fields["audio"], torch.ones(2400))
+ assert fields["audio_sample_rate"] == 24_000
+ assert fields["tts_text"] == "first text"
+
+
+def test_rollout_adapter_rejects_non_mono_waveform():
+ token_ids = [101, 102, 2150]
+ generated = torch.arange(3 * 16, dtype=torch.long).reshape(3, 16) + 1
+ generated[:, 0] = torch.tensor(token_ids)
+ policy = SimpleNamespace(
+ stage_id=0,
+ outputs=[SimpleNamespace(token_ids=token_ids)],
+ multimodal_output={"codes": {"audio": torch.cat((torch.zeros(12, 16), generated))}},
+ )
+ decoder = SimpleNamespace(
+ stage_id=1,
+ outputs=[],
+ multimodal_output={"audio": torch.ones(2, 2400), "sr": 24_000},
+ )
+ prompt = {"additional_information": {"text": ["first text"]}}
+
+ with pytest.raises(RuntimeError, match="one-dimensional mono waveform"):
+ Qwen3TTSRolloutAdapter.combine_engine_outputs([policy, decoder], prompt)
+
+
+def test_rollout_adapter_prepares_actor_policy_sequence():
+ codes = torch.arange(5 * 16, dtype=torch.long).reshape(5, 16)
+ output = SimpleNamespace(
+ prompt_ids=[9, 8],
+ response_ids=[7, 6, 5, 4, 3],
+ response_mask=[1] * 5,
+ response_logprobs=[-0.1, -0.2, -0.3, -0.4, -0.5],
+ extra_fields={
+ "tts_audio_codes": codes,
+ "tts_text": "first text",
+ "audio": torch.ones(2400),
+ "audio_sample_rate": 24_000,
+ },
+ )
+
+ result = Qwen3TTSRolloutAdapter.postprocess_agent_loop_output(
+ output,
+ tokenizer=_Tokenizer(),
+ response_length=3,
+ )
+
+ assert result is output
+ assert result.prompt_ids == [0]
+ assert result.response_ids == codes[:3, 0].tolist()
+ assert result.response_mask == [1, 1, 1]
+ assert result.response_logprobs == [-0.1, -0.2, -0.3]
+ assert "tts_audio_codes" not in result.extra_fields
+ assert "tts_text" not in result.extra_fields
+ replay = result.extra_fields[QWEN3_TTS_REPLAY_KEY]
+ torch.testing.assert_close(replay["audio_codes"], codes[:3])
+ assert replay["text_ids"]
+ assert result.extra_fields["audio_sample_rate"] == 24_000
+
+
+def test_ar_strategy_prepares_stage_specific_sampling_params():
+ class Adapter:
+ @staticmethod
+ def prepare_engine_prompt(**kwargs):
+ return {
+ "prompt_token_ids": [1, 1, 1, 1],
+ "additional_information": {"text": ["hello"]},
+ }
+
+ server = SimpleNamespace(
+ model_config=SimpleNamespace(),
+ config=SimpleNamespace(
+ max_model_len=64,
+ prompt_length=16,
+ response_length=8,
+ repetition_penalty=1.0,
+ ),
+ engine=SimpleNamespace(default_sampling_params_list=[SamplingParams(), SimpleNamespace(stage="decoder")]),
+ )
+ strategy = ARStrategy(server)
+ strategy._rollout_adapter = Adapter
+ strategy._rollout_output_modalities = ["latent", "audio"]
+ strategy._policy_stage_index = 0
+ strategy._policy_sampling_constraints = {}
+
+ prompt, params = strategy.preprocess_input(
+ [5, 6],
+ {"temperature": 0.8, "logprobs": True},
+ {},
+ None,
+ None,
+ )
+
+ assert prompt["additional_information"]["max_new_tokens"] == [8]
+ assert len(params) == 2
+ assert params[0].max_tokens == 8
+ assert params[0].temperature == pytest.approx(0.8)
+ assert params[0].logprobs == 0
+ assert params[1].stage == "decoder"
+
+
+@pytest.mark.parametrize(
+ ("adapter_prompt", "message"),
+ [
+ ({"additional_information": {"text": ["hello"]}}, "must contain prompt_token_ids"),
+ ([1, 2], "must return a dict or None"),
+ ],
+)
+def test_ar_strategy_rejects_invalid_adapter_prompt(adapter_prompt, message):
+ class Adapter:
+ @staticmethod
+ def prepare_engine_prompt(**kwargs):
+ return adapter_prompt
+
+ server = SimpleNamespace(
+ model_config=SimpleNamespace(),
+ config=SimpleNamespace(max_model_len=64, prompt_length=16, response_length=8),
+ )
+ strategy = ARStrategy(server)
+ strategy._rollout_adapter = Adapter
+
+ with pytest.raises((RuntimeError, TypeError), match=message):
+ strategy.preprocess_input([5, 6], {}, {}, None, None)
+
+
+@pytest.mark.asyncio
+async def test_ar_strategy_retains_requested_stage_outputs_and_targets_weight_sync():
+ completion = SimpleNamespace(token_ids=[7], logprobs=None, finish_reason="stop", num_preempted=0)
+ policy = SimpleNamespace(request_id="request-0", outputs=[completion])
+
+ class Engine:
+ def __init__(self):
+ self.generate_kwargs = None
+ self.rpc_kwargs = None
+
+ async def generate(self, **kwargs):
+ self.generate_kwargs = kwargs
+ yield policy
+
+ async def collective_rpc(self, **kwargs):
+ self.rpc_kwargs = kwargs
+ return "rpc-result"
+
+ class Adapter:
+ @staticmethod
+ def combine_engine_outputs(outputs, prompt):
+ assert outputs == [policy]
+ return policy, {"audio_sample_rate": 24_000}
+
+ server = object.__new__(vLLMOmniHttpServer)
+ server.engine = Engine()
+ server.global_steps = 3
+ strategy = ARStrategy(server)
+ strategy._rollout_output_modalities = ["latent", "audio"]
+ strategy._rollout_adapter = Adapter
+ strategy._weight_sync_stage_ids = [0]
+ server._generate_strategy = strategy
+
+ result = await strategy.run_generation({"prompt_token_ids": [1]}, SamplingParams(), "request-0", None, 0)
+ rpc_result = await server.collective_rpc("update_weights_from_ipc", kwargs={"base_sync_done": True})
+
+ assert result is policy
+ assert not hasattr(result, "_verl_omni_rollout_fields")
+ assert strategy._rollout_fields_by_request_id == {"request-0": {"audio_sample_rate": 24_000}}
+ assert server.engine.generate_kwargs["output_modalities"] == ["latent", "audio"]
+ assert server.engine.rpc_kwargs["stage_ids"] == [0]
+ assert rpc_result is None
+
+ output = strategy.process_output(result, SamplingParams(), {})
+ assert output.extra_fields == {"global_steps": 3, "audio_sample_rate": 24_000}
+ assert strategy._rollout_fields_by_request_id == {}
diff --git a/verl_omni/pipelines/__init__.py b/verl_omni/pipelines/__init__.py
index a1acf099d..ab3c9ca10 100644
--- a/verl_omni/pipelines/__init__.py
+++ b/verl_omni/pipelines/__init__.py
@@ -20,6 +20,7 @@
minimax_h3_diffusion_nft,
minimax_h3_flow_grpo,
qwen3_omni,
+ qwen3_tts,
qwen_image_diffusion_nft,
qwen_image_dpo,
qwen_image_dual_grpo,
@@ -37,6 +38,7 @@
from .minimax_h3_diffusion_nft import * # noqa: F401, F403
from .minimax_h3_flow_grpo import * # noqa: F401, F403
from .qwen3_omni import * # noqa: F401, F403
+from .qwen3_tts import * # noqa: F401, F403
from .qwen_image_diffusion_nft import * # noqa: F401, F403
from .qwen_image_dpo import * # noqa: F401, F403
from .qwen_image_dual_grpo import * # noqa: F401, F403
@@ -48,6 +50,7 @@
from .wan22_dance_grpo import * # noqa: F401, F403
__all__ = list(qwen3_omni.__all__)
+__all__ += list(qwen3_tts.__all__)
__all__ += list(qwen_image_flow_grpo.__all__)
__all__ += list(qwen_image_diffusion_nft.__all__)
__all__ += list(qwen_image_mix_grpo.__all__)
diff --git a/verl_omni/pipelines/qwen3_tts/__init__.py b/verl_omni/pipelines/qwen3_tts/__init__.py
new file mode 100644
index 000000000..5831f3727
--- /dev/null
+++ b/verl_omni/pipelines/qwen3_tts/__init__.py
@@ -0,0 +1,18 @@
+# Copyright 2026 Bytedance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+from .omni_rollout_adapter import Qwen3TTSRolloutAdapter
+from .talker_training_adapter import Qwen3TTSTalkerAdapter
+
+__all__ = ["Qwen3TTSTalkerAdapter", "Qwen3TTSRolloutAdapter"]
diff --git a/verl_omni/pipelines/qwen3_tts/omni_rollout_adapter.py b/verl_omni/pipelines/qwen3_tts/omni_rollout_adapter.py
new file mode 100644
index 000000000..266e0338e
--- /dev/null
+++ b/verl_omni/pipelines/qwen3_tts/omni_rollout_adapter.py
@@ -0,0 +1,264 @@
+# Copyright 2026 Bytedance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Qwen3-TTS two-stage rollout adapter."""
+
+import hashlib
+from dataclasses import replace
+from functools import lru_cache
+from types import SimpleNamespace
+
+import torch
+from vllm_omni.config.pipeline_registry import register_pipeline
+from vllm_omni.config.stage_config import PipelineConfig
+from vllm_omni.model_executor.models.qwen3_tts.pipeline import QWEN3_TTS_PIPELINE
+
+from verl_omni.pipelines.model_base import OmniRolloutPipelineBase
+from verl_omni.pipelines.qwen3_tts.rollout_utils import QWEN3_TTS_REPLAY_KEY, align_audio_codes
+from verl_omni.pipelines.qwen3_tts.talker_forward import (
+ NUM_CODEBOOKS,
+ TEXT_PROMPT_TRAILER_TOKENS,
+ build_assistant_text,
+ load_speaker_xvector,
+ require_auto_language,
+)
+
+_PIPELINE_ID = "qwen3_tts_rl"
+
+
+def prepare_code2wav_input_for_policy_replay(
+ source_outputs: list,
+ prompt=None,
+ _requires_multimodal_data: bool = False,
+) -> list:
+ """Size Code2Wav placeholders from the retained Talker policy trajectory.
+
+ The pinned sync engine keeps codec values on the worker connector while
+ the orchestrator sees only cumulative policy tokens. Code2Wav still needs
+ a placeholder sized to 16 codebooks for every generated codec frame.
+ """
+ from vllm_omni.model_executor.stage_input_processors.qwen3_tts import talker2code2wav_token_only
+
+ normalized_outputs = []
+ for source_output in source_outputs:
+ completions = getattr(source_output, "outputs", None)
+ if not getattr(source_output, "finished", False):
+ normalized_outputs.append(source_output)
+ continue
+ if not completions:
+ raise RuntimeError("Qwen3-TTS Talker finished without a completion.")
+
+ completion = completions[0]
+ token_ids = list(getattr(completion, "cumulative_token_ids", None) or [])
+ if len(token_ids) < 2:
+ raise RuntimeError("Qwen3-TTS Talker produced no codec frames for Code2Wav.")
+ frame_count = len(token_ids) - 1
+ # This tensor supplies only the placeholder shape. The worker
+ # connector sends the actual 16-codebook values to Code2Wav.
+ multimodal_output = {
+ "codes": {"audio": torch.ones((frame_count, NUM_CODEBOOKS), dtype=torch.long)},
+ }
+
+ completion_proxy = SimpleNamespace(
+ cumulative_token_ids=token_ids,
+ multimodal_output=multimodal_output,
+ )
+ normalized_outputs.append(
+ SimpleNamespace(
+ finished=source_output.finished,
+ outputs=[completion_proxy],
+ )
+ )
+
+ return talker2code2wav_token_only(
+ normalized_outputs,
+ prompt=prompt,
+ _requires_multimodal_data=_requires_multimodal_data,
+ )
+
+
+QWEN3_TTS_RL_PIPELINE = PipelineConfig(
+ model_type=_PIPELINE_ID,
+ model_arch=QWEN3_TTS_PIPELINE.model_arch,
+ stages=(
+ replace(
+ QWEN3_TTS_PIPELINE.stages[0],
+ final_output=True,
+ final_output_type="latent",
+ sampling_constraints={
+ **QWEN3_TTS_PIPELINE.stages[0].sampling_constraints,
+ "min_tokens": 2,
+ },
+ ),
+ replace(
+ QWEN3_TTS_PIPELINE.stages[1],
+ sync_process_input_func=f"{__name__}.prepare_code2wav_input_for_policy_replay",
+ ),
+ ),
+)
+
+
+@lru_cache(maxsize=4)
+def _load_speaker_vector(path: str) -> list[float]:
+ return load_speaker_xvector(path).reshape(-1).tolist()
+
+
+@OmniRolloutPipelineBase.register(_PIPELINE_ID)
+class Qwen3TTSRolloutAdapter(OmniRolloutPipelineBase):
+ supports_async_chunk = False
+
+ @classmethod
+ def _check_mode(cls, pipeline_mode):
+ if pipeline_mode != "full":
+ raise ValueError("Qwen3-TTS RL supports only pipeline_mode='full'.")
+
+ @classmethod
+ def build_stage_configs(cls, pipeline_mode="full"):
+ cls._check_mode(pipeline_mode)
+ return list(QWEN3_TTS_RL_PIPELINE.stages)
+
+ @classmethod
+ def get_pipeline_id(cls, pipeline_mode="full"):
+ cls._check_mode(pipeline_mode)
+ return _PIPELINE_ID
+
+ @classmethod
+ def ensure_pipeline_registered(cls, pipeline_mode="full"):
+ cls._check_mode(pipeline_mode)
+ register_pipeline(QWEN3_TTS_RL_PIPELINE)
+
+ @classmethod
+ def weight_sync_stage_ids(cls, pipeline_mode="full"):
+ """Sync actor weights only to stage 0; stage 1 is the frozen decoder."""
+ cls._check_mode(pipeline_mode)
+ return [0]
+
+ @classmethod
+ def get_stage_engine_extras(cls, stage_id, pipeline_mode="full"):
+ cls._check_mode(pipeline_mode)
+ if stage_id == 0:
+ return {}
+ if stage_id == 1:
+ return {"max_model_len": 65536, "max_num_batched_tokens": 65536}
+ raise ValueError(f"Qwen3-TTS has no rollout stage {stage_id}.")
+
+ @classmethod
+ def postprocess_agent_loop_output(cls, output, *, tokenizer, response_length):
+ """Map the 16-codebook rollout to codec-0 policy tokens and replay fields."""
+ extra = output.extra_fields
+ if "tts_audio_codes" not in extra or "tts_text" not in extra:
+ raise RuntimeError("Qwen3-TTS rollout did not return codec codes and text.")
+ if QWEN3_TTS_REPLAY_KEY in extra:
+ raise RuntimeError(f"Qwen3-TTS rollout unexpectedly returned reserved field {QWEN3_TTS_REPLAY_KEY!r}.")
+ codes, text = extra["tts_audio_codes"], extra["tts_text"]
+ if not isinstance(text, str):
+ raise TypeError(f"Qwen3-TTS rollout text must be a string, got {type(text).__name__}.")
+ codes = torch.as_tensor(codes, dtype=torch.long)
+ if codes.ndim != 2 or codes.shape[-1] != 16:
+ raise ValueError("Qwen3-TTS codec codes must have shape (frames, 16).")
+ codes = codes[:response_length]
+ policy_ids = codes[:, 0].tolist()
+ if not policy_ids:
+ raise RuntimeError("Qwen3-TTS rollout returned an empty codec trajectory.")
+ if output.response_logprobs is not None:
+ if len(output.response_logprobs) < len(policy_ids):
+ raise RuntimeError("Qwen3-TTS rollout logprobs are shorter than the policy trajectory.")
+ output.response_logprobs = output.response_logprobs[: len(policy_ids)]
+ text_ids = tokenizer(build_assistant_text(text), return_tensors="pt", padding=False)["input_ids"]
+ text_ids = torch.as_tensor(text_ids, dtype=torch.long)
+ if text_ids.ndim == 1:
+ text_ids = text_ids.unsqueeze(0)
+ if text_ids.ndim != 2 or text_ids.shape[1] <= TEXT_PROMPT_TRAILER_TOKENS:
+ raise ValueError("Qwen3-TTS assistant text tokenization returned an invalid sequence.")
+ extra[QWEN3_TTS_REPLAY_KEY] = {
+ "text_ids": text_ids[:, :-TEXT_PROMPT_TRAILER_TOKENS].reshape(-1).tolist(),
+ "audio_codes": codes,
+ }
+ del extra["tts_audio_codes"]
+ del extra["tts_text"]
+ output.prompt_ids = [0]
+ output.response_ids = policy_ids
+ output.response_mask = [1] * len(policy_ids)
+ return output
+
+ @classmethod
+ def prepare_engine_prompt(cls, prompt_ids, model_config, multi_modal_data, mm_processor_kwargs=None):
+ """Build the Base-task prompt and fixed-speaker conditioning for vLLM-Omni."""
+ text = model_config.tokenizer.decode(prompt_ids, skip_special_tokens=True).strip()
+ if not text:
+ raise ValueError("Qwen3-TTS received an empty text prompt.")
+ speaker_path = model_config.override_config.get("tts_spk_embed_path")
+ if not speaker_path:
+ raise ValueError("Qwen3-TTS GRPO requires tts_spk_embed_path for the validated non-streaming replay.")
+ language = require_auto_language(model_config.override_config.get("tts_language"))
+ additional_information = {
+ "task_type": ["Base"],
+ "text": [text],
+ "language": [language],
+ "non_streaming_mode": [True],
+ "x_vector_only_mode": [True],
+ "voice_clone_prompt": [{"ref_spk_embedding": _load_speaker_vector(speaker_path)}],
+ }
+ assistant_ids = model_config.tokenizer(build_assistant_text(text), padding=False)["input_ids"]
+ assistant_ids = torch.as_tensor(assistant_ids).reshape(-1).tolist()
+ prompt_length = len(assistant_ids) + 2
+ identity = "\0".join((text, str(language), str(speaker_path))).encode()
+ return {
+ "prompt_token_ids": [1] * prompt_length,
+ "additional_information": additional_information,
+ "cache_salt": hashlib.sha256(identity).hexdigest(),
+ }
+
+ @classmethod
+ def combine_engine_outputs(cls, outputs, prompt):
+ """Combine stage-0 policy tokens with codec and waveform outputs."""
+ policy_outputs = [output for output in outputs if output.stage_id == 0]
+ decoder_outputs = [output for output in outputs if output.stage_id == 1]
+ if not policy_outputs:
+ raise RuntimeError("Qwen3-TTS rollout produced no stage-0 policy output.")
+ if not decoder_outputs:
+ raise RuntimeError("Qwen3-TTS rollout produced no stage-1 decoder output.")
+
+ policy_output = policy_outputs[-1]
+ decoder_output = decoder_outputs[-1]
+ if len(policy_output.outputs) != 1:
+ raise RuntimeError(
+ f"Qwen3-TTS stage 0 must return exactly one completion, got {len(policy_output.outputs)}."
+ )
+ token_ids = list(policy_output.outputs[0].token_ids)
+ if not token_ids:
+ raise RuntimeError("Qwen3-TTS rollout returned an empty stage-0 policy trajectory.")
+
+ try:
+ audio_codes = torch.as_tensor(policy_output.multimodal_output["codes"]["audio"]).detach().cpu()
+ waveform = torch.as_tensor(decoder_output.multimodal_output["audio"]).detach().cpu().float()
+ sample_rate = torch.as_tensor(decoder_output.multimodal_output["sr"])
+ except (KeyError, TypeError, ValueError, RuntimeError) as error:
+ raise RuntimeError("Qwen3-TTS rollout output does not match the pinned two-stage contract.") from error
+ if waveform.ndim != 1:
+ raise RuntimeError("Qwen3-TTS stage 1 must return a one-dimensional mono waveform.")
+ if waveform.numel() == 0:
+ raise RuntimeError("Qwen3-TTS stage 1 returned an empty waveform.")
+ if sample_rate.numel() != 1:
+ raise RuntimeError("Qwen3-TTS stage 1 must return one scalar sample rate.")
+ sample_rate_value = float(sample_rate.item())
+ if sample_rate_value <= 0 or not sample_rate_value.is_integer():
+ raise RuntimeError(f"Qwen3-TTS stage 1 returned an invalid sample rate: {sample_rate_value!r}.")
+
+ fields = {
+ "tts_audio_codes": align_audio_codes(audio_codes, token_ids),
+ "tts_text": prompt["additional_information"]["text"][0],
+ "audio": waveform,
+ "audio_sample_rate": int(sample_rate_value),
+ }
+ return policy_output, fields
diff --git a/verl_omni/pipelines/qwen3_tts/rollout_utils.py b/verl_omni/pipelines/qwen3_tts/rollout_utils.py
new file mode 100644
index 000000000..24ce2949a
--- /dev/null
+++ b/verl_omni/pipelines/qwen3_tts/rollout_utils.py
@@ -0,0 +1,66 @@
+# Copyright 2026 Bytedance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Helpers for validating Qwen3-TTS rollout outputs."""
+
+import torch
+
+QWEN3_TTS_REPLAY_KEY = "qwen3_tts_talker_replay"
+
+
+def align_audio_codes(audio_codes: torch.Tensor, token_ids: list[int]) -> torch.Tensor:
+ """Recover residual codebooks using codec-0 policy tokens as an exact invariant.
+
+ The engine can prepend placeholder rows and need not emit residual codes
+ for the last sampled token, because that token is never consumed as the
+ context for another policy token. Thus ``token_ids[:-1]`` must match
+ exactly; a final residual row is retained only when the engine provides it.
+ """
+ if audio_codes.ndim != 2 or audio_codes.shape[-1] != 16:
+ raise ValueError("Qwen3-TTS codec codes must have shape (frames, 16).")
+ if not token_ids:
+ return audio_codes[:0].long()
+ raw_codes = audio_codes.long()
+ policy_ids = torch.as_tensor(token_ids, dtype=torch.long, device=raw_codes.device)
+ required = len(token_ids) - 1
+ candidates: list[tuple[int, int]] = []
+ if required == 0:
+ candidates.append((0, 0))
+ else:
+ for start in range(raw_codes.shape[0] - required + 1):
+ if torch.equal(raw_codes[start : start + required, 0], policy_ids[:required]):
+ has_final = int(
+ raw_codes.shape[0] - start >= len(token_ids)
+ and raw_codes[start + required, 0] == policy_ids[required]
+ )
+ candidates.append((required + has_final, start))
+ if not candidates:
+ raise RuntimeError(
+ "Could not exactly align Qwen3-TTS residual codebooks with the sampled codec-0 policy: "
+ f"response_length={len(token_ids)}, raw_codec_rows={raw_codes.shape[0]}."
+ )
+
+ copy_length = max(length for length, _ in candidates)
+ best_starts = [start for length, start in candidates if length == copy_length]
+ if len(best_starts) != 1:
+ raise RuntimeError(
+ "Ambiguous Qwen3-TTS codec alignment: multiple residual-codebook spans match the sampled policy."
+ )
+ start = best_starts[0]
+ codes = raw_codes.new_zeros((len(token_ids), 16))
+ if copy_length:
+ codes[:copy_length] = raw_codes[start : start + copy_length]
+ codes[:, 0] = policy_ids
+ if required and not torch.equal(codes[:required, 0], policy_ids[:required]):
+ raise RuntimeError("Qwen3-TTS codec alignment invariant failed after recovery.")
+ return codes
diff --git a/verl_omni/pipelines/qwen3_tts/talker_forward.py b/verl_omni/pipelines/qwen3_tts/talker_forward.py
new file mode 100644
index 000000000..a1acc763a
--- /dev/null
+++ b/verl_omni/pipelines/qwen3_tts/talker_forward.py
@@ -0,0 +1,250 @@
+# Copyright 2026 Bytedance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Teacher-forced codec-0 forward math for Qwen3-TTS."""
+
+import json
+from dataclasses import dataclass
+
+import torch
+
+NUM_CODEBOOKS = 16
+TEXT_PROMPT_TRAILER_TOKENS = 5
+
+
+def build_assistant_text(text: str) -> str:
+ return f"<|im_start|>assistant\n{text}<|im_end|>\n<|im_start|>assistant\n"
+
+
+def load_speaker_xvector(path: str) -> torch.Tensor:
+ with open(path) as file:
+ vector = json.load(file)
+ return torch.tensor(vector, dtype=torch.float32).reshape(1, -1)
+
+
+@dataclass
+class TalkerTokens:
+ tts_pad: int
+ tts_bos: int
+ tts_eos: int
+ codec_pad: int
+ codec_bos: int
+ codec_eos: int
+ codec_nothink: int
+ codec_think_bos: int
+ codec_think_eos: int
+
+ @classmethod
+ def from_config(cls, config):
+ talker = config.talker_config
+ return cls(
+ int(config.tts_pad_token_id),
+ int(config.tts_bos_token_id),
+ int(config.tts_eos_token_id),
+ int(talker.codec_pad_id),
+ int(talker.codec_bos_id),
+ int(talker.codec_eos_token_id),
+ int(talker.codec_nothink_id),
+ int(talker.codec_think_bos_id),
+ int(talker.codec_think_eos_id),
+ )
+
+
+@dataclass
+class TalkerBatch:
+ input_ids: torch.Tensor
+ codec_ids: torch.Tensor
+ text_embedding_mask: torch.Tensor
+ codec_embedding_mask: torch.Tensor
+ codec_mask: torch.Tensor
+ attention_mask: torch.Tensor
+ codec_lens: list[int]
+ logit_start: list[int]
+ speaker_slots: list[int]
+
+
+def build_talker_batch(
+ text_ids,
+ audio_codes,
+ tokens,
+ *,
+ sub_codebook_vocab: int,
+ device=None,
+) -> TalkerBatch:
+ if not text_ids or len(text_ids) != len(audio_codes):
+ raise ValueError("Qwen3-TTS teacher forcing requires matching non-empty text and codec batches.")
+ if sub_codebook_vocab <= 0:
+ raise ValueError(f"sub_codebook_vocab must be positive, got {sub_codebook_vocab}.")
+ if any(ids.reshape(-1).numel() < 3 for ids in text_ids):
+ raise ValueError("Qwen3-TTS teacher-forcing text sequences must contain at least three prefix tokens.")
+ if any(codes.ndim != 2 or codes.shape[-1] != NUM_CODEBOOKS for codes in audio_codes):
+ raise ValueError(f"Qwen3-TTS codec codes must have shape (frames, {NUM_CODEBOOKS}).")
+ text_lens = [int(ids.reshape(-1).shape[0]) for ids in text_ids]
+ codec_lens = [int(codes.shape[0]) for codes in audio_codes]
+ speaker_slot = 6
+ text_start = 8
+ seq_len = max(t + c for t, c in zip(text_lens, codec_lens, strict=True)) + 8
+ batch_size = len(text_ids)
+ input_ids = torch.zeros((batch_size, seq_len, 2), dtype=torch.long, device=device)
+ codec_ids = torch.zeros((batch_size, seq_len, NUM_CODEBOOKS), dtype=torch.long, device=device)
+ text_mask = torch.zeros((batch_size, seq_len), dtype=torch.bool, device=device)
+ codec_embedding_mask = torch.zeros_like(text_mask)
+ codec_mask = torch.zeros_like(text_mask)
+ attention_mask = torch.zeros((batch_size, seq_len), dtype=torch.long, device=device)
+
+ for index, (sample_text, sample_codes) in enumerate(zip(text_ids, audio_codes, strict=True)):
+ ids = sample_text.reshape(-1).to(device=device, dtype=torch.long)
+ codes = sample_codes.to(device=device, dtype=torch.long)
+ if torch.any((codes[:, 1:] < 0) | (codes[:, 1:] >= sub_codebook_vocab)):
+ raise ValueError("Qwen3-TTS residual codec IDs are outside the code-predictor vocabulary.")
+ text_len, codec_len = text_lens[index], codec_lens[index]
+
+ input_ids[index, :3, 0] = ids[:3]
+ input_ids[index, 3:speaker_slot, 0] = tokens.tts_pad
+ input_ids[index, speaker_slot, 0] = tokens.tts_pad
+ input_ids[index, speaker_slot + 1, 0] = tokens.tts_bos
+ input_ids[index, text_start : text_start + text_len - 3, 0] = ids[3:]
+ input_ids[index, text_start + text_len - 3, 0] = tokens.tts_eos
+ input_ids[index, text_start + text_len - 2 : text_start + text_len + codec_len, 0] = tokens.tts_pad
+ text_mask[index, : text_start + text_len + codec_len] = True
+
+ input_ids[index, 3:speaker_slot, 1] = torch.tensor(
+ [tokens.codec_nothink, tokens.codec_think_bos, tokens.codec_think_eos], device=device
+ )
+ input_ids[index, speaker_slot + 1, 1] = tokens.codec_pad
+ input_ids[index, text_start : text_start + text_len - 2, 1] = tokens.codec_pad
+ input_ids[index, text_start + text_len - 2, 1] = tokens.codec_bos
+ start = text_start + text_len - 1
+ input_ids[index, start : start + codec_len, 1] = codes[:, 0]
+ input_ids[index, start + codec_len, 1] = tokens.codec_eos
+ codec_ids[index, start : start + codec_len] = codes
+ codec_embedding_mask[index, 3 : text_start + text_len + codec_len] = True
+ codec_embedding_mask[index, speaker_slot] = False
+ codec_mask[index, start : start + codec_len] = True
+ attention_mask[index, : text_start + text_len + codec_len] = True
+
+ return TalkerBatch(
+ input_ids,
+ codec_ids,
+ text_mask.unsqueeze(-1),
+ codec_embedding_mask.unsqueeze(-1),
+ codec_mask,
+ attention_mask,
+ codec_lens,
+ [text_start + text_len - 2 for text_len in text_lens],
+ [speaker_slot] * batch_size,
+ )
+
+
+def require_auto_language(language) -> str:
+ """Limit RL training to the prompt layout validated by the actor forward."""
+ normalized = str(language).strip()
+ if normalized.lower() != "auto":
+ raise ValueError(
+ "Qwen3-TTS RL currently supports only tts_language=Auto; "
+ "language-specific codec prefixes require a separately validated actor forward."
+ )
+ return "Auto"
+
+
+def codec0_input_embeddings(talker, batch: TalkerBatch, speaker_embedding: torch.Tensor) -> torch.Tensor:
+ ids = batch.input_ids
+ text_embeddings = talker.text_projection(talker.model.text_embedding(ids[:, :, 0]))
+ codec_embeddings = talker.model.codec_embedding(ids[:, :, 1]) * batch.codec_embedding_mask
+ codec_embeddings = codec_embeddings.clone()
+ sample_indices = torch.arange(codec_embeddings.shape[0], device=codec_embeddings.device)
+ codec_embeddings[sample_indices, batch.speaker_slots] = speaker_embedding.to(codec_embeddings.dtype)
+ embeddings = text_embeddings * batch.text_embedding_mask + codec_embeddings
+ sub_embeddings = talker.code_predictor.get_input_embeddings()
+ codec_mask = batch.codec_mask.unsqueeze(-1)
+ for codebook in range(1, NUM_CODEBOOKS):
+ embeddings += sub_embeddings[codebook - 1](batch.codec_ids[:, :, codebook]) * codec_mask
+ return embeddings
+
+
+def codec0_logits(talker, batch: TalkerBatch, speaker_embedding: torch.Tensor) -> torch.Tensor:
+ embeddings = codec0_input_embeddings(talker, batch, speaker_embedding)
+ output = talker(
+ inputs_embeds=embeddings[:, :-1],
+ attention_mask=batch.attention_mask[:, :-1],
+ use_cache=False,
+ output_hidden_states=False,
+ )
+ return output.logits
+
+
+def mask_codec0_logits(logits: torch.Tensor, codebook_vocab: int, codec_eos_token_id: int) -> torch.Tensor:
+ """Match the codec-token vocabulary exposed by the rollout model."""
+ if not 1 < codebook_vocab <= logits.shape[-1]:
+ raise ValueError(f"codebook_vocab must be in [2, {logits.shape[-1]}], got {codebook_vocab}.")
+ if not 0 <= codec_eos_token_id < logits.shape[-1]:
+ raise ValueError(f"codec_eos_token_id must be in [0, {logits.shape[-1]}), got {codec_eos_token_id}.")
+ valid = torch.zeros(logits.shape[-1], dtype=torch.bool, device=logits.device)
+ valid[1:codebook_vocab] = True
+ valid[codec_eos_token_id] = True
+ return logits.masked_fill(~valid, -1e4)
+
+
+def tts_actor_logits(
+ model,
+ input_ids,
+ attention_mask,
+ tts_text_ids,
+ tts_audio_codes,
+ response_len,
+ text_len,
+ speaker_embedding,
+) -> torch.Tensor:
+ batch_size, output_len = input_ids.shape
+ texts, codes, response_starts = [], [], []
+ for index in range(batch_size):
+ response_size, text_size = int(response_len[index]), int(text_len[index])
+ response_start = int(attention_mask[index].sum()) - response_size
+ if response_start < 1 or response_start + response_size > output_len:
+ raise ValueError("Invalid Qwen3-TTS response alignment.")
+ policy_ids = input_ids[index, response_start : response_start + response_size].long()
+ codec0_ids = tts_audio_codes[index, :response_size, 0].long()
+ if not torch.equal(policy_ids, codec0_ids):
+ mismatch = int(torch.nonzero(policy_ids != codec0_ids, as_tuple=False)[0])
+ raise RuntimeError(
+ f"Qwen3-TTS codec trajectory is not aligned with actor labels: sample={index}, frame={mismatch}."
+ )
+ texts.append(tts_text_ids[index, :text_size].long())
+ codes.append(tts_audio_codes[index, :response_size].long())
+ response_starts.append(response_start)
+
+ talker = model.talker
+ sub_vocab = int(talker.code_predictor.get_input_embeddings()[0].num_embeddings)
+ tokens = TalkerTokens.from_config(model.config)
+ batch = build_talker_batch(
+ texts,
+ codes,
+ tokens,
+ device=input_ids.device,
+ sub_codebook_vocab=sub_vocab,
+ )
+ logits = mask_codec0_logits(
+ codec0_logits(talker, batch, speaker_embedding),
+ sub_vocab,
+ int(model.config.talker_config.codec_eos_token_id),
+ )
+ if int(input_ids.min()) < 0 or int(input_ids.max()) >= logits.shape[-1]:
+ raise ValueError("Qwen3-TTS actor token IDs are outside the codec-0 logit vocabulary.")
+ output_vocab = logits.shape[-1]
+ aligned = logits.new_zeros((batch_size, output_len, output_vocab))
+ for index, response_start in enumerate(response_starts):
+ codec_len = batch.codec_lens[index]
+ target = slice(response_start - 1, response_start - 1 + codec_len)
+ source = batch.logit_start[index]
+ aligned[index, target] = logits[index, source : source + codec_len]
+ return aligned
diff --git a/verl_omni/pipelines/qwen3_tts/talker_training_adapter.py b/verl_omni/pipelines/qwen3_tts/talker_training_adapter.py
new file mode 100644
index 000000000..82c5b7954
--- /dev/null
+++ b/verl_omni/pipelines/qwen3_tts/talker_training_adapter.py
@@ -0,0 +1,184 @@
+# Copyright 2026 Bytedance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+"""Qwen3-TTS talker actor adapter."""
+
+import logging
+import types
+from collections.abc import Mapping
+from typing import Any
+
+import torch
+from transformers import AutoModelForTextToWaveform
+from verl.utils import tensordict_utils as tu
+
+from verl_omni.pipelines.model_base import OmniModelBase
+from verl_omni.pipelines.qwen3_tts.rollout_utils import QWEN3_TTS_REPLAY_KEY
+from verl_omni.pipelines.qwen3_tts.talker_forward import (
+ load_speaker_xvector,
+ require_auto_language,
+ tts_actor_logits,
+)
+
+
+def _speaker_embedding(model, batch_size, device, dtype):
+ return model._verl_tts_speaker_embedding.to(device=device, dtype=dtype).expand(batch_size, -1)
+
+
+def _qwen3_tts_forward(
+ self,
+ input_ids=None,
+ attention_mask=None,
+ tts_text_ids=None,
+ tts_audio_codes=None,
+ response_len=None,
+ text_len=None,
+ **kwargs,
+):
+ from transformers.modeling_outputs import CausalLMOutputWithPast
+
+ required_inputs = (input_ids, attention_mask, tts_text_ids, tts_audio_codes, response_len, text_len)
+ if any(value is None for value in required_inputs):
+ raise RuntimeError("Qwen3-TTS forward is missing exact rollout codec fields.")
+ speaker = _speaker_embedding(self, input_ids.shape[0], input_ids.device, next(self.talker.parameters()).dtype)
+ return CausalLMOutputWithPast(
+ logits=tts_actor_logits(
+ self,
+ input_ids,
+ attention_mask,
+ tts_text_ids,
+ tts_audio_codes,
+ response_len,
+ text_len,
+ speaker,
+ )
+ )
+
+
+def _get_input_embeddings(self):
+ return self.talker.model.codec_embedding
+
+
+def _set_input_embeddings(self, value):
+ self.talker.model.codec_embedding = value
+
+
+@OmniModelBase.register("Qwen3TTSForConditionalGeneration", stage="talker")
+class Qwen3TTSTalkerAdapter(OmniModelBase):
+ auto_model_class = AutoModelForTextToWaveform
+
+ @classmethod
+ def register_auto_classes(cls) -> None:
+ from qwen_tts.core.models.configuration_qwen3_tts import Qwen3TTSConfig
+ from qwen_tts.core.models.modeling_qwen3_tts import Qwen3TTSForConditionalGeneration
+ from transformers import AutoConfig, AutoModelForTextToWaveform
+
+ AutoConfig.register("qwen3_tts", Qwen3TTSConfig, exist_ok=True)
+ AutoModelForTextToWaveform.register(
+ Qwen3TTSConfig,
+ Qwen3TTSForConditionalGeneration,
+ exist_ok=True,
+ )
+
+ @classmethod
+ def get_strip_modules(cls, model_config):
+ return ["speaker_encoder", "speech_tokenizer", "code2wav"]
+
+ @classmethod
+ def configure_model(cls, module, model_config):
+ if getattr(model_config, "use_remove_padding", False):
+ raise ValueError("Qwen3-TTS Talker training requires actor_rollout_ref.model.use_remove_padding=false.")
+ module = super().configure_model(module, model_config)
+ module.config.tts_spk_embed_path = model_config.override_config.get("tts_spk_embed_path")
+ module.config.tts_language = require_auto_language(model_config.override_config.get("tts_language"))
+ if not module.config.tts_spk_embed_path:
+ raise ValueError("Qwen3-TTS GRPO requires tts_spk_embed_path for the validated non-streaming replay.")
+ module._verl_tts_speaker_embedding = load_speaker_xvector(module.config.tts_spk_embed_path)
+ module.forward = types.MethodType(_qwen3_tts_forward, module)
+ module.get_input_embeddings = types.MethodType(_get_input_embeddings, module)
+ module.set_input_embeddings = types.MethodType(_set_input_embeddings, module)
+ module._no_split_modules = ["Qwen3TTSTalkerDecoderLayer", "Qwen3TTSDecoderLayer"]
+ trainable = 0
+ for name, parameter in module.named_parameters():
+ parameter.requires_grad_(name.startswith(("talker.model.", "talker.codec_head.")))
+ trainable += int(parameter.requires_grad)
+ logging.getLogger(__name__).info("Qwen3-TTS talker adapter enabled %d trainable parameter tensors", trainable)
+ return module
+
+ @classmethod
+ def configure_processor(cls, model_path: str, model_config) -> Any:
+ return None
+
+ @classmethod
+ def configure_tokenizer(cls, model_path: str, model_config):
+ from transformers import AutoTokenizer
+
+ tokenizer = AutoTokenizer.from_pretrained(model_path, trust_remote_code=model_config.trust_remote_code)
+ tokenizer.chat_template = "{% for message in messages %}{{ message['content'] }}{% endfor %}"
+ if tokenizer.pad_token_id is None:
+ if tokenizer.eos_token_id is None:
+ raise ValueError("Qwen3-TTS tokenizer must define either pad_token_id or eos_token_id.")
+ tokenizer.pad_token_id = tokenizer.eos_token_id
+ model_config.hf_config.talker_config.tie_word_embeddings = False
+ return tokenizer
+
+ @classmethod
+ def prepare_model_inputs(cls, model_inputs, micro_batch, model_config):
+ del model_config
+ # The online V1 trainer retains AgentLoopOutput.extra_fields as one mapping per sample.
+ sample_extra_fields = tu.get(micro_batch, "extra_fields")
+ if sample_extra_fields is None:
+ raise RuntimeError(f"Qwen3-TTS actor inputs require the {QWEN3_TTS_REPLAY_KEY!r} replay payload.")
+ if not isinstance(sample_extra_fields, list) or any(
+ not isinstance(item, Mapping) for item in sample_extra_fields
+ ):
+ raise TypeError("Qwen3-TTS actor extra_fields must be a list of mappings.")
+ if len(sample_extra_fields) != model_inputs["input_ids"].shape[0]:
+ raise RuntimeError("Qwen3-TTS actor extra_fields do not match the actor batch size.")
+ if any(QWEN3_TTS_REPLAY_KEY not in item for item in sample_extra_fields):
+ raise RuntimeError(f"Qwen3-TTS actor inputs require the {QWEN3_TTS_REPLAY_KEY!r} replay payload.")
+
+ fields = [item[QWEN3_TTS_REPLAY_KEY] for item in sample_extra_fields]
+ if any(not isinstance(item, Mapping) for item in fields):
+ raise TypeError(f"Qwen3-TTS {QWEN3_TTS_REPLAY_KEY!r} must be a list of mappings.")
+ if any("text_ids" not in item or "audio_codes" not in item for item in fields):
+ raise RuntimeError("Qwen3-TTS replay payloads must contain text_ids and audio_codes.")
+
+ texts = [torch.as_tensor(item["text_ids"], dtype=torch.long).reshape(-1) for item in fields]
+ codes = [torch.as_tensor(item["audio_codes"], dtype=torch.long) for item in fields]
+ if any(item.numel() < 3 for item in texts):
+ raise ValueError("Qwen3-TTS actor text fields must contain at least three prefix tokens.")
+ if any(item.ndim != 2 or item.shape[-1] != 16 for item in codes):
+ raise ValueError("Qwen3-TTS codec codes must have shape (frames, 16).")
+ if any(item.shape[0] == 0 for item in codes):
+ raise ValueError("Qwen3-TTS actor codec fields must contain at least one frame.")
+ device = model_inputs["input_ids"].device
+ text_buffer = torch.zeros((len(fields), max(item.numel() for item in texts)), dtype=torch.long, device=device)
+ code_buffer = torch.zeros(
+ (len(fields), max(item.shape[0] for item in codes), 16), dtype=torch.long, device=device
+ )
+ text_lens = torch.empty(len(fields), dtype=torch.long, device=device)
+ response_lens = torch.empty_like(text_lens)
+ for index, (text, code) in enumerate(zip(texts, codes, strict=True)):
+ text_buffer[index, : text.numel()] = text.to(device)
+ code_buffer[index, : code.shape[0]] = code.to(device)
+ text_lens[index], response_lens[index] = text.numel(), code.shape[0]
+ model_inputs.update(
+ {
+ "tts_text_ids": text_buffer,
+ "tts_audio_codes": code_buffer,
+ "text_len": text_lens,
+ "response_len": response_lens,
+ }
+ )
+ return model_inputs