diff --git a/docs/serving/video_stream_api.md b/docs/serving/video_stream_api.md index 78575e0fc..101289fc0 100644 --- a/docs/serving/video_stream_api.md +++ b/docs/serving/video_stream_api.md @@ -60,10 +60,24 @@ WebSocket /v1/video/chat/stream | `max_frames` | integer, 1-256 | `50` | Maximum retained frame buffer size. Oldest frames are evicted first. | | `system_prompt` | string or null | null | Optional custom system prompt. | | `use_audio_in_video` | bool | `true` | Include streamed audio chunks in multimodal video understanding when audio is present. | -| `sampling_params_list` | list or null | null | Optional per-stage sampling parameter overrides. | +| `sampling_params_list` | list or null | null | Optional per-stage parameter dictionaries. Each provided entry replaces that stage's deployment sampling settings. | | `enable_frame_filter` | bool | `true` | Enable EVS near-duplicate frame filtering. | | `frame_filter_threshold` | float, 0.0-1.0 | `0.95` | EVS similarity threshold. Higher keeps more frames; lower drops more near-duplicates. | +Sampling parameter bugfix: `sampling_params_list` is now forwarded to the engine; +earlier versions accepted this field but silently used deployment defaults instead. +Each provided entry constructs a fresh `SamplingParams` for that stage in pipeline +order; it is not merged with that stage's deployment defaults. Fields absent from a +provided entry use `SamplingParams` constructor defaults. Only omitted trailing +stages keep their deployment defaults. An omitted, `null`, or empty list keeps the +engine defaults. Invalid sampling parameters return an `error` when the query is +submitted. + +For example, `[{"temperature": 0.2, "max_tokens": 64}]` configures the thinker while +the talker and code2wav retain their deployment defaults. With only +`[{"temperature": 0.2}]`, the thinker's `max_tokens` and `top_p` use constructor +defaults, not its YAML values. + ### Legacy Aliases The server accepts these legacy field names and rewrites them before validation. New clients should send the canonical names above. diff --git a/tests/entrypoints/openai_api/test_serving_video_stream.py b/tests/entrypoints/openai_api/test_serving_video_stream.py index 58ea2b57f..f7267a061 100644 --- a/tests/entrypoints/openai_api/test_serving_video_stream.py +++ b/tests/entrypoints/openai_api/test_serving_video_stream.py @@ -12,11 +12,15 @@ import json import threading import weakref from typing import Any +from unittest.mock import AsyncMock, MagicMock import pytest import torch from PIL import Image +from vllm.sampling_params import RequestOutputKind, SamplingParams +from tests.helpers.serving_chat import build_serving_chat +from vllm_omni.config.stage_config import StageConfig from vllm_omni.entrypoints.openai import video_stream_base, video_stream_envs from vllm_omni.entrypoints.openai.serving_video_stream import ( QwenOmniStreamingVideoHandler, @@ -141,6 +145,130 @@ async def test_receive_config_accepts_client_legacy_aliases(): assert config.frame_filter_threshold == 0.87 +@pytest.fixture +def run_sampling_session(monkeypatch): + async def run(config_overrides): + engine = MagicMock() + engine.stage_configs = [ + StageConfig(stage_id=index, model_stage=name) + for index, name in enumerate(("thinker", "talker", "code2wav")) + ] + engine.default_sampling_params_list = [ + SamplingParams(temperature=0.4, max_tokens=64, top_p=0.85), + SamplingParams(temperature=0.7, max_tokens=96), + SamplingParams(temperature=0.8, max_tokens=128), + ] + + async def generate(**kwargs): + yield _text_result("answer") + + engine.generate.side_effect = generate + handler = QwenOmniStreamingVideoHandler( + chat_service=build_serving_chat(engine_client=engine), + engine_client=engine, + ) + monkeypatch.setattr(handler, "_preprocess_to_engine_prompt", AsyncMock(return_value={"prompt": "video"})) + ws = MockWebSocket( + [ + json.dumps( + { + "type": "session.config", + "model": "test", + "modalities": ["text"], + "enable_frame_filter": False, + **config_overrides, + } + ), + json.dumps({"type": "video.frame", "data": _b64(_make_jpeg())}), + json.dumps({"type": "video.query", "text": "describe"}), + json.dumps({"type": "video.done"}), + ] + ) + await handler.handle_session(ws) + return ws, engine + + return run + + +@pytest.mark.asyncio +@pytest.mark.parametrize("async_chunk_mode", ["on", "off"]) +@pytest.mark.parametrize("num_stage_overrides", [1, 3]) +async def test_video_sampling_params_reach_engine( + run_sampling_session, monkeypatch, async_chunk_mode, num_stage_overrides +): + monkeypatch.setenv("VLLM_VIDEO_ASYNC_CHUNK", async_chunk_mode) + overrides = [ + {"temperature": 0.2, "max_tokens": 1, "seed": 42}, + {"temperature": 0.6, "max_tokens": 12}, + {"temperature": 0.9, "max_tokens": 24}, + ][:num_stage_overrides] + + ws, engine = await run_sampling_session({"sampling_params_list": overrides}) + + assert not [msg for msg in ws.sent if msg["type"] == "error"] + engine.generate.assert_called_once() + params = engine.generate.call_args.kwargs["sampling_params_list"] + assert len(params) == 3 + for index, param in enumerate(params): + assert isinstance(param, SamplingParams) + default = engine.default_sampling_params_list[index] + expected = ( + overrides[index] + if index < len(overrides) + else { + "temperature": default.temperature, + "max_tokens": default.max_tokens, + } + ) + assert param.temperature == expected["temperature"] + assert param.max_tokens == expected["max_tokens"] + assert param.output_kind == RequestOutputKind.DELTA + assert param is not default + assert params[0].seed == 42 + assert all(param.output_kind == RequestOutputKind.CUMULATIVE for param in engine.default_sampling_params_list) + assert next(msg for msg in ws.sent if msg["type"] == "response.text.done")["text"] == "answer" + + +@pytest.mark.asyncio +async def test_video_sampling_params_provided_stage_uses_constructor_defaults(run_sampling_session): + ws, engine = await run_sampling_session({"sampling_params_list": [{"temperature": 0.2}]}) + + assert not [msg for msg in ws.sent if msg["type"] == "error"] + engine.generate.assert_called_once() + params = engine.generate.call_args.kwargs["sampling_params_list"] + constructor_defaults = SamplingParams() + assert params[0].temperature == 0.2 + for field in ("max_tokens", "top_p"): + assert getattr(params[0], field) == getattr(constructor_defaults, field) + assert getattr(params[0], field) != getattr(engine.default_sampling_params_list[0], field) + assert len(params) == 3 + for param, default in zip(params[1:], engine.default_sampling_params_list[1:]): + assert param.temperature == default.temperature + assert param.max_tokens == default.max_tokens + assert param.top_p == default.top_p + assert param is not default + + +@pytest.mark.asyncio +@pytest.mark.parametrize("config_overrides", [{}, {"sampling_params_list": None}, {"sampling_params_list": []}]) +async def test_video_sampling_params_omitted_preserve_engine_defaults(run_sampling_session, config_overrides): + ws, engine = await run_sampling_session(config_overrides) + + assert not [msg for msg in ws.sent if msg["type"] == "error"] + engine.generate.assert_called_once() + assert engine.generate.call_args.kwargs.get("sampling_params_list") is None + + +@pytest.mark.asyncio +@pytest.mark.parametrize("invalid_params", [{"temperature": -1}, {"max_tokens": 0}, {"unknown_sampling_option": 1}]) +async def test_video_sampling_params_invalid_do_not_reach_engine(run_sampling_session, invalid_params): + ws, engine = await run_sampling_session({"sampling_params_list": [invalid_params]}) + + engine.generate.assert_not_called() + assert any(msg["type"] == "error" for msg in ws.sent) + assert ws.sent[-1]["type"] == "session.done" + + @pytest.mark.asyncio async def test_video_frame_ack_reports_receiver_buffer_state(): ws = MockWebSocket( diff --git a/vllm_omni/entrypoints/openai/video_stream_base.py b/vllm_omni/entrypoints/openai/video_stream_base.py index 977a0b065..8885e9a33 100644 --- a/vllm_omni/entrypoints/openai/video_stream_base.py +++ b/vllm_omni/entrypoints/openai/video_stream_base.py @@ -51,6 +51,7 @@ from vllm_omni.entrypoints.openai.video_frame_filter import FrameSimilarityFilte from vllm_omni.entrypoints.openai.video_stream_context import ( text_only_message, ) +from vllm_omni.entrypoints.utils import coerce_param_message_types from vllm_omni.outputs import OmniRequestOutput logger = init_logger(__name__) @@ -660,11 +661,13 @@ class OmniStreamingVideoHandler: request_kwargs["mm_processor_kwargs"] = { "use_audio_in_video": True, } - if config.sampling_params_list: - request_kwargs["sampling_params_list"] = config.sampling_params_list - + sampling_kwargs: dict[str, Any] = {} try: chat_request = ChatCompletionRequest(**request_kwargs) + if config.sampling_params_list: + params = self._chat_service._to_sampling_params_list(config.sampling_params_list) + # Explicit params must retain the engine's streaming output mode. + sampling_kwargs["sampling_params_list"] = coerce_param_message_types(params, is_streaming=True) except Exception as e: await self._send_error(websocket, f"Failed to build request: {e}") return @@ -704,6 +707,7 @@ class OmniStreamingVideoHandler: prompt=engine_prompt, request_id=request_id, output_modalities=config.modalities, + **sampling_kwargs, ) async for output in result_gen: