diff --git a/docs/serving/realtime_duplex_api.md b/docs/serving/realtime_duplex_api.md index c18b416c7..a20750090 100644 --- a/docs/serving/realtime_duplex_api.md +++ b/docs/serving/realtime_duplex_api.md @@ -768,7 +768,10 @@ events with `server_event_seq > 40` are replayed in order) ``` with camera frames (omni video; rides the append that closes a 1 s model -unit; each entry is a **bare** base64 JPEG/PNG — no `data:` URL prefix) +unit; each entry is a **bare** base64 JPEG/PNG — no `data:` URL prefix). +Only a model whose Stage 0 interleaves a frame track at unit boundaries takes +this field; a turn model has no such boundary and rejects it, expecting +`conversation.item.create` with `input_image` content instead. ```json { diff --git a/examples/online_serving/minicpmo/README.md b/examples/online_serving/minicpmo/README.md index 67b268a77..ffd50167f 100644 --- a/examples/online_serving/minicpmo/README.md +++ b/examples/online_serving/minicpmo/README.md @@ -316,3 +316,12 @@ them unless `--allow-invalid-clock` is explicit. [`examples/offline_inference/minicpmo/`](../../offline_inference/minicpmo/) - Recipe: [`recipes/OpenBMB/MiniCPM-o-4_5.md`](../../../recipes/OpenBMB/MiniCPM-o-4_5.md) + +### Shared realtime UI implementation + +The `python -m examples.online_serving.minicpmo.realtime_web` command remains +available with the same MiniCPM defaults and required `--ref-audio`. Its assets +now live in [the shared realtime UI](../realtime_web/README.md), with a dedicated +`minicpm-native` profile preserving native duplex input, playback ACKs and camera +frames. Qwen3 uses a separate profile in the same shell, supporting manual STT +turns or Server VAD turns with speech interruption. diff --git a/examples/online_serving/minicpmo/realtime_web/server.py b/examples/online_serving/minicpmo/realtime_web/server.py index ee78fff74..1ef12d186 100644 --- a/examples/online_serving/minicpmo/realtime_web/server.py +++ b/examples/online_serving/minicpmo/realtime_web/server.py @@ -1,185 +1,23 @@ -"""Static host and same-origin Realtime proxy for the full-duplex demo.""" +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project -from __future__ import annotations +"""Backward-compatible MiniCPM entry point for the shared realtime UI.""" -import argparse -import asyncio -import base64 -import contextlib -import hashlib -import json -import logging -from pathlib import Path -from urllib.parse import urlencode +from examples.online_serving.realtime_web.server import ( + APP_DIR, + STATIC_DIR, + _join_ws_url, + build_app, +) +from examples.online_serving.realtime_web.server import ( + main as _main, +) -import uvicorn -import websockets -from fastapi import FastAPI, WebSocket, WebSocketDisconnect -from fastapi.responses import HTMLResponse, Response -from fastapi.staticfiles import StaticFiles - -logger = logging.getLogger(__name__) -APP_DIR = Path(__file__).parent / "app" -STATIC_DIR = APP_DIR / "static" - - -def _join_ws_url(base: str, path: str, query: str) -> str: - return base.rstrip("/") + path + (("?" + query) if query else "") - - -async def _pump_client_to_backend(client: WebSocket, backend) -> None: - try: - while True: - message = await client.receive() - if message["type"] == "websocket.disconnect": - await backend.close() - return - if message.get("text") is not None: - await backend.send(message["text"]) - elif message.get("bytes") is not None: - await backend.send(message["bytes"]) - except (WebSocketDisconnect, websockets.ConnectionClosed): - with contextlib.suppress(Exception): - await backend.close() - - -async def _pump_backend_to_client(client: WebSocket, backend) -> None: - try: - async for message in backend: - if isinstance(message, bytes): - await client.send_bytes(message) - else: - await client.send_text(message) - except (WebSocketDisconnect, websockets.ConnectionClosed): - return - except RuntimeError as exc: - if "websocket.send" in str(exc) and "websocket.close" in str(exc): - return - raise - - -def _expected_proxy_close(exc: BaseException) -> bool: - if isinstance(exc, (WebSocketDisconnect, websockets.ConnectionClosed, asyncio.CancelledError)): - return True - return isinstance(exc, RuntimeError) and "websocket.send" in str(exc) and "websocket.close" in str(exc) - - -def build_app( - *, - ws_backend: str = "ws://127.0.0.1:8099", - model: str = "openbmb/MiniCPM-o-4_5", - public_realtime_url: str | None = None, - ref_audio: str | None = None, -) -> FastAPI: - app = FastAPI(title="Experimental Full-Duplex Web Demo") - index_path = APP_DIR / "index.html" - app_version_hash = hashlib.sha256() - for asset_path in ( - STATIC_DIR / "app.js", - STATIC_DIR / "pcm_worklet.js", - STATIC_DIR / "playback_worklet.js", - ): - app_version_hash.update(asset_path.read_bytes()) - app_version = app_version_hash.hexdigest()[:12] - - ref_audio_uri: str | None = None - if ref_audio: - ref_path = Path(ref_audio) - if not ref_path.is_file(): - raise SystemExit(f"--ref-audio not found: {ref_audio}") - encoded = base64.b64encode(ref_path.read_bytes()).decode("ascii") - ref_audio_uri = f"data:audio/wav;base64,{encoded}" - - @app.get("/", response_class=HTMLResponse) - def index() -> HTMLResponse: - config = json.dumps( - { - "model": model, - "realtimePath": public_realtime_url or "v1/realtime", - "refAudio": ref_audio_uri, - "appVersion": app_version, - }, - ensure_ascii=True, - ) - html = ( - index_path.read_text(encoding="utf-8") - .replace("__FULL_DUPLEX_CONFIG__", config) - .replace("__FULL_DUPLEX_APP_VERSION__", app_version) - ) - return HTMLResponse(html, headers={"Cache-Control": "no-store"}) - - @app.get("/healthz") - def healthz() -> Response: - return Response(content="ok", media_type="text/plain") - - @app.websocket("/v1/realtime") - async def realtime_proxy(websocket: WebSocket) -> None: - await websocket.accept() - query = urlencode(websocket.query_params.multi_items()) - backend_url = _join_ws_url(ws_backend, "/v1/realtime", query) - logger.info("Proxying Realtime WebSocket to %s", backend_url) - try: - async with websockets.connect( - backend_url, - max_size=64 * 1024 * 1024, - ) as backend: - tasks = { - asyncio.create_task(_pump_client_to_backend(websocket, backend)), - asyncio.create_task(_pump_backend_to_client(websocket, backend)), - } - done, pending = await asyncio.wait( - tasks, - return_when=asyncio.FIRST_COMPLETED, - ) - for task in pending: - task.cancel() - await asyncio.gather(*pending, return_exceptions=True) - for result in await asyncio.gather(*done, return_exceptions=True): - if isinstance(result, BaseException) and not _expected_proxy_close(result): - raise result - except (WebSocketDisconnect, websockets.ConnectionClosed): - return - except Exception: - logger.exception("Realtime WebSocket proxy failed") - with contextlib.suppress(Exception): - await websocket.close(code=1011) - - app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") - return app +__all__ = ["APP_DIR", "STATIC_DIR", "_join_ws_url", "build_app", "main"] def main() -> None: - parser = argparse.ArgumentParser(description=__doc__) - parser.add_argument("--host", default="0.0.0.0") - parser.add_argument("--port", type=int, default=7862) - parser.add_argument("--ws-backend", default="ws://127.0.0.1:8099") - parser.add_argument( - "--public-realtime-url", - help="Browser-visible ws:// or wss:// Realtime URL; defaults to the same-origin proxy.", - ) - parser.add_argument("--model", default="openbmb/MiniCPM-o-4_5") - parser.add_argument( - "--ref-audio", - required=True, - help=( - "Reference voice wav for TTS voice cloning, e.g. the " - "official MiniCPM-o-Demo assets/ref_audio/ref_minicpm_signature.wav" - ), - ) - args = parser.parse_args() - - logging.basicConfig(level=logging.INFO) - uvicorn.run( - build_app( - ws_backend=args.ws_backend, - model=args.model, - public_realtime_url=args.public_realtime_url, - ref_audio=args.ref_audio, - ), - host=args.host, - port=args.port, - log_level="info", - ) + _main(default_profile="minicpm-native") if __name__ == "__main__": diff --git a/examples/online_serving/qwen3_omni/README.md b/examples/online_serving/qwen3_omni/README.md index 8d6014db2..5eb9a2042 100644 --- a/examples/online_serving/qwen3_omni/README.md +++ b/examples/online_serving/qwen3_omni/README.md @@ -320,24 +320,17 @@ python openai_realtime_client.py \ Server VAD uses the pinned Silero v6.2 ONNX artifact from `istupakov/silero-vad-onnx` and never downloads model files while processing a session. Make the artifact available in the Hugging Face cache before startup, or configure a local artifact in the deployment YAML: ```yaml +base_config: /path/to/vllm-omni/vllm_omni/deploy/qwen3_omni_duplex.yaml duplex_session: server_vad_model_path: /models/silero_vad.onnx ``` -Start from the bundled Qwen3-Omni deployment YAML, add the fields above at the top level, and serve that complete configuration: - -```bash -cp vllm_omni/deploy/qwen3_omni_moe.yaml /path/to/qwen3_omni_server_vad.yaml -# Edit /path/to/qwen3_omni_server_vad.yaml and add duplex_session. -vllm serve Qwen/Qwen3-Omni-30B-A3B-Instruct \ - --omni \ - --port 8091 \ - --deploy-config /path/to/qwen3_omni_server_vad.yaml -``` - -Keep Qwen's default `session_mode: turn`; `duplex_session` enables the Realtime handler without changing scheduler -semantics. Bare `/v1/realtime` selects that handler. The client uses `?duplex=0` only for the existing non-Server-VAD -wire flow; `?duplex=1` remains a supported compatibility alias. +For the engine-owned plugin, use an overlay based on +`vllm_omni/deploy/qwen3_omni_duplex.yaml`, with `duplex_session.server_vad_model_path` +pointing to your local artifact. See the [complete duplex setup](../realtime_web/README.md#qwen3-with-server-vad-automatic-turns). +The default `qwen3_omni_moe.yaml` remains a turn deployment. Duplex mode serves +`/v1/realtime?duplex=1` and `/v1/chat/completions`; the legacy STT handler requires +a turn deployment. The Python client supports the following command-line arguments: @@ -584,3 +577,53 @@ The gradio script supports the following arguments: - `--ip`: Host/IP for Gradio server (default: 127.0.0.1) - `--port`: Port for Gradio server (default: 7861) - `--share`: Share the Gradio demo publicly (creates a public link) + +## Browser voice call (shared realtime UI) + +Run these commands from the repository root with the vLLM-Omni environment activated. + +### Without VAD: manually submit each turn + +Start the backend: + +```bash +vllm serve Qwen/Qwen3-Omni-30B-A3B-Instruct --omni --port 8091 +``` + +In another terminal, check backend readiness and start the UI: + +```bash +curl --fail http://127.0.0.1:8091/health +python -m examples.online_serving.qwen3_omni.realtime_web \ + --backend ws://127.0.0.1:8091 --stt --port 7863 +``` + +Open `http://localhost:7863`, start a session, speak, and press **Send turn**. +`--stt` is the default. Each turn uses a separate connection without shared model +history; the conversation shown in the browser is a local log. + +### With VAD: automatically submit after silence + +Start the backend with a Server VAD deployment overlay and a compatible Silero +ONNX artifact, following the +[complete VAD setup](../realtime_web/README.md#qwen3-with-server-vad-automatic-turns). +After its health check succeeds, start the UI: + +```bash +python -m examples.online_serving.qwen3_omni.realtime_web \ + --backend ws://127.0.0.1:8091 --vad --port 7863 +``` + +Speak and pause for 500 ms to submit automatically. `--vad` configures the client; +it does not load a VAD model or change the backend deployment by itself. + +VAD mode keeps uploading audio during replies and supports speech interruption. +The engine-owned Qwen duplex plugin maintains conversation history and tracks +played replies through playback acknowledgements. Each committed utterance starts +a new generation request; the model does not use native streaming-input KV decoding. +STT mode pauses microphone upload while the reply is generated and played. +In VAD mode, click **Camera** to upload sampled frames with your spoken question; +see [camera setup and limits](../realtime_web/README.md#qwen-vad-camera-input). +STT remains audio-only. To switch back to manual turns, use the default turn +deployment, restart the UI with `--stt`, refresh the page, and reconnect. Stop the existing UI before reusing port 7863. +See the [shared UI guide](../realtime_web/README.md) for HTTPS access and testing. diff --git a/examples/online_serving/qwen3_omni/openai_realtime_client.py b/examples/online_serving/qwen3_omni/openai_realtime_client.py index 0a98d69a0..2a6bca0fe 100644 --- a/examples/online_serving/qwen3_omni/openai_realtime_client.py +++ b/examples/online_serving/qwen3_omni/openai_realtime_client.py @@ -130,7 +130,7 @@ async def run_client( "prefix_padding_ms": 300, "silence_duration_ms": 500, "create_response": True, - "interrupt_response": False, + "interrupt_response": True, }, } }, @@ -233,6 +233,7 @@ async def run_client( continue if event_type == "response.done" and server_vad: + await ws.send(json.dumps({"type": "session.close"})) break if event_type == "error": diff --git a/examples/online_serving/qwen3_omni/realtime_web/__init__.py b/examples/online_serving/qwen3_omni/realtime_web/__init__.py new file mode 100644 index 000000000..14c5daad2 --- /dev/null +++ b/examples/online_serving/qwen3_omni/realtime_web/__init__.py @@ -0,0 +1,4 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +"""Qwen3-Omni turn-based browser call UI.""" diff --git a/examples/online_serving/qwen3_omni/realtime_web/__main__.py b/examples/online_serving/qwen3_omni/realtime_web/__main__.py new file mode 100644 index 000000000..698179528 --- /dev/null +++ b/examples/online_serving/qwen3_omni/realtime_web/__main__.py @@ -0,0 +1,7 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +from examples.online_serving.realtime_web.server import main + +if __name__ == "__main__": + main(default_profile="qwen3-turn") diff --git a/examples/online_serving/realtime_web/README.md b/examples/online_serving/realtime_web/README.md new file mode 100644 index 000000000..854aa4718 --- /dev/null +++ b/examples/online_serving/realtime_web/README.md @@ -0,0 +1,221 @@ +# Shared Realtime Web UI + +One browser shell provides microphone capture, speaker playback, a level meter, +conversation text, and an event log. Model profiles own the wire protocol. +This implements [RFC #7222](https://github.com/vllm-project/vllm-omni/issues/7222). + +| Profile | Turn control | Camera | Playback acknowledgement | +| --- | --- | --- | --- | +| `minicpm-native` | Model-controlled listen/speak, continuous audio input | Frames accompany audio | Yes | +| `qwen3-turn --stt` (default) | User presses **Send turn** | No | No | +| `qwen3-turn --vad` | Server detects trailing silence; speech interrupts replies | Sampled frames with each spoken turn | Yes | + +Qwen3 VAD uses an engine-owned duplex plugin: microphone upload continues +while replies stream, and speech can interrupt generation and playback. The +checkpoint still generates committed turns; this is not native streaming-input +KV decoding. STT remains explicit-turn and audio-only. + +## MiniCPM: existing command stays valid + +Start your MiniCPM backend using the [existing deployment instructions](../minicpmo/README.md), then: + +```bash +python -m examples.online_serving.minicpmo.realtime_web \ + --ws-backend ws://127.0.0.1:8099 \ + --ref-audio /path/to/ref_minicpm_signature.wav +``` + +The compatibility wrapper serves the shared assets with `minicpm-native`. +The current native query, `extra_body`, reference voice, continuous microphone +upload, camera frames and playback acknowledgements are retained. + +## Qwen3: explicit-turn STT adapter + +Start the backend, then the UI in a second terminal: + +```bash +vllm serve Qwen/Qwen3-Omni-30B-A3B-Instruct --omni --port 8091 + +python -m examples.online_serving.qwen3_omni.realtime_web \ + --backend ws://127.0.0.1:8091 --stt --port 7863 +``` + +Open `http://localhost:7863`, start a session, speak, and press **Send turn**. +Wait for the answer to finish playing before speaking again. The UI opens a +fresh STT connection for each turn; the displayed conversation is a local log, +and **previous turns are not provided as model history**. The STT endpoint +accepts only model selection, so system-prompt controls are hidden in this mode. + +The adapter follows the shipped `qwen3_omni/openai_realtime_client.py` and +`vllm_omni/entrypoints/openai/realtime_connection.py`: + +1. Explicit `duplex=0` selects the legacy STT handler on a turn deployment. + It does not enable STT on a duplex deployment. +2. Send `{type: "session.update", model: ...}` and `commit(final=false)`. +3. Stream mono PCM16 at 16 kHz; **Send turn** flushes buffered audio, then sends `commit(final=true)`. +4. `response.output_audio.delta.audio` contains PCM; `response.output_audio.done` + terminates this adapter, including a response with no audio. +5. The local implementation emits **model output** on `transcription.*`, so the + UI displays it as assistant text. This differs from the early RFC's suggested + user-transcript mapping. The VAD adapter also supports `response.output_text.*` + and audio-transcript events without duplicating the answer. + +## Qwen3 with Server VAD (automatic turns) + +Run commands from the repository root with the vLLM-Omni environment activated. +Prepare the pinned Silero v6.2 ONNX artifact as described in the +[Qwen3 Server VAD instructions](../qwen3_omni/README.md#realtime-websocket-client-openai_realtime_clientpy). +The backend requires ONNX Runtime and a compatible local artifact; setting +`--vad` on the UI alone does not enable VAD on the backend. + +Create an overlay named `qwen3_vad.yaml`, replacing both paths with absolute paths +on the backend host: + +```yaml +base_config: /path/to/vllm-omni/vllm_omni/deploy/qwen3_omni_duplex.yaml +duplex_session: + server_vad_model_path: /path/to/silero_vad.onnx +``` + +The base configuration enables duplex mode and sets the per-prompt audio and image limits. + +Start the backend: + +```bash +vllm serve Qwen/Qwen3-Omni-30B-A3B-Instruct --omni --port 8091 \ + --deploy-config qwen3_vad.yaml +``` + +In a second terminal, wait until the health check succeeds, then start the UI: + +```bash +curl --fail http://127.0.0.1:8091/health +python -m examples.online_serving.qwen3_omni.realtime_web \ + --backend ws://127.0.0.1:8091 --vad --port 7863 +``` + +Open `http://localhost:7863` and start a session. Speak and pause: the server +commits the turn automatically after 500 ms of silence. No **Send turn** action +is needed. Defaults are a speech threshold of 0.5 and 300 ms of prefix padding. +Speak again to interrupt an answer. The engine owns input buffering, VAD, +request cancellation, playback acknowledgements, and conversation history. +Each committed utterance starts a normal Thinker → Talker → Code2Wav request; +it does not append input to a live Qwen KV cache. The capability response reports +`implementation_level: turn_based_duplex` and `supports_core_resumable_request: false`. + +The plugin retains up to four audio inputs (including the current turn), pruning +older audio at an 8 MiB base64 budget; one current utterance is bounded by the +engine's pending-input limit. The UI reports playback progress both on normal +drain and on interruption. Qwen does not provide text/audio alignment, so assistant +text enters history only after the complete audio response has been generated and +acknowledged as played. A partially played answer contributes no guessed text +prefix. Image conversation items remain available until deleted. Model context limits still apply. +The bundled duplex deploy allows four audio inputs and eight images per prompt. +Use headphones to avoid speaker audio triggering VAD interruption. Reference +voices and tool calls are not supported by this plugin. + +### Switching between VAD and manual turns + +| Mode | Backend | UI flag | Submit a turn | +| --- | --- | --- | --- | +| Without VAD | Default Qwen turn deployment | `--stt` (default) | Press **Send turn** | +| With VAD | Server VAD configuration and Silero artifact | `--vad` | Pause after speaking | + +Stop the existing UI process before starting another on the same port, then +refresh the browser and reconnect. To use STT, start a backend with the default +turn deployment and restart the UI with `--stt`. The duplex backend exposes +Realtime duplex and ChatCompletion; it does not expose the legacy STT handler. +To enable VAD on a default backend, first restart that backend with the overlay. +Make `--backend` match the actual backend port; the UI and backend use separate +ports. `Connection refused` means the target backend is unavailable: check its +startup log and `/health` before connecting. + +VAD connects with `duplex=1` and sends nested `session.audio.input` configuration +with `create_response: true` and `interrupt_response: true`. It sends playback +ACKs when speaker audio drains. `response.audio.done` / `response.output_audio.done` +only drain playback; `response.done` terminates a turn. Backpressure displays a +notice without pausing the microphone. Failed handshakes time out after 15 seconds; +Silero initialization failures display a specific error. + +After updating the backend Python code, restart the backend, refresh the UI, and +start a new session so the plugin and browser assets use the updated code. A +successful `/health` check confirms readiness, not which deployment mode is active. + +## Qwen VAD camera input + +After starting a session with `--vad`, click **Camera** and allow access. The UI +sends one JPEG per second, resized to a longest side of 448 pixels, independently +of microphone uploads (including while muted). Each frame is a user conversation +item using the [OpenAI Realtime image input](https://developers.openai.com/api/docs/guides/realtime-conversations#image-inputs) shape: + +```json +{"type":"conversation.item.create","item":{"id":"camera_1","type":"message","role":"user","content":[{"type":"input_image","image_url":"data:image/jpeg;base64,..."}]}} +``` + +Images become conversation context immediately; they do not start a response. +Speak and let VAD commit your audio, or send `response.create` after adding an +image/text item. A manual `input_audio_buffer.commit` only commits audio; follow +it with `response.create` to generate. Images persist across audio commits, +interruptions, and audio-buffer clears. Remove them with `conversation.item.delete`. +The browser retains eight camera items and deletes the oldest before adding a +ninth. Camera-off stops capture but leaves those items in the conversation. + +This deployment accepts JPEG/PNG Base64 data URLs with a total image-context +budget of eight images and 4 MiB; external image URLs are not supported. Clients +must delete old image items when reaching the limit. There is no precise +video/audio timestamp alignment or native streaming-video KV update. + +### Protocol compatibility boundary + +Qwen VAD uses `session.update`, 24 kHz PCM16 `input_audio_buffer.append`, +`conversation.item.create` with `input_image`/`input_text`, and `response.create`. +The engine resamples input to Qwen's 16 kHz processing rate. Qwen rejects the +custom `video_frames` field on an append: it answers whole turns, so it has no +audio unit for a frame track to align to. MiniCPM keeps that field for its +native camera protocol, where Stage 0 interleaves frames at unit boundaries. + +This is a supported subset, not full OpenAI Realtime API compatibility. +`duplex=1`, `overlap_policy`, `playback.ack`, and `session.close` remain vLLM-Omni +extensions. The legacy `--stt` mode is unchanged and is not the standard Realtime +conversation API. This demo does not add WebRTC, semantic VAD, or tool calling. + +## Shared host options + +The shared entry point can select either profile explicitly: + +```bash +python -m examples.online_serving.realtime_web --profile qwen3-turn \ + --backend ws://127.0.0.1:8091 --stt --port 7863 +``` + +- `--backend` / `--ws-backend`: backend WebSocket origin (not the full `/v1/realtime` URL). +- `--model`: override the wrapper's model name. +- `--public-realtime-url`: optional browser-visible WebSocket URL; otherwise the + static host proxies `/v1/realtime` on the same origin. +- `--host`, `--port`: UI bind address and port (default port 7862). +- `--ref-audio`: required for MiniCPM; rejected for Qwen3. + +Microphone and camera access require `localhost` or HTTPS. For a remote backend, +use an SSH tunnel to the UI host or an HTTPS reverse proxy with WebSocket support. +An explicit `wss://` public URL retains its scheme. + +## Validation + +```bash +node --test tests/examples/test_realtime_web_profiles.cjs +pytest tests/examples/test_minicpmo_realtime_web_server.py \ + tests/examples/test_minicpmo_realtime_web_static.py \ + tests/examples/test_realtime_web_profiles.py +pytest -m cpu tests/model_executor/models/qwen3_omni/test_duplex_plugin.py \ + tests/engine/duplex/test_session_runner.py \ + tests/engine/test_duplex_orchestrator.py +``` + +These CPU tests cover profile messages, two STT turns, VAD terminal/drain ordering, +backpressure, native capture during playback, server config injection, and shared +worklets. Backend regressions cover current audio inclusion in history-based +prompts, delayed playback ACK ordering, and no extra generation during silence +after a completed VAD turn. Hardware acceptance still requires a MiniCPM call +with camera/barge-in and two Qwen turns against each enabled backend, including +speech interruption in VAD mode; simulated tests do not +establish model quality or hardware latency. diff --git a/examples/online_serving/realtime_web/__init__.py b/examples/online_serving/realtime_web/__init__.py new file mode 100644 index 000000000..9e82f8967 --- /dev/null +++ b/examples/online_serving/realtime_web/__init__.py @@ -0,0 +1,4 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +"""Shared browser audio UI and model-specific realtime profiles.""" diff --git a/examples/online_serving/realtime_web/__main__.py b/examples/online_serving/realtime_web/__main__.py new file mode 100644 index 000000000..618f3c542 --- /dev/null +++ b/examples/online_serving/realtime_web/__main__.py @@ -0,0 +1,7 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +from .server import main + +if __name__ == "__main__": + main() diff --git a/examples/online_serving/minicpmo/realtime_web/app/index.html b/examples/online_serving/realtime_web/app/index.html similarity index 76% rename from examples/online_serving/minicpmo/realtime_web/app/index.html rename to examples/online_serving/realtime_web/app/index.html index 46ab15907..a38a731e4 100644 --- a/examples/online_serving/minicpmo/realtime_web/app/index.html +++ b/examples/online_serving/realtime_web/app/index.html @@ -4,37 +4,35 @@ - MiniCPM-o 4.5 Voice + Omni Realtime Voice
-

Experimental full-duplex runtime

-

MiniCPM-o 4.5 Voice

+

+

Omni Realtime Voice

Offline
+

+
-
+
- +
Mic @@ -53,7 +51,7 @@

Live transcript

Conversation

- Model-controlled listen / speak +

Start a session and speak naturally.

@@ -73,7 +71,10 @@
- + + + + diff --git a/examples/online_serving/minicpmo/realtime_web/app/static/app.js b/examples/online_serving/realtime_web/app/static/app.js similarity index 59% rename from examples/online_serving/minicpmo/realtime_web/app/static/app.js rename to examples/online_serving/realtime_web/app/static/app.js index 4d2b78247..4d9215ba0 100644 --- a/examples/online_serving/minicpmo/realtime_web/app/static/app.js +++ b/examples/online_serving/realtime_web/app/static/app.js @@ -1,8 +1,10 @@ (() => { 'use strict'; - const config = window.FULL_DUPLEX_CONFIG || {}; + const config = window.OMNI_REALTIME_CONFIG || window.FULL_DUPLEX_CONFIG || {}; + const profile = window.OmniRealtimeProfiles[config.profile || 'minicpm-native'](config); const callButton = document.getElementById('callButton'); + const sendTurnButton = document.getElementById('sendTurnButton'); const muteButton = document.getElementById('muteButton'); const cameraButton = document.getElementById('cameraButton'); const cameraPreview = document.getElementById('cameraPreview'); @@ -27,20 +29,26 @@ const INITIAL_PLAYBACK_BUFFER_MS = 400; const SESSION_CLOSE_TIMEOUT_MS = 1000; - // Default prompts mirroring the official MiniCPM-o-Demo presets - // (assets/presets/{omni,audio_duplex}/*.yaml). - const PROMPT_PRESETS = { - omni: 'Streaming Omni Conversation.', - chinese_call: '扮演一个具有以上声音特征的助手。请认真、高质量地回复用户的问题。' - + '请用高自然度的方式和用户聊天。你处于双工模式,可以一边听、一边说。' - + '你是由面壁智能开发的人工智能助手:面壁小钢炮。', - english_call: 'Replicate the tone and style from the input audio. Your task is to be ' - + 'a helpful assistant using this voice pattern. Please answer the user\'s questions ' - + 'seriously and in a high quality. Please chat with the user in a high naturalness ' - + 'style. You are in duplex mode, where you can listen and speak at the same time.', - }; + const PROMPT_PRESETS = profile.presets; + document.title = profile.title; + document.getElementById('pageTitle').textContent = profile.title; + document.getElementById('pageEyebrow').textContent = profile.eyebrow; + document.getElementById('profileDescription').textContent = profile.description; + document.getElementById('policyLabel').textContent = profile.policy; + cameraButton.hidden = !profile.camera; + sendTurnButton.hidden = !profile.clientCommit; + promptPreset.replaceChildren(); + for (const name of [...Object.keys(PROMPT_PRESETS), 'custom']) { + const option = document.createElement('option'); + option.value = name; + option.textContent = name.replaceAll('_', ' '); + promptPreset.appendChild(option); + } + systemPromptInput.value = Object.values(PROMPT_PRESETS)[0]; + document.getElementById('promptControls').hidden = !profile.instructions; let socket = null; + let closingSocket = null; let mediaStream = null; let captureContext = null; let captureNode = null; @@ -65,6 +73,63 @@ let liveUserTurn = null; let liveAssistantTurn = null; let sessionCloseResolver = null; + let responseComplete = false; + let playbackComplete = true; + let turnSubmitted = false; + let echoTimer = null; + let sessionGeneration = 0; + let audioChain = Promise.resolve(); + let assistantTextChannel = null; + let connectionReady = false; + let turnCounter = 0; + let stopping = null; + let turnTimeout = null; + + function markBusy() { + assistantActive = true; + if (profile.halfDuplex) pendingCapture = []; + } + + function armTurnTimeout() { + clearTimeout(turnTimeout); + turnTimeout = window.setTimeout(() => { + failSession('Timed out waiting for the model response. Check the backend event log and restart the session.'); + }, 120000); + } + + async function failSession(message) { + appendLog(message, true); + await stopSession({ terminal: false }); + setConnection('Error', 'error'); + runtimeDetail.textContent = message; + } + + function finishResponseIfReady() { + if (!responseComplete || !playbackComplete || echoTimer !== null) return; + const generation = sessionGeneration; + echoTimer = window.setTimeout(async () => { + echoTimer = null; + if (!running || generation !== sessionGeneration) return; + if (profile.reconnectEachTurn) { + connectionReady = false; + const previous = socket; + socket = null; + if (previous) { previous.onclose = null; previous.onmessage = null; previous.close(); } + try { await openSocket(); } + catch (error) { await failSession(error.message); return; } + if (!running || generation !== sessionGeneration) return; + } + assistantActive = false; + responseComplete = false; + turnSubmitted = false; + currentResponseId = null; + responseHasAudio = false; + assistantTextChannel = null; + if (profile.halfDuplex) pendingCapture = []; + sendTurnButton.disabled = !profile.clientCommit; + setModel(profile.waiting); + }, ECHO_GUARD_MS); + } function staticAssetUrl(path) { const version = String(config.appVersion || '').trim(); @@ -82,12 +147,7 @@ } function realtimeUrl() { - const url = new URL(config.realtimePath, window.location.href); - url.protocol = window.location.protocol === 'https:' ? 'wss:' : 'ws:'; - url.searchParams.set('duplex', '1'); - url.searchParams.set('model', config.model || 'openbmb/MiniCPM-o-4_5'); - url.searchParams.set('autostart', '0'); - return url.toString(); + return profile.url(config, window.location.href); } function setConnection(label, kind) { @@ -250,7 +310,7 @@ } function microphoneUploadEnabled() { - return running && !muted; + return running && connectionReady && !muted && (!profile.halfDuplex || !assistantActive); } function flushCapture() { @@ -267,19 +327,9 @@ offset += chunk.length; } pendingCapture = []; - const pcm = resampleInt16(merged, captureRate, INPUT_RATE); - const appendEvent = { - type: 'input_audio_buffer.append', - audio: int16ToBase64(pcm), - format: 'pcm16', - sample_rate_hz: INPUT_RATE, - }; - // Omni duplex: ~1 fps camera frame rides the audio append (official - // MiniCPM-o-Demo contract: one base64 JPEG per ~1 s chunk). - if (cameraPendingFrame) { - appendEvent.video_frames = [cameraPendingFrame]; - cameraPendingFrame = null; - } + const pcm = resampleInt16(merged, captureRate, profile.inputSampleRate || INPUT_RATE); + const appendEvent = profile.append(int16ToBase64(pcm), cameraPendingFrame); + cameraPendingFrame = null; socket.send(JSON.stringify(appendEvent)); } @@ -287,13 +337,17 @@ currentResponseId = responseId || currentResponseId; responseHasAudio = false; assistantActive = true; - setModel('Speaking'); + responseComplete = false; + playbackComplete = true; + if (profile.halfDuplex) pendingCapture = []; + setModel(profile.halfDuplex ? 'Thinking' : 'Speaking'); } function feedPlayback(decoded, responseId) { if (!decoded || !decoded.pcm || decoded.pcm.length === 0 || !playbackNode) return; const pcm = resampleInt16(decoded.pcm, decoded.sourceRate, playbackRate); responseHasAudio = true; + playbackComplete = false; assistantActive = true; setPlayback('Buffering'); playbackNode.port.postMessage({ @@ -305,6 +359,7 @@ } function requestPlaybackDrain(responseId) { + if (!responseHasAudio) { playbackComplete = true; finishResponseIfReady(); return; } if (!playbackNode) return; playbackNode.port.postMessage({ type: 'drain', responseId: responseId || currentResponseId }); } @@ -314,90 +369,118 @@ if (!responseId && playedMs > 0) appendLog('playback ack skipped: missing response id', true); return; } - socket.send(JSON.stringify({ - type: 'playback.ack', - response_id: responseId, - item_id: `item_${responseId}`, - played_ms: playedMs, - committed_ms: playedMs, - })); + const ack = profile.ack(responseId, playedMs); + if (ack) socket.send(JSON.stringify(ack)); } function playbackDrained(message) { const responseId = message.responseId || currentResponseId; - sendPlaybackAck(responseId, Number(message.playedMs) || 0); + if (currentResponseId && responseId !== currentResponseId) return; + if (profile.playbackAck) sendPlaybackAck(responseId, Number(message.playedMs) || 0); setPlayback('Idle'); - if (message.underrunMs > 0) { - appendLog(`playback underrun ${message.underrunMs} ms`); - } - window.setTimeout(() => { - assistantActive = false; - currentResponseId = null; - responseHasAudio = false; - if (running) setModel('Listening'); - }, ECHO_GUARD_MS); + playbackComplete = true; + if (message.underrunMs > 0) appendLog(`playback underrun ${message.underrunMs} ms`); + if (!profile.halfDuplex && !profile.waitForResponseDone) responseComplete = true; + finishResponseIfReady(); } - function handleEvent(event) { + async function handleAudioEvent(action) { + markBusy(); + currentResponseId = action.responseId || currentResponseId || `turn-${turnCounter}`; + setModel('Speaking'); + const generation = sessionGeneration; + const decoded = await decodeAudioDelta(action.event); + if (generation === sessionGeneration) feedPlayback(decoded, currentResponseId); + } + + function handleTranscriptEvent(action) { + if (action.role === 'assistant') { + // Qwen may emit both text and audio-transcript representations. + // Display one channel per response instead of duplicating the answer. + if (profile.deduplicateTranscript && assistantTextChannel && action.channel !== assistantTextChannel) return; + if (action.text) assistantTextChannel = action.channel || assistantTextChannel; + } + if (action.kind === 'text') addTranscript(action.role, action.text); + else if (profile.deduplicateTranscript && action.role === 'assistant') { + if (action.text) { + const turn = ensureTurn('assistant'); + turn.value = action.text; + turn.text.textContent = action.text; + } + } else finishTranscript(action.role, action.text); + } + + async function handleEvent(event) { appendEventLog(event); - const responseId = responseIdOf(event); - switch (event.type) { - case 'session.created': - case 'session.updated': + const action = profile.mapEvent(event); + const responseId = action.responseId; + switch (action.kind) { + case 'connected': setConnection('Connected', 'online'); - setModel('Listening'); + if (!assistantActive) setModel(profile.waiting); break; - case 'response.listen': + case 'interrupt': + sessionGeneration += 1; + // A cancelled response never reaches 'done', so nothing else would + // disarm the turn watchdog it left armed. + clearTimeout(turnTimeout); + if (playbackNode) playbackNode.port.postMessage({ type: 'clear' }); + responseComplete = true; + playbackComplete = true; + responseHasAudio = false; assistantActive = false; - setModel('Listening'); + currentResponseId = null; + assistantTextChannel = null; + finishTranscript('assistant'); + setPlayback('Idle'); + setModel(profile.waiting); break; - case 'response.created': - case 'response.speak': + case 'listen': + assistantActive = false; + setModel(profile.waiting); + break; + case 'begin': + if (profile.halfDuplex) armTurnTimeout(); + if (echoTimer !== null) { clearTimeout(echoTimer); echoTimer = null; } beginAssistant(responseId); break; - case 'response.output_audio.delta': - currentResponseId = responseId || currentResponseId; - assistantActive = true; - setModel('Speaking'); - decodeAudioDelta(event) - .then((decoded) => feedPlayback(decoded, responseId)) - .catch((error) => appendLog(`audio decode failed: ${error.message || error}`, true)); + case 'audio': + await handleAudioEvent(action); break; - case 'response.output_audio.done': + case 'drain': requestPlaybackDrain(responseId); break; - case 'response.output_audio_transcript.delta': - addTranscript('assistant', event.delta || ''); + case 'text': case 'text-final': + handleTranscriptEvent(action); break; - case 'response.output_audio_transcript.done': - finishTranscript('assistant', event.transcript || ''); - break; - case 'conversation.item.input_audio_transcription.delta': - addTranscript('user', event.delta || ''); - break; - case 'conversation.item.input_audio_transcription.completed': - finishTranscript('user', event.transcript || ''); - break; - case 'response.done': + case 'done': + clearTimeout(turnTimeout); + responseComplete = true; finishTranscript('assistant'); - if (!responseHasAudio) requestPlaybackDrain(responseId); + requestPlaybackDrain(responseId); + finishResponseIfReady(); break; - case 'playback.acknowledged': - { - const acknowledgement = event.event || event; - runtimeDetail.textContent = `Playback committed ${acknowledgement.committed_ms || 0} ms`; - } + case 'backpressure': + armTurnTimeout(); + markBusy(); + setModel('Thinking / Speaking'); + runtimeDetail.textContent = action.message; break; - case 'session.closed': + case 'ack': + runtimeDetail.textContent = `Playback committed ${action.committedMs} ms`; + break; + case 'closed': if (sessionCloseResolver) sessionCloseResolver(); sessionCloseResolver = null; break; case 'error': - setConnection('Error', 'error'); - runtimeDetail.textContent = String(event.error || event.code || 'Server error'); - break; - default: + if (action.fatal) { await failSession(action.message); break; } + // A rejected frame or a refused update costs one operation, not the + // call: report it and keep listening. + appendLog(action.message, true); + runtimeDetail.textContent = action.message; break; + default: break; } } @@ -406,8 +489,15 @@ playbackRate = playbackContext.sampleRate; await playbackContext.audioWorklet.addModule(staticAssetUrl('static/playback_worklet.js')); playbackNode = new AudioWorkletNode(playbackContext, 'fullduplex-pcm-playback'); + const currentPlaybackNode = playbackNode; playbackNode.port.onmessage = (message) => { + if (playbackNode !== currentPlaybackNode) return; if (message.data.type === 'playback-started') setPlayback('Playing'); + else if (message.data.type === 'playback-stopped') { + // The old response may report after the next one has started. ACK its + // own cursor without changing the current response's UI state. + if (profile.playbackAck) sendPlaybackAck(message.data.responseId, Number(message.data.playedMs) || 0); + } else if (message.data.type === 'playback-drained') playbackDrained(message.data); else if (message.data.type === 'playback-underrun') { runtimeDetail.textContent = `Playback underrun ${message.data.underrunMs || 0} ms`; @@ -451,48 +541,69 @@ function openSocket() { return new Promise((resolve, reject) => { const url = realtimeUrl(); - socket = new WebSocket(url); + const current = new WebSocket(url); + socket = current; + connectionReady = false; + turnCounter += 1; let settled = false; - socket.onopen = () => { + const timer = window.setTimeout(() => { + if (settled) return; settled = true; - const extraBody = { - auto_response: true, - }; - const session = { - modalities: ['audio', 'text'], - voice: 'default', - extra_body: extraBody, - }; - // Reference voice for TTS cloning, provided by the server via - // --ref-audio (mirrors the official demo's default ref audio). - if (config.refAudio) session.ref_audio = config.refAudio; - const instructions = systemPromptInput ? systemPromptInput.value.trim() : ''; - if (instructions) session.instructions = instructions; - socket.send(JSON.stringify({ type: 'session.update', session })); - runtimeDetail.textContent = `${captureRate} Hz capture / ${playbackRate} Hz playback`; + current.onclose = null; + current.close(); + reject(new Error(`Session handshake timed out. ${profile.connectionHint}`)); + }, 15000); + const rejectOnce = (message) => { + clearTimeout(timer); + if (!settled) { settled = true; reject(new Error(message)); } + }; + current.onopen = () => { + const instructions = systemPromptInput.value.trim(); + for (const event of profile.initialMessages(config, instructions)) current.send(JSON.stringify(event)); appendLog(`websocket open ${url}`); - resolve(); }; - socket.onmessage = (message) => { - if (typeof message.data !== 'string') return; - try { - handleEvent(JSON.parse(message.data)); - } catch (error) { - appendLog(`invalid server event: ${error.message || error}`, true); + current.onmessage = (message) => { + if (typeof message.data !== 'string' || (socket !== current && closingSocket !== current)) return; + let event; + try { event = JSON.parse(message.data); } + catch (error) { + if (socket !== current) return; + const detail = `Invalid server event: ${error.message}`; + if (!settled) rejectOnce(detail); + else void failSession(detail); + return; } - }; - socket.onerror = () => { - if (!settled) { + // Shutdown has detached the active socket. Only its close acknowledgement + // may pass, directly: the event queue can itself be awaiting cleanup. + if (closingSocket === current) { + if (event?.type === 'session.closed' && sessionCloseResolver) sessionCloseResolver(); + return; + } + const action = profile.mapEvent(event); + if (action.kind === 'error' && !settled) { + rejectOnce(action.message); + return; + } + if (event.type === profile.readyEvent && !settled) { + clearTimeout(timer); settled = true; - reject(new Error(`WebSocket connection failed: ${url}`)); + connectionReady = true; + runtimeDetail.textContent = `${captureRate} Hz capture / ${playbackRate} Hz playback`; + resolve(); } + // Serialize decoding with terminal events: a drain must never overtake + // an asynchronously decoded WAV chunk. + audioChain = audioChain.then(() => { + if (socket === current || action.kind === 'closed') return handleEvent(event); + }).catch((error) => failSession(`Server event failed: ${error.message}`)); }; - socket.onclose = (event) => { + current.onerror = () => rejectOnce(`WebSocket connection failed. ${profile.connectionHint}`); + current.onclose = (event) => { + clearTimeout(timer); + connectionReady = false; appendLog(`websocket closed code=${event.code}`); - if (running) { - setConnection('Disconnected', 'error'); - stopSession(); - } + if (!settled) rejectOnce(`Connection closed before session ready. ${profile.connectionHint}`); + else if (running && socket === current) failSession(`Backend disconnected (${event.code}). ${profile.connectionHint}`); }; }); } @@ -515,46 +626,59 @@ setConnection('Connecting', 'connecting'); runtimeDetail.textContent = 'Requesting microphone access'; try { + sessionGeneration += 1; + audioChain = Promise.resolve(); await openPlayback(); await openCapture(); await openSocket(); running = true; muted = false; assistantActive = false; - sendTimer = window.setInterval(flushCapture, SEND_INTERVAL_MS); + sendTimer = window.setInterval(flushCapture, profile.sendIntervalMs || SEND_INTERVAL_MS); startClock(); callButton.textContent = 'End session'; callButton.classList.add('is-active'); muteButton.disabled = false; - cameraButton.disabled = false; + cameraButton.disabled = !profile.camera; + sendTurnButton.disabled = !profile.clientCommit; setConnection('Connected', 'online'); - setModel('Listening'); + setModel(profile.waiting); appendLog('session started'); } catch (error) { appendLog(`start failed: ${error.message || error}`, true); + await stopSession({ terminal: false }); setConnection('Error', 'error'); runtimeDetail.textContent = String(error.message || error); - await stopSession(); } finally { callButton.disabled = false; } } async function startCamera() { - if (cameraStream) return; + if (!profile.camera || cameraStream) return; cameraStream = await navigator.mediaDevices.getUserMedia({ video: true, audio: false }); cameraPreview.srcObject = cameraStream; cameraPreview.style.display = ''; await cameraPreview.play().catch(() => {}); // Official omni-duplex cadence: one JPEG (quality 0.7) per ~1 s chunk, // no client-side resize (the server normalizes at scale_resolution=448). - cameraTimer = window.setInterval(() => { + const captureCameraFrame = () => { if (!cameraStream || cameraPreview.videoWidth === 0) return; - cameraCanvas.width = cameraPreview.videoWidth; - cameraCanvas.height = cameraPreview.videoHeight; - cameraCanvas.getContext('2d').drawImage(cameraPreview, 0, 0); - cameraPendingFrame = cameraCanvas.toDataURL('image/jpeg', 0.7).split(',')[1]; - }, 1000); + const scale = profile.cameraMaxDimension + ? Math.min(1, profile.cameraMaxDimension / Math.max(cameraPreview.videoWidth, cameraPreview.videoHeight)) : 1; + cameraCanvas.width = Math.max(1, Math.round(cameraPreview.videoWidth * scale)); + cameraCanvas.height = Math.max(1, Math.round(cameraPreview.videoHeight * scale)); + cameraCanvas.getContext('2d').drawImage(cameraPreview, 0, 0, cameraCanvas.width, cameraCanvas.height); + const frame = cameraCanvas.toDataURL('image/jpeg', 0.7).split(',')[1]; + if (profile.imageMessages) { + if (connectionReady && socket?.readyState === WebSocket.OPEN) { + for (const event of profile.imageMessages(frame)) socket.send(JSON.stringify(event)); + } + } else cameraPendingFrame = frame; + }; + // Do not make the first spoken turn race a one-second timer. + captureCameraFrame(); + cameraTimer = window.setInterval(captureCameraFrame, 1000); cameraButton.textContent = 'Camera off'; cameraButton.classList.add('is-active'); appendLog('camera on (1 fps omni frames)'); @@ -590,15 +714,32 @@ const finish = () => { if (done) return; done = true; + clearTimeout(timer); if (sessionCloseResolver === finish) sessionCloseResolver = null; resolve(); }; sessionCloseResolver = finish; - window.setTimeout(finish, timeoutMs); + const timer = window.setTimeout(finish, timeoutMs); }); } - async function stopSession({ terminal = true } = {}) { + function stopSession(options = {}) { + if (stopping) return stopping; + stopping = cleanupSession(options).finally(() => { stopping = null; }); + return stopping; + } + + async function cleanupSession({ terminal = true } = {}) { + sessionGeneration += 1; + connectionReady = false; + clearTimeout(echoTimer); + clearTimeout(turnTimeout); + echoTimer = null; + responseComplete = false; + playbackComplete = true; + turnSubmitted = false; + assistantTextChannel = null; + sendTurnButton.disabled = true; running = false; assistantActive = false; pendingCapture = []; @@ -607,15 +748,16 @@ sendTimer = null; clockTimer = null; if (socket) { - const closingSocket = socket; + closingSocket = socket; socket = null; closingSocket.onclose = null; - if (terminal && closingSocket.readyState === WebSocket.OPEN) { + if (terminal && profile.closeSession && closingSocket.readyState === WebSocket.OPEN) { const closed = waitForSessionClosed(closingSocket, SESSION_CLOSE_TIMEOUT_MS); closingSocket.send(JSON.stringify({ type: 'session.close' })); await closed; } closingSocket.close(1000, 'client stop'); + closingSocket = null; } if (playbackNode) playbackNode.port.postMessage({ type: 'clear' }); if (mediaStream) { @@ -632,6 +774,8 @@ playbackNode = null; currentResponseId = null; responseHasAudio = false; + finishTranscript('user'); + finishTranscript('assistant'); meterFill.style.width = '0%'; sessionTimer.textContent = '00:00'; callButton.textContent = 'Start session'; @@ -659,6 +803,18 @@ if (running) stopSession(); else startSession(); }); + sendTurnButton.addEventListener('click', () => { + if (!running || !connectionReady || !profile.clientCommit || turnSubmitted) return; + flushCapture(); + for (const event of profile.commitMessages()) socket.send(JSON.stringify(event)); + turnSubmitted = true; + responseComplete = false; + markBusy(); + armTurnTimeout(); + sendTurnButton.disabled = true; + setModel('Thinking'); + appendLog('turn submitted'); + }); muteButton.addEventListener('click', toggleMute); clearLogButton.addEventListener('click', () => { eventLog.textContent = ''; diff --git a/examples/online_serving/minicpmo/realtime_web/app/static/pcm_worklet.js b/examples/online_serving/realtime_web/app/static/pcm_worklet.js similarity index 100% rename from examples/online_serving/minicpmo/realtime_web/app/static/pcm_worklet.js rename to examples/online_serving/realtime_web/app/static/pcm_worklet.js diff --git a/examples/online_serving/minicpmo/realtime_web/app/static/playback_worklet.js b/examples/online_serving/realtime_web/app/static/playback_worklet.js similarity index 95% rename from examples/online_serving/minicpmo/realtime_web/app/static/playback_worklet.js rename to examples/online_serving/realtime_web/app/static/playback_worklet.js index f0d3d42ae..8aaa131c6 100644 --- a/examples/online_serving/minicpmo/realtime_web/app/static/playback_worklet.js +++ b/examples/online_serving/realtime_web/app/static/playback_worklet.js @@ -39,6 +39,13 @@ class FullDuplexPcmPlayback extends AudioWorkletProcessor { } this.notifyIfDrained(); } else if (message.type === 'clear') { + if (this.activeResponseId) { + this.port.postMessage({ + type: 'playback-stopped', + responseId: this.activeResponseId, + playedMs: Math.round((this.playedFrames * 1000) / sampleRate), + }); + } this.queue = []; this.offset = 0; this.playedFrames = 0; diff --git a/examples/online_serving/realtime_web/app/static/profiles/common.js b/examples/online_serving/realtime_web/app/static/profiles/common.js new file mode 100644 index 000000000..feefbaa74 --- /dev/null +++ b/examples/online_serving/realtime_web/app/static/profiles/common.js @@ -0,0 +1,56 @@ +/* Shared wire helpers. Profiles translate server events into shell actions. */ +(function (root) { + 'use strict'; + const profiles = root.OmniRealtimeProfiles = root.OmniRealtimeProfiles || {}; + profiles.url = (config, location) => { + const url = new URL(config.realtimePath || 'v1/realtime', location); + if (url.protocol === 'http:') url.protocol = 'ws:'; + if (url.protocol === 'https:') url.protocol = 'wss:'; + return url; + }; + /* Errors that leave the session unusable, so the shell has to tear it down. + Everything else costs one operation --- a rejected image, a refused + session.update, a late playback ack --- and the call carries on. + 'runtime_append_failed' is deliberately absent: the engine reports it and + keeps the session, unlike 'runtime_append_task_failed' which kills it. */ + profiles.fatalErrorCodes = new Set([ + 'unknown_session', + 'session_closed', + 'resource_exhausted', + 'unsupported', + 'unsupported_turn_detection', + 'server_vad_unavailable', + 'runtime_append_task_failed', + 'internal_error', + ]); + profiles.event = (event) => { + const responseId = event.response_id || event.response?.id || null; + const action = (kind, extra = {}) => ({ kind, responseId, ...extra }); + switch (event.type) { + case 'session.created': case 'session.updated': return action('connected'); + case 'response.created': return action('begin'); + case 'response.audio.delta': case 'response.output_audio.delta': + return action('audio', { event: { ...event, delta: event.delta || event.audio || event.response?.audio } }); + case 'response.audio.done': case 'response.output_audio.done': return action('drain'); + case 'response.audio_transcript.delta': case 'response.output_audio_transcript.delta': + return action('text', { role: 'assistant', channel: 'audio', text: event.delta || '' }); + case 'response.audio_transcript.done': case 'response.output_audio_transcript.done': + return action('text-final', { role: 'assistant', channel: 'audio', text: event.transcript || '' }); + case 'conversation.item.input_audio_transcription.delta': + return action('text', { role: 'user', text: event.delta || '' }); + case 'conversation.item.input_audio_transcription.completed': + return action('text-final', { role: 'user', text: event.transcript || '' }); + case 'response.done': return action('done'); + case 'session.closed': return action('closed'); + case 'error': { + const code = event.code || event.error?.code; + return action('error', { + code, + fatal: profiles.fatalErrorCodes.has(code), + message: typeof event.error === 'string' ? event.error : event.error?.message || event.message || code || 'Server error', + }); + } + default: return action('ignore'); + } + }; +})(globalThis); diff --git a/examples/online_serving/realtime_web/app/static/profiles/minicpm_native.js b/examples/online_serving/realtime_web/app/static/profiles/minicpm_native.js new file mode 100644 index 000000000..801d549be --- /dev/null +++ b/examples/online_serving/realtime_web/app/static/profiles/minicpm_native.js @@ -0,0 +1,55 @@ +(function (root) { + 'use strict'; + const profiles = root.OmniRealtimeProfiles; + profiles['minicpm-native'] = () => ({ + title: 'MiniCPM-o 4.5 Voice', eyebrow: 'Experimental full-duplex runtime', + policy: 'Model-controlled listen / speak', + description: 'Speak naturally. The model can listen and speak at the same time. Camera frames are sent with your audio.', + waiting: 'Listening', camera: true, playbackAck: true, clientCommit: false, + halfDuplex: false, closeSession: true, reconnectEachTurn: false, + readyEvent: 'session.updated', instructions: true, sendIntervalMs: 200, + presets: { + omni: 'Streaming Omni Conversation.', + chinese_call: '扮演一个具有以上声音特征的助手。请认真、高质量地回复用户的问题。' + + '请用高自然度的方式和用户聊天。你处于双工模式,可以一边听、一边说。' + + '你是由面壁智能开发的人工智能助手:面壁小钢炮。', + english_call: 'Replicate the tone and style from the input audio. Your task is to be ' + + 'a helpful assistant using this voice pattern. Please answer the user\'s questions ' + + 'seriously and in a high quality. Please chat with the user in a high naturalness ' + + 'style. You are in duplex mode, where you can listen and speak at the same time.', + }, + url(config, location) { + const url = profiles.url(config, location); + url.searchParams.set('duplex', '1'); + url.searchParams.set('model', config.model || 'openbmb/MiniCPM-o-4_5'); + url.searchParams.set('autostart', '0'); + return url.toString(); + }, + initialMessages(config, instructions) { + const session = { modalities: ['audio', 'text'], voice: 'default', + extra_body: { auto_response: true } }; + if (config.refAudio) session.ref_audio = config.refAudio; + if (instructions) session.instructions = instructions; + return [{ type: 'session.update', session }]; + }, + append(audio, frame) { + const event = { type: 'input_audio_buffer.append', audio, format: 'pcm16', sample_rate_hz: 16000 }; + if (frame) event.video_frames = [frame]; + return event; + }, + commitMessages: () => [], + ack(responseId, playedMs) { + return { type: 'playback.ack', response_id: responseId, item_id: `item_${responseId}`, + played_ms: playedMs, committed_ms: playedMs }; + }, + mapEvent(event) { + if (event.type === 'response.listen') return { kind: 'listen' }; + if (event.type === 'response.speak') return { kind: 'begin', responseId: event.response_id || event.response?.id }; + if (event.type === 'playback.acknowledged') { + return { kind: 'ack', committedMs: (event.event || event).committed_ms || 0 }; + } + return profiles.event(event); + }, + connectionHint: 'Check the MiniCPM duplex deployment and reference voice configuration.', + }); +})(globalThis); diff --git a/examples/online_serving/realtime_web/app/static/profiles/qwen3_turn.js b/examples/online_serving/realtime_web/app/static/profiles/qwen3_turn.js new file mode 100644 index 000000000..d6df32edd --- /dev/null +++ b/examples/online_serving/realtime_web/app/static/profiles/qwen3_turn.js @@ -0,0 +1,100 @@ +(function (root) { + 'use strict'; + const profiles = root.OmniRealtimeProfiles; + profiles['qwen3-turn'] = (config) => { + const vad = config.adapter === 'vad'; + let imageSeq = 0; + let imageItems = []; + // The engine refuses a ninth image (or 4 MiB of them) for the whole + // session, so the camera track retires its own oldest frame instead. + const MAX_IMAGE_ITEMS = 8; + const hint = 'This Qwen deploy needs session_mode: duplex and Server VAD support; use --stt or enable Server VAD.'; + return { + title: 'Qwen3-Omni Voice', eyebrow: vad ? 'Engine-owned duplex call' : 'Turn-based realtime call', + policy: vad ? 'Server VAD • interruptible responses' : 'Send turn • one response at a time', + description: vad + ? 'Pause to send your turn; speak again to interrupt. Microphone upload continues during playback. Requires the Qwen duplex plugin and Silero. Optional camera frames accompany each spoken turn.' + : 'Speak, then press Send turn. Microphone upload pauses while the model answers. Each turn uses a fresh connection; conversation history is not carried between turns. No barge-in or camera input.', + waiting: 'Waiting for you', camera: vad, cameraMaxDimension: 448, playbackAck: vad, clientCommit: !vad, + waitForResponseDone: true, halfDuplex: !vad, deduplicateTranscript: true, closeSession: vad, reconnectEachTurn: !vad, + inputSampleRate: vad ? 24000 : 16000, + readyEvent: vad ? 'session.updated' : 'session.created', instructions: vad, sendIntervalMs: 200, + presets: { assistant: 'You are a helpful assistant. Answer clearly and concisely.' }, + url(config, location) { + const url = profiles.url(config, location); + for (const key of ['native_duplex', 'minicpmo45_native_duplex', 'autostart']) url.searchParams.delete(key); + // Explicit opt-out matches the local file client, including servers + // that auto-select the duplex endpoint when the query is omitted. + url.searchParams.set('duplex', vad ? '1' : '0'); + url.searchParams.set('model', config.model || 'Qwen/Qwen3-Omni-30B-A3B-Instruct'); + return url.toString(); + }, + initialMessages(config, instructions) { + imageItems = []; + if (!vad) return [ + { type: 'session.update', model: config.model }, + { type: 'input_audio_buffer.commit', final: false }, + ]; + const session = { model: config.model, overlap_policy: 'barge_in_on_speech', audio: { input: { + format: { type: 'audio/pcm', rate: 24000 }, + turn_detection: { type: 'server_vad', threshold: 0.5, prefix_padding_ms: 300, + silence_duration_ms: 500, create_response: true, interrupt_response: true }, + } } }; + if (instructions) session.instructions = instructions; + return [{ type: 'session.update', session }]; + }, + append(audio) { + return vad ? { type: 'input_audio_buffer.append', audio } + : { type: 'input_audio_buffer.append', audio, format: 'pcm16', sample_rate_hz: 16000 }; + }, + // Qwen is a turn model: a frame is a picture added to the conversation, + // not a track interleaved into the audio. That is exactly the OpenAI + // Realtime image interface, so the camera sends conversation items + // rather than an extension field on the append. + imageMessages(frame) { + if (!vad) return []; + const messages = []; + if (imageItems.length >= MAX_IMAGE_ITEMS) { + // Retire before creating: the engine counts what is already stored. + messages.push({ type: 'conversation.item.delete', item_id: imageItems.shift() }); + } + const id = `camera_${++imageSeq}`; + imageItems.push(id); + messages.push({ type: 'conversation.item.create', item: { + id, type: 'message', role: 'user', + content: [{ type: 'input_image', image_url: `data:image/jpeg;base64,${frame}` }], + } }); + return messages; + }, + commitMessages: () => vad ? [] : [{ type: 'input_audio_buffer.commit', final: true }], + ack: (responseId, playedMs) => vad ? { type: 'playback.ack', response_id: responseId, + item_id: `item_${responseId}`, played_ms: playedMs, committed_ms: playedMs } : null, + mapEvent(event) { + if (event.type === 'playback.acknowledged') return { kind: 'ack', committedMs: (event.event || event).committed_ms || 0 }; + if (event.type === 'response.output_text.delta' || event.type === 'transcription.delta') { + // On the shipped STT route transcription.* is model-generated text, + // not a separate ASR transcript of the user. See realtime_connection.py. + return { kind: 'text', role: 'assistant', channel: 'text', text: event.delta || '' }; + } + if (event.type === 'response.output_text.done' || event.type === 'transcription.done') { + return { kind: 'text-final', role: 'assistant', channel: 'text', text: event.text || '' }; + } + if (vad && (event.type === 'output_audio_buffer.cleared' || + (event.type === 'response.done' && event.response?.status === 'cancelled'))) { + return { kind: 'interrupt' }; + } + if (event.type === 'input_audio_buffer.committed') return { kind: 'begin' }; + if (event.type === 'input_audio_buffer.cleared') return { kind: 'backpressure', message: 'Input cleared. Model is answering; please wait.' }; + if (!vad && event.type === 'response.output_audio.done') return { kind: 'done' }; + const action = profiles.event(event); + if (action.kind === 'error') { + if (action.code === 'input_backpressure') return { ...action, kind: 'backpressure', message: 'Input buffer is full; please repeat dropped speech after the response.' }; + if (action.code === 'server_vad_initialization_failed') action.message = 'Server VAD could not load Silero. Configure the server artifact or restart this UI with --stt.'; + else if (vad && action.code === 'unsupported') action.message += ` ${hint}`; + } + return action; + }, + connectionHint: vad ? hint : 'Check the Qwen /v1/realtime STT endpoint and model name.', + }; + }; +})(globalThis); diff --git a/examples/online_serving/minicpmo/realtime_web/app/static/styles.css b/examples/online_serving/realtime_web/app/static/styles.css similarity index 98% rename from examples/online_serving/minicpmo/realtime_web/app/static/styles.css rename to examples/online_serving/realtime_web/app/static/styles.css index 9b080176b..bf25c3eff 100644 --- a/examples/online_serving/minicpmo/realtime_web/app/static/styles.css +++ b/examples/online_serving/realtime_web/app/static/styles.css @@ -125,3 +125,6 @@ button:disabled { opacity: .46; cursor: default; } @media (prefers-reduced-motion: reduce) { *, *::before, *::after { animation-duration: .01ms !important; animation-iteration-count: 1 !important; } } + +/* Profile-specific controls may have inline flex styles. */ +[hidden] { display: none !important; } diff --git a/examples/online_serving/realtime_web/server.py b/examples/online_serving/realtime_web/server.py new file mode 100644 index 000000000..85e4b63d5 --- /dev/null +++ b/examples/online_serving/realtime_web/server.py @@ -0,0 +1,229 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +"""Static host and same-origin proxy for the shared realtime browser UI.""" + +from __future__ import annotations + +import argparse +import asyncio +import base64 +import contextlib +import hashlib +import json +import logging +from pathlib import Path +from urllib.parse import urlencode + +import uvicorn +import websockets +from fastapi import FastAPI, WebSocket, WebSocketDisconnect +from fastapi.responses import HTMLResponse, Response +from fastapi.staticfiles import StaticFiles + +logger = logging.getLogger(__name__) +APP_DIR = Path(__file__).parent / "app" +STATIC_DIR = APP_DIR / "static" + + +def _join_ws_url(base: str, path: str, query: str) -> str: + return base.rstrip("/") + path + (("?" + query) if query else "") + + +async def _pump_client_to_backend(client: WebSocket, backend) -> None: + try: + while True: + message = await client.receive() + if message["type"] == "websocket.disconnect": + await backend.close() + return + if message.get("text") is not None: + await backend.send(message["text"]) + elif message.get("bytes") is not None: + await backend.send(message["bytes"]) + except (WebSocketDisconnect, websockets.ConnectionClosed): + with contextlib.suppress(Exception): + await backend.close() + + +async def _pump_backend_to_client(client: WebSocket, backend) -> None: + try: + async for message in backend: + if isinstance(message, bytes): + await client.send_bytes(message) + else: + await client.send_text(message) + except (WebSocketDisconnect, websockets.ConnectionClosed): + return + except RuntimeError as exc: + if "websocket.send" in str(exc) and "websocket.close" in str(exc): + return + raise + + +def _expected_proxy_close(exc: BaseException) -> bool: + if isinstance(exc, (WebSocketDisconnect, websockets.ConnectionClosed, asyncio.CancelledError)): + return True + return isinstance(exc, RuntimeError) and "websocket.send" in str(exc) and "websocket.close" in str(exc) + + +def build_app( + *, + ws_backend: str | None = None, + model: str | None = None, + public_realtime_url: str | None = None, + ref_audio: str | None = None, + profile: str = "minicpm-native", + adapter: str = "stt", +) -> FastAPI: + if profile not in {"minicpm-native", "qwen3-turn"}: + raise ValueError(f"Unknown profile: {profile}") + if adapter not in {"stt", "vad"}: + raise ValueError(f"Unknown adapter: {adapter}") + if profile == "qwen3-turn" and ref_audio: + raise ValueError("Qwen3 profile does not accept --ref-audio") + native = profile == "minicpm-native" + ws_backend = ws_backend or ("ws://127.0.0.1:8099" if native else "ws://127.0.0.1:8091") + model = model or ("openbmb/MiniCPM-o-4_5" if native else "Qwen/Qwen3-Omni-30B-A3B-Instruct") + app = FastAPI(title="Omni Realtime Web UI") + index_path = APP_DIR / "index.html" + app_version_hash = hashlib.sha256() + for asset_path in ( + STATIC_DIR / "app.js", + STATIC_DIR / "pcm_worklet.js", + STATIC_DIR / "playback_worklet.js", + *sorted((STATIC_DIR / "profiles").glob("*.js")), + ): + app_version_hash.update(asset_path.read_bytes()) + app_version = app_version_hash.hexdigest()[:12] + + ref_audio_uri: str | None = None + if ref_audio: + ref_path = Path(ref_audio) + if not ref_path.is_file(): + raise SystemExit(f"--ref-audio not found: {ref_audio}") + encoded = base64.b64encode(ref_path.read_bytes()).decode("ascii") + ref_audio_uri = f"data:audio/wav;base64,{encoded}" + + @app.get("/", response_class=HTMLResponse) + def index() -> HTMLResponse: + config = json.dumps( + { + "model": model, + "profile": profile, + "adapter": adapter, + "realtimePath": public_realtime_url or "v1/realtime", + "refAudio": ref_audio_uri, + "appVersion": app_version, + }, + ensure_ascii=True, + ).replace("<", "\\u003c") + html = ( + index_path.read_text(encoding="utf-8") + .replace("__FULL_DUPLEX_CONFIG__", config) + .replace("__FULL_DUPLEX_APP_VERSION__", app_version) + ) + return HTMLResponse(html, headers={"Cache-Control": "no-store"}) + + @app.get("/healthz") + def healthz() -> Response: + return Response(content="ok", media_type="text/plain") + + @app.websocket("/v1/realtime") + async def realtime_proxy(websocket: WebSocket) -> None: + await websocket.accept() + query = urlencode(websocket.query_params.multi_items()) + backend_url = _join_ws_url(ws_backend, "/v1/realtime", query) + logger.info("Proxying Realtime WebSocket to %s", backend_url) + try: + async with websockets.connect( + backend_url, + max_size=64 * 1024 * 1024, + ) as backend: + tasks = { + asyncio.create_task(_pump_client_to_backend(websocket, backend)), + asyncio.create_task(_pump_backend_to_client(websocket, backend)), + } + done, pending = await asyncio.wait( + tasks, + return_when=asyncio.FIRST_COMPLETED, + ) + for task in pending: + task.cancel() + await asyncio.gather(*pending, return_exceptions=True) + for result in await asyncio.gather(*done, return_exceptions=True): + if isinstance(result, BaseException) and not _expected_proxy_close(result): + raise result + except (WebSocketDisconnect, websockets.ConnectionClosed): + return + except (OSError, TimeoutError) as exc: + message = ( + f"Cannot connect to realtime backend {ws_backend}. " + "Check that vllm serve is running and --backend matches its --port. " + "For containers or remote hosts, use the backend address reachable from the UI server." + ) + logger.warning("%s (%s)", message, exc) + with contextlib.suppress(Exception): + await websocket.send_json({"type": "error", "code": "backend_unavailable", "error": message}) + await websocket.close(code=1011) + except Exception: + logger.exception("Realtime WebSocket proxy failed") + with contextlib.suppress(Exception): + await websocket.close(code=1011) + finally: + with contextlib.suppress(Exception): + await websocket.close() + + app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") + return app + + +def main(default_profile: str = "minicpm-native") -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--host", default="0.0.0.0") + parser.add_argument("--port", type=int, default=7862) + parser.add_argument("--profile", choices=("minicpm-native", "qwen3-turn"), default=default_profile) + parser.add_argument("--ws-backend", "--backend", dest="ws_backend") + mode = parser.add_mutually_exclusive_group() + mode.add_argument("--stt", dest="adapter", action="store_const", const="stt") + mode.add_argument("--vad", dest="adapter", action="store_const", const="vad") + parser.set_defaults(adapter="stt") + parser.add_argument( + "--public-realtime-url", + help="Browser-visible ws:// or wss:// Realtime URL; defaults to the same-origin proxy.", + ) + parser.add_argument("--model") + parser.add_argument( + "--ref-audio", + help=( + "Reference voice wav for TTS voice cloning, e.g. the " + "official MiniCPM-o-Demo assets/ref_audio/ref_minicpm_signature.wav" + ), + ) + args = parser.parse_args() + native = args.profile == "minicpm-native" + if native and not args.ref_audio: + parser.error("--ref-audio is required for the minicpm-native profile") + if not native and args.ref_audio: + parser.error("--ref-audio is only supported by minicpm-native") + args.model = args.model or ("openbmb/MiniCPM-o-4_5" if native else "Qwen/Qwen3-Omni-30B-A3B-Instruct") + args.ws_backend = args.ws_backend or ("ws://127.0.0.1:8099" if native else "ws://127.0.0.1:8091") + + logging.basicConfig(level=logging.INFO) + uvicorn.run( + build_app( + ws_backend=args.ws_backend, + model=args.model, + public_realtime_url=args.public_realtime_url, + ref_audio=args.ref_audio, + profile=args.profile, + adapter=args.adapter, + ), + host=args.host, + port=args.port, + log_level="info", + ) + + +if __name__ == "__main__": + main() diff --git a/tests/config/test_environment_variables.py b/tests/config/test_environment_variables.py index b73062689..eb073f0c7 100644 --- a/tests/config/test_environment_variables.py +++ b/tests/config/test_environment_variables.py @@ -167,7 +167,7 @@ def test_inventory_matches_reviewed_snapshot_counts(): EnvironmentVariableCategory.PUBLIC_OMNI: 28, EnvironmentVariableCategory.INHERITED_VLLM: 20, EnvironmentVariableCategory.PLATFORM_EXTERNAL: 27, - EnvironmentVariableCategory.MODEL_SPECIFIC: 64, + EnvironmentVariableCategory.MODEL_SPECIFIC: 65, EnvironmentVariableCategory.BENCHMARK_TRANSITIONAL: 20, EnvironmentVariableCategory.INTERNAL: 2, } @@ -181,7 +181,7 @@ def test_inventory_matches_reviewed_snapshot_counts(): ModelEnvironmentVariableDisposition.PROMOTE: 38, ModelEnvironmentVariableDisposition.REQUEST_SCOPE: 6, ModelEnvironmentVariableDisposition.EXTERNAL: 0, - ModelEnvironmentVariableDisposition.INTERNALIZE: 15, + ModelEnvironmentVariableDisposition.INTERNALIZE: 16, ModelEnvironmentVariableDisposition.DEPRECATE_REMOVE: 5, } diff --git a/tests/engine/duplex/test_engine_session.py b/tests/engine/duplex/test_engine_session.py index fccaa59e8..c90291a81 100644 --- a/tests/engine/duplex/test_engine_session.py +++ b/tests/engine/duplex/test_engine_session.py @@ -567,10 +567,10 @@ def test_overlap_policy_defaults_and_invalid_values_to_listen_only(): assert DuplexSessionConfig.from_event({"session": {"overlap_policy": "bogus"}}).overlap_policy == "listen_only" -def test_capabilities_as_dict_reports_model_native_duplex_constants(): +def test_capabilities_as_dict_reports_declared_implementation(): caps = DuplexCapabilities().as_dict() - assert caps["implementation_level"] == "model_native_duplex" + assert caps["implementation_level"] == "turn_based_duplex" assert caps["input_modes"] == ["append_audio_chunk"] assert caps["supports_kv_lease"] is False assert caps["supports_core_kv_lease"] is False diff --git a/tests/engine/duplex/test_session_manager.py b/tests/engine/duplex/test_session_manager.py index 4f1962ee5..bccce18f1 100644 --- a/tests/engine/duplex/test_session_manager.py +++ b/tests/engine/duplex/test_session_manager.py @@ -196,7 +196,7 @@ class FakePlugin(DuplexModelPlugin): def capabilities(self, *, max_sessions: int) -> DuplexCapabilities: del max_sessions - return DuplexCapabilities(supports_input_append=True) + return DuplexCapabilities(supports_input_append=True, supports_core_resumable_request=True) def validate_client_extra_body(self, extra_body: object) -> None: pass @@ -408,7 +408,9 @@ async def test_open_answers_with_capabilities_and_emits_session_created() -> Non assert result.control_id == "open-sid-open" assert result.session_id == "sid-open" assert result.lease_generation == 0 - assert result.capabilities == DuplexCapabilities(supports_input_append=True) + assert result.capabilities == DuplexCapabilities( + supports_input_append=True, supports_core_resumable_request=True + ) assert result.public_session is not None assert result.public_session["id"] == "sid-open" assert result.public_session["voice"] == "test" diff --git a/tests/engine/duplex/test_vad_backend.py b/tests/engine/duplex/test_vad_backend.py index 8a22259ac..f3163ff4e 100644 --- a/tests/engine/duplex/test_vad_backend.py +++ b/tests/engine/duplex/test_vad_backend.py @@ -107,8 +107,8 @@ def test_endpoint_decisions_match_the_serving_side_policy( # --------------------------------------------------------------------------- # -def test_a_loud_frame_does_not_restart_the_silence_timer() -> None: - """Silero v6.2 hysteresis: once silence is running, a loud frame only delays it. +def test_hysteresis_band_does_not_restart_the_silence_timer() -> None: + """Scores between the negative and activation thresholds preserve the timer. The frame at 0.45 sits above the negative threshold (0.35) but below the activation threshold. It must not cancel the pending endpoint, or a speaker @@ -164,9 +164,14 @@ def _pcm16_b64(samples: np.ndarray) -> str: def test_pcm16_input_is_accepted_alongside_float32() -> None: config = SileroVADConfig(threshold=0.5) - tone = np.zeros(FRAME * 2, dtype=np.float32) + tone: np.ndarray = np.zeros(FRAME * 2, dtype=np.float32) seen: list[int] = [] - vad = SileroStreamingVAD(config, frame_scorer=lambda frame: seen.append(frame.size) or 0.0) + + def _score_frame(frame: np.ndarray) -> float: + seen.append(frame.size) + return 0.0 + + vad = SileroStreamingVAD(config, frame_scorer=_score_frame) vad.process_base64(_pcm16_b64(tone), fmt="pcm16", sample_rate_hz=SAMPLE_RATE_HZ) assert seen == [FRAME, FRAME] @@ -272,7 +277,7 @@ def test_the_onnx_backend_holds_the_upstream_silero_v62_contract(monkeypatch, tm intra_op_num_threads = 0 class FakeInferenceSession: - def __init__(self, path: str, *, providers: list[str], sess_options: object) -> None: + def __init__(self, path: str, *, providers: list[str], sess_options: FakeSessionOptions) -> None: self.providers = providers self.sess_options = sess_options self.calls: list[dict[str, np.ndarray]] = [] @@ -313,7 +318,7 @@ def test_the_onnx_backend_holds_the_upstream_silero_v62_contract(monkeypatch, tm assert session.calls[0]["sr"].item() == SAMPLE_RATE_HZ state = backend.new_state() - first_frame = np.arange(backend.frame_samples, dtype=np.float32) + first_frame: np.ndarray = np.arange(backend.frame_samples, dtype=np.float32) probability, state = backend.infer(first_frame, state) first_call = session.calls[1] @@ -377,3 +382,17 @@ def test_the_documented_server_vad_fields_are_all_accepted() -> None: ) is None ) + + +@pytest.mark.parametrize("speech_probability", [0.5, 0.9]) +def test_resumed_speech_cancels_pending_silence(speech_probability): + config = SileroVADConfig(threshold=0.5, prefix_padding_ms=0, silence_duration_ms=96, min_speech_duration_ms=32) + # 32 ms pause, resumed speech, then a full 96 ms pause: one utterance. + scores = [0.9, 0.1, speech_probability, speech_probability, 0.1, 0.1, 0.1] + assert _drive(config, scores) == [("start", 0), ("stop", 224)] + + +def test_short_pauses_do_not_split_continuous_speech_into_repeated_turns(): + config = SileroVADConfig(threshold=0.5, prefix_padding_ms=0, silence_duration_ms=500, min_speech_duration_ms=32) + scores = [0.9] + [0.1, 0.9, 0.9, 0.9] * 20 + assert _drive(config, scores) == [("start", 0)] diff --git a/tests/engine/test_duplex_orchestrator.py b/tests/engine/test_duplex_orchestrator.py index 5b8130c36..703af6a35 100644 --- a/tests/engine/test_duplex_orchestrator.py +++ b/tests/engine/test_duplex_orchestrator.py @@ -515,3 +515,48 @@ async def test_a_dead_replica_closes_the_sessions_it_was_serving() -> None: messages = [output_q.get_nowait() for _ in range(output_q.qsize())] types = [getattr(getattr(m, "event", None), "type", type(m).__name__) for m in messages] assert types[-1] in {"session.expired", "session.closed"}, f"the client needs a terminal event, got {types}" + + +@pytest.mark.asyncio +async def test_turn_plugin_processes_multimodal_prompt_before_stage_submission(monkeypatch): + from vllm_omni.engine.orchestrator import build_engine_core_request_from_tokens + from vllm_omni.model_executor.models.qwen3_omni.duplex.plugin import Qwen3OmniDuplexPlugin + + orchestrator, clients, rpc_q, _ = _build() + plugin = Qwen3OmniDuplexPlugin(_encode_audio) + plugin.processor = SimpleNamespace(apply_chat_template=lambda messages, **kwargs: "audio prompt") + orchestrator.plugin = orchestrator.session_manager.plugin = plugin + seen = [] + + def process_inputs(**kwargs): + seen.append(kwargs) + return build_engine_core_request_from_tokens( + request_id=kwargs["request_id"], + prompt={"prompt_token_ids": [1, 2]}, + params=kwargs["params"], + model_config=orchestrator.stage_pools[0].stage_vllm_config.model_config, + resumable=kwargs["resumable"], + ) + + monkeypatch.setattr( + orchestrator, "_get_stage_input_processor", lambda stage_id: SimpleNamespace(process_inputs=process_inputs) + ) + await orchestrator._dispatch_message( + OpenDuplexSessionMessage( + control_id="open-qwen", + session_id=SESSION_ID, + session_config=DuplexSessionConfig(model="qwen", modalities=["text", "audio"]), + ) + ) + assert (await rpc_q.get()).ok + try: + await _submit(orchestrator, _append_audio()) + assert not clients[0].add_request_calls + await _submit(orchestrator, commands.Commit(final=True, create_response=True)) + assert len(seen) == 1 + assert seen[0]["prompt"]["multi_modal_data"]["audio"][0][0].shape == (16000,) + submitted = clients[0].add_request_calls[0][0] + assert submitted.resumable is False + assert not orchestrator.request_states[submitted.request_id].streaming.enabled + finally: + await _close(orchestrator, rpc_q) diff --git a/tests/entrypoints/openai_api/test_api_server_guards.py b/tests/entrypoints/openai_api/test_api_server_guards.py index ebc08662f..43d88d692 100644 --- a/tests/entrypoints/openai_api/test_api_server_guards.py +++ b/tests/entrypoints/openai_api/test_api_server_guards.py @@ -47,6 +47,7 @@ from argparse import Namespace from contextlib import asynccontextmanager from dataclasses import dataclass from types import SimpleNamespace +from typing import Any import pytest from fastapi import FastAPI, HTTPException @@ -564,7 +565,7 @@ async def test_timestamp_middleware_stamps_http_and_passes_websocket(monkeypatch Fails if HTTP requests lose ``request_timestamp``, or WebSocket scopes start getting stamped / mutated incorrectly. """ - captured: dict[str, object] = {} + captured: dict[str, Any] = {} def fake_build_openai_app(args, supported_tasks): return FastAPI() @@ -592,8 +593,8 @@ async def test_timestamp_middleware_stamps_http_and_passes_websocket(monkeypatch await api_server.omni_run_server_worker("127.0.0.1:8000", _FakeSocket(), _minimal_args()) middleware = captured["served_app"] - http_scope = {"type": "http", "state": {}} - ws_scope = {"type": "websocket"} + http_scope: dict[str, Any] = {"type": "http", "state": {}} + ws_scope: dict[str, Any] = {"type": "websocket"} async def _receive(): return {"type": "http.disconnect"} @@ -920,11 +921,10 @@ async def test_pure_diffusion_app_state_key_snapshot(monkeypatch) -> None: engine = _FakeEngineClient(stage_configs=[stage]) def _for_diffusion_factory(label: str): - @classmethod def _factory(cls, *args, **kwargs): return _marker(label) - return _factory + return classmethod(_factory) monkeypatch.setattr(api_server.OmniOpenAIServingChat, "for_diffusion", _for_diffusion_factory("chat")) monkeypatch.setattr(api_server.OmniOpenAIServingChatBatch, "for_diffusion", _for_diffusion_factory("chat_batch")) @@ -965,13 +965,11 @@ async def test_pure_diffusion_speech_forwards_media_access_args(monkeypatch) -> speech_kwargs = {} def _for_diffusion_factory(label: str): - @classmethod def _factory(cls, *args, **kwargs): return _marker(label) - return _factory + return classmethod(_factory) - @classmethod def _speech_factory(cls, *args, **kwargs): speech_kwargs.update(kwargs) return _marker("speech") @@ -985,7 +983,7 @@ async def test_pure_diffusion_speech_forwards_media_access_args(monkeypatch) -> ) monkeypatch.setattr(api_server.OmniOpenAIServingVideo, "for_diffusion", _for_diffusion_factory("video")) monkeypatch.setattr(api_server.OmniStreamingVideoOutputHandler, "__init__", lambda self, *a, **k: None) - monkeypatch.setattr(api_server.OmniOpenAIServingSpeech, "for_diffusion", _speech_factory) + monkeypatch.setattr(api_server.OmniOpenAIServingSpeech, "for_diffusion", classmethod(_speech_factory)) monkeypatch.setattr( api_server.ServingRealtimeRobotOpenPI, "create_policy_server", diff --git a/tests/entrypoints/openai_api/test_duplex_api_server.py b/tests/entrypoints/openai_api/test_duplex_api_server.py index 9cf58b114..0d9eebe5b 100644 --- a/tests/entrypoints/openai_api/test_duplex_api_server.py +++ b/tests/entrypoints/openai_api/test_duplex_api_server.py @@ -490,3 +490,31 @@ async def test_a_model_that_does_not_declare_chat_completions_does_not_get_the_r assert state.openai_serving_chat is None assert isinstance(state.openai_serving_duplex, OmniDuplexSessionHandler) + + +def test_qwen_plugin_preserves_custom_turn_deployment_default(monkeypatch, tmp_path): + from vllm_omni.model_executor.models.qwen3_omni.pipeline import QWEN3_OMNI_PIPELINE + + monkeypatch.setattr(StageConfigFactory, "get_pipeline_config", lambda **kwargs: QWEN3_OMNI_PIPELINE) + deploy = tmp_path / "qwen.yaml" + deploy.write_text("stages: []\n") + assert not api_server._should_serve_duplex("qwen", {"deploy_config": str(deploy)}) + deploy.write_text("session_mode: duplex\nstages: []\n") + assert api_server._should_serve_duplex("qwen", {"deploy_config": str(deploy)}) + + +@pytest.mark.parametrize("flag", ["1", "true", "on"]) +def test_turn_deployment_rejects_explicit_duplex_without_falling_back_to_stt(flag: str) -> None: + app = _duplex_app(None) + # Even an available STT service must not silently accept a duplex request. + app.state.openai_serving_realtime = object() + with TestClient(app) as client: + with client.websocket_connect(f"/v1/realtime?duplex={flag}") as websocket: + assert websocket.receive_json() == { + "type": "error", + "code": "unsupported", + "error": "VAD realtime is not enabled", + } + with pytest.raises(WebSocketDisconnect) as exc: + websocket.receive_text() + assert exc.value.code == 1008 diff --git a/tests/examples/test_minicpmo_realtime_web_server.py b/tests/examples/test_minicpmo_realtime_web_server.py index 6c35760a7..44f04eb1b 100644 --- a/tests/examples/test_minicpmo_realtime_web_server.py +++ b/tests/examples/test_minicpmo_realtime_web_server.py @@ -1,3 +1,6 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + import hashlib import importlib.util import sys @@ -70,7 +73,12 @@ def test_build_app_injects_public_realtime_url(): def test_build_app_versions_client_bundle_and_disables_index_cache(): client = TestClient(build_app()) expected_version_hash = hashlib.sha256() - for asset_name in ("app.js", "pcm_worklet.js", "playback_worklet.js"): + for asset_name in ( + "app.js", + "pcm_worklet.js", + "playback_worklet.js", + *[str(p.relative_to(STATIC_DIR)) for p in sorted((STATIC_DIR / "profiles").glob("*.js"))], + ): expected_version_hash.update((STATIC_DIR / asset_name).read_bytes()) expected_version = expected_version_hash.hexdigest()[:12] @@ -97,3 +105,41 @@ def test_build_app_exposes_realtime_websocket_and_static_assets(): assert client.get("/static/app.js").status_code == 200 assert client.get("/static/pcm_worklet.js").status_code == 200 assert client.get("/static/playback_worklet.js").status_code == 200 + + +def test_qwen_config_does_not_inject_reference_voice(): + client = TestClient(build_app(profile="qwen3-turn", adapter="vad", model="Qwen/Qwen3-Omni-30B-A3B-Instruct")) + page = client.get("/").text + assert '"profile": "qwen3-turn"' in page + assert '"adapter": "vad"' in page + assert '"refAudio": null' in page + for name in ("common", "minicpm_native", "qwen3_turn"): + assert client.get(f"/static/profiles/{name}.js").status_code == 200 + with pytest.raises(ValueError, match="ref-audio"): + build_app(profile="qwen3-turn", ref_audio="unused.wav") + + +def test_config_cannot_close_its_script_element(): + page = TestClient(build_app(model="")).get("/").text + assert "