diff --git a/docs/serving/README.md b/docs/serving/README.md index 4aa470c6c..86793d3c8 100644 --- a/docs/serving/README.md +++ b/docs/serving/README.md @@ -154,18 +154,16 @@ All six routes are model- or configuration-dependent. In particular, pipeline declares a `duplex_plugin` and its deploy configuration sets `session_mode: duplex` (with `session_mode: turn` the same model boots the ordinary turn-based serving stack instead). On a duplex server a stock -Realtime client needs no vendor query parameter; `duplex=0` selects the -turn-based Realtime handler, which a duplex server does not mount, so that -connection is refused with `Realtime API is not available`. Such a server -serves the websocket route plus `POST /v1/chat/completions`, `/v1/models` -and `/health`, and no other turn-based HTTP route. The chat route is the -ordinary chat service running on the duplex engine: a request is a -turn-based generation on the same stages, served alongside the live -websocket sessions; it opens no duplex session and holds no -`duplex_session.max_sessions` slot -- see [Full Duplex](full_duplex_api.md). -On a server that is not duplex, an explicit `?duplex=1` is refused rather -than answered by the turn-based handler, so a client that requested duplex -never silently gets the other protocol. +Realtime client needs no vendor query parameter; `duplex=0` opts out of native +duplex and selects the turn-based Realtime handler. It runs generation on the +same stages, alongside live websocket sessions, and opens no duplex session +or consumes a `duplex_session.max_sessions` slot -- see +[Full Duplex](full_duplex_api.md). +Such a server serves the websocket route plus `POST /v1/chat/completions`, +`/v1/models` and `/health`, and no other turn-based HTTP route. On a server +that is not duplex, an explicit `?duplex=1` is refused rather than answered by +the turn-based handler, so a client that requested duplex never silently gets +the other protocol. ## Related Endpoints diff --git a/examples/online_serving/qwen3_omni/README.md b/examples/online_serving/qwen3_omni/README.md index 5eb9a2042..8e56ce01f 100644 --- a/examples/online_serving/qwen3_omni/README.md +++ b/examples/online_serving/qwen3_omni/README.md @@ -277,61 +277,6 @@ cd examples/online_serving/qwen3_omni python examples/online_serving/openai_chat_completion_client_for_multimodal_generation.py --model Qwen/Qwen3-Omni-30B-A3B-Instruct --query-type use_image --port 8091 --host "localhost" ``` -#### Realtime WebSocket client (`openai_realtime_client.py`) - -[`openai_realtime_client.py`](./openai_realtime_client.py) connects to **`ws://:/v1/realtime`**, streams a local WAV as **PCM16 mono @ 16 kHz** in fixed-size chunks, and receives **`response.output_audio.*`**, **`response.output_text.*`**, and transcript events. By default it uses explicit `input_audio_buffer.commit`; pass **`--server-vad`** to let the server detect speech endpoints and commit the turn automatically. The client concatenates audio deltas and writes **`--output-wav`** (model output is typically **24 kHz**). Optional **`--delta-dump-dir`** saves each delta as `delta_000001.wav`, … for debugging. - -Streaming input works well for translation-style use cases; if the Thinker runs while input is still incomplete, consider limiting **`max_tokens`** in your session / server defaults to avoid over-generation. - -**Dependencies:** - -```bash -pip install websockets -``` - -**From this directory** (`examples/online_serving/qwen3_omni`): - -```bash -python openai_realtime_client.py \ - --url ws://localhost:8091/v1/realtime \ - --model Qwen/Qwen3-Omni-30B-A3B-Instruct \ - --input-wav /path/to/input_16k_mono.wav \ - --output-wav realtime_output.wav \ - --server-vad \ - --delta-dump-dir ./rt_delta_wavs -``` - -**Arguments:** - -| Flag | Default | Description | -| ------ | --------- | ------------- | -| `--url` | `ws://localhost:8091/v1/realtime` | Full WebSocket URL including path | -| `--model` | `Qwen/Qwen3-Omni-30B-A3B-Instruct` | Must match the served model (sent in `session.update`) | -| `--input-wav` | *(required)* | Input WAV: mono, 16-bit PCM, **16 kHz** | -| `--output-wav` | `realtime_output.wav` | Output path for concatenated reply audio | -| `--output-text` | *(optional)* | If set, write final transcription text to this path | -| `--chunk-ms` | `200` | Size of each uploaded audio chunk (milliseconds of audio) | -| `--send-delay-ms` | `0` | Delay between chunk sends (simulate realtime upload) | -| `--delta-dump-dir` | *(optional)* | Directory to write per-`response.output_audio.delta` WAV files | -| `--server-vad` | disabled | Enable server-side Silero VAD instead of sending explicit commits | -| `--num-requests` | `1` | Number of sequential sessions (see `--concurrency`) | -| `--concurrency` | `1` | Max concurrent WebSocket sessions when `--num-requests` > 1 | - -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 -``` - -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: - `--query-type` (or `-q`): Query type (default: `use_video`). Options: `text`, `use_audio`, `use_image`, `use_video` @@ -358,6 +303,13 @@ python examples/online_serving/openai_chat_completion_client_for_multimodal_gene bash run_curl_multimodal_generation.sh use_image ``` +### Realtime WebSocket + +The [shared Realtime UI](../realtime_web/README.md#qwen3-explicit-turn-realtime) +supports explicit turns through `/v1/realtime?duplex=0`. The +[Server VAD setup](../realtime_web/README.md#qwen3-with-server-vad-automatic-turns) +uses the duplex endpoint. + ### FAQ ## Modality control @@ -582,26 +534,6 @@ The gradio script supports the following arguments: 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 @@ -610,7 +542,7 @@ ONNX artifact, following the After its health check succeeds, start the UI: ```bash -python -m examples.online_serving.qwen3_omni.realtime_web \ +python -m examples.online_serving.realtime_web --profile qwen3-turn \ --backend ws://127.0.0.1:8091 --vad --port 7863 ``` @@ -621,9 +553,7 @@ 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. +For explicit turns, use `--manual` with the shared UI as described in the +[shared UI guide](../realtime_web/README.md#qwen3-explicit-turn-realtime). diff --git a/examples/online_serving/realtime_web/README.md b/examples/online_serving/realtime_web/README.md index 7aba2ee10..28af5c361 100644 --- a/examples/online_serving/realtime_web/README.md +++ b/examples/online_serving/realtime_web/README.md @@ -7,14 +7,14 @@ This implements [RFC #7222](https://github.com/vllm-project/vllm-omni/issues/722 | 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 --manual` | Press **Send turn**; mic pauses while the reply plays | No | No | | `qwen3-turn --vad` | Server detects trailing silence; speech interrupts replies | Sampled frames with each spoken turn | Yes | | `aura-ptt` | Hold **Hold to talk**; release commits. Hold also stops local playback | Sticky frames with speech; vision-follow every 2 frames only while unlocked (locked from PTT release until text-final / listen) | 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. +KV decoding. ## MiniCPM: existing command stays valid @@ -62,45 +62,36 @@ just-committed speech turn on Stage0/1 (`max_num_seqs=1`). Audio still playing does not hold the next vision turn once text is done. MiniCPM / Qwen profiles do not set `pushToTalk`, so the PTT control stays hidden. -## Qwen3: explicit-turn STT adapter +## Qwen3: explicit-turn Realtime -Start the backend, then the UI in a second terminal: +The shared UI sends OpenAI Realtime events to `/v1/realtime?duplex=0` and keeps +conversation history on the same connection: + +Start the turn-based backend: ```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. +In another terminal, check readiness and start the UI: -The adapter follows the shipped `qwen3_omni/openai_realtime_client.py` and -`vllm_omni/entrypoints/openai/realtime_connection.py`: +```bash +curl --fail http://127.0.0.1:8091/health +python -m examples.online_serving.realtime_web --profile qwen3-turn \ + --backend ws://127.0.0.1:8091 --manual --port 7863 +``` - -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. +Speak, then press **Send turn**. The client commits the audio and requests a +response; recording pauses while the reply plays. ## 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. +The plugin uses the pinned Silero v6.2 ONNX artifact from +`istupakov/silero-vad-onnx` and does not download it while serving. Make it +available in the Hugging Face cache or set `server_vad_model_path` to a local +copy in the overlay below. The backend requires ONNX Runtime; setting `--vad` +on the UI alone does not enable VAD. Create an overlay named `qwen3_vad.yaml`, replacing both paths with absolute paths on the backend host: @@ -124,7 +115,7 @@ 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 \ +python -m examples.online_serving.realtime_web --profile qwen3-turn \ --backend ws://127.0.0.1:8091 --vad --port 7863 ``` @@ -148,21 +139,17 @@ 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 +### Using the Qwen3 web UI -| 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 | +The shared UI supports both explicit turns and Server VAD. Use the manual UI +command above with the turn-based deployment, or start the VAD deployment above +and launch the UI with `--vad`. 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. +refresh the browser and reconnect. To enable VAD on a default backend, 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 @@ -210,8 +197,7 @@ 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. +extensions. This demo does not add WebRTC, semantic VAD, or tool calling. ## Shared host options @@ -219,11 +205,14 @@ The shared entry point can select a 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://127.0.0.1:8091 --vad --port 7863 ``` +This Qwen3 UI profile requires the server-VAD deployment above. + - `--backend` / `--ws-backend`: backend WebSocket origin (not the full `/v1/realtime` URL). - `--model`: override the wrapper's model name. +- `--manual` / `--vad`: Qwen turn control; manual turns are the default. - `--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). @@ -245,7 +234,7 @@ pytest -m cpu tests/model_executor/models/qwen3_omni/test_duplex_plugin.py \ tests/engine/test_duplex_orchestrator.py ``` -These CPU tests cover profile messages, two STT turns, VAD terminal/drain ordering, +These CPU tests cover profile message serialization, 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 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 index 5c4fa9d6f..af52c2d0d 100644 --- a/examples/online_serving/realtime_web/app/static/profiles/qwen3_turn.js +++ b/examples/online_serving/realtime_web/app/static/profiles/qwen3_turn.js @@ -2,24 +2,24 @@ 'use strict'; const profiles = root.OmniRealtimeProfiles; profiles['qwen3-turn'] = (config) => { - const vad = config.adapter === 'vad'; + const vad = config.turnMode === 'vad'; let interruptOnSpeech = 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.'; + const hint = 'This Qwen deployment needs session_mode: duplex with Server VAD enabled.'; return { - title: 'Qwen3-Omni Voice', eyebrow: vad ? 'Engine-owned duplex call' : 'Turn-based realtime call', + 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.', + : 'Speak, then press Send turn. Microphone upload pauses while the model answers. Conversation history stays on the Realtime session. 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, + waitForResponseDone: true, halfDuplex: !vad, deduplicateTranscript: true, closeSession: vad, + inputSampleRate: 24000, + readyEvent: 'session.updated', instructions: true, sendIntervalMs: 200, presets: { assistant: 'You are a helpful assistant. Answer clearly and concisely.' }, url(config, location) { const url = profiles.url(config, location); @@ -33,22 +33,17 @@ initialMessages(config, instructions) { imageItems = []; interruptOnSpeech = vad; - 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: { + const session = { model: config.model, 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 }, + turn_detection: vad ? { type: 'server_vad', threshold: 0.5, prefix_padding_ms: 300, + silence_duration_ms: 500, create_response: true, interrupt_response: true } : null, } } }; + if (vad) session.overlap_policy = 'barge_in_on_speech'; + else session.type = 'realtime'; 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 }; - }, + append: (audio) => ({ type: 'input_audio_buffer.append', audio }), // 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 @@ -68,7 +63,10 @@ } }); return messages; }, - commitMessages: () => vad ? [] : [{ type: 'input_audio_buffer.commit', final: true }], + commitMessages: () => vad ? [] : [ + { type: 'input_audio_buffer.commit' }, + { type: 'response.create' }, + ], ack: (responseId, playedMs) => vad ? { type: 'playback.ack', response_id: responseId, item_id: `item_${responseId}`, played_ms: playedMs, committed_ms: playedMs } : null, mapEvent(event) { @@ -87,30 +85,28 @@ return { kind: 'interrupt' }; } 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. + if (event.type === 'response.output_text.delta') { return { kind: 'text', role: 'assistant', channel: 'text', text: event.delta || '' }; } - if (event.type === 'response.output_text.done' || event.type === 'transcription.done') { + if (event.type === 'response.output_text.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', responseId: event.response_id || event.response?.id || null }; } - if (event.type === 'input_audio_buffer.committed') return { kind: 'begin' }; + if (vad && event.type === 'input_audio_buffer.committed') return { kind: 'begin' }; + if (!vad && event.type === 'input_audio_buffer.cleared') return { kind: 'ignore' }; 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.'; + if (action.code === 'server_vad_initialization_failed') action.message = 'Server VAD could not load Silero. Configure the server artifact and restart the UI.'; else if (vad && action.code === 'unsupported') action.message += ` ${hint}`; } return action; }, - connectionHint: vad ? hint : 'Check the Qwen /v1/realtime STT endpoint and model name.', + connectionHint: vad ? hint : 'Check the Qwen turn-based /v1/realtime deployment and model name.', }; }; })(globalThis); diff --git a/examples/online_serving/realtime_web/server.py b/examples/online_serving/realtime_web/server.py index f6a10c4d3..60c3f1b74 100644 --- a/examples/online_serving/realtime_web/server.py +++ b/examples/online_serving/realtime_web/server.py @@ -74,12 +74,12 @@ def build_app( public_realtime_url: str | None = None, ref_audio: str | None = None, profile: str = "minicpm-native", - adapter: str = "stt", + turn_mode: str = "manual", ) -> FastAPI: if profile not in {"minicpm-native", "qwen3-turn", "aura-ptt"}: raise ValueError(f"Unknown profile: {profile}") - if adapter not in {"stt", "vad"}: - raise ValueError(f"Unknown adapter: {adapter}") + if turn_mode not in {"manual", "vad"}: + raise ValueError(f"Unknown turn mode: {turn_mode}") if profile == "qwen3-turn" and ref_audio: raise ValueError("Qwen3 profile does not accept --ref-audio") if profile == "aura-ptt" and ref_audio: @@ -116,7 +116,7 @@ def build_app( { "model": model, "profile": profile, - "adapter": adapter, + "turnMode": turn_mode, "realtimePath": public_realtime_url or "v1/realtime", "refAudio": ref_audio_uri, "appVersion": app_version, @@ -190,9 +190,9 @@ def main(default_profile: str = "minicpm-native") -> None: parser.add_argument("--profile", choices=("minicpm-native", "qwen3-turn", "aura-ptt"), 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") + mode.add_argument("--manual", dest="turn_mode", action="store_const", const="manual") + mode.add_argument("--vad", dest="turn_mode", action="store_const", const="vad") + parser.set_defaults(turn_mode="manual") parser.add_argument( "--public-realtime-url", help="Browser-visible ws:// or wss:// Realtime URL; defaults to the same-origin proxy.", @@ -225,7 +225,7 @@ def main(default_profile: str = "minicpm-native") -> None: public_realtime_url=args.public_realtime_url, ref_audio=args.ref_audio, profile=args.profile, - adapter=args.adapter, + turn_mode=args.turn_mode, ), host=args.host, port=args.port, diff --git a/tests/entrypoints/openai_api/test_api_server_guards.py b/tests/entrypoints/openai_api/test_api_server_guards.py index f59ebf7b5..7bae57d36 100644 --- a/tests/entrypoints/openai_api/test_api_server_guards.py +++ b/tests/entrypoints/openai_api/test_api_server_guards.py @@ -58,7 +58,7 @@ from starlette.requests import Request from starlette.websockets import WebSocketDisconnect from vllm.v1.engine.exceptions import EngineDeadError, EngineGenerateError -from vllm_omni.entrypoints.openai import api_server, realtime_connection +from vllm_omni.entrypoints.openai import api_server from vllm_omni.entrypoints.serve.utils import errors as serve_errors pytestmark = [pytest.mark.core_model, pytest.mark.cpu] @@ -670,21 +670,15 @@ async def test_realtime_route_defaults_to_configured_duplex_handler( async def handle_realtime_session(self, _websocket) -> None: calls.append("duplex") - class _LegacyConnection: - async def handle_connection(self) -> None: - calls.append("legacy") + async def _dispatch_turn_based(_websocket) -> None: + calls.append("legacy") - monkeypatch.setattr( - realtime_connection, - "RealtimeConnection", - lambda _websocket, _serving: _LegacyConnection(), - ) + monkeypatch.setattr(api_server, "dispatch_realtime_websocket", _dispatch_turn_based) query_params = {} if duplex_query is None else {"duplex": duplex_query} websocket = SimpleNamespace( app=SimpleNamespace( state=SimpleNamespace( openai_serving_duplex=_DuplexHandler(), - openai_serving_realtime=object(), ) ), query_params=query_params, diff --git a/tests/entrypoints/openai_api/test_duplex_api_server.py b/tests/entrypoints/openai_api/test_duplex_api_server.py index 31421f6ff..653c32321 100644 --- a/tests/entrypoints/openai_api/test_duplex_api_server.py +++ b/tests/entrypoints/openai_api/test_duplex_api_server.py @@ -18,8 +18,7 @@ from vllm.entrypoints.openai.models.protocol import BaseModelPath from vllm_omni.config import stage_config from vllm_omni.config.config_factory import StageConfigFactory -from vllm_omni.config.omni_config import VllmOmniConfig -from vllm_omni.config.stage_config import DeployConfig, DuplexSessionRuntimeConfig, PipelineConfig, StagePipelineConfig +from vllm_omni.config.stage_config import DuplexSessionRuntimeConfig, PipelineConfig, StagePipelineConfig from vllm_omni.engine.duplex.config import DuplexCapabilities from vllm_omni.entrypoints.duplex.serving import OmniDuplexSessionHandler from vllm_omni.entrypoints.duplex_omni import DuplexOmni @@ -58,14 +57,12 @@ _DUPLEX_APP_STATE_KEYS = { "openai_streaming_speech", "openai_streaming_video", "openai_streaming_video_output", - "openai_serving_realtime", "openai_serving_realtime_robot", "anthropic_serving_messages", "openai_serving_duplex", "enable_server_load_tracking", "server_load_metrics", } -#: Every turn-based service, plus the Realtime route that is not the duplex one. #: ``openai_serving_chat`` is absent: a duplex engine also serves chat, through #: the ordinary turn-based service, because DuplexOmni extends AsyncOmni. #: ``online_renderer`` is absent for the same reason -- the chat service needs it. @@ -370,11 +367,31 @@ def _duplex_app(handler: object) -> FastAPI: app = FastAPI() app.include_router(api_server.router) app.state.openai_serving_duplex = handler - # A duplex server serves no turn-based Realtime route. - app.state.openai_serving_realtime = None return app +def _turn_based_realtime_app(handler: object | None, mocker) -> FastAPI: + app = _duplex_app(handler) + app.state.args = Namespace() + app.state.openai_serving_models = _FakeModels( + base_model_paths=[BaseModelPath(name="omni-model", model_path="omni-model")] + ) + app.state.engine_client = mocker.Mock() + app.state.openai_serving_chat = mocker.Mock() + return app + + +def _assert_openai_realtime_connection(client: TestClient, path: str) -> None: + with client.websocket_connect(path) as websocket: + assert websocket.receive_json()["type"] == "session.created" + assert websocket.receive_json()["type"] == "conversation.created" + websocket.send_json({"type": "unknown_event"}) + error = websocket.receive_json() + assert error["type"] == "error" + assert error["error"]["type"] == "invalid_request_error" + assert error["error"]["code"] == "invalid_event" + + @pytest.mark.parametrize("path", ["/v1/duplex", "/v1/realtime?duplex=1"]) def test_both_duplex_routes_reach_the_session_handler(path: str) -> None: """``/v1/duplex`` is an alias of ``/v1/realtime?duplex=1``: same handler, same protocol.""" @@ -423,73 +440,21 @@ def test_realtime_without_a_duplex_flag_reaches_the_session_handler(query: str) @pytest.mark.parametrize("query", ["?duplex=0", "?duplex=false", "?duplex=off"]) -def test_realtime_opted_out_of_duplex_falls_through_to_the_turn_based_route(query: str) -> None: - """An explicit opt-out selects the legacy handler, which a duplex server does not mount.""" +def test_realtime_opted_out_of_duplex_falls_through_to_the_turn_based_route(query: str, mocker) -> None: handler = _RecordingHandler() + app = _turn_based_realtime_app(handler, mocker) - with TestClient(_duplex_app(handler)) as client: - with client.websocket_connect(f"/v1/realtime{query}") as websocket: - assert websocket.receive_json() == { - "type": "error", - "error": "Realtime API is not available", - "code": "unsupported", - } - with pytest.raises(WebSocketDisconnect): - websocket.receive_text() + with TestClient(app) as client: + _assert_openai_realtime_connection(client, f"/v1/realtime{query}") assert handler.queries == [] -@pytest.fixture -def qwen3_realtime_stages(): - from vllm_omni.model_executor.models.qwen3_omni.pipeline import QWEN3_OMNI_PIPELINE - - return VllmOmniConfig.from_pipeline_config( - QWEN3_OMNI_PIPELINE, user_deploy_config=DeployConfig(async_chunk=True) - ).stage_configs - - -def test_qwen3_realtime_recognizes_typed_stages(qwen3_realtime_stages) -> None: - from vllm_omni.entrypoints.duplex.openai import supports_qwen3_omni_realtime - - assert supports_qwen3_omni_realtime(qwen3_realtime_stages) - - -@pytest.mark.parametrize("topology", ["missing-stage", "duplicate-stage", "other-architecture"]) -def test_qwen3_realtime_rejects_other_topologies(qwen3_realtime_stages, topology: str) -> None: - from vllm_omni.entrypoints.duplex.openai import supports_qwen3_omni_realtime - - stages = list(qwen3_realtime_stages) - if topology == "missing-stage": - stages.pop() - elif topology == "duplicate-stage": - stages[-1] = stages[0] - else: - stages[0].model_config.model_arch = "Qwen2_5OmniForConditionalGeneration" - - assert not supports_qwen3_omni_realtime(stages) - - -def test_qwen3_turn_deployment_uses_conformant_realtime(qwen3_realtime_stages, mocker) -> None: - app = _duplex_app(None) - app.state.stage_configs = qwen3_realtime_stages - app.state.args = Namespace() - app.state.openai_serving_models = _FakeModels( - base_model_paths=[BaseModelPath(name="qwen3-omni", model_path="qwen3-omni")] - ) - app.state.engine_client = mocker.Mock( - get_tokenizer=mocker.AsyncMock(return_value=mocker.Mock(chat_template="{{ messages }}")) - ) +def test_turn_based_deployment_uses_openai_realtime(mocker) -> None: + app = _turn_based_realtime_app(None, mocker) with TestClient(app) as client: - with client.websocket_connect("/v1/realtime?model=qwen3-omni") as websocket: - assert websocket.receive_json()["type"] == "session.created" - assert websocket.receive_json()["type"] == "conversation.created" - websocket.send_json({"type": "unknown_event"}) - error = websocket.receive_json() - assert error["type"] == "error" - assert error["error"]["type"] == "invalid_request_error" - assert error["error"]["code"] == "invalid_event" + _assert_openai_realtime_connection(client, "/v1/realtime?model=omni-model") # --------------------------------------------------------------------------- # @@ -560,8 +525,6 @@ def test_qwen_plugin_preserves_custom_turn_deployment_default(monkeypatch, tmp_p @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() == { diff --git a/tests/entrypoints/openai_api/test_realtime_prompt_preflight.py b/tests/entrypoints/openai_api/test_realtime_prompt_preflight.py new file mode 100644 index 000000000..632efb212 --- /dev/null +++ b/tests/entrypoints/openai_api/test_realtime_prompt_preflight.py @@ -0,0 +1,289 @@ +# SPDX-License-Identifier: Apache-2.0 +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project + +from __future__ import annotations + +import base64 +import io +import json +import wave +from types import SimpleNamespace +from typing import Any + +import numpy as np +import pytest +from vllm.sampling_params import RequestOutputKind, SamplingParams + +from tests.helpers.serving_chat import build_serving_chat +from vllm_omni.entrypoints.openai.realtime.connection import ( + SAMPLE_RATE_HZ, + OpenAIFullDuplexConnection, + _ResolvedResponse, +) +from vllm_omni.entrypoints.openai.realtime.session import ActiveResponse + +pytestmark = [pytest.mark.core_model, pytest.mark.cpu] + + +class _FakeRenderer: + def get_tokenizer(self) -> object: + return object() + + def __init__(self) -> None: + self.raw_prompt_lengths: list[int] = [] + self.tokenization_params: list[Any] = [] + self.engine_inputs: list[dict[str, Any]] = [] + self.conversations: list[list[dict[str, Any]]] = [] + + async def render_chat_async( + self, + conversations: list[list[dict[str, Any]]], + _chat_params: Any, + tok_params: Any, + ): + self.conversations.append(conversations[0]) + raw_prompt_length = 2 * len(conversations[0]) + self.raw_prompt_lengths.append(raw_prompt_length) + self.tokenization_params.append(tok_params) + # Model-side expansion makes the rendered prompt much longer than its text tokens. + engine_input = {"prompt_token_ids": [0] * (10 + 20 * raw_prompt_length)} + self.engine_inputs.append(engine_input) + return conversations, [engine_input] + + +class _FakeWebSocket: + def __init__(self) -> None: + self.messages: list[str] = [] + + async def send_text(self, message: str) -> None: + self.messages.append(message) + + +def _make_connection( + *, + max_model_len: int = 100, + websocket: Any = None, + engine: Any = None, + chat_handler: Any = None, +) -> tuple[OpenAIFullDuplexConnection, _FakeRenderer]: + renderer = _FakeRenderer() + model_config = SimpleNamespace(max_model_len=max_model_len, multimodal_config=None) + + async def preprocess_chat(request: Any, messages: list[dict[str, Any]], **kwargs: Any): + tok_params = kwargs.get("tok_params") + if tok_params is None: + tok_params = request.build_tok_params(model_config) + return await renderer.render_chat_async( + [messages], + kwargs.get("default_template_kwargs"), + tok_params, + ) + + connection = OpenAIFullDuplexConnection( + websocket=websocket, + engine=engine if engine is not None else SimpleNamespace(model_config=model_config), + model_name="test-model", + chat_handler=chat_handler + if chat_handler is not None + else SimpleNamespace( + renderer=renderer, + chat_template=None, + chat_template_content_format="auto", + _effective_chat_template_kwargs=lambda _request: {}, + _preprocess_chat=preprocess_chat, + _fix_minicpmo45_audio_stream_output_kinds=lambda params, _modalities: params, + ), + ) + return connection, renderer + + +@pytest.mark.asyncio +async def test_preflight_truncates_by_rendered_token_count() -> None: + connection, renderer = _make_connection() + + first = SimpleNamespace( + id="first", + type="message", + role="user", + content=[SimpleNamespace(type="input_text", text="first")], + ) + second = SimpleNamespace( + id="second", + type="message", + role="user", + content=[SimpleNamespace(type="input_text", text="second")], + ) + items = [first, second] + response = _ResolvedResponse( + input=items, + instructions=None, + modalities=["text"], + max_output_tokens="inf", + tools=None, + tool_choice="none", + metadata=None, + ) + + engine_input = await connection._prepare_engine_input_with_auto_truncation(response) + + assert renderer.raw_prompt_lengths == [4, 2, 2] + assert [len(prompt["prompt_token_ids"]) for prompt in renderer.engine_inputs] == [90, 50, 50] + assert all(params.max_total_tokens is None for params in renderer.tokenization_params) + assert all(params.max_output_tokens == 0 for params in renderer.tokenization_params) + assert items == [second] + assert engine_input is renderer.engine_inputs[-1] + + +@pytest.mark.asyncio +async def test_audio_prompt_uses_standard_chat_audio_content() -> None: + audio = b"\x00\x00\x00\x40" + connection, renderer = _make_connection(websocket=_FakeWebSocket()) + connection.session.input_audio_buffer.extend(audio) + item = await connection._commit_audio_buffer() + + assert item is not None + assert base64.b64decode(item.content[0].audio) == audio + + await connection._build_full_prompt(items=[item]) + + audio_content = renderer.conversations[0][0]["content"][0]["input_audio"] + assert audio_content["format"] == "wav" + with wave.open(io.BytesIO(base64.b64decode(audio_content["data"]))) as wav: + assert wav.getframerate() == SAMPLE_RATE_HZ + assert wav.getnchannels() == 1 + assert wav.getsampwidth() == 2 + assert wav.readframes(2) == audio + + +@pytest.mark.asyncio +async def test_generated_audio_is_resampled_to_realtime_rate() -> None: + source_rate = 32_000 + waveform = np.sin(2 * np.pi * 440 * np.arange(source_rate) / source_rate).astype(np.float32) + outputs: list[Any] = [] + for chunk in np.split(waveform, 2): + multimodal_output = {"audio": chunk, "sample_rate": source_rate} + outputs.append( + SimpleNamespace( + final_output_type="audio", + multimodal_output=multimodal_output, + outputs=[SimpleNamespace(multimodal_output=multimodal_output)], + ) + ) + + async def generate(**_kwargs): + for output in outputs: + yield output + + model_config = SimpleNamespace(max_model_len=100, multimodal_config=None) + engine = SimpleNamespace( + model_config=model_config, + default_sampling_params_list=[], + generate=generate, + ) + websocket = _FakeWebSocket() + connection, _ = _make_connection(websocket=websocket, engine=engine) + response = _ResolvedResponse( + input=[], + instructions=None, + modalities=["audio"], + max_output_tokens="inf", + tools=None, + tool_choice="none", + metadata=None, + ) + active = ActiveResponse(response_id="resp_test", request_id="req_test") + connection.session.active_response = active + + await connection._run_response(active.response_id, response, {"prompt_token_ids": []}) + + events = [json.loads(message) for message in websocket.messages] + audio_deltas = [event for event in events if event["type"] == "response.output_audio.delta"] + pcm = b"".join(base64.b64decode(event["delta"]) for event in audio_deltas) + assert active.item_id is not None + assert len(pcm) == SAMPLE_RATE_HZ * 2 + assert connection.session.item_duration_ms[active.item_id] == pytest.approx(1000) + + +@pytest.mark.asyncio +async def test_minicpmo45_realtime_keeps_thinker_final_only() -> None: + model_arch = "MiniCPMO45OmniForConditionalGeneration" + stages = [ + SimpleNamespace(engine_args=SimpleNamespace(model_arch=model_arch, model_stage="llm")), + SimpleNamespace(engine_args=SimpleNamespace(model_arch=model_arch, model_stage="tts")), + ] + submitted_params: list[list[SamplingParams]] = [] + + async def generate(**kwargs): + submitted_params.append(kwargs["sampling_params_list"]) + yield SimpleNamespace(final_output_type="text", outputs=[]) + + model_config = SimpleNamespace(max_model_len=100, multimodal_config=None) + chat_handler = build_serving_chat() + engine = chat_handler.engine_client + engine.model_config = model_config + engine.stage_configs = stages + engine.default_sampling_params_list = [SamplingParams(), SamplingParams()] + engine.generate = generate + websocket = _FakeWebSocket() + connection, _ = _make_connection( + websocket=websocket, + engine=engine, + chat_handler=chat_handler, + ) + response = _ResolvedResponse( + input=[], + instructions=None, + modalities=["audio"], + max_output_tokens="inf", + tools=None, + tool_choice="none", + metadata=None, + ) + active = ActiveResponse(response_id="resp_test", request_id="req_test") + connection.session.active_response = active + + await connection._run_response(active.response_id, response, {"prompt_token_ids": []}) + + assert len(submitted_params) == 1 + assert [params.output_kind for params in submitted_params[0]] == [ + RequestOutputKind.FINAL_ONLY, + RequestOutputKind.DELTA, + ] + + +@pytest.mark.parametrize( + ("audio_end_ms", "expected"), + [(0, ""), (250, "0"), (500, "01"), (999, "012"), (1000, "0123")], +) +def test_truncate_transcript_uses_item_audio_duration(audio_end_ms: float, expected: str) -> None: + connection, _ = _make_connection() + connection._tokenizer = SimpleNamespace(decode=lambda ids, **_kwargs: "".join(map(str, ids))) + connection.session.item_duration_ms["item"] = 1000 + connection.session.item_token_ids["item"] = [0, 1, 2, 3] + + assert connection._truncate_transcript("item", audio_end_ms) == expected + + +def test_truncate_transcript_falls_back_without_tokens_or_duration() -> None: + connection, _ = _make_connection() + connection._tokenizer = SimpleNamespace(decode=lambda ids, **_kwargs: "".join(map(str, ids))) + connection.session.item_token_ids["missing_duration"] = [0, 1, 2, 3] + connection.session.item_duration_ms["missing_tokens"] = 1000 + + assert connection._truncate_transcript("missing_duration", 500) == "" + assert connection._truncate_transcript("missing_tokens", 500) == "" + + +@pytest.mark.asyncio +async def test_rejected_response_create_does_not_cancel_active_response() -> None: + websocket = _FakeWebSocket() + connection, _ = _make_connection(max_model_len=5, websocket=websocket) + active_response = SimpleNamespace(response_id="active", request_id="active-request") + connection.session.active_response = active_response + + await connection._handle_response_create(SimpleNamespace(event_id="evt_create", response=None)) + + assert connection.session.active_response is active_response + assert not connection._response_cancel_event.is_set() + assert len(websocket.messages) == 1 + assert "exceeds the model's input token limit" in websocket.messages[0] diff --git a/tests/entrypoints/test_realtime_connection_helpers.py b/tests/entrypoints/test_realtime_connection_helpers.py index e795aa92d..07946c6af 100644 --- a/tests/entrypoints/test_realtime_connection_helpers.py +++ b/tests/entrypoints/test_realtime_connection_helpers.py @@ -1,62 +1,17 @@ # SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM project -"""Unit tests for realtime streaming helpers (PR #2581 /v1/realtime path).""" +# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project +"""Unit tests for AsyncOmni input streaming parameter validation.""" from __future__ import annotations -import base64 - -import numpy as np import pytest -import torch from vllm.sampling_params import RequestOutputKind, SamplingParams from vllm_omni.entrypoints.async_omni import AsyncOmni -from vllm_omni.entrypoints.openai.realtime_connection import RealtimeConnection pytestmark = [pytest.mark.core_model, pytest.mark.cpu] -@pytest.fixture -def realtime_conn() -> RealtimeConnection: - return RealtimeConnection.__new__(RealtimeConnection) - - -class TestRealtimeConnectionTensorAndPcm: - def test_tensor_to_numpy_none(self) -> None: - assert RealtimeConnection._tensor_to_numpy(None) is None - - def test_tensor_to_numpy_1d_numpy(self) -> None: - arr = np.array([1.0, 2.0], dtype=np.float64) - out = RealtimeConnection._tensor_to_numpy(arr) - assert out is not None - assert out.dtype == np.float32 - assert out.shape == (2,) - - def test_tensor_to_numpy_2d_numpy_flattened(self) -> None: - arr = np.array([[0.5], [-0.5]], dtype=np.float32) - out = RealtimeConnection._tensor_to_numpy(arr) - assert out is not None - assert out.shape == (2,) - - def test_tensor_to_numpy_torch(self) -> None: - t = torch.tensor([[0.25, -0.25]], dtype=torch.float32) - out = RealtimeConnection._tensor_to_numpy(t) - assert out is not None - assert out.shape == (2,) - np.testing.assert_allclose(out, [0.25, -0.25], rtol=1e-5) - - def test_pcm16_b64_roundtrip(self) -> None: - audio = np.array([0.0, 1.0, -1.0], dtype=np.float32) - b64 = RealtimeConnection._pcm16_b64(audio) - raw = base64.b64decode(b64) - assert len(raw) == 6 - pcm = np.frombuffer(raw, dtype=np.int16) - assert pcm[0] == 0 - assert pcm[1] == 32767 - assert pcm[2] == -32767 - - class TestAsyncOmniStreamingParamsValidation: def test_accepts_streaming_friendly_params(self) -> None: p = SamplingParams( diff --git a/tests/examples/test_minicpmo_realtime_web_server.py b/tests/examples/test_minicpmo_realtime_web_server.py index 0f7149a34..b041435ea 100644 --- a/tests/examples/test_minicpmo_realtime_web_server.py +++ b/tests/examples/test_minicpmo_realtime_web_server.py @@ -108,10 +108,10 @@ def test_build_app_exposes_realtime_websocket_and_static_assets(): 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")) + client = TestClient(build_app(profile="qwen3-turn", turn_mode="vad", model="Qwen/Qwen3-Omni-30B-A3B-Instruct")) page = client.get("/").text assert '"profile": "qwen3-turn"' in page - assert '"adapter": "vad"' in page + assert '"turnMode": "vad"' in page assert '"refAudio": null' in page for name in ("common", "minicpm_native", "qwen3_turn", "aura_ptt"): assert client.get(f"/static/profiles/{name}.js").status_code == 200 diff --git a/tests/examples/test_realtime_web_profiles.cjs b/tests/examples/test_realtime_web_profiles.cjs index b69bd159a..f813c227a 100644 --- a/tests/examples/test_realtime_web_profiles.cjs +++ b/tests/examples/test_realtime_web_profiles.cjs @@ -28,40 +28,57 @@ test('MiniCPM keeps native flags, reference voice, camera, acknowledgements and assert.equal(native.halfDuplex, false); }); -test('STT uses the shipped commit sequence and audio payload, including each new turn', () => { - const stt = profiles['qwen3-turn'](config); - for (let turn = 0; turn < 2; turn++) { - assert.deepEqual(plain(stt.initialMessages(config, 'ignored')), [ - { type: 'session.update', model: config.model }, - { type: 'input_audio_buffer.commit', final: false }, - ]); - assert.deepEqual(plain(stt.commitMessages()), [{ type: 'input_audio_buffer.commit', final: true }]); - } - assert.equal(stt.reconnectEachTurn, true); - assert.equal(stt.mapEvent({ type: 'response.output_audio.delta', audio: 'PCM' }).event.delta, 'PCM'); - assert.equal(stt.mapEvent({ type: 'response.output_audio.done' }).kind, 'done'); - assert.equal(stt.mapEvent({ type: 'transcription.delta', delta: 'hello' }).role, 'assistant'); +test('manual turns use standard Realtime events and keep one session', () => { + const manual = profiles['qwen3-turn'](config); + const [update] = plain(manual.initialMessages(config, 'help')); + assert.equal(update.type, 'session.update'); + assert.deepEqual(update.session, { + type: 'realtime', + model: config.model, + audio: { input: { + format: { type: 'audio/pcm', rate: 24000 }, + turn_detection: null, + } }, + instructions: 'help', + }); + assert.equal(manual.readyEvent, 'session.updated'); + assert.equal(manual.inputSampleRate, 24000); + assert.deepEqual(plain(manual.append('PCM')), { type: 'input_audio_buffer.append', audio: 'PCM' }); + assert.deepEqual(plain(manual.commitMessages()), [ + { type: 'input_audio_buffer.commit' }, + { type: 'response.create' }, + ]); + assert.equal(manual.reconnectEachTurn, undefined); + assert.equal(manual.mapEvent({ type: 'response.output_audio.delta', delta: 'PCM' }).event.delta, 'PCM'); + assert.equal(manual.mapEvent({ type: 'response.output_audio.done' }).kind, 'drain'); + assert.equal(manual.mapEvent({ type: 'response.done' }).kind, 'done'); + assert.equal(manual.mapEvent({ type: 'response.output_audio_transcript.delta', delta: 'hello' }).role, 'assistant'); + assert.equal(manual.mapEvent({ type: 'input_audio_buffer.cleared' }).kind, 'ignore'); + assert.equal(manual.mapEvent({ type: 'transcription.delta', delta: 'hello' }).kind, 'ignore'); }); test('Qwen enables camera and playback ACK only for the duplex VAD profile', () => { - for (const adapter of ['stt', 'vad']) { - const p = profiles['qwen3-turn']({ ...config, adapter }); + for (const turnMode of ['manual', 'vad']) { + const p = profiles['qwen3-turn']({ ...config, turnMode }); const url = new URL(p.url({ ...config, realtimePath: 'wss://backend/v1/realtime?native_duplex=1&minicpmo45_native_duplex=1' }, 'http://localhost/')); assert.equal(url.protocol, 'wss:'); - assert.equal(url.searchParams.get('duplex'), adapter === 'vad' ? '1' : '0'); + assert.equal(url.searchParams.get('duplex'), turnMode === 'vad' ? '1' : '0'); assert.equal(url.searchParams.has('native_duplex'), false); assert.equal(url.searchParams.has('minicpmo45_native_duplex'), false); assert.equal(p.append('PCM', 'JPEG').video_frames, undefined); - assert.equal(plain(p.imageMessages('JPEG')).length, adapter === 'vad' ? 1 : 0); - assert.equal(p.ack('r', 100)?.type || null, adapter === 'vad' ? 'playback.ack' : null); - assert.equal(p.camera, adapter === 'vad'); + assert.equal(plain(p.imageMessages('JPEG')).length, turnMode === 'vad' ? 1 : 0); + assert.equal(p.ack('r', 100)?.type || null, turnMode === 'vad' ? 'playback.ack' : null); + assert.equal(p.camera, turnMode === 'vad'); assert.equal(p.mapEvent({ type: 'response.listen' }).kind, 'ignore'); } }); test('VAD uses nested format and interruptible endpoint detection', () => { - const vad = profiles['qwen3-turn']({ ...config, adapter: 'vad' }); + const vad = profiles['qwen3-turn']({ ...config, turnMode: 'vad' }); const [update] = vad.initialMessages(config, 'help'); + assert.equal(update.session.type, undefined); + assert.equal(update.session.overlap_policy, 'barge_in_on_speech'); + assert.equal(update.session.instructions, 'help'); assert.deepEqual(plain(update.session.audio.input), { format: { type: 'audio/pcm', rate: 24000 }, turn_detection: { type: 'server_vad', threshold: 0.5, prefix_padding_ms: 300, @@ -78,7 +95,7 @@ test('VAD uses nested format and interruptible endpoint detection', () => { assert.match(vad.mapEvent({ type: 'error', code: 'server_vad_initialization_failed' }).message, /Silero/); }); -function shell(profileName, adapter = 'stt', options = {}) { +function shell(profileName, turnMode = 'manual', options = {}) { const elements = new Map(); class Element { constructor() { this.style = {}; this.listeners = {}; this.children = []; this.value = ''; this.textContent = ''; this.classList = { add() {}, remove() {}, toggle() {} }; } @@ -103,7 +120,7 @@ function shell(profileName, adapter = 'stt', options = {}) { this.readyState = 1; this.sent = []; sockets.push(this); queueMicrotask(() => { this.onopen?.(); - if (!options.silent) this.onmessage?.({ data: JSON.stringify(options.error || { type: adapter === 'stt' && profileName === 'qwen3-turn' ? 'session.created' : 'session.updated' }) }); + if (!options.silent) this.onmessage?.({ data: JSON.stringify(options.error || { type: 'session.updated' }) }); }); } send(value) { this.sent.push(JSON.parse(value)); } @@ -141,7 +158,7 @@ function shell(profileName, adapter = 'stt', options = {}) { URL, document, AudioContext, AudioWorkletNode, WebSocket: Socket, navigator: { mediaDevices: { getUserMedia: async () => ({ getTracks: () => [{ stop() {} }] }) } }, location: { href: 'https://localhost/' }, - OMNI_REALTIME_CONFIG: { ...config, profile: profileName, adapter }, + OMNI_REALTIME_CONFIG: { ...config, profile: profileName, turnMode }, addEventListener() {}, setTimeout(fn, ms) { const id = ++timerId; timers.set(id, { fn, ms }); return id; }, clearTimeout(id) { timers.delete(id); }, @@ -174,26 +191,53 @@ function shell(profileName, adapter = 'stt', options = {}) { }; } -test('shared shell STT sends final commit and starts a fresh second turn only after playback', async () => { +test('shared shell sends standard manual turns and keeps the Realtime session open', async () => { const app = shell('qwen3-turn'); await app.ui.startSession(); assert.equal(app.ui.microphoneUploadEnabled(), true); + assert.deepEqual(app.sockets[0].sent[0], { + type: 'session.update', + session: { + type: 'realtime', + model: config.model, + audio: { input: { + format: { type: 'audio/pcm', rate: 24000 }, + turn_detection: null, + } }, + instructions: 'You are a helpful assistant. Answer clearly and concisely.', + }, + }); app.ui.capture(); app.ui.flushCapture(); + const append = app.sockets[0].sent.at(-1); + assert.equal(append.type, 'input_audio_buffer.append'); + assert.equal(typeof append.audio, 'string'); + assert.equal(append.format, undefined); + assert.equal(append.sample_rate_hz, undefined); app.elements.get('sendTurnButton').listeners.click(); - assert.equal(app.sockets[0].sent.at(-1).final, true); + assert.deepEqual(app.sockets[0].sent.slice(-2), [ + { type: 'input_audio_buffer.commit' }, + { type: 'response.create' }, + ]); assert.equal(app.ui.microphoneUploadEnabled(), false); - await app.ui.handleEvent({ type: 'response.output_audio.delta', audio: 'AAAAAA==' }); + await app.ui.handleEvent({ type: 'response.created', response: { id: 'turn-1' } }); + await app.ui.handleEvent({ type: 'response.output_audio.delta', delta: 'AAAAAA==', response_id: 'turn-1' }); await app.ui.handleEvent({ type: 'response.output_audio.done' }); - await app.echo(); - assert.equal(app.sockets.length, 1, 'wait for actual speaker drain'); + await app.ui.handleEvent({ type: 'response.done', response: { id: 'turn-1', status: 'completed' } }); app.ui.playbackDrained({ responseId: 'turn-1', playedMs: 1 }); await app.echo(); - assert.equal(app.sockets.length, 2); - assert.equal(app.sockets[1].sent[1].final, false); + assert.equal(app.sockets.length, 1, 'manual turns preserve the same Realtime session'); assert.equal(app.ui.microphoneUploadEnabled(), true); assert.equal(app.elements.get('cameraButton').hidden, true); assert.equal(app.sockets.flatMap(s => s.sent).some(e => e.type === 'playback.ack'), false); assert.equal(app.elements.get('pttButton').hidden, true); + + app.ui.capture(); app.ui.flushCapture(); + app.elements.get('sendTurnButton').listeners.click(); + assert.deepEqual(app.sockets[0].sent.slice(-2), [ + { type: 'input_audio_buffer.commit' }, + { type: 'response.create' }, + ]); + assert.equal(app.sockets.length, 1); await app.ui.stopSession({ terminal: false }); }); @@ -360,7 +404,7 @@ test('Qwen VAD interruption clears playback while continuing microphone upload', test('Qwen runtime errors do not suggest changing a working deployment', () => { - const p = profiles['qwen3-turn']({ adapter: 'vad' }); + const p = profiles['qwen3-turn']({ turnMode: 'vad' }); const message = 'playback.ack arrived after a later user input was committed.'; assert.equal(p.mapEvent({ type: 'error', code: 'playback_ack_too_late', error: message }).message, message); }); @@ -403,7 +447,7 @@ test('late interruption cursor acknowledges old response without finishing the n test('Qwen camera frames ride the OpenAI image interface, not the audio append', () => { - const p = profiles['qwen3-turn']({ ...config, adapter: 'vad' }); + const p = profiles['qwen3-turn']({ ...config, turnMode: 'vad' }); assert.deepEqual(plain(p.append('PCM', 'JPEG')), { type: 'input_audio_buffer.append', audio: 'PCM' }); const [created] = plain(p.imageMessages('JPEG')); @@ -415,7 +459,7 @@ test('Qwen camera frames ride the OpenAI image interface, not the audio append', }); test('the camera retires its oldest image instead of exhausting the session budget', () => { - const p = profiles['qwen3-turn']({ ...config, adapter: 'vad' }); + const p = profiles['qwen3-turn']({ ...config, turnMode: 'vad' }); const ids = []; for (let i = 0; i < 8; i++) { const messages = plain(p.imageMessages('JPEG')); @@ -543,7 +587,7 @@ test('AURA shell holds speech until PTT and opens one vision turn per two frames }); test('AURA hold-to-talk stops local playback and does not cancel the response', async () => { - const app = shell('aura-ptt', 'stt', { realPlayback: true }); + const app = shell('aura-ptt', 'manual', { realPlayback: true }); await app.ui.startSession(); receive(app, { type: 'response.created', response: { id: 'old' } }); receive(app, audioChunk('old')); @@ -679,7 +723,7 @@ test('late cancellation of an old Qwen response does not clear a newer response' }); test('Qwen speech interruption follows accepted turn detection; MiniCPM mapping stays unchanged', () => { - const qwen = profiles['qwen3-turn']({ adapter: 'vad' }); + const qwen = profiles['qwen3-turn']({ turnMode: 'vad' }); const speech = { type: 'input_audio_buffer.speech_started', item_id: 'next' }; assert.equal(qwen.mapEvent(speech).kind, 'interrupt'); qwen.mapEvent({ type: 'session.updated', session: { audio: { input: { turn_detection: { @@ -692,7 +736,7 @@ test('Qwen speech interruption follows accepted turn detection; MiniCPM mapping type: 'server_vad', interrupt_response: true, } } } } }); assert.equal(qwen.mapEvent(speech).kind, 'interrupt'); - assert.equal(profiles['qwen3-turn']({ adapter: 'stt' }).mapEvent(speech).kind, 'ignore'); + assert.equal(profiles['qwen3-turn']({ turnMode: 'manual' }).mapEvent(speech).kind, 'ignore'); assert.equal(profiles['minicpm-native']().mapEvent(speech).kind, 'ignore'); }); diff --git a/vllm_omni/entrypoints/duplex/openai.py b/vllm_omni/entrypoints/duplex/openai.py index b2fbf570d..24c9ed131 100644 --- a/vllm_omni/entrypoints/duplex/openai.py +++ b/vllm_omni/entrypoints/duplex/openai.py @@ -4,8 +4,6 @@ from __future__ import annotations import asyncio -from collections.abc import Sequence -from typing import TYPE_CHECKING from fastapi import WebSocket from vllm.logger import init_logger @@ -13,27 +11,12 @@ from vllm.utils import random_uuid from vllm_omni.entrypoints.duplex.warmup import DUPLEX_WARMUP_CLIENT_WAIT_S -if TYPE_CHECKING: - from vllm_omni.config.omni_config import BaseVllmOmniStageConfig - logger = init_logger(__name__) -_QWEN3_OMNI_REALTIME_ARCH = "Qwen3OmniMoeForConditionalGeneration" -_QWEN3_OMNI_REALTIME_STAGES = {"thinker", "talker", "code2wav"} - # Keep OpenAI connection imports in the handlers: openai/__init__.py imports # api_server, which imports this module. -def supports_qwen3_omni_realtime(stage_configs: Sequence[BaseVllmOmniStageConfig] | None) -> bool: - stages = stage_configs or () - return ( - len(stages) == len(_QWEN3_OMNI_REALTIME_STAGES) - and all(stage.model_config.model_arch == _QWEN3_OMNI_REALTIME_ARCH for stage in stages) - and {stage.model_stage for stage in stages} == _QWEN3_OMNI_REALTIME_STAGES - ) - - async def reject_realtime_websocket(websocket: WebSocket, message: str) -> None: await websocket.accept() await websocket.send_json( @@ -52,27 +35,21 @@ async def reject_realtime_websocket(websocket: WebSocket, message: str) -> None: await websocket.close(code=1008) -async def dispatch_generic_realtime_websocket(websocket: WebSocket) -> None: - """Fall back to the generic OpenAI Realtime API for non-Qwen3-Omni models.""" - serving = getattr(websocket.app.state, "openai_serving_realtime", None) - if serving is None: - await websocket.accept() - await websocket.send_json({"type": "error", "error": "Realtime API is not available", "code": "unsupported"}) - await websocket.close() - return - from vllm_omni.entrypoints.openai.realtime_connection import RealtimeConnection - - connection = RealtimeConnection(websocket, serving) - await connection.handle_connection() +async def _reject_unavailable_realtime_websocket(websocket: WebSocket) -> None: + await websocket.accept() + await websocket.send_json({"type": "error", "error": "Realtime API is not available", "code": "unsupported"}) + await websocket.close() async def dispatch_realtime_websocket(websocket: WebSocket) -> None: - """Handle Realtime sessions not already routed to the proprietary duplex - handler by the caller: Qwen3-Omni gets a conformant full-duplex connection, - everything else falls back to the generic turn-based Realtime API.""" + """Route a turn-based Realtime session to the OpenAI connection handler.""" state = websocket.app.state - if not supports_qwen3_omni_realtime(getattr(state, "stage_configs", None)): - await dispatch_generic_realtime_websocket(websocket) + if ( + getattr(state, "diffusion_engine", None) is not None + or getattr(state, "engine_client", None) is None + or getattr(state, "openai_serving_chat", None) is None + ): + await _reject_unavailable_realtime_websocket(websocket) return model_name = state.openai_serving_models.base_model_paths[0].name @@ -95,14 +72,13 @@ async def dispatch_realtime_websocket(websocket: WebSocket) -> None: DUPLEX_WARMUP_CLIENT_WAIT_S, ) - tokenizer = await state.engine_client.get_tokenizer() from vllm_omni.entrypoints.openai.realtime.connection import OpenAIFullDuplexConnection connection = OpenAIFullDuplexConnection( websocket=websocket, engine=state.engine_client, model_name=model_name, - tokenizer=tokenizer, + chat_handler=state.openai_serving_chat, tool_call_parser=getattr(state.args, "tool_call_parser", None), enable_auto_tool_choice=getattr(state.args, "enable_auto_tool_choice", False), ) diff --git a/vllm_omni/entrypoints/openai/api_server.py b/vllm_omni/entrypoints/openai/api_server.py index 189323b66..553da3fe6 100644 --- a/vllm_omni/entrypoints/openai/api_server.py +++ b/vllm_omni/entrypoints/openai/api_server.py @@ -73,7 +73,6 @@ from vllm.entrypoints.serve.utils.api_utils import ( ) from vllm.entrypoints.serve.utils.orca_metrics import metrics_header from vllm.entrypoints.serve.utils.request_logger import RequestLogger -from vllm.entrypoints.speech_to_text.realtime.serving import OpenAIServingRealtime from vllm.entrypoints.speech_to_text.transcription.serving import ( OpenAIServingTranscription, ) @@ -692,7 +691,6 @@ async def _init_duplex_app_state( "openai_streaming_speech", "openai_streaming_video", "openai_streaming_video_output", - "openai_serving_realtime", "openai_serving_realtime_robot", "anthropic_serving_messages", ): @@ -1227,12 +1225,6 @@ async def omni_init_app_state( else None ) state.openai_serving_duplex = None - state.openai_serving_realtime = OpenAIServingRealtime( - engine_client=engine_client, - models=state.openai_serving_models, - request_logger=request_logger, - ) - state.openai_serving_video = OmniOpenAIServingVideo( engine_client, model_name=served_model_names[0] if served_model_names else None, diff --git a/vllm_omni/entrypoints/openai/realtime/connection.py b/vllm_omni/entrypoints/openai/realtime/connection.py index b9e66287e..4831b6bef 100644 --- a/vllm_omni/entrypoints/openai/realtime/connection.py +++ b/vllm_omni/entrypoints/openai/realtime/connection.py @@ -5,10 +5,11 @@ from __future__ import annotations import asyncio import binascii +import io import json -import time import warnings -from dataclasses import dataclass +import wave +from dataclasses import dataclass, replace from typing import TYPE_CHECKING, Any import numpy as np @@ -21,12 +22,13 @@ from vllm.entrypoints.openai.chat_completion.protocol import ( ChatCompletionRequest, ChatCompletionToolsParam, ) +from vllm.exceptions import VLLMValidationError from vllm.logger import init_logger from vllm.sampling_params import RequestOutputKind, SamplingParams, StructuredOutputsParams from vllm.tool_parsers import ToolParserManager if TYPE_CHECKING: - from vllm.inputs import TokensPrompt + from vllm.inputs import EngineInput from vllm_omni.entrypoints.async_omni import AsyncOmni from vllm_omni.entrypoints.openai.realtime.session import ( @@ -37,27 +39,20 @@ from vllm_omni.entrypoints.openai.realtime.session import ( _gen_id, merge_session_config, ) +from vllm_omni.utils.audio import audio_chunk_sample_rate +from vllm_omni.utils.audio_resample import StreamingAudioResampler logger = init_logger(__name__) _CLIENT_EVENT_ADAPTER = TypeAdapter(types.RealtimeClientEvent) -# Qwen3-Omni's Realtime contract is mono signed PCM16 at 24 kHz. PCM16 is the -# sample width; 24 kHz is the sample rate. +# The Realtime PCM contract is mono signed PCM16 at 24 kHz. SAMPLE_RATE_HZ = 24000 BYTES_PER_SAMPLE_PCM16 = 2 # Application safety caps for one append and the complete pending input turn. MAX_AUDIO_APPEND_BYTES = 15 * 1024 * 1024 MAX_INPUT_AUDIO_BUFFER_BYTES = 64 * 1024 * 1024 -# The Qwen3-Omni chat template uses these literal special tokens for audio. -# The generic chat-template interface does not expose that placeholder as -# metadata, so it cannot be inferred reliably from the Realtime event schema. -AUDIO_PLACEHOLDER = "<|audio_start|><|audio_pad|><|audio_end|>" - -# Empirically calibrated for Qwen3-Omni from 8,808 ms / 23 thinker tokens. -QWEN3_OMNI_MS_PER_TOKEN = 383.0 - AUTO_TRUNCATION_TRIGGER_RATIO = 0.8 AUTO_TRUNCATION_TARGET_RATIO = 0.5 @@ -85,13 +80,15 @@ class OpenAIFullDuplexConnection: websocket: WebSocket, engine: AsyncOmni, model_name: str, - tokenizer: Any, + chat_handler: Any, tool_call_parser: str | None = None, enable_auto_tool_choice: bool = False, ): self.ws = websocket self.engine = engine self.model_name = model_name + self.chat_handler = chat_handler + self._tokenizer = chat_handler.renderer.get_tokenizer() self._tool_call_parser_name = tool_call_parser if enable_auto_tool_choice else None self.session = AudioFullDuplexSessionState() @@ -102,8 +99,6 @@ class OpenAIFullDuplexConnection: self._response_cancel_event = asyncio.Event() self._send_lock = asyncio.Lock() - self._tokenizer = self._resolve_tokenizer(tokenizer) - # ------------------------------------------------------------------ # # Lifecycle # # ------------------------------------------------------------------ # @@ -140,23 +135,6 @@ class OpenAIFullDuplexConnection: await self._cancel_active_response() logger.info("[realtime] connection closed, session_id=%s", self.session.session_id) - def _resolve_tokenizer(self, tokenizer: Any) -> Any: - if getattr(tokenizer, "chat_template", None): - return tokenizer - - input_processor = getattr(self.engine, "input_processor", None) - if input_processor is None: - return tokenizer - try: - from vllm.transformers_utils.processor import cached_processor_from_config - - processor = cached_processor_from_config(input_processor.model_config) - if getattr(processor, "apply_chat_template", None): - return processor - except Exception: - logger.warning("Could not load processor for chat templating") - return tokenizer - # ------------------------------------------------------------------ # # Event dispatch # # ------------------------------------------------------------------ # @@ -332,6 +310,16 @@ class OpenAIFullDuplexConnection: raise ValueError("PCM audio data must contain complete 16-bit samples") return decoded + @staticmethod + def _pcm16_wav_b64(audio: bytes) -> str: + with io.BytesIO() as buffer: + with wave.open(buffer, "wb") as wav: + wav.setnchannels(1) + wav.setsampwidth(BYTES_PER_SAMPLE_PCM16) + wav.setframerate(SAMPLE_RATE_HZ) + wav.writeframes(audio) + return base64.b64encode(buffer.getvalue()).decode("ascii") + async def _commit_audio_buffer( self, event_id: str | None = None, @@ -340,13 +328,12 @@ class OpenAIFullDuplexConnection: s = self.session if len(s.input_audio_buffer) == 0: return None - pcm16 = np.frombuffer(bytes(s.input_audio_buffer), dtype=np.int16) - audio_f32 = pcm16.astype(np.float32) / 32768.0 + audio = base64.b64encode(s.input_audio_buffer).decode("ascii") item = types.RealtimeConversationItemUserMessage( type="message", role="user", status="completed", - content=[{"type": "input_audio", "audio": self._pcm16_b64(audio_f32)}], + content=[{"type": "input_audio", "audio": audio}], ) try: s.insert_item(item) @@ -447,37 +434,24 @@ class OpenAIFullDuplexConnection: metadata=metadata, ) - async def _estimate_total_tokens( + async def _prepare_engine_input_with_auto_truncation( self, - tools: list | None, + response: _ResolvedResponse, *, - instructions: str | None = None, - items: list[Any] | None = None, - ) -> int: - t0 = time.monotonic() - prompt = await self._build_full_prompt(tools=tools, instructions=instructions, items=items) - total = len(prompt["prompt_token_ids"]) - audio_arrays = prompt.get("multi_modal_data", {}).get("audio", []) - if audio_arrays: - raw_tok = getattr(self._tokenizer, "tokenizer", self._tokenizer) - placeholder_len = len(raw_tok.encode(AUDIO_PLACEHOLDER, add_special_tokens=False)) - for arr, sr in audio_arrays: - total += self._qwen3_omni_audio_token_count(arr.shape[0], sr) - placeholder_len - elapsed = time.monotonic() - t0 - if elapsed > 0.05: - logger.warning( - "[realtime] _estimate_total_tokens took %.3fs (history=%d items, %d audio segments)", - elapsed, - len(items if items is not None else self.session.items), - len(audio_arrays), - ) - return total - - async def _maybe_truncate_history(self, response: _ResolvedResponse) -> bool: + preflight_only: bool = False, + ) -> EngineInput | None: s = self.session max_model_len = getattr(self.engine.model_config, "max_model_len", None) - if not max_model_len: - return True + persistent = response.input is None + items: list[Any] = s.items if response.input is None else response.input + staged_deletions: list[Any] = [] + if max_model_len is None: + return await self._build_full_prompt( + tools=response.tools, + instructions=response.instructions, + items=items, + skip_mm_cache=preflight_only, + ) truncation = s.config.truncation or "auto" ratio = 1.0 @@ -503,22 +477,36 @@ class OpenAIFullDuplexConnection: trigger = limit target = limit - persistent = response.input is None - items: list[Any] = s.items if response.input is None else response.input - total = await self._estimate_total_tokens( - response.tools, + # Budget checks are speculative and may render repeatedly while + # truncating, so keep them off the shared sender cache. Once accepted, + # render again through the normal cache path for engine submission. + engine_input = await self._build_full_prompt( + tools=response.tools, instructions=response.instructions, items=items, + skip_mm_cache=True, ) + total = len(engine_input["prompt_token_ids"]) if total <= trigger: - return True + if preflight_only: + return engine_input + return await self._build_full_prompt( + tools=response.tools, + instructions=response.instructions, + items=items, + ) if mode == "disabled": logger.warning( "[realtime] token budget exceeded (%d/%d) and truncation is disabled -- rejecting response.create", total, limit, ) - return False + return None + + if persistent: + # Preflight truncation is speculative. Work on a snapshot so a + # rejected response.create does not delete items from session history. + items = list(items) idx = 0 while total > target and idx < len(items): @@ -547,23 +535,36 @@ class OpenAIFullDuplexConnection: for remove_idx in sorted(remove_indexes, reverse=True): removed = items[remove_idx] if persistent: - s.remove_item(removed.id) - await self._send_event( - types.ConversationItemDeletedEvent( - event_id=_gen_id("evt"), - type="conversation.item.deleted", - item_id=removed.id, - ) - ) - else: - del items[remove_idx] - total = await self._estimate_total_tokens( - response.tools, + staged_deletions.append(removed) + del items[remove_idx] + engine_input = await self._build_full_prompt( + tools=response.tools, instructions=response.instructions, items=items, + skip_mm_cache=True, ) + total = len(engine_input["prompt_token_ids"]) - return total <= limit + if total > limit: + return None + if preflight_only: + return engine_input + engine_input = await self._build_full_prompt( + tools=response.tools, + instructions=response.instructions, + items=items, + ) + for removed in staged_deletions: + if removed.id is None or s.remove_item(removed.id) is None: + continue + await self._send_event( + types.ConversationItemDeletedEvent( + event_id=_gen_id("evt"), + type="conversation.item.deleted", + item_id=removed.id, + ) + ) + return engine_input async def _handle_response_create(self, event: types.ResponseCreateEvent): s = self.session @@ -582,7 +583,16 @@ class OpenAIFullDuplexConnection: await self._send_error(str(exc), "invalid_request_error", event_id=event.event_id) return - if not await self._maybe_truncate_history(response): + had_active_response = s.active_response is not None + try: + engine_input = await self._prepare_engine_input_with_auto_truncation( + response, + preflight_only=had_active_response, + ) + except VLLMValidationError as exc: + await self._send_error(str(exc), "invalid_request_error", event_id=event.event_id) + return + if engine_input is None: await self._send_error( "The response input exceeds the model's input token limit", "invalid_request_error", @@ -590,6 +600,22 @@ class OpenAIFullDuplexConnection: ) return + if had_active_response: + if s.active_response is not None: + await self._cancel_active_response() + try: + engine_input = await self._prepare_engine_input_with_auto_truncation(response) + except VLLMValidationError as exc: + await self._send_error(str(exc), "invalid_request_error", event_id=event.event_id) + return + if engine_input is None: + await self._send_error( + "The response input exceeds the model's input token limit", + "invalid_request_error", + event_id=event.event_id, + ) + return + response_id = _gen_id("resp") await self._send_event( types.ResponseCreatedEvent( @@ -599,13 +625,10 @@ class OpenAIFullDuplexConnection: ) ) - if s.active_response is not None: - await self._cancel_active_response() - s.active_response = ActiveResponse(response_id=response_id, request_id=f"rt-{response_id}") self._response_cancel_event.clear() - self._response_task = asyncio.create_task(self._run_response(response_id, response)) + self._response_task = asyncio.create_task(self._run_response(response_id, response, engine_input)) def _response_object( self, @@ -630,7 +653,12 @@ class OpenAIFullDuplexConnection: usage=usage, ) - async def _run_response(self, response_id: str, response: _ResolvedResponse): + async def _run_response( + self, + response_id: str, + response: _ResolvedResponse, + engine_input: EngineInput, + ): s = self.session active = s.active_response if active is None: @@ -638,7 +666,7 @@ class OpenAIFullDuplexConnection: completed = False try: - await self._run_response_inner(response_id, response, s, active) + await self._run_response_inner(response_id, response, s, active, engine_input) completed = True except asyncio.CancelledError: if self._response_cancel_event.is_set(): @@ -716,17 +744,11 @@ class OpenAIFullDuplexConnection: except Exception: logger.debug("Failed to send failure response.done for %s", response_id, exc_info=True) - async def _run_response_inner(self, response_id, response, s, active): + async def _run_response_inner(self, response_id, response, s, active, engine_input: EngineInput): previous_item_id = s.items[-1].id if s.items else None modalities = response.modalities is_audio = "audio" in modalities tools, tool_choice = response.tools, response.tool_choice - prompt = await self._build_full_prompt( - tools=tools, - instructions=response.instructions, - items=response.input, - ) - item_id = _gen_id("item") active.item_id = item_id output_index = 0 @@ -804,12 +826,13 @@ class OpenAIFullDuplexConnection: limit_reached = False usage = ResponseUsage() total_audio_samples = 0 + audio_resampler: StreamingAudioResampler | None = None previous_text = "" previous_token_ids: list[int] = [] pending_tool_calls: dict[int, dict[str, Any]] = {} next_output_index = 1 # 0 is the message item, reserved above - # Qwen's talker speaks tool markup, so stop audio after detecting a call. + # Tool-call markup is control output, so suppress its audio after detection. tool_call_seen = False async def emit_content_delta(piece: str) -> None: @@ -843,6 +866,26 @@ class OpenAIFullDuplexConnection: ) ) + async def emit_audio_delta(chunk: np.ndarray) -> None: + nonlocal total_audio_samples + if not chunk.size: + return + # Honor the requested modality and suppress audio after a tool call. + if not is_audio or tool_call_seen: + return + total_audio_samples += chunk.shape[0] + await self._send_event( + types.ResponseAudioDeltaEvent( + event_id=_gen_id("evt"), + type="response.output_audio.delta", + response_id=response_id, + item_id=item_id, + output_index=output_index, + content_index=content_index, + delta=self._pcm16_b64(chunk), + ) + ) + async def handle_tool_parser_delta(delta_msg) -> None: nonlocal next_output_index, tool_call_seen if delta_msg is None: @@ -911,8 +954,11 @@ class OpenAIFullDuplexConnection: sp.structured_outputs = StructuredOutputsParams(structural_tag=structural_tag_json) thinker_params_configured = True + sampling_params_list = self.chat_handler._fix_minicpmo45_audio_stream_output_kinds( + sampling_params_list, modalities + ) gen = self.engine.generate( - prompt=prompt, + prompt=engine_input, request_id=active.request_id, sampling_params_list=sampling_params_list, output_modalities=modalities, @@ -927,31 +973,14 @@ class OpenAIFullDuplexConnection: output_type = getattr(output, "final_output_type", "text") if output_type == "audio": audio_chunks = self._extract_audio_deltas(output) - for chunk in audio_chunks: - total_audio_samples += chunk.shape[0] - # is_audio guard is defense-in-depth: generate() is - # now given output_modalities=modalities above, so - # the engine shouldn't produce audio-typed output - # for a text-only response -- but don't rely on - # that alone to honor the client's explicit - # request; never flip session state or forward - # audio it didn't ask for. - if not is_audio: - continue - if tool_call_seen: - continue - b64 = self._pcm16_b64(chunk) - await self._send_event( - types.ResponseAudioDeltaEvent( - event_id=_gen_id("evt"), - type="response.output_audio.delta", - response_id=response_id, - item_id=item_id, - output_index=output_index, - content_index=content_index, - delta=b64, + if audio_chunks: + if audio_resampler is None: + audio_resampler = StreamingAudioResampler( + audio_chunk_sample_rate(output), + SAMPLE_RATE_HZ, ) - ) + for chunk in audio_chunks: + await emit_audio_delta(audio_resampler.process(chunk)) continue if output.outputs: @@ -966,11 +995,8 @@ class OpenAIFullDuplexConnection: delta_text = first_out.text or "" delta_token_ids = list(first_out.token_ids) usage.output_tokens += len(delta_token_ids) - # Raw thinker token stream, in talker-consumption order -- - # this is what _qwen3_omni_truncate_transcript correlates - # against codec frames, independent of any tool-parser - # stripping applied to full_text/full_transcript below - # (the talker speaks the raw stream, tool tags included). + # Keep the raw token stream for approximate transcript + # truncation, independent of any tool-parser stripping. full_token_ids.extend(delta_token_ids) if output.prompt_token_ids: @@ -1021,6 +1047,10 @@ class OpenAIFullDuplexConnection: if self._response_cancel_event.is_set(): cancelled = True + if audio_resampler is not None and not cancelled: + tail = audio_resampler.process(np.empty(0, dtype=np.float32), final=True) + await emit_audio_delta(tail) + status = "cancelled" if cancelled else "incomplete" if limit_reached else "completed" if is_audio: @@ -1170,11 +1200,9 @@ class OpenAIFullDuplexConnection: s.item_duration_ms[item_obj.id] = total_audio_samples / SAMPLE_RATE_HZ * 1000 # Skip storing for tool-call responses: full_token_ids is # the raw thinker stream (tool-call tags included), but - # this item's transcript/text is the tool-parser-stripped - # content -- the two no longer line up token-for-token, - # so _qwen3_omni_truncate_transcript falls back to - # blanking (today's behavior) rather than risk splicing - # raw text into a truncated transcript. + # this item's transcript/text is tool-parser-stripped, + # so truncation falls back to a blank transcript rather + # than splice raw tool markup into it. if not pending_tool_calls: s.item_token_ids[item_obj.id] = full_token_ids await self._send_conversation_item_added_and_done(history_item, previous_item_id) @@ -1488,16 +1516,10 @@ class OpenAIFullDuplexConnection: ) return - # Per spec, truncating audio must not leave text in context the user - # never heard -- but rather than blanking the transcript entirely - # (the previous, spec-minimum behavior), reconstruct the prefix that - # *was* actually heard using Qwen3-Omni's fixed talker text/frame - # correlation (see _qwen3_omni_truncate_transcript). This keeps the - # model's own memory of what it said in sync with what the user - # actually heard, instead of wiping it and confusing later turns. - # Falls back to "" (today's behavior) when we don't have a captured - # token stream for this item (e.g. it included a tool call). - truncated_text = self._qwen3_omni_truncate_transcript(item_id, audio_end_ms) + # Preserve only the transcript prefix proportional to the audio heard. + # Exact token/audio alignment is model-specific, so fall back to an + # empty transcript when token IDs or duration are unavailable. + truncated_text = self._truncate_transcript(item_id, audio_end_ms) part = new_content[content_index] if hasattr(part, "transcript"): new_content[content_index] = { @@ -1602,45 +1624,33 @@ class OpenAIFullDuplexConnection: resolved.append(referenced) return resolved - def _qwen3_omni_truncate_transcript(self, item_id: str, audio_end_ms: float) -> str: - """Approximate the transcript heard before ``audio_end_ms``.""" + def _truncate_transcript(self, item_id: str, audio_end_ms: float) -> str: + """Estimate the transcript prefix heard before ``audio_end_ms``.""" token_ids = self.session.item_token_ids.get(item_id) - if not token_ids: + duration_ms = self.session.item_duration_ms.get(item_id) + if not token_ids or duration_ms is None or duration_ms <= 0: return "" - tokens_heard = round(audio_end_ms / QWEN3_OMNI_MS_PER_TOKEN) - tokens_heard = min(tokens_heard, len(token_ids)) + tokens_heard = int(len(token_ids) * audio_end_ms / duration_ms) if tokens_heard <= 0: return "" raw_tok = getattr(self._tokenizer, "tokenizer", self._tokenizer) return raw_tok.decode(token_ids[:tokens_heard], skip_special_tokens=True) - @staticmethod - def _qwen3_omni_audio_token_count(num_samples: int, sample_rate: int) -> int: - """Return Qwen3-Omni's expanded token count for an audio segment.""" - from vllm.model_executor.models.qwen3_omni_moe_thinker import ( - _get_feat_extract_output_lengths, - ) - - num_samples_16k = round(num_samples * 16000 / sample_rate) - input_lengths = num_samples_16k // 160 - output_lengths = _get_feat_extract_output_lengths(input_lengths) - return int(output_lengths) + 2 # + <|audio_start|> + <|audio_end|> - async def _build_full_prompt( self, tools: list | None = None, *, instructions: str | None = None, items: list | None = None, - ) -> TokensPrompt: - """Render the effective instructions and items into an engine prompt.""" - from vllm.inputs import TokensPrompt + skip_mm_cache: bool = False, + ) -> EngineInput: + """Render the effective conversation through normal chat preprocessing.""" + chat_handler = self.chat_handler s = self.session effective_instructions = instructions if instructions is not None else s.config.instructions effective_items = items if items is not None else s.items messages: list[dict[str, Any]] = [] - audio_arrays: list[tuple[np.ndarray, int]] = [] converted_tools = self._convert_tools(tools) if tools else None if effective_instructions: @@ -1648,11 +1658,7 @@ class OpenAIFullDuplexConnection: for item in effective_items: if item.type == "function_call": - # "" not None: this model's chat template only handles - # message.content as a string or a list (line 55/59 of its - # Jinja template unconditionally iterates non-string content - # assuming it's a list) -- None crashes with "'NoneType' - # object is not iterable" (confirmed in production logs). + # Some chat templates iterate message content unconditionally. messages.append( { "role": "assistant", @@ -1673,17 +1679,23 @@ class OpenAIFullDuplexConnection: role = getattr(item, "role", None) if role == "user": - parts_text = [] + content = [] for part in item.content: if part.type == "input_audio" and part.audio: - parts_text.append(AUDIO_PLACEHOLDER) audio_bytes = self._decode_pcm16(part.audio, MAX_INPUT_AUDIO_BUFFER_BYTES) - pcm16 = np.frombuffer(audio_bytes, dtype=np.int16) - audio_arrays.append((pcm16.astype(np.float32) / 32768.0, SAMPLE_RATE_HZ)) + content.append( + { + "type": "input_audio", + "input_audio": { + "data": self._pcm16_wav_b64(audio_bytes), + "format": "wav", + }, + } + ) elif part.type == "input_text" and part.text: - parts_text.append(part.text) - if parts_text: - messages.append({"role": "user", "content": "".join(parts_text)}) + content.append({"type": "text", "text": part.text}) + if content: + messages.append({"role": "user", "content": content}) elif role == "assistant": text = self._assistant_item_text(item) if text: @@ -1693,21 +1705,24 @@ class OpenAIFullDuplexConnection: if text: messages.append({"role": "system", "content": text}) - chat_template_kwargs: dict[str, Any] = { - "tokenize": False, - "add_generation_prompt": True, - "add_special_tokens": False, - } - if converted_tools: - chat_template_kwargs["tools"] = converted_tools - text = self._tokenizer.apply_chat_template(messages, **chat_template_kwargs) - raw_tok = getattr(self._tokenizer, "tokenizer", self._tokenizer) - token_ids = raw_tok.encode(text, add_special_tokens=False) - - prompt_data = TokensPrompt(prompt_token_ids=token_ids) - if audio_arrays: - prompt_data["multi_modal_data"] = {"audio": audio_arrays} - return prompt_data + request = ChatCompletionRequest(model=self.model_name, messages=messages) + tok_params = replace( + request.build_tok_params(self.engine.model_config), + max_total_tokens=None, + ) + _, (engine_input,) = await chat_handler._preprocess_chat( + request, + messages, + default_template=request.chat_template or chat_handler.chat_template, + default_template_content_format=chat_handler.chat_template_content_format, + default_template_kwargs=chat_handler._effective_chat_template_kwargs(request), + tool_dicts=converted_tools, + tok_params=tok_params, + skip_mm_cache=skip_mm_cache, + ) + if engine_input.get("prompt_token_ids") is None: + raise RuntimeError("Realtime renderer did not return prompt token IDs") + return engine_input # ------------------------------------------------------------------ # # Audio output processing # diff --git a/vllm_omni/entrypoints/openai/realtime_connection.py b/vllm_omni/entrypoints/openai/realtime_connection.py deleted file mode 100644 index 3d37fbe37..000000000 --- a/vllm_omni/entrypoints/openai/realtime_connection.py +++ /dev/null @@ -1,229 +0,0 @@ -# SPDX-License-Identifier: Apache-2.0 -# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project - -from __future__ import annotations - -import asyncio -import base64 -import json -from collections.abc import AsyncGenerator, Mapping -from typing import cast -from uuid import uuid4 - -import numpy as np -from vllm.entrypoints.serve.engine.protocol import UsageInfo -from vllm.entrypoints.speech_to_text.realtime.connection import RealtimeConnection as VllmRealtimeConnection -from vllm.entrypoints.speech_to_text.realtime.protocol import TranscriptionDelta, TranscriptionDone -from vllm.logger import init_logger - -from vllm_omni.entrypoints.async_omni import AsyncOmni -from vllm_omni.entrypoints.utils import coerce_param_message_types - -logger = init_logger(__name__) - - -class RealtimeConnection(VllmRealtimeConnection): - """Omni realtime connection with audio-only server events. - - Reuses upstream vLLM websocket/session lifecycle and only customizes - generation output handling to emit audio deltas. - """ - - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - self.engine = cast(AsyncOmni, self.serving.engine_client) - self._realtime_audio_ref: np.ndarray | None = None - - async def start_generation(self): - await super().start_generation() - - @staticmethod - def _tensor_to_numpy(value) -> np.ndarray | None: - if value is None: - return None - if isinstance(value, np.ndarray): - arr = value - elif hasattr(value, "detach"): - arr = value.detach().float().cpu().numpy() - else: - try: - arr = np.asarray(value) - except Exception: - return None - if arr.ndim > 1: - arr = arr.reshape(-1) - return arr.astype(np.float32, copy=False) - - @staticmethod - def _numpy_audio_prefix_match(prev: np.ndarray, curr: np.ndarray) -> bool: - n = prev.shape[0] - if n == 0: - return True - if curr.shape[0] < n: - return False - return bool(np.allclose(curr[:n], prev, rtol=1e-3, atol=2e-4)) - - def _raw_waveform_to_deltas(self, arr: np.ndarray) -> list[np.ndarray]: - """Convert one streaming PCM f32 chunk into incremental piece(s) for the client. - - Some engine paths emit a growing cumulative waveform each step; others emit - true per-step deltas. We support both without duplicating audio on the client. - """ - if arr.size == 0: - return [] - ref = self._realtime_audio_ref - if ref is None: - self._realtime_audio_ref = arr.copy() - return [arr] - if self._numpy_audio_prefix_match(ref, arr): - delta = arr[ref.shape[0] :] - self._realtime_audio_ref = arr.copy() - return [delta] if delta.size > 0 else [] - # True per-step delta (not a prefix extension of what we have seen). - self._realtime_audio_ref = np.concatenate([ref, arr]) - return [arr] - - def _extract_audio_chunks(self, output) -> tuple[list[np.ndarray], int]: - mm = getattr(output, "multimodal_output", None) - if mm is None: - return [], 24000 - # Support both MultimodalPayload and plain dict - if not isinstance(mm, Mapping): - return [], 24000 - - sr = mm.get("sr") or mm.get("sample_rate") or mm.get("audio_sample_rate") or 24000 - if isinstance(sr, (list, tuple)) and sr: - sr = sr[-1] - if hasattr(sr, "item"): - sr = sr.item() - sample_rate_hz = int(sr) - key = "audio" if "audio" in mm else ("model_outputs" if "model_outputs" in mm else None) - if key is None: - return [], sample_rate_hz - - raw_audio = mm.get(key) - chunks: list[np.ndarray] = [] - if isinstance(raw_audio, (list, tuple)): - if len(raw_audio) > 0: - arr = self._tensor_to_numpy(raw_audio[-1]) - if arr is not None and arr.size > 0: - chunks.extend(self._raw_waveform_to_deltas(arr)) - else: - arr = self._tensor_to_numpy(raw_audio) - if arr is not None and arr.size > 0: - chunks.extend(self._raw_waveform_to_deltas(arr)) - return chunks, sample_rate_hz - - @staticmethod - def _pcm16_b64(audio_f32: np.ndarray) -> str: - clipped = np.clip(audio_f32, -1.0, 1.0) - pcm16 = (clipped * 32767.0).astype(np.int16) - return base64.b64encode(pcm16.tobytes()).decode("utf-8") - - async def _run_generation( - self, - streaming_input_gen: AsyncGenerator, - input_stream: asyncio.Queue[list[int]], - ): - request_id = f"rt-{self.connection_id}-{uuid4()}" - sent_audio = False - audio_done_sent = False - full_text = "" - prompt_token_ids_len = 0 - completion_tokens_len = 0 - self._realtime_audio_ref = None - - # Coerce cumulative outputs to delta outputs; this ensures - # we don't emit redundant MM data & drain after emitting. - sampling_params_list = list(self.engine.default_sampling_params_list) - sampling_params_list = coerce_param_message_types( - sampling_params_list, - is_streaming=True, - ) - - result_gen = None - try: - result_gen = self.engine.generate( - prompt=streaming_input_gen, - request_id=request_id, - sampling_params_list=sampling_params_list, - ) - - async for output in result_gen: - stage_id = getattr(output, "stage_id", None) - if stage_id == 0 and output.outputs: - first_output = output.outputs[0] - new_token_ids = list(first_output.token_ids) - if new_token_ids: - input_stream.put_nowait(new_token_ids) - - if output.prompt_token_ids: - prompt_token_ids_len = max( - prompt_token_ids_len, - len(output.prompt_token_ids), - ) - - delta_text = first_output.text or "" - full_text += delta_text - completion_tokens_len += len(new_token_ids) - - if delta_text: - await self.send(TranscriptionDelta(delta=delta_text)) - - audio_chunks, sample_rate = self._extract_audio_chunks(output) - - for chunk in audio_chunks: - sent_audio = True - await self.send_json( - { - "type": "response.output_audio.delta", - "audio": self._pcm16_b64(chunk), - "format": "pcm16", - "sample_rate_hz": sample_rate, - } - ) - - if not self._is_connected: - break - - usage = UsageInfo( - prompt_tokens=prompt_token_ids_len, - completion_tokens=completion_tokens_len, - total_tokens=prompt_token_ids_len + completion_tokens_len, - ) - await self.send(TranscriptionDone(text=full_text, usage=usage)) - - if sent_audio: - await self.send_json({"type": "response.output_audio.done", "has_audio": True}) - audio_done_sent = True - except Exception as e: - logger.exception("Error in generation: %s", e) - await self.send_error(str(e), "processing_error") - finally: - # Close the generator explicitly so AsyncOmni.generate's cleanup - # (input-pump cancellation and engine-side abort) runs now rather - # than whenever the event loop garbage-collects the async - # generator; the delay window is where a disconnected session - # keeps cycling through the stages (issue #4271). - if result_gen is not None: - try: - await result_gen.aclose() - except Exception: - logger.exception("Failed to close realtime result generator") - # Always send terminal event so clients don't hang forever. - if self._is_connected and not audio_done_sent: - try: - await self.send_json({"type": "response.output_audio.done", "has_audio": sent_audio}) - except Exception: - logger.exception("Failed to send response.output_audio.done") - while not self.audio_queue.empty(): - self.audio_queue.get_nowait() - - async def send_json(self, payload: dict): - try: - await self.websocket.send_text(json.dumps(payload)) - except Exception: - # A failed send means the client is gone; flag it so the - # generation loop stops instead of retrying into a dead socket. - self._is_connected = False - raise diff --git a/vllm_omni/entrypoints/openai/serving_chat.py b/vllm_omni/entrypoints/openai/serving_chat.py index 9ed6ee47e..bae1a3e63 100644 --- a/vllm_omni/entrypoints/openai/serving_chat.py +++ b/vllm_omni/entrypoints/openai/serving_chat.py @@ -105,6 +105,7 @@ from vllm.parser.mistral import MistralToolCall from vllm.reasoning import ReasoningParser from vllm.renderers import BaseRenderer, merge_kwargs from vllm.renderers.inputs import TokPrompt +from vllm.renderers.params import TokenizeParams from vllm.sampling_params import RequestOutputKind, SamplingParams from vllm.tokenizers import TokenizerLike from vllm.tokenizers import TokenizerLike as AnyTokenizer @@ -977,6 +978,8 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): continue_final_message: bool = False, documents: list[dict[str, str]] | None = None, add_special_tokens: bool = False, + tok_params: TokenizeParams | None = None, + skip_mm_cache: bool = False, ) -> tuple[list[ConversationMessage], list[TokPrompt]]: if renderer is None: renderer = self.renderer @@ -995,7 +998,8 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): ), ) - tok_params = request.build_tok_params(self.model_config) + if tok_params is None: + tok_params = request.build_tok_params(self.model_config) mm_config = self.model_config.multimodal_config chat_params = request.build_chat_params( default_template, @@ -1020,6 +1024,7 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin): prompt_extras={ k: v for k in ("mm_processor_kwargs", "cache_salt") if (v := getattr(request, k, None)) is not None }, + skip_mm_cache=skip_mm_cache, ) tokenizer = renderer.get_tokenizer()