[Bugfix] Fix Qwen3-ASR transcription streaming postprocessing (#42478)

Signed-off-by: JooHo Lee <BWAAEEEK@users.noreply.github.com>
Signed-off-by: JooHo Lee <jooho414@gmail.com>
Co-authored-by: JooHo Lee <BWAAEEEK@users.noreply.github.com>

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("lish<asr", False) == ""
+    assert post_processor.process_delta("_text>Hello", 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("lish<asr_text>Hello", 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("lish<asr", finish_reason=None)
+        yield _request_output("_text>Hello", finish_reason=None)
+        yield _request_output("")
+
+    async def gen_world() -> AsyncGenerator[RequestOutput, None]:
+        yield _request_output(" language Eng", finish_reason=None)
+        yield _request_output("lish<asr_text>world")
+
+    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."""

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<asr_text>" 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:

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,

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 = "<asr_text>"
 # 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}<asr_text>{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 <asr_text> 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