From c523026a8ef656f12f2b0707d1ce0dce77a4bdff Mon Sep 17 00:00:00 2001 From: Sy03 <1370724210@qq.com> Date: Sat, 4 Jul 2026 12:35:08 +0800 Subject: [PATCH] Fix Qwen3-TTS no-async-chunk payload splitting Signed-off-by: Sy03 <1370724210@qq.com> --- .../openai_api/test_serving_speech.py | 20 ++++++++ tests/worker/test_gpu_ar_model_runner.py | 46 +++++++++++++++++++ .../entrypoints/openai/serving_speech.py | 12 +++-- vllm_omni/worker/gpu_ar_model_runner.py | 3 +- 4 files changed, 76 insertions(+), 5 deletions(-) diff --git a/tests/entrypoints/openai_api/test_serving_speech.py b/tests/entrypoints/openai_api/test_serving_speech.py index 0c2006536..fd1685821 100644 --- a/tests/entrypoints/openai_api/test_serving_speech.py +++ b/tests/entrypoints/openai_api/test_serving_speech.py @@ -3394,6 +3394,26 @@ class TestTTSAsyncOffloading: mock_coerce.assert_called_once_with(qwen3_tts_server.engine_client.default_sampling_params_list, True) + def test_prepare_speech_generation_no_async_chunk_stream_uses_final_only( + self, qwen3_tts_server, mocker: MockerFixture + ): + """Full-payload TTS streaming should not request delta multimodal outputs.""" + qwen3_tts_server.engine_client.model_config.async_chunk = False + qwen3_tts_server._validate_tts_request = mocker.MagicMock(return_value=None) + qwen3_tts_server._build_tts_params = mocker.MagicMock( + return_value={"text": ["hello"], "task_type": ["CustomVoice"], "speaker": ["Vivian"]} + ) + qwen3_tts_server._estimate_prompt_len_async = mocker.AsyncMock(return_value=512) + mock_coerce = mocker.patch( + "vllm_omni.entrypoints.openai.serving_speech.coerce_param_message_types", + return_value=qwen3_tts_server.engine_client.default_sampling_params_list, + ) + request = OpenAICreateSpeechRequest(input="hello", stream_format="audio", response_format="pcm") + + asyncio.run(qwen3_tts_server._prepare_speech_generation(request)) + + mock_coerce.assert_called_once_with(qwen3_tts_server.engine_client.default_sampling_params_list, False) + def test_prepare_speech_generation_qwen3_voicedesign_non_streaming_mode_false( self, qwen3_tts_server, mocker: MockerFixture ): diff --git a/tests/worker/test_gpu_ar_model_runner.py b/tests/worker/test_gpu_ar_model_runner.py index a37c9b993..d1642a248 100644 --- a/tests/worker/test_gpu_ar_model_runner.py +++ b/tests/worker/test_gpu_ar_model_runner.py @@ -342,6 +342,52 @@ def test_build_omni_output_skips_hidden_when_model_opts_out(monkeypatch): assert output.multimodal_outputs is None +def test_build_omni_output_splits_mm_by_scheduled_tokens_when_hidden_is_tail_only(monkeypatch): + runner = _make_async_output_runner(engine_output_type="latent") + runner.model.omni_pooler_payload_include_hidden = False + runner._async_chunk = False + runner.requests = {"r1": object(), "r2": object(), "r3": object()} + + monkeypatch.setattr( + GPUARModelRunner, + "_resolve_pooler_payload_req_ids", + lambda self, req_ids: ("latent", req_ids), + ) + monkeypatch.setattr(GPUARModelRunner, "_should_accumulate_full_payload_output", lambda self: False) + monkeypatch.setattr(GPUARModelRunner, "get_omni_connector_output", lambda self: None) + monkeypatch.setattr(GPUARModelRunner, "_process_additional_information_updates", lambda *args, **kwargs: None) + + codes = torch.arange(48, dtype=torch.long).reshape(3, 16) + output = GPUARModelRunner._build_omni_model_runner_output_from_snapshot( + runner, + scheduler_output=SimpleNamespace( + total_num_scheduled_tokens=3, + num_scheduled_tokens={"r1": 1, "r2": 1, "r3": 1}, + ), + hidden_states=torch.tensor([[1.0]]), + staged_hidden_states_cpu=None, + multimodal_outputs={"codes": {"audio": codes}}, + req_ids_output_copy=["r1", "r2", "r3"], + req_id_to_index_output_copy={"r1": 0, "r2": 1, "r3": 2}, + valid_sampled_token_ids=[[101], [102], [103]], + logprobs_lists=None, + prompt_logprobs_dict={}, + num_nans_in_logits=None, + kv_connector_output=None, + ec_connector_output=None, + cudagraph_stats=None, + kv_extracted_req_ids=None, + seq_len=1, + num_scheduled_tokens_np=np.array([1, 1, 1], dtype=np.int32), + query_start_loc_cpu=torch.tensor([0, 1, 2], dtype=torch.long), + ) + + assert output.inter_stage_outputs is not None + assert torch.equal(output.inter_stage_outputs[0]["codes.audio"], codes[0:1]) + assert torch.equal(output.inter_stage_outputs[1]["codes.audio"], codes[1:2]) + assert torch.equal(output.inter_stage_outputs[2]["codes.audio"], codes[2:3]) + + def test_async_snapshot_payload_omits_hidden_when_model_opts_out(): runner = _make_async_output_runner() runner.model.omni_pooler_payload_include_hidden = False diff --git a/vllm_omni/entrypoints/openai/serving_speech.py b/vllm_omni/entrypoints/openai/serving_speech.py index 72654738f..76d1aa76e 100644 --- a/vllm_omni/entrypoints/openai/serving_speech.py +++ b/vllm_omni/entrypoints/openai/serving_speech.py @@ -3495,12 +3495,16 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin): request_id = request_id or f"speech-{random_uuid()}" qwen3_ref_audio_warmup_artifact_key: str | None = None - # If this is a streaming request, we need to coerce - # cumulative outputs to delta outputs; this ensures - # we don't emit redundant MM data & drain after emitting. + # If this is a streaming request with real async chunks, we need to + # coerce cumulative outputs to delta outputs; this ensures we don't + # emit redundant MM data & drain after emitting. Full-payload + # (async_chunk=False) TTS has no incremental audio chunks, so keep + # FINAL_ONLY semantics and let the streaming response send the final + # waveform once. # list() makes a copy to avoid mutating the params. sampling_params_list = list(self.engine_client.default_sampling_params_list) - is_streaming_request = request.is_streaming() + async_chunk = getattr(self.engine_client.model_config, "async_chunk", True) + is_streaming_request = request.is_streaming() and bool(async_chunk) sampling_params_list = coerce_param_message_types(sampling_params_list, is_streaming_request) # Build prompt + tts_params via the per-model adapter (RFC #4327). Every diff --git a/vllm_omni/worker/gpu_ar_model_runner.py b/vllm_omni/worker/gpu_ar_model_runner.py index c1812b8c9..2a49f3862 100644 --- a/vllm_omni/worker/gpu_ar_model_runner.py +++ b/vllm_omni/worker/gpu_ar_model_runner.py @@ -1601,6 +1601,7 @@ class GPUARModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin): # The actual multimodal wire transport uses multimodal_outputs instead. pooler_output: list[dict[str, object]] | None = None if needs_pooler_payload: + mm_seq_len = int(scheduler_output.total_num_scheduled_tokens) mm_cpu = None if self.omni_prefix_cache is not None: ( @@ -1656,7 +1657,7 @@ class GPUARModelRunner(OmniGPUModelRunner, OmniConnectorModelRunnerMixin): mm_cpu=mm_cpu, audio_sparse_output=audio_sparse_output, sparse_mm_index=sparse_mm_index, - seq_len=seq_len, + seq_len=mm_seq_len, ) pooler_output.append(flatten_payload(payload))