From 8d83fb59e9f635b778eaa93041f8d6fd84a2a17c Mon Sep 17 00:00:00 2001 From: Mike Qiu Date: Mon, 1 Jun 2026 21:51:29 +0800 Subject: [PATCH] [BugFix] Fix two stop reason for multimodal output (#3374) Signed-off-by: Mike_Qiu Signed-off-by: Mike Qiu Co-authored-by: Mike_Qiu Co-authored-by: Claude Opus 4.7 Co-authored-by: Hongsheng Liu --- .../entrypoints/test_stream_finish_reason.py | 460 ++++++++++++++++++ vllm_omni/entrypoints/openai/serving_chat.py | 59 ++- 2 files changed, 515 insertions(+), 4 deletions(-) create mode 100644 tests/entrypoints/test_stream_finish_reason.py diff --git a/tests/entrypoints/test_stream_finish_reason.py b/tests/entrypoints/test_stream_finish_reason.py new file mode 100644 index 000000000..52cc32215 --- /dev/null +++ b/tests/entrypoints/test_stream_finish_reason.py @@ -0,0 +1,460 @@ +# SPDX-License-Identifier: Apache-2.0 +"""Tests for multi-modal streaming finish_reason behavior (commit 44c799bc). + +Verifies that the /v1/chat/completions streaming endpoint emits exactly one +finish_reason="stop" per choice when multiple output modalities (text, audio) +are active, complying with the OpenAI streaming spec. + +Key invariants tested: + - Single modality (text only): last chunk carries finish_reason="stop" + - Multi-modality (text+audio): only the final modality chunk carries + finish_reason="stop"; earlier finishing modalities emit finish_reason=null + - n>1 with multi-modality: each choice independently tracks its own + modality state, so each choice gets exactly one "stop" + - Engine skips a declared modality: fallback stop chunk is emitted at + stream end so the client always receives finish_reason="stop" + - voice/speaker parameter compatibility in chat completions +""" + +import enum +import json +from unittest.mock import MagicMock + +import pytest + +# Python 3.10 compat: StrEnum was added in 3.11 +if not hasattr(enum, "StrEnum"): + + class _StrEnum(str, enum.Enum): + """Minimal StrEnum backport for Python 3.10.""" + + enum.StrEnum = _StrEnum # type: ignore[attr-defined] + +from vllm.entrypoints.openai.chat_completion.protocol import ( + ChatCompletionRequest, + ChatCompletionResponseStreamChoice, +) +from vllm.entrypoints.openai.engine.protocol import DeltaMessage +from vllm.entrypoints.openai.models.serving import OpenAIServingModels +from vllm.outputs import CompletionOutput, RequestOutput + +from vllm_omni.entrypoints.openai.serving_chat import OmniOpenAIServingChat +from vllm_omni.outputs import OmniRequestOutput + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_text_omni_output( + request_id: str = "test-req", + text: str = "hello", + token_ids: list[int] | None = None, + finish_reason: str | None = None, + index: int = 0, + num_prompt_tokens: int = 3, +) -> OmniRequestOutput: + """Build an OmniRequestOutput wrapping a text RequestOutput.""" + if token_ids is None: + token_ids = [10, 11, 12] + res = RequestOutput( + request_id=request_id, + prompt="test", + prompt_token_ids=list(range(num_prompt_tokens)), + prompt_logprobs=None, + outputs=[ + CompletionOutput( + index=index, + text=text, + token_ids=token_ids, + cumulative_logprob=0.0, + logprobs=None, + finish_reason=finish_reason, + stop_reason=None, + ) + ], + finished=finish_reason is not None, + ) + return OmniRequestOutput( + request_id=request_id, + final_output_type="text", + request_output=res, + finished=finish_reason is not None, + ) + + +def _make_audio_omni_output( + request_id: str = "test-req", + index: int = 0, + num_prompt_tokens: int = 3, +) -> OmniRequestOutput: + """Build an OmniRequestOutput for audio (no torch dependency).""" + res = RequestOutput( + request_id=request_id, + prompt="test", + prompt_token_ids=list(range(num_prompt_tokens)), + prompt_logprobs=None, + outputs=[ + CompletionOutput( + index=index, + text="", + token_ids=[], + cumulative_logprob=0.0, + logprobs=None, + finish_reason="stop", + stop_reason=None, + ) + ], + finished=True, + ) + return OmniRequestOutput( + request_id=request_id, + final_output_type="audio", + request_output=res, + finished=True, + ) + + +def _mock_audio_choices(index: int = 0, role: str = "assistant"): + return [ + ChatCompletionResponseStreamChoice( + index=index, + delta=DeltaMessage(role=role, content="dGVzdA=="), + logprobs=None, + finish_reason="stop", + ) + ] + + +def _build_serving_chat(): + """Create a minimal OmniOpenAIServingChat for testing.""" + mock_engine = MagicMock() + mock_engine.errored = False + + models = OpenAIServingModels( + engine_client=mock_engine, + base_model_paths=[], + ) + mock_render = MagicMock() + + instance = OmniOpenAIServingChat( + engine_client=mock_engine, + models=models, + response_role="assistant", + openai_serving_render=mock_render, + request_logger=None, + chat_template=None, + chat_template_content_format="auto", + ) + instance._create_audio_choice = MagicMock( + side_effect=lambda omni_res, role, request, stream=False: _mock_audio_choices( + index=omni_res.request_output.outputs[0].index, + role=role, + ) + ) + return instance + + +def _make_request(modalities: list[str], n: int = 1) -> ChatCompletionRequest: + req = ChatCompletionRequest( + model="test-model", + messages=[{"role": "user", "content": "hello"}], + n=n, + stream=True, + ) + req.modalities = modalities # type: ignore[attr-defined] + return req + + +def _parse_sse_chunks(lines: list[str]) -> list[dict]: + """Parse SSE lines into JSON dicts.""" + prefix = "data: " + chunks = [] + for line in lines: + line = line.strip() + if not line.startswith(prefix): + continue + payload = line[len(prefix) :].strip() + if payload == "[DONE]": + continue + try: + chunks.append(json.loads(payload)) + except json.JSONDecodeError: + pass + return chunks + + +async def _collect_stream(gen): + result = [] + async for item in gen: + result.append(item) + return result + + +# --------------------------------------------------------------------------- +# Tests: finish_reason correctness +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_single_modality_text_only_one_stop(): + """Text-only streaming: exactly one chunk has finish_reason='stop'.""" + serving_chat = _build_serving_chat() + request = _make_request(modalities=["text"]) + + async def result_generator(): + yield _make_text_omni_output(text="he", token_ids=[10, 11], finish_reason=None) + yield _make_text_omni_output(text="llo", token_ids=[12], finish_reason="stop") + + raw_lines = await _collect_stream( + serving_chat.chat_completion_stream_generator( + request=request, + result_generator=result_generator(), + request_id="test-req", + model_name="test-model", + conversation=[], + tokenizer=MagicMock(), + request_metadata=MagicMock(), + ) + ) + + chunks = _parse_sse_chunks(raw_lines) + finish_reasons = [c["choices"][0]["finish_reason"] for c in chunks if c.get("choices")] + + assert finish_reasons[-1] == "stop" + assert finish_reasons.count("stop") == 1 + for fr in finish_reasons[:-1]: + assert fr is None + + +@pytest.mark.asyncio +async def test_multi_modal_text_audio_only_last_stop(): + """text+audio: text finish sends finish_reason=null, audio sends stop.""" + serving_chat = _build_serving_chat() + request = _make_request(modalities=["text", "audio"]) + + async def result_generator(): + yield _make_text_omni_output(text="he", token_ids=[10, 11], finish_reason=None) + yield _make_text_omni_output(text="llo", token_ids=[12], finish_reason="stop") + yield _make_audio_omni_output() + + raw_lines = await _collect_stream( + serving_chat.chat_completion_stream_generator( + request=request, + result_generator=result_generator(), + request_id="test-req", + model_name="test-model", + conversation=[], + tokenizer=MagicMock(), + request_metadata=MagicMock(), + ) + ) + + chunks = _parse_sse_chunks(raw_lines) + finish_reasons = [ch["finish_reason"] for c in chunks for ch in c.get("choices", [])] + + assert finish_reasons.count("stop") == 1 + assert finish_reasons[-1] == "stop" + + # The text finish chunk must have finish_reason=None + for idx, c in enumerate(chunks): + for ch in c.get("choices", []): + if c.get("modality") == "text" and ch.get("delta", {}).get("content") == "lo": + assert ch["finish_reason"] is None + + +@pytest.mark.asyncio +async def test_multi_modal_n2_independent_per_choice(): + """n=2 with text+audio: each choice gets exactly one stop, at the end.""" + serving_chat = _build_serving_chat() + request = _make_request(modalities=["text", "audio"], n=2) + + async def result_generator(): + yield _make_text_omni_output(text="A", token_ids=[10], finish_reason=None, index=0) + yield _make_text_omni_output(text="B", token_ids=[20], finish_reason=None, index=1) + yield _make_text_omni_output(text="", token_ids=[11], finish_reason="stop", index=0) + yield _make_text_omni_output(text="", token_ids=[21], finish_reason="stop", index=1) + yield _make_audio_omni_output(index=0) + yield _make_audio_omni_output(index=1) + + raw_lines = await _collect_stream( + serving_chat.chat_completion_stream_generator( + request=request, + result_generator=result_generator(), + request_id="test-req", + model_name="test-model", + conversation=[], + tokenizer=MagicMock(), + request_metadata=MagicMock(), + ) + ) + + chunks = _parse_sse_chunks(raw_lines) + per_choice: dict[int, list] = {} + for c in chunks: + for ch in c.get("choices", []): + per_choice.setdefault(ch["index"], []).append(ch["finish_reason"]) + + for idx, reasons in per_choice.items(): + assert reasons.count("stop") == 1, f"Choice {idx} has {reasons.count('stop')} stops" + assert reasons[-1] == "stop", f"Choice {idx} last reason is {reasons[-1]}" + + +@pytest.mark.asyncio +async def test_single_modality_audio_only_one_stop(): + """Audio-only streaming: the audio chunk carries finish_reason='stop'.""" + serving_chat = _build_serving_chat() + request = _make_request(modalities=["audio"]) + + async def result_generator(): + yield _make_audio_omni_output() + + raw_lines = await _collect_stream( + serving_chat.chat_completion_stream_generator( + request=request, + result_generator=result_generator(), + request_id="test-req", + model_name="test-model", + conversation=[], + tokenizer=MagicMock(), + request_metadata=MagicMock(), + ) + ) + + chunks = _parse_sse_chunks(raw_lines) + finish_reasons = [ch["finish_reason"] for c in chunks for ch in c.get("choices", [])] + + assert finish_reasons.count("stop") == 1 + assert finish_reasons[-1] == "stop" + + +# --------------------------------------------------------------------------- +# Tests: fallback stop chunk when declared modality is not produced +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_declared_modality_not_produced_emits_fallback_stop(): + """If request.modalities declares ["text","audio"] but engine only produces + text, a fallback stop chunk is emitted at stream end.""" + serving_chat = _build_serving_chat() + request = _make_request(modalities=["text", "audio"]) + + async def result_generator(): + # Engine only produces text, no audio output at all + yield _make_text_omni_output(text="hi", token_ids=[10], finish_reason=None) + yield _make_text_omni_output(text="!", token_ids=[11], finish_reason="stop") + + raw_lines = await _collect_stream( + serving_chat.chat_completion_stream_generator( + request=request, + result_generator=result_generator(), + request_id="test-req", + model_name="test-model", + conversation=[], + tokenizer=MagicMock(), + request_metadata=MagicMock(), + ) + ) + + chunks = _parse_sse_chunks(raw_lines) + finish_reasons = [ch["finish_reason"] for c in chunks for ch in c.get("choices", [])] + + # Text finish is suppressed (audio not seen yet), but fallback stop + # chunk must appear at end. + assert finish_reasons.count("stop") == 1, f"Expected 1 stop, got {finish_reasons}" + assert finish_reasons[-1] == "stop" + + +@pytest.mark.asyncio +async def test_declared_modality_not_produced_text_finish_suppressed(): + """When text finishes but audio (declared in modalities) never appears, + the text finish chunk has finish_reason=null (suppressed).""" + serving_chat = _build_serving_chat() + request = _make_request(modalities=["text", "audio"]) + + async def result_generator(): + yield _make_text_omni_output(text="hi", token_ids=[10], finish_reason=None) + yield _make_text_omni_output(text="!", token_ids=[11], finish_reason="stop") + # No audio output — stream ends + + raw_lines = await _collect_stream( + serving_chat.chat_completion_stream_generator( + request=request, + result_generator=result_generator(), + request_id="test-req", + model_name="test-model", + conversation=[], + tokenizer=MagicMock(), + request_metadata=MagicMock(), + ) + ) + + chunks = _parse_sse_chunks(raw_lines) + + # Find the text finish chunk (content "!") + for c in chunks: + for ch in c.get("choices", []): + if c.get("modality") == "text" and ch.get("delta", {}).get("content") == "!": + # Text finish should be suppressed because audio hasn't appeared + assert ch["finish_reason"] is None + + +# --------------------------------------------------------------------------- +# Tests: voice/speaker parameter compatibility +# --------------------------------------------------------------------------- + + +class TestVoiceSpeakerCompat: + """Tests for voice/speaker parameter handling in chat completions.""" + + def test_voice_parameter_takes_priority(self): + """When both voice and speaker are provided via extra_body, voice wins.""" + req = ChatCompletionRequest( + model="test-model", + messages=[{"role": "user", "content": "hello"}], + ) + # Pydantic model_extra collects unknown fields + req.voice = "alloy" # type: ignore[attr-defined] + req.speaker = "vivian" # type: ignore[attr-defined] + + # voice takes priority: getattr(request, "voice", None) returns "alloy" + speaker = getattr(req, "voice", None) or getattr(req, "speaker", None) + assert speaker == "alloy" + + def test_speaker_fallback_when_no_voice(self): + """When only speaker is provided, it is used.""" + req = ChatCompletionRequest( + model="test-model", + messages=[{"role": "user", "content": "hello"}], + ) + req.speaker = "vivian" # type: ignore[attr-defined] + + speaker = getattr(req, "voice", None) or getattr(req, "speaker", None) + assert speaker == "vivian" + + def test_neither_voice_nor_speaker(self): + """When neither is provided, result is None.""" + req = ChatCompletionRequest( + model="test-model", + messages=[{"role": "user", "content": "hello"}], + ) + + speaker = getattr(req, "voice", None) or getattr(req, "speaker", None) + assert speaker is None + + def test_empty_voice_falls_back_to_speaker(self): + """Empty string voice falls back to speaker.""" + req = ChatCompletionRequest( + model="test-model", + messages=[{"role": "user", "content": "hello"}], + ) + req.voice = "" # type: ignore[attr-defined] + req.speaker = "vivian" # type: ignore[attr-defined] + + # Empty string is falsy, so speaker is used + speaker = getattr(req, "voice", None) or getattr(req, "speaker", None) + assert speaker == "vivian" diff --git a/vllm_omni/entrypoints/openai/serving_chat.py b/vllm_omni/entrypoints/openai/serving_chat.py index 7e1d3027b..2247f7394 100644 --- a/vllm_omni/entrypoints/openai/serving_chat.py +++ b/vllm_omni/entrypoints/openai/serving_chat.py @@ -364,10 +364,11 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): if raw_request: raw_request.state.request_metadata = request_metadata - output_modalities = getattr(request, "modalities", self.engine_client.output_modalities) - request.modalities = ( - output_modalities if output_modalities is not None else self.engine_client.output_modalities - ) + # Some models will return a list like ["text", None, "audio"], better + # to strip None in the list + engine_output_modalities = [x for x in self.engine_client.output_modalities if x is not None] + output_modalities = getattr(request, "modalities", engine_output_modalities) + request.modalities = output_modalities if output_modalities is not None else engine_output_modalities num_inference_steps = None cfg_text_scale = None @@ -986,6 +987,9 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): num_choices = 1 if request.n is None else request.n previous_num_tokens = [0] * num_choices finish_reason_sent = [False] * num_choices + modality_finished: list[set[str]] = [set() for _ in range(num_choices)] + modality_seen: list[set[str]] = [set() for _ in range(num_choices)] + stop_reason_emitted: list[bool] = [False] * num_choices num_prompt_tokens = 0 num_cached_tokens = None if self.use_harmony: @@ -1051,6 +1055,13 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): logger.warning(f"final output type: {final_output_type} is not needed by the request") continue + # Track which modalities have actually appeared in the stream. + # This is used to determine when all *produced* modalities have + # finished, which may be a subset of request.modalities when the + # engine does not produce every requested modality. + for output in res.outputs: + modality_seen[output.index].add(final_output_type) + if omni_res.metrics: last_metrics = omni_res.metrics @@ -1581,6 +1592,16 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): finish_reason_ = "tool_calls" else: finish_reason_ = output.finish_reason if output.finish_reason else "stop" + # Only emit finish_reason on the last modality to + # comply with OpenAI streaming spec: exactly one + # chunk per choice carries finish_reason="stop". + modality_finished[i].add("text") + if modality_seen[i] < set(request.modalities) or not all( + m in modality_finished[i] for m in modality_seen[i] + ): + finish_reason_ = None + else: + stop_reason_emitted[i] = True choice_data = ChatCompletionResponseStreamChoice( index=i, delta=delta_message, @@ -1654,6 +1675,17 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): role = self.get_chat_request_role(request) choices_data = self._create_audio_choice(omni_res, role, request, stream=True) + # Only emit finish_reason on the last modality to + # comply with OpenAI streaming spec. + for choice in choices_data: + if choice.finish_reason is not None: + modality_finished[choice.index].add("audio") + if modality_seen[choice.index] < set(request.modalities) or not all( + m in modality_finished[choice.index] for m in modality_seen[choice.index] + ): + choice.finish_reason = None + else: + stop_reason_emitted[choice.index] = True # Record per-chunk PCM byte count + arrival timestamp for # audio_underrun_s / audio_continuity_ok_total at finalize. if req_state is not None and req_state.request_arrival_ts > 0: @@ -1683,6 +1715,25 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): logger.warning(f"Unsupported streaming final output type: {final_output_type}") continue + # Fallback: if a choice had modalities finish but no finish_reason="stop" + # was emitted (e.g. request.modalities included a modality the engine + # never produced), emit a final stop chunk for that choice. + for i in range(num_choices): + if modality_finished[i] and not stop_reason_emitted[i]: + stop_choice = ChatCompletionResponseStreamChoice( + index=i, + delta=DeltaMessage(), + finish_reason="stop", + ) + stop_chunk = OmniChatCompletionStreamResponse( + id=request_id, + object=chunk_object_type, + created=created_time, + choices=[stop_choice], + model=model_name, + ) + data = stop_chunk.model_dump_json(exclude_unset=True) + yield f"data: {data}\n\n" # Emit audio_underrun_s + audio_continuity_ok_total once per # request after the audio chunk stream is exhausted. The # captured reference (req_state_audio_ref) outlives the inner