From a07765c6bd102bc1beaa5226d0fe5291b30a8524 Mon Sep 17 00:00:00 2001 From: JooHo Lee <96564470+BWAAEEEK@users.noreply.github.com> Date: Thu, 9 Jul 2026 16:33:27 +0900 Subject: [PATCH] [Bugfix] Fix Qwen3-ASR transcription streaming postprocessing (#42478) Signed-off-by: JooHo Lee Signed-off-by: JooHo Lee Co-authored-by: JooHo Lee --- .../test_transcription_inter_chunk_spacing.py | 121 +++++++++++++++++- .../speech_to_text/base/serving.py | 21 ++- vllm/model_executor/models/interfaces.py | 20 +++ vllm/model_executor/models/qwen3_asr.py | 62 +++++++-- 4 files changed, 207 insertions(+), 17 deletions(-) diff --git a/tests/entrypoints/speech_to_text/transcription/test_transcription_inter_chunk_spacing.py b/tests/entrypoints/speech_to_text/transcription/test_transcription_inter_chunk_spacing.py index 3175a9e6fb..7e51e5be44 100644 --- a/tests/entrypoints/speech_to_text/transcription/test_transcription_inter_chunk_spacing.py +++ b/tests/entrypoints/speech_to_text/transcription/test_transcription_inter_chunk_spacing.py @@ -32,7 +32,11 @@ from vllm.entrypoints.speech_to_text.transcription.protocol import Transcription from vllm.entrypoints.speech_to_text.transcription.serving import ( OpenAIServingTranscription, ) -from vllm.model_executor.models.interfaces import SupportsTranscription +from vllm.model_executor.models.interfaces import ( + StreamingTranscriptionPostProcessor, + SupportsTranscription, +) +from vllm.model_executor.models.qwen3_asr import Qwen3ASRForConditionalGeneration from vllm.outputs import CompletionOutput, RequestOutput # --- Unit: helper + protocol ------------------------------------------------- @@ -58,6 +62,63 @@ def test_asr_inter_chunk_separator_matches_protocol(language, expected_sep): assert sep == expected_sep +def test_qwen3_asr_stream_processor_passes_plain_text_without_prefix(): + post_processor = ( + Qwen3ASRForConditionalGeneration.get_streaming_post_processor_cls()() + ) + + assert post_processor.process_delta("Hello", False) == "Hello" + assert post_processor.process_delta(" world", True) == " world" + + +def test_qwen3_asr_stream_processor_buffers_prefix_with_leading_space(): + post_processor = ( + Qwen3ASRForConditionalGeneration.get_streaming_post_processor_cls()() + ) + + assert post_processor.process_delta(" language Eng", False) == "" + assert post_processor.process_delta("lishHello", True) == "Hello" + + +def test_qwen3_asr_stream_processor_keeps_independent_state(): + processor_cls = Qwen3ASRForConditionalGeneration.get_streaming_post_processor_cls() + first_processor = processor_cls() + second_processor = processor_cls() + + assert first_processor.process_delta(" language Eng", False) == "" + assert second_processor.process_delta("plain text", True) == "plain text" + assert first_processor.process_delta("lishHello", True) == "Hello" + + +def test_qwen3_asr_stream_processor_emits_finished_incomplete_prefix(): + post_processor = ( + Qwen3ASRForConditionalGeneration.get_streaming_post_processor_cls()() + ) + + assert ( + post_processor.process_delta(" language English", True) == " language English" + ) + + +def test_qwen3_asr_stream_processor_stops_buffering_long_plain_prefix(): + post_processor = ( + Qwen3ASRForConditionalGeneration.get_streaming_post_processor_cls()() + ) + text = " language " + ("x" * 50) + + assert post_processor.process_delta(text, False) == text + + +def test_qwen3_asr_stream_processor_stops_buffering_prefix_with_newline(): + post_processor = ( + Qwen3ASRForConditionalGeneration.get_streaming_post_processor_cls()() + ) + text = " language English\nhello" + + assert post_processor.process_delta(text, False) == text + + def test_joined_chunks_english_has_space_between(): sep = asr_inter_chunk_separator("en", SupportsTranscription.no_space_languages) assert sep.join(["hello", "world"]) == "hello world" @@ -90,8 +151,14 @@ class _StubTranscriptionModel: def post_process_output(cls, text: str) -> str: return text + @classmethod + def get_streaming_post_processor_cls( + cls, + ) -> type[StreamingTranscriptionPostProcessor]: + return StreamingTranscriptionPostProcessor -def _request_output(text: str) -> RequestOutput: + +def _request_output(text: str, finish_reason: str | None = "stop") -> RequestOutput: return RequestOutput( request_id="rid", prompt=None, @@ -104,7 +171,7 @@ def _request_output(text: str) -> RequestOutput: token_ids=(1, 2, 3), cumulative_logprob=None, logprobs=None, - finish_reason="stop", + finish_reason=finish_reason, ) ], finished=True, @@ -141,6 +208,9 @@ async def test_transcription_stream_generator_english_inserts_space_between_chun serving = OpenAIServingTranscription.__new__(OpenAIServingTranscription) serving.enable_force_include_usage = False serving.model_cls = _StubTranscriptionModel + serving.streaming_post_processor_cls = ( + _StubTranscriptionModel.get_streaming_post_processor_cls() + ) serving.task_type = "transcribe" request = SimpleNamespace( model="stub-model", @@ -178,6 +248,9 @@ async def test_transcription_stream_generator_chinese_no_space_between_chunks(): serving = OpenAIServingTranscription.__new__(OpenAIServingTranscription) serving.enable_force_include_usage = False serving.model_cls = _StubTranscriptionModel + serving.streaming_post_processor_cls = ( + _StubTranscriptionModel.get_streaming_post_processor_cls() + ) serving.task_type = "transcribe" request = SimpleNamespace( model="stub-model", @@ -203,6 +276,48 @@ async def test_transcription_stream_generator_chinese_no_space_between_chunks(): assert combined == "你好世界" +@pytest.mark.asyncio +async def test_transcription_stream_generator_strips_qwen3_asr_prefix_per_chunk(): + async def gen_hello() -> AsyncGenerator[RequestOutput, None]: + yield _request_output("language Eng", finish_reason=None) + yield _request_output("lishHello", finish_reason=None) + yield _request_output("") + + async def gen_world() -> AsyncGenerator[RequestOutput, None]: + yield _request_output(" language Eng", finish_reason=None) + yield _request_output("lishworld") + + serving = OpenAIServingTranscription.__new__(OpenAIServingTranscription) + serving.enable_force_include_usage = False + serving.model_cls = Qwen3ASRForConditionalGeneration + serving.streaming_post_processor_cls = ( + Qwen3ASRForConditionalGeneration.get_streaming_post_processor_cls() + ) + serving.task_type = "transcribe" + request = SimpleNamespace( + model="stub-qwen3-asr", + stream_include_usage=False, + stream_continuous_usage_stats=False, + ) + + out_lines: list[str] = [] + agen = OpenAIServingTranscription.transcription_stream_generator( + serving, + request=request, + result_generator=[gen_hello(), gen_world()], + request_id="test-qwen3-asr", + request_metadata=RequestResponseMetadata(request_id="test-qwen3-asr"), + audio_duration_s=1.0, + separator=" ", + ) + async for line in agen: + out_lines.append(line) + + combined = "".join(_sse_delta_contents("".join(out_lines))) + assert combined == "Hello world" + + @pytest.mark.asyncio async def test_create_transcription_non_streaming_joins_chunks_by_language(): """``create_transcription`` uses the same separator logic as the helper.""" diff --git a/vllm/entrypoints/speech_to_text/base/serving.py b/vllm/entrypoints/speech_to_text/base/serving.py index 15be7dc3f5..7c703f2535 100644 --- a/vllm/entrypoints/speech_to_text/base/serving.py +++ b/vllm/entrypoints/speech_to_text/base/serving.py @@ -114,6 +114,9 @@ class SpeechToTextBaseServing(GenerateBaseServing): self.asr_config = self.model_cls.get_speech_to_text_config( self.model_config, task_type ) + self.streaming_post_processor_cls = ( + self.model_cls.get_streaming_post_processor_cls() + ) self.enable_force_include_usage = enable_force_include_usage @@ -692,6 +695,7 @@ class SpeechToTextBaseServing(GenerateBaseServing): try: for result_generator in list_result_generator: beginning_of_chunk = True + post_processor = self.streaming_post_processor_cls() async for res in result_generator: # On first result. if res.prompt_token_ids is not None: @@ -709,19 +713,24 @@ class SpeechToTextBaseServing(GenerateBaseServing): assert len(res.outputs) == 1 output = res.outputs[0] + output_text = post_processor.process_delta( + output.text, output.finish_reason is not None + ) + # dont add separator to the first chunk if ( result_generator is not list_result_generator[0] and beginning_of_chunk + and output_text ): - output.text = separator + output.text + output_text = separator + output_text beginning_of_chunk = False - # TODO: For models that output structured formats (e.g., - # Qwen3-ASR with "language X" prefix), streaming - # would need buffering to strip the prefix properly since - # deltas may split the tag across chunks. - delta_message = DeltaMessage(content=output.text) + if output.finish_reason is None and not output_text: + completion_tokens += len(output.token_ids) + continue + + delta_message = DeltaMessage(content=output_text) completion_tokens += len(output.token_ids) if output.finish_reason is None: diff --git a/vllm/model_executor/models/interfaces.py b/vllm/model_executor/models/interfaces.py index f1d6d56373..32534f8fbb 100644 --- a/vllm/model_executor/models/interfaces.py +++ b/vllm/model_executor/models/interfaces.py @@ -70,6 +70,13 @@ The output embeddings must be one of the following formats: """ +class StreamingTranscriptionPostProcessor: + """Stateful streaming post-processor for transcription deltas.""" + + def process_delta(self, text_delta: str, finished: bool) -> str: + return text_delta + + def _require_is_multimodal(is_multimodal: Tensor | None) -> Tensor: """ A helper function to be used in the context of @@ -1198,6 +1205,19 @@ class SupportsTranscription(Protocol): """ return text + @classmethod + def get_streaming_post_processor_cls( + cls, + ) -> type[StreamingTranscriptionPostProcessor]: + """ + Return a stateful post-processor class for streaming output deltas. + + Each instance receives the next decoded text delta and whether the + request output is final. It returns the cleaned delta that should be + sent to the client. + """ + return StreamingTranscriptionPostProcessor + @classmethod def get_language_detection_prompt( cls, diff --git a/vllm/model_executor/models/qwen3_asr.py b/vllm/model_executor/models/qwen3_asr.py index 1c2001dcda..1e0523dd5f 100644 --- a/vllm/model_executor/models/qwen3_asr.py +++ b/vllm/model_executor/models/qwen3_asr.py @@ -38,6 +38,7 @@ from vllm.inputs import ModalityData, MultiModalDataDict, PromptType, TokensProm from vllm.logger import init_logger from vllm.model_executor.models.interfaces import ( MultiModalEmbeddings, + StreamingTranscriptionPostProcessor, SupportsLoRA, SupportsMRoPE, SupportsMultiModal, @@ -93,6 +94,53 @@ logger = init_logger(__name__) _ASR_TEXT_TAG = "" # User-supplied `prompt` / `response_prefix` must not inject extra ChatML turns. _CHATML_LIKE_TOKEN = re.compile(r"<\|[^|]+\|>") +_LANGUAGE_PREFIX = "language " +_MAX_STREAMING_PREFIX_CHARS = 50 + + +def _post_process_qwen3_asr_output(text: str) -> str: + if not text or _ASR_TEXT_TAG not in text: + return text + + _, text_part = text.rsplit(_ASR_TEXT_TAG, 1) + return text_part + + +class Qwen3ASRStreamingPostProcessor(StreamingTranscriptionPostProcessor): + def __init__(self) -> None: + self.raw_text = "" + self.emitted_text = "" + self.is_structured_output: bool | None = None + + def process_delta(self, text_delta: str, finished: bool) -> str: + self.raw_text += text_delta + + if self.is_structured_output is None: + text_without_leading_space = self.raw_text.lstrip() + maybe_structured_output = ( + text_without_leading_space == "" + or _LANGUAGE_PREFIX.startswith(text_without_leading_space) + or ( + text_without_leading_space.startswith(_LANGUAGE_PREFIX) + and len(text_without_leading_space) < _MAX_STREAMING_PREFIX_CHARS + and "\n" not in text_without_leading_space + ) + ) + if maybe_structured_output and _ASR_TEXT_TAG not in self.raw_text: + if not finished: + return "" + self.is_structured_output = False + else: + self.is_structured_output = _ASR_TEXT_TAG in self.raw_text + + processed_text = ( + _post_process_qwen3_asr_output(self.raw_text) + if self.is_structured_output + else self.raw_text + ) + new_text = processed_text[len(self.emitted_text) :] + self.emitted_text = processed_text + return new_text def _sanitize_transcription_user_text(text: str) -> str: @@ -632,12 +680,10 @@ class Qwen3ASRForConditionalGeneration( The model outputs in format: "language {lang}{transcription}" This method strips the language prefix and asr_text tags. """ - if not text: - return "" + return _post_process_qwen3_asr_output(text) - if _ASR_TEXT_TAG not in text: - return text - - # Split on tag and take the transcription part - _, text_part = text.rsplit(_ASR_TEXT_TAG, 1) - return text_part + @classmethod + def get_streaming_post_processor_cls( + cls, + ) -> type[StreamingTranscriptionPostProcessor]: + return Qwen3ASRStreamingPostProcessor