| 1 | # SPDX-License-Identifier: Apache-2.0 |
| 2 | # SPDX-FileCopyrightText: Copyright contributors to the vLLM project |
| 3 | |
| 4 | import asyncio |
| 5 | import json |
| 6 | import warnings |
| 7 | |
| 8 | import numpy as np |
| 9 | import pybase64 as base64 |
| 10 | import pytest |
| 11 | import websockets |
| 12 | |
| 13 | from tests.utils import ROCM_EXTRA_ARGS, RemoteOpenAIServer |
| 14 | from vllm.assets.audio import AudioAsset |
| 15 | from vllm.multimodal.media.audio import load_audio |
| 16 | |
| 17 | # Increase engine iteration timeout for ROCm where first-use JIT compilation |
| 18 | # can exceed the default 60s, causing a silent deadlock in feed_tokens. |
| 19 | REALTIME_ENV_OVERRIDES = { |
| 20 | "VLLM_ENGINE_ITERATION_TIMEOUT_S": "600", |
| 21 | } |
| 22 | |
| 23 | MISTRAL_FORMAT_ARGS = [ |
| 24 | "--tokenizer_mode", |
| 25 | "mistral", |
| 26 | "--config_format", |
| 27 | "mistral", |
| 28 | "--load_format", |
| 29 | "mistral", |
| 30 | ] + ROCM_EXTRA_ARGS |
| 31 | |
| 32 | MODEL_NAME = "mistralai/Voxtral-Mini-4B-Realtime-2602" |
| 33 | |
| 34 | |
| 35 | def _get_websocket_url(server: RemoteOpenAIServer) -> str: |
| 36 | """Convert HTTP URL to WebSocket URL for realtime endpoint.""" |
| 37 | http_url = server.url_root |
| 38 | ws_url = http_url.replace("http://", "ws://") |
| 39 | return f"{ws_url}/v1/realtime" |
| 40 | |
| 41 | |
| 42 | async def receive_event(ws, timeout: float = 60.0) -> dict: |
| 43 | """Receive and parse JSON event from WebSocket.""" |
| 44 | message = await asyncio.wait_for(ws.recv(), timeout=timeout) |
| 45 | return json.loads(message) |
| 46 | |
| 47 | |
| 48 | async def send_event(ws, event: dict) -> None: |
| 49 | """Send JSON event to WebSocket.""" |
| 50 | await ws.send(json.dumps(event)) |
| 51 | |
| 52 | |
| 53 | @pytest.fixture |
| 54 | def mary_had_lamb_audio_chunks() -> list[str]: |
| 55 | """Audio split into ~1 second chunks for streaming.""" |
| 56 | path = AudioAsset("mary_had_lamb").get_local_path() |
| 57 | audio, _ = load_audio(str(path), sr=16000, mono=True) |
| 58 | |
| 59 | # Split into ~0.1 second chunks (1600 samples at 16kHz) |
| 60 | chunk_size = 1600 |
| 61 | chunks = [] |
| 62 | for i in range(0, len(audio), chunk_size): |
| 63 | chunk = audio[i : i + chunk_size] |
| 64 | chunk_int16 = (chunk * 32767).astype(np.int16) |
| 65 | chunk_bytes = chunk_int16.tobytes() |
| 66 | chunks.append(base64.b64encode(chunk_bytes).decode("utf-8")) |
| 67 | |
| 68 | return chunks |
| 69 | |
| 70 | |
| 71 | @pytest.mark.asyncio |
| 72 | @pytest.mark.parametrize("model_name", [MODEL_NAME]) |
| 73 | async def test_multi_chunk_streaming(model_name, mary_had_lamb_audio_chunks): |
| 74 | """Test streaming multiple audio chunks before committing.""" |
| 75 | server_args = ["--enforce-eager", "--max-model-len", "2048"] |
| 76 | |
| 77 | if model_name.startswith("mistralai"): |
| 78 | server_args += MISTRAL_FORMAT_ARGS |
| 79 | |
| 80 | with RemoteOpenAIServer( |
| 81 | model_name, server_args, env_dict=REALTIME_ENV_OVERRIDES |
| 82 | ) as remote_server: |
| 83 | ws_url = _get_websocket_url(remote_server) |
| 84 | async with websockets.connect(ws_url) as ws: |
| 85 | # Receive session.created |
| 86 | event = await receive_event(ws, timeout=30.0) |
| 87 | assert event["type"] == "session.created" |
| 88 | |
| 89 | await send_event(ws, {"type": "session.update", "model": model_name}) |
| 90 | |
| 91 | # Wait for the server to acknowledge the session update. |
| 92 | try: |
| 93 | while True: |
| 94 | event = await receive_event(ws, timeout=5.0) |
| 95 | if event["type"] == "session.updated": |
| 96 | break |
| 97 | except TimeoutError: |
| 98 | warnings.warn( |
| 99 | f"session.updated not received within {5.0}s after " |
| 100 | "session.update. The server may not implement this event.", |
| 101 | stacklevel=2, |
| 102 | ) |
| 103 | |
| 104 | # (ROCm) Warm-up: send a non-final commit (required to start |
| 105 | # transcription) with a small audio chunk to trigger aiter |
| 106 | # compilation on first use. |
| 107 | await send_event(ws, {"type": "input_audio_buffer.commit"}) |
| 108 | await send_event( |
| 109 | ws, |
| 110 | { |
| 111 | "type": "input_audio_buffer.append", |
| 112 | "audio": mary_had_lamb_audio_chunks[0], |
| 113 | }, |
| 114 | ) |
| 115 | await send_event(ws, {"type": "input_audio_buffer.commit", "final": True}) |
| 116 | |
| 117 | # (ROCm) Drain all warm-up responses with generous timeout for |
| 118 | # JIT compilation |
| 119 | warmup_done = False |
| 120 | while not warmup_done: |
| 121 | event = await receive_event(ws, timeout=600.0) |
| 122 | if event["type"] in ("transcription.done", "error"): |
| 123 | warmup_done = True |
| 124 | |
| 125 | # Now send the real test audio |
| 126 | await send_event(ws, {"type": "input_audio_buffer.commit"}) |
| 127 | |
| 128 | # Send multiple audio chunks |
| 129 | for chunk in mary_had_lamb_audio_chunks: |
| 130 | await send_event( |
| 131 | ws, {"type": "input_audio_buffer.append", "audio": chunk} |
| 132 | ) |
| 133 | |
| 134 | # Send commit to end |
| 135 | await send_event(ws, {"type": "input_audio_buffer.commit", "final": True}) |
| 136 | |
| 137 | # Collect transcription deltas |
| 138 | full_text = "" |
| 139 | done_received = False |
| 140 | |
| 141 | while not done_received: |
| 142 | event = await receive_event(ws, timeout=60.0) |
| 143 | |
| 144 | if event["type"] == "transcription.delta": |
| 145 | full_text += event["delta"] |
| 146 | elif event["type"] == "transcription.done": |
| 147 | done_received = True |
| 148 | assert "text" in event |
| 149 | elif event["type"] == "error": |
| 150 | pytest.fail(f"Received error: {event}") |
| 151 | |
| 152 | # Verify transcription contains expected content |
| 153 | assert event["type"] == "transcription.done" |
| 154 | assert event["text"] == full_text |
| 155 | assert full_text == ( |
| 156 | " First words I spoke in the original phonograph." |
| 157 | " A little piece of practical poetry. Mary had a little lamb," |
| 158 | " it sleeps with quite a flow, and everywhere that Mary went," |
| 159 | " the lamb was sure to go." |
| 160 | ) or full_text == ( |
| 161 | " First words I spoke in the original phonograph." |
| 162 | " A little piece of practical poetry. Mary had a little lamb," |
| 163 | " it squeaked with quite a flow, and everywhere that Mary went," |
| 164 | " the lamb was sure to go." |
| 165 | ) |
| 166 | |
| 167 | |
| 168 | @pytest.mark.asyncio |
| 169 | @pytest.mark.parametrize("model_name", [MODEL_NAME]) |
| 170 | async def test_empty_commit_does_not_crash_engine( |
| 171 | model_name, mary_had_lamb_audio_chunks |
| 172 | ): |
| 173 | """Test that committing without audio does not crash the engine. |
| 174 | |
| 175 | Regression test for https://github.com/vllm-project/vllm/issues/34532. |
| 176 | An empty commit (no prior input_audio_buffer.append) used to trigger |
| 177 | ``AssertionError: For realtime you must provide a multimodal_embedding |
| 178 | at every step`` which killed the entire engine process, disconnecting |
| 179 | every connected client. |
| 180 | """ |
| 181 | server_args = ["--enforce-eager", "--max-model-len", "2048"] |
| 182 | |
| 183 | if model_name.startswith("mistralai"): |
| 184 | server_args += MISTRAL_FORMAT_ARGS |
| 185 | |
| 186 | with RemoteOpenAIServer( |
| 187 | model_name, server_args, env_dict=REALTIME_ENV_OVERRIDES |
| 188 | ) as remote_server: |
| 189 | ws_url = _get_websocket_url(remote_server) |
| 190 | |
| 191 | # --- First connection: empty commit (no audio appended) ---------- |
| 192 | async with websockets.connect(ws_url) as ws: |
| 193 | event = await receive_event(ws, timeout=30.0) |
| 194 | assert event["type"] == "session.created" |
| 195 | |
| 196 | await send_event(ws, {"type": "session.update", "model": model_name}) |
| 197 | |
| 198 | try: |
| 199 | while True: |
| 200 | event = await receive_event(ws, timeout=5.0) |
| 201 | if event["type"] == "session.updated": |
| 202 | break |
| 203 | except TimeoutError: |
| 204 | warnings.warn( |
| 205 | f"session.updated not received within {5.0}s after " |
| 206 | "session.update. The server may not implement this event.", |
| 207 | stacklevel=2, |
| 208 | ) |
| 209 | |
| 210 | # Start generation without sending any audio |
| 211 | await send_event(ws, {"type": "input_audio_buffer.commit"}) |
| 212 | |
| 213 | # Immediately signal end-of-audio |
| 214 | await send_event(ws, {"type": "input_audio_buffer.commit", "final": True}) |
| 215 | |
| 216 | # We should get *some* response (error or empty transcription), |
| 217 | # but the engine must NOT crash. |
| 218 | # (ROCm) Use generous timeout for first request (aiter JIT compilation) |
| 219 | event = await receive_event(ws, timeout=360.0) |
| 220 | assert event["type"] in ( |
| 221 | "error", |
| 222 | "transcription.done", |
| 223 | "transcription.delta", |
| 224 | ) |
| 225 | |
| 226 | # --- Second connection: normal transcription --------------------- |
| 227 | # Verifies the engine is still alive after the empty commit above. |
| 228 | async with websockets.connect(ws_url) as ws: |
| 229 | event = await receive_event(ws, timeout=30.0) |
| 230 | assert event["type"] == "session.created" |
| 231 | |
| 232 | await send_event(ws, {"type": "session.update", "model": model_name}) |
| 233 | |
| 234 | try: |
| 235 | while True: |
| 236 | event = await receive_event(ws, timeout=5.0) |
| 237 | if event["type"] == "session.updated": |
| 238 | break |
| 239 | except TimeoutError: |
| 240 | warnings.warn( |
| 241 | f"session.updated not received within {5.0}s after " |
| 242 | "session.update. The server may not implement this event.", |
| 243 | stacklevel=2, |
| 244 | ) |
| 245 | |
| 246 | # Start transcription |
| 247 | await send_event(ws, {"type": "input_audio_buffer.commit"}) |
| 248 | |
| 249 | for chunk in mary_had_lamb_audio_chunks: |
| 250 | await send_event( |
| 251 | ws, {"type": "input_audio_buffer.append", "audio": chunk} |
| 252 | ) |
| 253 | |
| 254 | await send_event(ws, {"type": "input_audio_buffer.commit", "final": True}) |
| 255 | |
| 256 | done_received = False |
| 257 | while not done_received: |
| 258 | event = await receive_event(ws, timeout=60.0) |
| 259 | if event["type"] == "transcription.done": |
| 260 | done_received = True |
| 261 | elif event["type"] == "error": |
| 262 | pytest.fail(f"Engine error after empty commit: {event}") |
| 263 | assert done_received |
| 264 | |
| 265 | |
| 266 | @pytest.mark.asyncio |
| 267 | @pytest.mark.parametrize("model_name", [MODEL_NAME]) |
| 268 | async def test_session_update_invalid_model_returns_error(model_name): |
| 269 | """Test that session.update with an invalid model returns an error.""" |
| 270 | server_args = ["--enforce-eager", "--max-model-len", "2048"] |
| 271 | |
| 272 | if model_name.startswith("mistralai"): |
| 273 | server_args += MISTRAL_FORMAT_ARGS |
| 274 | |
| 275 | with RemoteOpenAIServer( |
| 276 | model_name, server_args, env_dict=REALTIME_ENV_OVERRIDES |
| 277 | ) as remote_server: |
| 278 | ws_url = _get_websocket_url(remote_server) |
| 279 | async with websockets.connect(ws_url) as ws: |
| 280 | event = await receive_event(ws, timeout=30.0) |
| 281 | assert event["type"] == "session.created" |
| 282 | |
| 283 | # Send session.update with a model that doesn't exist |
| 284 | await send_event( |
| 285 | ws, |
| 286 | {"type": "session.update", "model": "nonexistent-model"}, |
| 287 | ) |
| 288 | |
| 289 | event = await receive_event(ws, timeout=10.0) |
| 290 | assert event["type"] == "error" |
| 291 | assert "nonexistent-model" in event["error"] |
| 292 | |
| 293 | |
| 294 | @pytest.mark.asyncio |
| 295 | @pytest.mark.parametrize("model_name", [MODEL_NAME]) |
| 296 | async def test_commit_without_session_update_returns_error(model_name): |
| 297 | """Test that committing before validating the model returns an error |
| 298 | and does not fall through to processing.""" |
| 299 | server_args = ["--enforce-eager", "--max-model-len", "2048"] |
| 300 | |
| 301 | if model_name.startswith("mistralai"): |
| 302 | server_args += MISTRAL_FORMAT_ARGS |
| 303 | |
| 304 | with RemoteOpenAIServer( |
| 305 | model_name, server_args, env_dict=REALTIME_ENV_OVERRIDES |
| 306 | ) as remote_server: |
| 307 | ws_url = _get_websocket_url(remote_server) |
| 308 | async with websockets.connect(ws_url) as ws: |
| 309 | event = await receive_event(ws, timeout=30.0) |
| 310 | assert event["type"] == "session.created" |
| 311 | |
| 312 | # Send commit without sending session.update first |
| 313 | await send_event( |
| 314 | ws, |
| 315 | {"type": "input_audio_buffer.commit", "final": True}, |
| 316 | ) |
| 317 | |
| 318 | event = await receive_event(ws, timeout=10.0) |
| 319 | assert event["type"] == "error" |
| 320 | assert "model_not_validated" in event.get("code", "") |