From b44cf97b314f17cb237873ebe4eeccedba5b3cdb Mon Sep 17 00:00:00 2001 From: Deep Shah Date: Tue, 22 Sep 2026 11:19:16 -0600 Subject: [PATCH] [Bugfix] Preserve all video-stream audio deltas (#7532) Signed-off-by: Deep Shah --- docs/serving/video_stream_api.md | 2 +- .../openai_api/test_video_audio_delta.py | 242 ++++++++++++++++++ tests/entrypoints/test_async_omni.py | 42 +++ .../entrypoints/openai/video_stream_base.py | 81 ++---- 4 files changed, 308 insertions(+), 59 deletions(-) create mode 100644 tests/entrypoints/openai_api/test_video_audio_delta.py diff --git a/docs/serving/video_stream_api.md b/docs/serving/video_stream_api.md index 23aa4cb08..22362aac6 100644 --- a/docs/serving/video_stream_api.md +++ b/docs/serving/video_stream_api.md @@ -101,7 +101,7 @@ The server accepts these legacy field names and rewrites them before validation. | Variable | Values | Default | Description | | -------- | ------ | ------- | ----------- | | `VLLM_VIDEO_ASYNC_CHUNK` | `on`, `off` | `on` | Wire-level streaming switch. `off` buffers server-side deltas and emits coalesced outputs at the end of a query. | -| `VLLM_VIDEO_AUDIO_DELTA_MODE` | `fast`, `slow` | `fast` | Audio delta extraction strategy. `fast` emits only newly produced chunks; `slow` recomputes from accumulated audio and exists for A/B verification. | +| `VLLM_VIDEO_AUDIO_DELTA_MODE` | `fast`, `slow` | `fast` | Both settings forward every fresh engine audio delta. `slow` is retained as a compatibility alias for `fast`. | ## EVS Semantics diff --git a/tests/entrypoints/openai_api/test_video_audio_delta.py b/tests/entrypoints/openai_api/test_video_audio_delta.py new file mode 100644 index 000000000..123ee7424 --- /dev/null +++ b/tests/entrypoints/openai_api/test_video_audio_delta.py @@ -0,0 +1,242 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""Preserve every sample from the engine's fresh DELTA audio payloads.""" + +import asyncio +import base64 +import io +import wave +from types import SimpleNamespace + +import numpy as np +import pytest +import torch +from vllm.sampling_params import RequestOutputKind, SamplingParams +from vllm.v1.engine import FinishReason + +from tests.helpers.serving_chat import build_serving_chat +from vllm_omni.entrypoints.omni_base import OmniBase +from vllm_omni.entrypoints.openai.serving_video_stream import QwenOmniStreamingVideoHandler +from vllm_omni.entrypoints.openai.video_stream_base import ( + _CODEC_FRAME_SAMPLES, + OmniStreamingVideoHandler, + StreamingVideoSessionConfig, +) +from vllm_omni.outputs import OmniRequestOutput +from vllm_omni.outputs.mm_outputs import MultimodalCompletionOutput, MultimodalPayload +from vllm_omni.outputs.output_processor import OmniRequestState + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +def _audio_result(payload): + completion = MultimodalCompletionOutput( + index=0, + text="", + token_ids=[], + cumulative_logprob=None, + logprobs=None, + multimodal_output=MultimodalPayload.from_dict({"audio": payload}), + ) + return OmniRequestOutput( + outputs=[completion], + final_output_type="audio", + ) + + +def _pcm(encoded): + assert encoded is not None + with wave.open(io.BytesIO(base64.b64decode(encoded)), "rb") as wav: + assert (wav.getframerate(), wav.getnchannels(), wav.getsampwidth()) == (24000, 1, 2) + return np.frombuffer(wav.readframes(wav.getnframes()), dtype=" 2 * _CODEC_FRAME_SAMPLES else 0 + np.testing.assert_array_equal(_pcm(encoded), np.arange(start, first_length)) + encoded, seen = OmniStreamingVideoHandler._delta_fast(later, seen) + np.testing.assert_array_equal(_pcm(encoded), np.arange(5000, 10000)) + assert seen == 2 + + +@pytest.mark.parametrize("explicit", [False, True]) +def test_only_omitted_sampling_parameters_are_coerced(explicit): + base = object.__new__(OmniBase) + base.engine = SimpleNamespace(num_stages=1) + params = SamplingParams(output_kind=RequestOutputKind.CUMULATIVE) + base.default_sampling_params_list = [params] + base.sampling_constraints_list = [{}] + resolved = base.resolve_sampling_params_list([params] if explicit else None, allow_delta_coercion=True) + expected = RequestOutputKind.CUMULATIVE if explicit else RequestOutputKind.DELTA + assert resolved[0].output_kind == expected + assert params.output_kind == RequestOutputKind.CUMULATIVE + + +def _state(output_kind): + return OmniRequestState( + request_id="audio-delta", + external_req_id="audio-delta", + parent_req=None, + request_index=0, + lora_request=None, + prompt=None, + prompt_token_ids=[0], + prompt_embeds=None, + logprobs_processor=None, + detokenizer=None, + max_tokens_param=None, + arrival_time=0.0, + queue=None, + log_stats=False, + stream_interval=1, + output_kind=output_kind, + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("wire_mode", ["on", "off"]) +@pytest.mark.parametrize("delta_mode", ["fast", "slow"]) +@pytest.mark.parametrize("batch_later_chunks", [False, True]) +@pytest.mark.parametrize("finish_with_payload", [False, True]) +@pytest.mark.parametrize("explicit_sampling", [False, True]) +@pytest.mark.parametrize("lengths", [(5000, 5000, 5000), (0, 5000, 0, 2000, 5000)]) +async def test_real_producer_deltas_reach_video_client( + monkeypatch, wire_mode, delta_mode, batch_later_chunks, finish_with_payload, explicit_sampling, lengths +): + monkeypatch.setenv("VLLM_VIDEO_ASYNC_CHUNK", wire_mode) + monkeypatch.setenv("VLLM_VIDEO_AUDIO_DELTA_MODE", delta_mode) + chunks = [] + offset = 0 + for length in lengths: + # Qwen's generation runner hands off [1, samples] CPU tensors. + chunks.append(torch.arange(offset, offset + length, dtype=torch.float32).reshape(1, -1) / 32768) + offset += length + groups = [[chunks[0]], chunks[1:]] if batch_later_chunks else [[chunk] for chunk in chunks] + state = _state(RequestOutputKind.DELTA) + snapshots = [] + produced = [] + # Complete production before consumption: retained DELTA snapshots must + # survive subsequent accumulation and draining under consumer backpressure. + for index, group in enumerate(groups): + for chunk in group: + state.add_multimodal_tensor(chunk, mm_type="audio") + finish = FinishReason.STOP if finish_with_payload and index == len(groups) - 1 else None + result = state.make_request_output([], None, finish, None) + assert result is not None + assert "audio" not in state.mm_accumulated + snapshots.append(result.outputs[0].multimodal_output["audio"]) + produced.append(OmniRequestOutput.from_stage_output(result, final_output_type="audio")) + if not finish_with_payload: + result = state.make_request_output([], None, FinishReason.STOP, None) + assert result is not None + produced.append(OmniRequestOutput.from_stage_output(result, final_output_type="audio")) + + class Engine: + async def generate(self, *, prompt, request_id, output_modalities, sampling_params_list=None): + if explicit_sampling: + assert sampling_params_list is not None + assert sampling_params_list[0].output_kind == RequestOutputKind.DELTA + else: + assert sampling_params_list is None + # AsyncOmni coerces omitted parameters to DELTA as well. + for result in produced: + yield result + + class Handler(QwenOmniStreamingVideoHandler): + async def _preprocess_to_engine_prompt(self, request): + return {"prompt": "describe"} + + class WebSocket: + def __init__(self): + self.sent = [] + + async def send_json(self, message): + self.sent.append(message) + + ws = WebSocket() + handler = Handler(chat_service=build_serving_chat(), engine_client=Engine()) + config = StreamingVideoSessionConfig( + model="test", + modalities=["text", "audio"], + sampling_params_list=[{"temperature": 0, "output_kind": 0}] if explicit_sampling else None, + ) + await handler._process_query_engine( + ws, + config, + [], + bytearray(), + [], + "describe", + "audio-delta", + asyncio.Event(), + {}, + ) + assert not any(message["type"] == "error" for message in ws.sent) + audio = [message for message in ws.sent if message["type"] == "response.output_audio.delta"] + nonempty_groups = sum(any(chunk.numel() for chunk in group) for group in groups) + assert len(audio) == (nonempty_groups if wire_mode == "on" else 1) + np.testing.assert_array_equal( + np.concatenate([_pcm(message["data"]) for message in audio]), + np.arange(_CODEC_FRAME_SAMPLES, sum(lengths)), + ) + assert sum(message["type"] == "response.output_audio.done" for message in ws.sent) == 1 + assert torch.equal(snapshots[0], chunks[0]) + last = torch.cat(snapshots[-1], dim=-1) if isinstance(snapshots[-1], list) else snapshots[-1] + assert torch.equal(last, torch.cat(groups[-1], dim=-1)) + + +def test_cumulative_producer_is_a_different_contract(): + # The fix must not change the producer globally: explicit CUMULATIVE users + # still receive accumulated history, not fresh deltas. + state = _state(RequestOutputKind.CUMULATIVE) + chunks = _chunks() + for chunk in chunks: + state.add_multimodal_tensor(chunk, mm_type="audio") + result = state.make_request_output([], None, None, None) + assert result is not None + assert "audio" in state.mm_accumulated + payload = result.outputs[0].multimodal_output["audio"] + assert isinstance(payload, torch.Tensor) + assert torch.equal(payload, torch.cat(chunks)) diff --git a/tests/entrypoints/test_async_omni.py b/tests/entrypoints/test_async_omni.py index 2379db78e..fdac862ca 100644 --- a/tests/entrypoints/test_async_omni.py +++ b/tests/entrypoints/test_async_omni.py @@ -14,6 +14,8 @@ from tests.helpers.stage_config import get_deploy_config_path from vllm_omni.engine.async_omni_engine import AsyncOmniEngine from vllm_omni.entrypoints import async_omni as async_omni_mod from vllm_omni.entrypoints.async_omni import AsyncOmni +from vllm_omni.entrypoints.utils import coerce_param_message_types +from vllm_omni.model_executor.models.qwen3_omni.pipeline import QWEN3_OMNI_PIPELINE from vllm_omni.outputs import OmniRequestOutput pytestmark = [pytest.mark.core_model] @@ -613,6 +615,46 @@ def test_output_kind_is_preserved_with_explicit_sampling_params(output_kind): assert captured_params[0].output_kind == output_kind +@pytest.mark.cpu +@pytest.mark.parametrize( + "input_kind", [None, RequestOutputKind.CUMULATIVE, RequestOutputKind.DELTA, RequestOutputKind.FINAL_ONLY] +) +def test_qwen_video_sampling_keeps_all_three_stages_delta(input_kind): + """Real generate/resolution preserves the video handler's DELTA contract.""" + captured = [] + + async def submit(*, sampling_params_list, **kwargs): + captured.extend(sampling_params_list) + + async def run(): + omni = get_async_omni_instance(fake_add_request=submit) + # The common fixture bypasses resolution; this test must execute it. + del omni.resolve_sampling_params_list + omni.engine.num_stages = 3 + omni.default_sampling_params_list = [SamplingParams(output_kind=RequestOutputKind.CUMULATIVE) for _ in range(3)] + omni.sampling_constraints_list = omni._get_sampling_constraints_list(QWEN3_OMNI_PIPELINE.stages) + params = None + if input_kind is not None: + # The video handler applies this coercion to explicit overrides. + params = coerce_param_message_types( + [SamplingParams(output_kind=input_kind) for _ in range(3)], is_streaming=True + ) + async for _ in omni.generate( + prompt={"prompt": "video"}, + request_id="video-sampling", + sampling_params_list=params, + output_modalities=["text", "audio"], + ): + pass + assert len(captured) == 3 + assert all(param.output_kind == RequestOutputKind.DELTA for param in captured) + assert captured[1].stop_token_ids == [2150] + assert [param.detokenize for param in captured] == [True, False, True] + assert all(param.output_kind == RequestOutputKind.CUMULATIVE for param in omni.default_sampling_params_list) + + asyncio.run(run()) + + # End to end tests for ensuring internal manipulation of request ID # in diffusion / Omni models don't leak back to the user. # diff --git a/vllm_omni/entrypoints/openai/video_stream_base.py b/vllm_omni/entrypoints/openai/video_stream_base.py index 52ae9d894..e5893545d 100644 --- a/vllm_omni/entrypoints/openai/video_stream_base.py +++ b/vllm_omni/entrypoints/openai/video_stream_base.py @@ -693,8 +693,8 @@ class OmniStreamingVideoHandler: text_parts: list[str] = [] text_done_sent = False audio_chunk_count = 0 - # Number of per-step tensors in OmniRequestOutput.audio_data already - # drained. Used by the fast path to skip already-emitted history. + # Count emitted tensors only to identify the first audio emission. + # Default and explicit sampling parameters use DELTA audio outputs. audio_chunks_drained = 0 previous_text = "" interrupted = False @@ -784,9 +784,9 @@ class OmniStreamingVideoHandler: audio_data = self._get_audio_data(output) if audio_data is not None: if isinstance(audio_data, list): - audio_tail_tensors = list(audio_data) + audio_tail_tensors.extend(audio_data) else: - audio_tail_tensors = [audio_data] + audio_tail_tensors.append(audio_data) else: delta_text, previous_text = self._extract_text_delta( output, @@ -874,16 +874,12 @@ class OmniStreamingVideoHandler: ) -> tuple[str | None, int]: """Return (base64 WAV of new samples, updated chunks_drained). - `chunks_drained` is the number of per-step tensors in - ``audio_data`` that have already been emitted. Each engine step appends - one tensor, so new samples are ``audio_data[chunks_drained:]`` — no - matter how many steps accumulated between reads (handles backpressure - cleanly, unlike a simple ``audio_data[-1]``). - - Two paths, selected at runtime by ``VLLM_VIDEO_AUDIO_DELTA_MODE``: - * fast — only D2H the new tail. Per-call cost ∝ new chunks. - * slow — full cat + D2H each call. Per-call cost ∝ total history. - Retained for A/B; remove once downstream callers confirm. + Explicit sampling parameters are coerced to DELTA above, and AsyncOmni + applies the same coercion to defaults. The output processor drains audio + after each result, so every tensor in the current payload is new. + `chunks_drained` only identifies the first emission for the existing + leading-artifact trim. + The legacy slow setting remains accepted as an equivalent alias. """ audio_data = cls._get_audio_data(result) if audio_data is None: @@ -913,23 +909,20 @@ class OmniStreamingVideoHandler: audio_data, chunks_drained: int, ) -> tuple[str | None, int]: - """Emit only tensors appended since the last call.""" - # Single tensor: output_processor hands us one tensor before it becomes a - # list (see output_processor.py:89). Treat it as chunk #0. - if not isinstance(audio_data, list): - if chunks_drained >= 1: - return None, chunks_drained - tail_np = cls._tensor_to_1d_np(audio_data) - return cls._encode_tail(tail_np, chunks_drained, new_drained=1, is_first=True) - - n = len(audio_data) - if n <= chunks_drained: + """Encode every fresh tensor in this DELTA payload.""" + new_chunks = audio_data if isinstance(audio_data, list) else [audio_data] + if not new_chunks: return None, chunks_drained - - new_chunks = audio_data[chunks_drained:] tail = new_chunks[0] if len(new_chunks) == 1 else torch.cat(new_chunks, dim=-1) tail_np = cls._tensor_to_1d_np(tail) - return cls._encode_tail(tail_np, chunks_drained, new_drained=n, is_first=(chunks_drained == 0)) + if tail_np is None or len(tail_np) == 0: + return None, chunks_drained + return cls._encode_tail( + tail_np, + chunks_drained, + new_drained=chunks_drained + len(new_chunks), + is_first=(chunks_drained == 0), + ) @classmethod def _delta_slow( @@ -937,36 +930,8 @@ class OmniStreamingVideoHandler: audio_data, chunks_drained: int, ) -> tuple[str | None, int]: - """Pre-fix behaviour: concat everything each call and slice on CPU.""" - if isinstance(audio_data, list): - if not audio_data: - return None, chunks_drained - audio_tensor = torch.cat(audio_data, dim=-1) - new_drained = len(audio_data) - else: - audio_tensor = audio_data - new_drained = 1 - - full_np = cls._tensor_to_1d_np(audio_tensor) - if full_np is None: - return None, chunks_drained - # chunks_drained doesn't map directly to sample offset without tracking - # per-chunk lengths, so we re-derive: replay the tail that corresponds - # to chunks appended since last call by slicing off the part produced - # by the already-drained prefix. For slow path this is intentionally - # wasteful — the point is to reproduce the pre-fix hot loop. - if chunks_drained == 0: - tail_np = full_np - else: - # Recover prefix length by re-concatenating the already-drained - # prefix tensors (cost intentionally identical to the baseline - # implementation this was lifted from). - if isinstance(audio_data, list) and chunks_drained < len(audio_data): - prefix_len = sum(int(t.shape[-1]) for t in audio_data[:chunks_drained]) - tail_np = full_np[prefix_len:] - else: - tail_np = full_np[0:0] - return cls._encode_tail(tail_np, chunks_drained, new_drained=new_drained, is_first=(chunks_drained == 0)) + """Compatibility alias for the DELTA payload contract.""" + return cls._delta_fast(audio_data, chunks_drained) @classmethod def _encode_tail(