From b22dabe27017109f8034ccb6619c352514a7a2a8 Mon Sep 17 00:00:00 2001 From: tzhouam Date: Wed, 9 Sep 2026 16:21:31 +0800 Subject: [PATCH] [Edge][WP5] Keep DELTA outputs for talker-only (codec-token) deployments The speech path forced FINAL_ONLY for Qwen3-TTS deployments without async chunks, so a talker-only server emitted one engine output per request and all codec frames arrived at once (measured: 1 engine output per request, first codes at output #1). A talker-only stage (final_output + latent) now keeps DELTA outputs so codec frames stream per step. Co-Authored-By: Claude Fable 5.1 --- .../openai_api/test_codec_stream_protocol.py | 18 ++++++++++++++++++ vllm_omni/entrypoints/openai/serving_speech.py | 17 ++++++++++++++++- 2 files changed, 34 insertions(+), 1 deletion(-) diff --git a/tests/entrypoints/openai_api/test_codec_stream_protocol.py b/tests/entrypoints/openai_api/test_codec_stream_protocol.py index bc914a474..e2029fe9e 100644 --- a/tests/entrypoints/openai_api/test_codec_stream_protocol.py +++ b/tests/entrypoints/openai_api/test_codec_stream_protocol.py @@ -181,3 +181,21 @@ def test_input_cancel_mid_stream_aborts_request(mocker): assert payload["cancelled"] is True assert "codec.done" in types and types[-1] == "session.done" svc.engine_client.abort.assert_awaited() + + +@pytest.mark.core_model +@pytest.mark.cpu +def test_talker_only_deployment_detection(): + from types import SimpleNamespace + + from vllm_omni.entrypoints.openai.serving_speech import OmniOpenAIServingSpeech + + probe = OmniOpenAIServingSpeech.__new__(OmniOpenAIServingSpeech) + probe._tts_stage = SimpleNamespace(final_output=True, final_output_type="latent") + assert probe._is_talker_only_deployment() is True + probe._tts_stage = SimpleNamespace(final_output=False, final_output_type="latent") + assert probe._is_talker_only_deployment() is False + probe._tts_stage = SimpleNamespace(final_output=True, final_output_type="audio") + assert probe._is_talker_only_deployment() is False + probe._tts_stage = None + assert probe._is_talker_only_deployment() is False diff --git a/vllm_omni/entrypoints/openai/serving_speech.py b/vllm_omni/entrypoints/openai/serving_speech.py index fc80d3b11..1d867771b 100644 --- a/vllm_omni/entrypoints/openai/serving_speech.py +++ b/vllm_omni/entrypoints/openai/serving_speech.py @@ -586,6 +586,15 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin): return stage return None + def _is_talker_only_deployment(self) -> bool: + """True when the TTS stage is the final stage and emits latents (codec tokens) per step.""" + stage = self._tts_stage + if stage is None: + return False + return bool(getattr(stage, "final_output", False)) and ( + str(getattr(stage, "final_output_type", "") or "").lower() == "latent" + ) + def _detect_tts_model_type(self) -> str | None: """Detect TTS model type from the resolved stage's deployment metadata. @@ -3096,7 +3105,13 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin): # list() makes a copy to avoid mutating the params. sampling_params_list = list(self.engine_client.default_sampling_params_list) async_chunk = getattr(self.model_config, "async_chunk", True) - qwen3_full_payload = self._tts_model_type == "qwen3_tts" and not bool(async_chunk) + # A talker-only (codec-token) deployment has no code2wav stage and no + # async chunks either, but its per-step ``codes.audio`` payloads are the + # stream itself, so it must keep DELTA outputs (otherwise the engine + # emits one FINAL_ONLY output and every codec frame arrives at once). + qwen3_full_payload = ( + self._tts_model_type == "qwen3_tts" and not bool(async_chunk) and not self._is_talker_only_deployment() + ) is_streaming_request = request.is_streaming() and not qwen3_full_payload sampling_params_list = coerce_param_message_types(sampling_params_list, is_streaming_request)