anodyine/vllm-omni
1
0 Fork 0
Code Issues 4 Pull Requests 2 Actions Packages Projects Releases Wiki Activity

[Realtime] Emit current (non-beta) OpenAI audio/transcript event names (#7339)

Browse Source 
Signed-off-by: Nick Cao <ncao@redhat.com>
Co-authored-by: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Nick CaoandClaude Sonnet 5 authored and GitHub committed 2026-09-11 08:50:12 +08:00
1 parent 5c69fc2a8a
commit 498c41b8d6
34 files changed
+251 -189
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

docs/design/fullduplex-personaplex.md
+2 -2
View File
@@ -20,7 +20,7 @@ The target is the engine-native path:
-> PersonaPlex Talker
-> streaming PersonaPlex Code2Wav
-> PersonaPlex data-plane projector
-> response.audio.delta + response.audio_transcript.delta
-> response.output_audio.delta + response.output_audio_transcript.delta
```
The standalone `/api/chat` and `/v1/audio/duplex` server that accompanied the
@@ -197,7 +197,7 @@ model-neutral native results:
It owns per-request audio and text cursors so a cumulative output cannot replay
old audio. The generic projector turns each native result into the Realtime
`response.audio.delta` / `response.audio_transcript.delta` pair (and
`response.output_audio.delta` / `response.output_audio_transcript.delta` pair (and
`response.output_text.delta` for text) under one `response_id`; the full
mapping is the name map in the
[Realtime Duplex API](../serving/realtime_duplex_api.md) serving guide.
docs/design/fullduplex.md
+7 -7
View File
@@ -606,11 +606,11 @@ scoped to one `response_id` unless stated otherwise.
| Public event | Trigger and ordering | Cardinality |
| --- | --- | --- |
| `response.created` | First event for a visible assistant response; precedes its output-item, content-part, speak, text, and audio events. | Exactly once for every visible response. |
| `response.speak` | The model selected speak; emitted after `response.created` and no later than the first `response.audio.delta`. It carries decision metadata, not transcript text. | At most once. |
| `response.audio.delta` | Carries one ordered audio chunk. Every emitted audio delta is followed by a `response.audio_transcript.delta` for the same chunk. | Zero or more. |
| `response.audio_transcript.delta` | Append-only transcript contribution paired with an audio delta; it may be empty for a text-less audio unit and must not repeat or overlap earlier text. | One per audio delta. |
| `response.audio.done` | Closes the audio stream after its final delta and before response terminal events. It is omitted for a response with no audio or transcript. | At most once. |
| `response.audio_transcript.done` | Contains the exact concatenation of all non-overlapping transcript deltas for the response. | At most once, and only for a non-empty transcript. |
| `response.speak` | The model selected speak; emitted after `response.created` and no later than the first `response.output_audio.delta`. It carries decision metadata, not transcript text. | At most once. |
| `response.output_audio.delta` | Carries one ordered audio chunk. Every emitted audio delta is followed by a `response.output_audio_transcript.delta` for the same chunk. | Zero or more. |
| `response.output_audio_transcript.delta` | Append-only transcript contribution paired with an audio delta; it may be empty for a text-less audio unit and must not repeat or overlap earlier text. | One per audio delta. |
| `response.output_audio.done` | Closes the audio stream after its final delta and before response terminal events. It is omitted for a response with no audio or transcript. | At most once. |
| `response.output_audio_transcript.done` | Contains the exact concatenation of all non-overlapping transcript deltas for the response. | At most once, and only for a non-empty transcript. |
| `response.output_item.done` / `conversation.item.done` | Finalize the assistant item after content-part terminal events and before `response.done`. | At most once each. |
| `response.done` | Terminal event for a created response; follows audio, transcript, content-part, and item terminal events. | Exactly once for every created response that reaches a terminal state. |
| `rate_limits.updated` | Compatibility event emitted immediately after `response.done`; the current payload has an empty rate-limit list. | Once per emitted `response.done`. |
@@ -625,8 +625,8 @@ response and never reuse that response ID for later model turns.
Compatibility changes must update this table and its protocol contract tests in
the same change. The golden transcript test requires joined
`response.audio_transcript.delta` values to equal
`response.audio_transcript.done.transcript`; response tests separately enforce
`response.output_audio_transcript.delta` values to equal
`response.output_audio_transcript.done.transcript`; response tests separately enforce
at-most-once `response.speak` and terminal event cardinality.
### Session internal ledgers
docs/serving/full_duplex_api.md
+1 -1
View File
@@ -86,7 +86,7 @@ A typical `/v1/realtime?duplex=1` session follows this lifecycle:
3. Send `input_audio_buffer.append` events while microphone audio arrives.
4. Send `input_audio_buffer.commit` at a user-turn boundary when required by
the model policy.
5. Consume `response.created`, transcript deltas, `response.audio.delta`, and
5. Consume `response.created`, transcript deltas, `response.output_audio.delta`, and
`response.done` or `response.listen` events.
6. Send `playback.ack` after audio has been played when the session advertises
playback acknowledgement support.
docs/serving/realtime_duplex_api.md
+23 -17
View File
@@ -311,13 +311,19 @@ work unmodified. The overlap falls into the three tiers used by the event catalo
| Direction | Messages |
| --- | --- |
| Client → server | `session.update`, `input_audio_buffer.append` / `.commit` / `.clear`, `conversation.item.create` / `.retrieve` / `.truncate` / `.delete`, `response.create`, `response.cancel`, `output_audio_buffer.clear` (OpenAI: WebRTC-only; here also a WebSocket event) |
| Server → client | `error` (same `{type, code, message, event_id, param}` envelope and the same classes `invalid_request_error` / `server_error` / `rate_limit_error`), `session.created`, `session.updated`, `input_audio_buffer.committed` / `.cleared` / `.speech_started` / `.speech_stopped`, `conversation.item.created` / `.added` / `.done` / `.deleted` / `.retrieved` / `.truncated`, `conversation.item.input_audio_transcription.completed`, `response.created`, `response.done`, `response.output_item.added` / `.done`, `response.content_part.added` / `.done`, `response.audio.delta` / `.done`, `response.audio_transcript.delta` / `.done`, `response.output_text.delta` / `.done`, `response.function_call_arguments.delta` / `.done`, `rate_limits.updated`, `output_audio_buffer.cleared` |
| Server → client | `error` (same `{type, code, message, event_id, param}` envelope and the same classes `invalid_request_error` / `server_error` / `rate_limit_error`), `session.created`, `session.updated`, `input_audio_buffer.committed` / `.cleared` / `.speech_started` / `.speech_stopped`, `conversation.item.created` / `.added` / `.done` / `.deleted` / `.retrieved` / `.truncated`, `conversation.item.input_audio_transcription.completed`, `response.created`, `response.done`, `response.output_item.added` / `.done`, `response.content_part.added` / `.done`, `response.output_audio.delta` / `.done`, `response.output_audio_transcript.delta` / `.done`, `response.output_text.delta` / `.done`, `response.function_call_arguments.delta` / `.done`, `rate_limits.updated`, `output_audio_buffer.cleared` |
| Session fields | `model`, `modalities` / `output_modalities`, `instructions`, `voice`, `input_audio_format` / `output_audio_format` (beta spelling) **and** `audio.input.format` / `audio.output.format` objects (GA spelling), `turn_detection.server_vad {threshold, prefix_padding_ms, silence_duration_ms}`, `input_audio_transcription`, `input_audio_noise_reduction`, `tools`, `tool_choice`, `temperature`, `max_response_output_tokens`, `speed`, `tracing` |
| Object shapes | `realtime.session`, `realtime.item`, `realtime.response`; item content parts `input_text` / `input_audio` / `output_text` / `output_audio`; `function_call` and `function_call_output` items; `previous_item_id` chaining; `output_index` / `content_index` addressing |
| Sequencing | `response.created → output_item.added → content_part.added → deltas → *.done → content_part.done → output_item.done → response.done → rate_limits.updated`; `speech_started → speech_stopped → committed` |
The server deliberately emits and accepts **both** the beta and the GA
spellings at once (`response.audio.delta` alongside `output_modalities`,
Audio/transcript output uses only the current (non-beta) OpenAI event names
(`response.output_audio.delta`, not the deprecated `response.audio.delta`),
matching `openai.types.realtime` in the official `openai` Python SDK and what
clients built against it (e.g. `livekit-plugins-openai`'s non-Azure code path)
expect; a client still on the old beta SDK types would not recognize these
events. Session fields and `conversation.item.*` events, in contrast, are
deliberately dual: the server emits and accepts **both** the beta and the GA
spellings at once (`output_modalities` alongside `modalities`,
`conversation.item.added` alongside `conversation.item.created`) so either
generation of OpenAI client parses the stream.
@@ -329,7 +335,7 @@ A stock client ignores the extra keys; the extensions are additive.
| --- | --- |
| `session.created` / `session.updated` | top-level `incarnation`, `attachment_generation`, `resume_token`; inside `session`: `state`, `turn_state`, `epoch`, `turn_id`, `active_request_id`, `active_response_id`, `active_response_turn_id`, `overlap_policy`, `overlap_*_ms/rms`, `playback_commit_policy`, `playback`, `capabilities`, `ref_audio`, `extra_body`, `idle_timeout_s`, `response_format` |
| `response.created` / `response.done` / `response.listen` | `response_id` at top level; the raw duplex event under `response.metadata` (`duplex_event`); `status_details.reason` uses vLLM reasons (`barge_in`, `client_cancelled`, `new_response`, …) |
| `response.audio.delta` | `format`, `sample_rate_hz`, `metadata{session_id, epoch, model_speak, end_of_turn, audio_duration_ms, audio_text_marks, playback}` |
| `response.output_audio.delta` | `format`, `sample_rate_hz`, `metadata{session_id, epoch, model_speak, end_of_turn, audio_duration_ms, audio_text_marks, playback}` |
| `response.speak` (inserted before the first delta) | not an OpenAI event, but rides the OpenAI response envelope (`response_id`, `item_id`, `output_index`, `content_index`) |
| `input_audio_buffer.append` | `is_speech`, `video_frames` (+ `max_slice_nums`), `duration_ms`, `audio_end_ms`, per-event `format` / `sample_rate_hz` |
| `input_audio_buffer.commit` | `final`, `response_create`, `is_speech:false` (silence declaration) |
@@ -443,12 +449,12 @@ table lives in the normative contract section of [Full-Duplex Runtime (MiniCPM-o
| `response.output_item.added` | 1 | Assistant item (or `function_call` item) attached to the response. |
| `response.content_part.added` | 1 | Audio/text content part opened on the assistant item. |
| `response.speak` | 3 | Model chose to speak (native lane); at most once per response, before the first audio delta. |
| `response.audio.delta` | 2 | One ordered audio chunk. OpenAI name; adds `format`, `sample_rate_hz`, `metadata{epoch, model_speak, end_of_turn, audio_text_marks, playback}`. |
| `response.audio_transcript.delta` | 1 | Transcript text paired one-to-one with each audio delta. |
| `response.output_audio.delta` | 2 | One ordered audio chunk. OpenAI name; adds `format`, `sample_rate_hz`, `metadata{epoch, model_speak, end_of_turn, audio_text_marks, playback}`. |
| `response.output_audio_transcript.delta` | 1 | Transcript text paired one-to-one with each audio delta. |
| `response.output_text.delta` | 1 | Text delta for text-modality responses. |
| `response.output_text.done` | 1 | Final text of a text-modality response. |
| `response.audio.done` | 1 | Audio stream closed for the response. |
| `response.audio_transcript.done` | 1 | Full transcript (concatenation of the deltas). |
| `response.output_audio.done` | 1 | Audio stream closed for the response. |
| `response.output_audio_transcript.done` | 1 | Full transcript (concatenation of the deltas). |
| `response.content_part.done` | 1 | Content part finalized with its transcript/text. |
| `response.output_item.done` | 1 | Assistant item finalized. |
| `response.done` | 2 | Terminal event; `status` ∈ `completed` \| `cancelled` \| `failed`. OpenAI name; vLLM-Omni `status_details.reason` values and raw duplex event under `metadata`. |
@@ -902,11 +908,11 @@ deferred commit during an active response (`event.response_create_deferred`)
}
```
`S→C response.audio.delta` (one ~1 s unit, 24 kHz PCM16)
`S→C response.output_audio.delta` (one ~1 s unit, 24 kHz PCM16)
```json
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": "resp_01",
"item_id": "item_resp_01",
"output_index": 0,
@@ -926,10 +932,10 @@ deferred commit during an active response (`event.response_create_deferred`)
}
```
`S→C response.audio_transcript.delta` (exactly one per audio delta)
`S→C response.output_audio_transcript.delta` (exactly one per audio delta)
```json
{"type": "response.audio_transcript.delta", "response_id": "resp_01", "item_id": "item_resp_01", "output_index": 0, "content_index": 0, "delta": "It is sunny"}
{"type": "response.output_audio_transcript.delta", "response_id": "resp_01", "item_id": "item_resp_01", "output_index": 0, "content_index": 0, "delta": "It is sunny"}
```
`S→C response.output_text.delta` / `response.output_text.done` (text modality)
@@ -942,14 +948,14 @@ deferred commit during an active response (`event.response_create_deferred`)
{"type": "response.output_text.done", "response_id": "resp_02", "item_id": "item_resp_02", "output_index": 0, "content_index": 0, "text": "Tomorrow looks cloudy."}
```
`S→C response.audio.done` / `response.audio_transcript.done`
`S→C response.output_audio.done` / `response.output_audio_transcript.done`
```json
{"type": "response.audio.done", "response_id": "resp_01", "item_id": "item_resp_01", "output_index": 0, "content_index": 0}
{"type": "response.output_audio.done", "response_id": "resp_01", "item_id": "item_resp_01", "output_index": 0, "content_index": 0}
```
```json
{"type": "response.audio_transcript.done", "response_id": "resp_01", "item_id": "item_resp_01", "output_index": 0, "content_index": 0, "transcript": "It is sunny, 24 degrees."}
{"type": "response.output_audio_transcript.done", "response_id": "resp_01", "item_id": "item_resp_01", "output_index": 0, "content_index": 0, "transcript": "It is sunny, 24 degrees."}
```
`S→C response.content_part.done`
@@ -1191,8 +1197,8 @@ Realtime projector renames and fans out these events before they reach a
| `session.create` (`open_session`, `session.config`) | first `session.update` |
| `input.committed` | `conversation.item.added` / `conversation.item.created`, `input_audio_buffer.committed`, `conversation.item.input_audio_transcription.completed`, `conversation.item.done` |
| `input.cancelled` | `input_audio_buffer.cleared` |
| `response.output_audio.delta` (with `audio_transcript`) | optional `response.speak`, then `response.audio.delta` and `response.audio_transcript.delta` |
| `response.output_audio.done` / `response.output_text.done` | `response.audio.done` / `response.output_text.done` |
| `response.output_audio.delta` (with `audio_transcript`) | optional `response.speak`, then `response.output_audio.delta` and `response.output_audio_transcript.delta` |
| `response.output_audio.done` / `response.output_text.done` | `response.output_audio.done` / `response.output_text.done` |
| `response.text.delta` | `response.output_text.delta` |
| `response.message` (chat-fallback raw chunk) | passed through |
| `audio.cancelled` | optional `output_audio_buffer.cleared`, then the cancelled terminal events and `response.done` with `status: "cancelled"` |
examples/online_serving/barge_in_client.py
+1 -1
View File
@@ -65,7 +65,7 @@ from vllm_omni.clients.duplex import ( # noqa: E402
)
_TRANSCRIPT_DELTA_EVENTS = {
"response.audio_transcript.delta",
"response.output_audio_transcript.delta",
"response.output_text.delta",
"response.text.delta",
}
examples/online_serving/minicpmo/realtime_duplex_demo.py
+5 -5
View File
@@ -63,14 +63,14 @@ class _StreamingOutputWriter:
def handle(self, event: dict[str, object]) -> None:
event_type = event.get("type")
if event_type in {
"response.audio_transcript.delta",
"response.output_audio_transcript.delta",
"response.output_text.delta",
}:
delta = event.get("delta")
if isinstance(delta, str) and delta:
print(delta, end="", file=sys.stderr, flush=True)
return
if event_type != "response.audio.delta":
if event_type != "response.output_audio.delta":
return
delta = event.get("delta") or event.get("audio")
@@ -509,12 +509,12 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
audio = collector.audio_bytes()
first_text_at_s = collector.first_received_at(
"response.audio_transcript.delta",
"response.output_audio_transcript.delta",
"response.output_text.delta",
after_s=commit_sent_at_s,
)
first_audio_at_s = collector.first_received_at(
"response.audio.delta",
"response.output_audio.delta",
after_s=commit_sent_at_s,
)
response_created_at_s = collector.first_received_at(
@@ -536,7 +536,7 @@ async def run_demo(args: argparse.Namespace) -> dict[str, object]:
for event in collector.events
if event.get("type")
in {
"response.audio_transcript.delta",
"response.output_audio_transcript.delta",
"response.output_text.delta",
}
]
examples/online_serving/minicpmo/realtime_web/app/static/app.js
+4 -4
View File
@@ -356,7 +356,7 @@
case 'response.speak':
beginAssistant(responseId);
break;
case 'response.audio.delta':
case 'response.output_audio.delta':
currentResponseId = responseId || currentResponseId;
assistantActive = true;
setModel('Speaking');
@@ -364,13 +364,13 @@
.then((decoded) => feedPlayback(decoded, responseId))
.catch((error) => appendLog(`audio decode failed: ${error.message || error}`, true));
break;
case 'response.audio.done':
case 'response.output_audio.done':
requestPlaybackDrain(responseId);
break;
case 'response.audio_transcript.delta':
case 'response.output_audio_transcript.delta':
addTranscript('assistant', event.delta || '');
break;
case 'response.audio_transcript.done':
case 'response.output_audio_transcript.done':
finishTranscript('assistant', event.transcript || '');
break;
case 'conversation.item.input_audio_transcription.delta':
tests/benchmarks/patch/test_patch.py
+2 -2
View File
@@ -168,7 +168,7 @@ async def test_seed_tts_realtime_duplex_exports_per_request_metrics(monkeypatch)
)
self.events.add(
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": response_id,
"delta": f"turn {self.response_count}",
},
@@ -176,7 +176,7 @@ async def test_seed_tts_realtime_duplex_exports_per_request_metrics(monkeypatch)
)
self.events.add(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": response_id,
"delta": base64.b64encode(b"\x00\x00" * 2400).decode(),
"sample_rate_hz": 24_000,
tests/benchmarks/test_omniinteract.py
+3 -3
View File
@@ -65,7 +65,7 @@ def _audio(
value: int = 1,
) -> dict[str, object]:
event: dict[str, object] = {
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": response_id,
"format": "pcm16",
"delta": base64.b64encode(bytes((value, 0)) * samples).decode(),
@@ -76,7 +76,7 @@ def _audio(
def _text(response_id: str = "r1", value: str = "hello") -> dict[str, object]:
return {"type": "response.audio_transcript.delta", "response_id": response_id, "delta": value}
return {"type": "response.output_audio_transcript.delta", "response_id": response_id, "delta": value}
def _listen(*, buffering: bool = False) -> dict[str, object]:
@@ -427,7 +427,7 @@ def test_artifacts_publish_official_bundle_and_sparse_deferred_state(tmp_path: P
assert result.success and result.eligible_for_official_eval
assert summary["scene_type"] == "multi_turn"
assert context and context.spans
assert all("delta" not in event for event in context.events if event["type"] == "response.audio.delta")
assert all("delta" not in event for event in context.events if event["type"] == "response.output_audio.delta")
oi.publish_deferred_case_artifacts(tmp_path / "out", case, result)
directory = oi._output_dir(tmp_path / "out", case)
assert all((directory / name).is_file() for name in oi.SUCCESS_ARTIFACTS)
tests/clients/test_duplex_client.py
+12 -12
View File
@@ -279,7 +279,7 @@ async def test_response_handle_flow():
sock.feed({"type": "response.speak", "response_id": "resp-1", "server_event_seq": 3})
sock.feed(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": "resp-1",
"delta": _b64(chunk),
"sample_rate_hz": 24_000,
@@ -288,7 +288,7 @@ async def test_response_handle_flow():
)
sock.feed(
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": "resp-1",
"delta": "hi there",
"server_event_seq": 5,
@@ -370,7 +370,7 @@ async def test_terminal_listen_keeps_spoken_decision():
sock.feed({"type": "response.speak", "response_id": "resp-1", "server_event_seq": 3})
sock.feed(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": "resp-1",
"delta": _b64(chunk),
"sample_rate_hz": 24_000,
@@ -418,7 +418,7 @@ async def test_slow_audio_consumer_drops_oldest_instead_of_stalling():
)
for payload in (b"c1", b"c2", b"c3"):
# _feed must never block the reader, even against a full queue.
handle._feed(AudioDelta({"type": "response.audio.delta", "delta": _b64(payload)}))
handle._feed(AudioDelta({"type": "response.output_audio.delta", "delta": _b64(payload)}))
handle._finish(None)
chunks = [chunk async for chunk in handle.audio()]
assert chunks == [b"c3"] # oldest chunks were dropped, the sentinel landed
@@ -561,7 +561,7 @@ def test_event_collector_accumulates_audio():
collector = EventCollector()
collector.add({"type": "response.created", "response": {"id": "r1"}}, received_at_s=1.0)
collector.add(
{"type": "response.audio.delta", "response_id": "r1", "delta": _b64(b"ab"), "sample_rate_hz": 16_000},
{"type": "response.output_audio.delta", "response_id": "r1", "delta": _b64(b"ab"), "sample_rate_hz": 16_000},
received_at_s=1.1,
)
collector.add(
@@ -632,7 +632,7 @@ def test_event_collector_partitions_audio_by_response():
collector.add({"type": "response.created", "response": {"id": "resp-a"}})
collector.add(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": "resp-a",
"delta": base64.b64encode(b"audio-a").decode("ascii"),
"sample_rate_hz": 16_000,
@@ -643,7 +643,7 @@ def test_event_collector_partitions_audio_by_response():
assert collector.audio_bytes("resp-a") == b"audio-a"
assert collector.output_sample_rate_hz == 16_000
assert collector.first_received_at("response.created") is not None
assert collector.last_received_at("response.audio.delta") is not None
assert collector.last_received_at("response.output_audio.delta") is not None
def test_event_collector_reports_engine_token_and_audio_intervals():
@@ -664,7 +664,7 @@ def test_event_collector_reports_engine_token_and_audio_intervals():
for received_at_s, cumulative_audio_ms in ((10.2, 80), (10.25, 160), (10.36, 240)):
collector.add(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": "resp-a",
"delta": base64.b64encode(b"audio").decode("ascii"),
"sample_rate_hz": 16_000,
@@ -676,11 +676,11 @@ def test_event_collector_reports_engine_token_and_audio_intervals():
received_at_s=received_at_s,
)
collector.add(
{"type": "response.audio_transcript.delta", "response_id": "resp-a", "delta": ""},
{"type": "response.output_audio_transcript.delta", "response_id": "resp-a", "delta": ""},
received_at_s=10.1,
)
collector.add(
{"type": "response.audio_transcript.delta", "response_id": "resp-a", "delta": "hello"},
{"type": "response.output_audio_transcript.delta", "response_id": "resp-a", "delta": "hello"},
received_at_s=10.15,
)
collector.add(
@@ -752,7 +752,7 @@ def test_response_timing_ignores_unowned_session_level_metrics():
)
collector.add(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": "resp-a",
"delta": base64.b64encode(b"audio").decode("ascii"),
"metadata": {
@@ -887,7 +887,7 @@ def test_reference_audio_data_url(tmp_path):
def test_event_collector_response_text_joins_deltas_per_response():
collector = EventCollector()
collector.add({"type": "response.created", "response_id": "r1"})
collector.add({"type": "response.audio_transcript.delta", "response_id": "r1", "delta": "he"})
collector.add({"type": "response.output_audio_transcript.delta", "response_id": "r1", "delta": "he"})
collector.add({"type": "response.output_text.delta", "response_id": "r2", "delta": "other"})
collector.add({"type": "response.text.delta", "response_id": "r1", "delta": "llo"})
assert collector.response_text("r1") == "hello"
tests/e2e/features/fullduplex/test_client.py
+6 -6
View File
@@ -241,7 +241,7 @@ def test_realtime_event_collector_partitions_audio_by_response():
collector.add({"type": "response.created", "response": {"id": "resp-a"}})
collector.add(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": "resp-a",
"delta": base64.b64encode(b"audio-a").decode("ascii"),
"sample_rate_hz": 16_000,
@@ -252,7 +252,7 @@ def test_realtime_event_collector_partitions_audio_by_response():
assert collector.audio_bytes("resp-a") == b"audio-a"
assert collector.output_sample_rate_hz == 16_000
assert collector.first_received_at("response.created") is not None
assert collector.last_received_at("response.audio.delta") is not None
assert collector.last_received_at("response.output_audio.delta") is not None
def test_realtime_event_collector_reports_engine_token_and_audio_intervals():
@@ -273,7 +273,7 @@ def test_realtime_event_collector_reports_engine_token_and_audio_intervals():
for received_at_s, cumulative_audio_ms in ((10.2, 80), (10.25, 160), (10.36, 240)):
collector.add(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": "resp-a",
"delta": base64.b64encode(b"audio").decode("ascii"),
"sample_rate_hz": 16_000,
@@ -286,7 +286,7 @@ def test_realtime_event_collector_reports_engine_token_and_audio_intervals():
)
collector.add(
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": "resp-a",
"delta": "",
},
@@ -294,7 +294,7 @@ def test_realtime_event_collector_reports_engine_token_and_audio_intervals():
)
collector.add(
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": "resp-a",
"delta": "hello",
},
@@ -369,7 +369,7 @@ def test_response_timing_ignores_unowned_session_level_metrics():
)
collector.add(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": "resp-a",
"delta": base64.b64encode(b"audio").decode("ascii"),
"metadata": {
tests/e2e/features/fullduplex/test_omni_duplex_eval_cli.py
+1 -1
View File
@@ -119,7 +119,7 @@ async def test_generate_exercises_realtime_socket_and_media_clock(tmp_path: Path
await websocket.send(
json.dumps(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": "r1",
"delta": base64.b64encode(b"\0\0" * 240).decode(),
"sample_rate_hz": 24_000,
tests/e2e/online_serving/helpers/minicpmo_realtime_duplex_scenarios.py
+21 -17
View File
@@ -213,7 +213,7 @@ class DemoState:
item_id = item.get("id")
if isinstance(item_id, str) and item_id not in self.assistant_item_ids:
self.assistant_item_ids.append(item_id)
elif event_type == "response.audio.delta":
elif event_type == "response.output_audio.delta":
delta = event.get("delta") or event.get("audio")
if isinstance(delta, str) and delta:
try:
@@ -386,8 +386,8 @@ class DemoState:
"response.output_item.added",
"response.content_part.added",
"response.speak",
"response.audio.delta",
"response.audio.done",
"response.output_audio.delta",
"response.output_audio.done",
"response.content_part.done",
"response.output_item.done",
"response.done",
@@ -402,8 +402,8 @@ class DemoState:
)
if index is None and event_type not in {
"response.speak",
"response.audio.delta",
"response.audio.done",
"response.output_audio.delta",
"response.output_audio.done",
}:
return {}
if index is not None:
@@ -451,13 +451,13 @@ class DemoState:
"response.output_item.added",
"response.content_part.added",
"response.speak",
"response.audio.delta",
"response.audio.done",
"response.output_audio.delta",
"response.output_audio.done",
"response.content_part.done",
"response.output_item.done",
"response.done",
]
if "response.audio.delta" not in indices_by_type:
if "response.output_audio.delta" not in indices_by_type:
listen_index = self.first_index(
"response.listen",
lambda event: (
@@ -515,7 +515,7 @@ class DemoState:
def model_speak_before_audio_ok(self) -> bool:
speak_index = self.first_index("response.speak")
audio_index = self.first_index("response.audio.delta")
audio_index = self.first_index("response.output_audio.delta")
return speak_index is not None and audio_index is not None and speak_index < audio_index
def response_done(self, response_id: str | None) -> bool:
@@ -532,7 +532,7 @@ class DemoState:
return sum(
1
for event in self.events
if event.get("type") == "response.audio.delta" and self._event_response_id(event) == response_id
if event.get("type") == "response.output_audio.delta" and self._event_response_id(event) == response_id
)
def response_playback_sent_ms(self, response_id: str | None) -> int:
@@ -565,14 +565,16 @@ class DemoState:
return "".join(
str(event.get("delta", ""))
for event in self.events
if event.get("type") == "response.audio_transcript.delta" and self._event_response_id(event) == response_id
if event.get("type") == "response.output_audio_transcript.delta"
and self._event_response_id(event) == response_id
)
def response_transcript_done(self, response_id: str) -> list[str]:
return [
str(event.get("transcript", ""))
for event in self.events
if event.get("type") == "response.audio_transcript.done" and self._event_response_id(event) == response_id
if event.get("type") == "response.output_audio_transcript.done"
and self._event_response_id(event) == response_id
]
def completed_response_ids(self) -> list[str]:
@@ -603,7 +605,7 @@ class DemoState:
return 0
stale = 0
for index, event in enumerate(self.events):
if event.get("type") != "response.audio.delta":
if event.get("type") != "response.output_audio.delta":
continue
metadata = event.get("metadata")
if not isinstance(metadata, dict):
@@ -1126,7 +1128,7 @@ async def _send_clean_turn(
state,
lambda: state.response_audio_delta_count(response_id) > 0,
timeout_s=timeout_s,
label=f"{transcript} response.audio.delta",
label=f"{transcript} response.output_audio.delta",
)
await _wait_for(
state,
@@ -1341,7 +1343,7 @@ async def _send_listen_only_overlap_pair(
state,
lambda: state.response_audio_delta_count(first_response_id) > 0,
timeout_s=timeout_s,
label=f"{transcripts[0]} response.audio.delta",
label=f"{transcripts[0]} response.output_audio.delta",
)
if state.response_done(first_response_id):
raise RuntimeError("first response completed before overlap input could start")
@@ -1600,7 +1602,9 @@ async def run_demo(args: DemoArgs) -> dict[str, object]:
transcript_hints_enabled=transcript_hints_enabled and not continuous_input,
)
model_speak_event_ok = state.model_speak_before_audio_ok()
realtime_audio_lifecycle_ok = state.count("response.audio.delta") > 0 and state.count("response.audio.done") > 0
realtime_audio_lifecycle_ok = (
state.count("response.output_audio.delta") > 0 and state.count("response.output_audio.done") > 0
)
completed_response_ids = state.completed_response_ids()
observed_turn_response_ids = [response_id for response_id in turn_response_ids if isinstance(response_id, str)]
expected_empty_response_ids = {
@@ -1625,7 +1629,7 @@ async def run_demo(args: DemoArgs) -> dict[str, object]:
state.count("response.created") == state.count("response.done")
and len(state.response_ids) == len(completed_response_ids)
and len(completed_response_ids) == len(set(completed_response_ids))
and state.count("response.audio.done") <= state.count("response.done")
and state.count("response.output_audio.done") <= state.count("response.done")
and _response_cardinality_ok(
completed_response_ids,
expected_turns=expected_turns,
tests/e2e/online_serving/nemotron_voicechat_realtime_duplex.py
+4 -4
View File
@@ -215,7 +215,7 @@ async def run(args: argparse.Namespace) -> dict[str, object]:
client.events.count("response.function_call_arguments.done") > 0
if args.expect_function_call
else (
client.events.count("response.audio.delta") >= args.minimum_audio_chunks
client.events.count("response.output_audio.delta") >= args.minimum_audio_chunks
and (
args.allow_incomplete_response
or client.events.count("response.done") > completed_responses_at_commit
@@ -279,10 +279,10 @@ async def run(args: argparse.Namespace) -> dict[str, object]:
transcript = "".join(
str(event.get("delta", ""))
for event in later
if event.get("type") == "response.audio_transcript.delta"
if event.get("type") == "response.output_audio_transcript.delta"
).lower()
return (
any(event.get("type") == "response.audio.delta" for event in later)
any(event.get("type") == "response.output_audio.delta" for event in later)
and any(event.get("type") == "response.done" for event in later)
and (args.expected_post_tool_text is None or args.expected_post_tool_text.lower() in transcript)
)
@@ -297,7 +297,7 @@ async def run(args: argparse.Namespace) -> dict[str, object]:
await asyncio.sleep(args.drain_s)
audio = client.events.audio_bytes()
audio_events = _events(client, "response.audio.delta")
audio_events = _events(client, "response.output_audio.delta")
rates = {event.get("sample_rate_hz") for event in audio_events}
if not args.expect_function_call and audio and rates != {OUTPUT_SAMPLE_RATE_HZ}:
raise AssertionError(f"unexpected output sample rates: {rates}")
tests/e2e/online_serving/personaplex_realtime_duplex.py
+3 -3
View File
@@ -244,7 +244,7 @@ def _audio_frame_stats(
def _response_ids(client: RawRealtimeProbe) -> set[str]:
return {
str(event["response_id"])
for event in _events(client, "response.audio.delta")
for event in _events(client, "response.output_audio.delta")
if isinstance(event.get("response_id"), str)
}
@@ -268,7 +268,7 @@ def _session_result(
raise AssertionError(f"output audio is non-finite or silent: samples={pcm.size}, rms={rms}")
rates = {
int(event["sample_rate_hz"])
for event in _events(client, "response.audio.delta")
for event in _events(client, "response.output_audio.delta")
if isinstance(event.get("sample_rate_hz"), int)
}
if rates != {SAMPLE_RATE_HZ}:
@@ -289,7 +289,7 @@ def _session_result(
)
runtime_metadata = [
metadata["vllm_omni"]
for event in _events(client, "response.audio.delta")
for event in _events(client, "response.output_audio.delta")
if isinstance((metadata := event.get("metadata")), dict) and isinstance(metadata.get("vllm_omni"), dict)
]
if not any(
tests/e2e/online_serving/run_minicpmo_realtime_duplex_demo_pair.py
+4 -1
View File
@@ -1,3 +1,6 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
"""Run two MiniCPM-o Realtime duplex demo processes concurrently.
This E2E driver covers the same process boundary users exercise manually:
@@ -20,7 +23,7 @@ from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[3]
DEMO_PATH = REPO_ROOT / "examples/online_serving/minicpmo/realtime_duplex_demo.py"
AUDIO_DELTA_EVENTS = {"response.audio.delta", "response.output_audio.delta"}
AUDIO_DELTA_EVENTS = {"response.output_audio.delta"}
def _canonical_path(path: str) -> Path:
tests/e2e/online_serving/run_minicpmo_realtime_duplex_server_vad.py
+10 -3
View File
@@ -1,3 +1,6 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
import base64
from pathlib import Path
@@ -41,7 +44,10 @@ async def run_server_vad_interrupt(args) -> dict[str, object]:
next(e for e in client.events.events if e.get("type") == "response.created")
)
assert target_id
await until(lambda: _events(client.events.events, target_id, "response.audio.delta"), "response.audio.delta")
await until(
lambda: _events(client.events.events, target_id, "response.output_audio.delta"),
"response.output_audio.delta",
)
cursor = len(client.events.events)
await client.stream_pcm16(interrupt + bytes(16_000 * 2 * 800 // 1000), chunk_ms=args.chunk_ms, realtime=True)
await until(lambda: _events(client.events.events, target_id, "response.done"), "cancelled response.done")
@@ -64,10 +70,11 @@ async def run_server_vad_interrupt(args) -> dict[str, object]:
terminal = done[0] if len(done) == 1 else None
trailing = events[events.index(terminal) + 1 :] if terminal is not None else events
stale = any(
e.get("type") == "response.audio.delta" and RealtimeEventCollector.response_id(e) == target_id for e in trailing
e.get("type") == "response.output_audio.delta" and RealtimeEventCollector.response_id(e) == target_id
for e in trailing
)
followup_audio = any(
e.get("type") == "response.audio.delta" and RealtimeEventCollector.response_id(e) != target_id
e.get("type") == "response.output_audio.delta" and RealtimeEventCollector.response_id(e) != target_id
for e in events[cursor:]
)
ok = (
tests/e2e/online_serving/run_minicpmo_realtime_duplex_soft_interrupt.py
+5 -2
View File
@@ -1,3 +1,6 @@
# SPDX-License-Identifier: Apache-2.0
# SPDX-FileCopyrightText: Copyright contributors to the vLLM-Omni project
"""Validate MiniCPM-o Realtime duplex soft-interrupt delta streaming.
This E2E driver runs the public ``realtime_duplex_demo.py`` against a live
@@ -21,9 +24,9 @@ from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[3]
DEMO_PATH = REPO_ROOT / "examples/online_serving/minicpmo/realtime_duplex_demo.py"
AUDIO_DELTA_EVENTS = {"response.audio.delta", "response.output_audio.delta"}
AUDIO_DELTA_EVENTS = {"response.output_audio.delta"}
TRANSCRIPT_DELTA_EVENTS = {
"response.audio_transcript.delta",
"response.output_audio_transcript.delta",
"response.output_text.delta",
}
tests/e2e/online_serving/test_minicpmo_realtime_duplex_drivers.py
+80 -45
View File
@@ -168,9 +168,9 @@ def test_realtime_duplex_demo_pair_launches_demo_processes_concurrently(tmp_path
"events = [",
" {'type': 'session.created'},",
" {'type': 'response.created', 'response': {'id': response_id}},",
" {'type': 'response.audio.delta', 'response_id': response_id, "
" {'type': 'response.output_audio.delta', 'response_id': response_id, "
"'delta': 'AAAA', 'sample_rate_hz': 24000, '_client_received_at_s': 10.1},",
" {'type': 'response.audio.delta', 'response_id': response_id, "
" {'type': 'response.output_audio.delta', 'response_id': response_id, "
"'delta': 'AAAA', 'sample_rate_hz': 24000, '_client_received_at_s': 10.6},",
" {'type': 'response.done', 'response_id': response_id},",
" {'type': 'session.closed'},",
@@ -237,8 +237,8 @@ def test_realtime_duplex_demo_pair_rejects_false_green_audio_contract(tmp_path):
response_id = "resp-a"
events = [
{"type": "response.created", "response": {"id": response_id}},
{"type": "response.audio.delta", "response_id": response_id, "delta": "AAAA", "sample_rate_hz": 24000},
{"type": "response.audio.delta", "response_id": "resp-other", "delta": "AAAA", "sample_rate_hz": 24000},
{"type": "response.output_audio.delta", "response_id": response_id, "delta": "AAAA", "sample_rate_hz": 24000},
{"type": "response.output_audio.delta", "response_id": "resp-other", "delta": "AAAA", "sample_rate_hz": 24000},
{"type": "response.done", "response_id": response_id},
]
(output / "events.jsonl").write_text(
@@ -459,25 +459,25 @@ def test_realtime_duplex_soft_interrupt_accepts_multi_delta_handoff_sequence(tmp
{"type": "response.created", "response": {"id": first_response_id}, "_client_received_at_s": 2.0},
{"type": "response.speak", "response_id": first_response_id, "_client_received_at_s": 2.0},
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": first_response_id,
"delta": "AAAA",
"_client_received_at_s": 2.1,
},
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": first_response_id,
"delta": "中国古代四大发明",
"_client_received_at_s": 2.1,
},
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": first_response_id,
"delta": "AAAA",
"_client_received_at_s": 2.6,
},
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": first_response_id,
"delta": "是造纸术。",
"_client_received_at_s": 2.6,
@@ -486,25 +486,25 @@ def test_realtime_duplex_soft_interrupt_accepts_multi_delta_handoff_sequence(tmp
{"type": "response.created", "response": {"id": second_response_id}, "_client_received_at_s": 4.0},
{"type": "response.speak", "response_id": second_response_id, "_client_received_at_s": 4.0},
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": second_response_id,
"delta": "AAAA",
"_client_received_at_s": 4.1,
},
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": second_response_id,
"delta": "一加一等于",
"_client_received_at_s": 4.1,
},
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": second_response_id,
"delta": "AAAA",
"_client_received_at_s": 4.6,
},
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": second_response_id,
"delta": "二。",
"_client_received_at_s": 4.6,
@@ -561,25 +561,25 @@ def test_realtime_duplex_soft_interrupt_accepts_followup_done_after_commit(tmp_p
{"type": "response.created", "response": {"id": first_response_id}, "_client_received_at_s": 2.0},
{"type": "response.speak", "response_id": first_response_id, "_client_received_at_s": 2.0},
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": first_response_id,
"delta": "AAAA",
"_client_received_at_s": 2.1,
},
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": first_response_id,
"delta": "中国古代四大发明",
"_client_received_at_s": 2.1,
},
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": first_response_id,
"delta": "AAAA",
"_client_received_at_s": 2.6,
},
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": first_response_id,
"delta": "是造纸术。",
"_client_received_at_s": 2.6,
@@ -589,25 +589,25 @@ def test_realtime_duplex_soft_interrupt_accepts_followup_done_after_commit(tmp_p
{"type": "response.created", "response": {"id": second_response_id}, "_client_received_at_s": 3.5},
{"type": "response.speak", "response_id": second_response_id, "_client_received_at_s": 3.5},
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": second_response_id,
"delta": "AAAA",
"_client_received_at_s": 3.6,
},
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": second_response_id,
"delta": "一加一等于",
"_client_received_at_s": 3.6,
},
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": second_response_id,
"delta": "AAAA",
"_client_received_at_s": 3.7,
},
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": second_response_id,
"delta": "二。",
"_client_received_at_s": 3.7,
@@ -653,8 +653,18 @@ def test_realtime_duplex_soft_interrupt_model_policy_accepts_single_response(tmp
events = [
{"type": "response.listen", "_client_received_at_s": 1.0},
{"type": "response.created", "response": {"id": response_id}, "_client_received_at_s": 2.0},
{"type": "response.audio.delta", "response_id": response_id, "delta": "AAAA", "_client_received_at_s": 2.1},
{"type": "response.audio.delta", "response_id": response_id, "delta": "AAAA", "_client_received_at_s": 2.15},
{
"type": "response.output_audio.delta",
"response_id": response_id,
"delta": "AAAA",
"_client_received_at_s": 2.1,
},
{
"type": "response.output_audio.delta",
"response_id": response_id,
"delta": "AAAA",
"_client_received_at_s": 2.15,
},
{"type": "response.done", "response_id": response_id, "_client_received_at_s": 2.2},
{"type": "response.listen", "_client_received_at_s": 2.3},
{"type": "input_audio_buffer.committed", "_client_received_at_s": 3.0},
@@ -692,10 +702,25 @@ def test_realtime_duplex_soft_interrupt_model_policy_accepts_commit_during_speak
events = [
{"type": "response.listen", "_client_received_at_s": 1.0},
{"type": "response.created", "response": {"id": response_id}, "_client_received_at_s": 2.0},
{"type": "response.audio.delta", "response_id": response_id, "delta": "AAAA", "_client_received_at_s": 2.1},
{"type": "response.audio.delta", "response_id": response_id, "delta": "AAAA", "_client_received_at_s": 2.2},
{
"type": "response.output_audio.delta",
"response_id": response_id,
"delta": "AAAA",
"_client_received_at_s": 2.1,
},
{
"type": "response.output_audio.delta",
"response_id": response_id,
"delta": "AAAA",
"_client_received_at_s": 2.2,
},
{"type": "input_audio_buffer.committed", "_client_received_at_s": 3.0},
{"type": "response.audio.delta", "response_id": response_id, "delta": "AAAA", "_client_received_at_s": 4.0},
{
"type": "response.output_audio.delta",
"response_id": response_id,
"delta": "AAAA",
"_client_received_at_s": 4.0,
},
{"type": "response.done", "response_id": response_id, "_client_received_at_s": 5.0},
]
(output / "events.jsonl").write_text(
@@ -729,8 +754,18 @@ def test_realtime_duplex_soft_interrupt_response_required_rejects_single_respons
events = [
{"type": "response.listen", "_client_received_at_s": 1.0},
{"type": "response.created", "response": {"id": response_id}, "_client_received_at_s": 2.0},
{"type": "response.audio.delta", "response_id": response_id, "delta": "AAAA", "_client_received_at_s": 2.1},
{"type": "response.audio.delta", "response_id": response_id, "delta": "AAAA", "_client_received_at_s": 2.15},
{
"type": "response.output_audio.delta",
"response_id": response_id,
"delta": "AAAA",
"_client_received_at_s": 2.1,
},
{
"type": "response.output_audio.delta",
"response_id": response_id,
"delta": "AAAA",
"_client_received_at_s": 2.15,
},
{"type": "response.done", "response_id": response_id, "_client_received_at_s": 2.2},
{"type": "response.listen", "_client_received_at_s": 2.3},
{"type": "input_audio_buffer.committed", "_client_received_at_s": 3.0},
@@ -767,19 +802,19 @@ def test_realtime_duplex_soft_interrupt_reports_text_expectation_without_gating(
{"type": "response.listen", "_client_received_at_s": 1.0},
{"type": "response.created", "response": {"id": first_response_id}, "_client_received_at_s": 2.0},
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": first_response_id,
"delta": "AAAA",
"_client_received_at_s": 2.1,
},
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": first_response_id,
"delta": "AAAA",
"_client_received_at_s": 2.2,
},
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": first_response_id,
"delta": "一加一等于二",
"_client_received_at_s": 2.2,
@@ -788,19 +823,19 @@ def test_realtime_duplex_soft_interrupt_reports_text_expectation_without_gating(
{"type": "response.listen", "_client_received_at_s": 2.4},
{"type": "response.created", "response": {"id": second_response_id}, "_client_received_at_s": 3.0},
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": second_response_id,
"delta": "AAAA",
"_client_received_at_s": 3.1,
},
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": second_response_id,
"delta": "AAAA",
"_client_received_at_s": 3.2,
},
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": second_response_id,
"delta": "不知道",
"_client_received_at_s": 3.2,
@@ -834,7 +869,7 @@ def test_realtime_duplex_soft_interrupt_reports_text_expectation_without_gating(
events = [
event
for event in events
if event.get("response_id") != second_response_id or event["type"] != "response.audio_transcript.delta"
if event.get("response_id") != second_response_id or event["type"] != "response.output_audio_transcript.delta"
]
(output / "events.jsonl").write_text(
"".join(demo.json.dumps(event) + "\n" for event in events),
@@ -1123,7 +1158,7 @@ def test_realtime_duplex_demo_partitions_timing_by_response_identity():
)
state.add(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": response_id,
"delta": base64.b64encode(b"audio").decode("ascii"),
"metadata": {
@@ -1142,7 +1177,7 @@ def test_realtime_duplex_demo_partitions_timing_by_response_identity():
)
state.add(
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": response_id,
"delta": "audio",
},
@@ -1251,8 +1286,8 @@ def test_realtime_duplex_demo_distinguishes_late_and_missing_commit():
{"type": "response.output_item.added", "response_id": response_id},
{"type": "response.content_part.added", "response_id": response_id},
{"type": "response.speak", "response_id": response_id},
{"type": "response.audio.delta", "response_id": response_id, "delta": "AAAA"},
{"type": "response.audio.done", "response_id": response_id},
{"type": "response.output_audio.delta", "response_id": response_id, "delta": "AAAA"},
{"type": "response.output_audio.done", "response_id": response_id},
{"type": "response.content_part.done", "response_id": response_id},
{"type": "response.output_item.done", "response_id": response_id},
{"type": "response.done", "response_id": response_id},
@@ -1314,7 +1349,7 @@ def test_realtime_duplex_demo_distinct_inputs_compare_audio_content():
def _add_response_transcript(state, response_id, *, transcript, audio=True):
state.add(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": response_id,
"delta": "YQ==" if audio else "",
}
@@ -1322,14 +1357,14 @@ def _add_response_transcript(state, response_id, *, transcript, audio=True):
if transcript:
state.add(
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": response_id,
"delta": transcript,
}
)
state.add(
{
"type": "response.audio_transcript.done",
"type": "response.output_audio_transcript.done",
"response_id": response_id,
"transcript": transcript,
}
@@ -1404,7 +1439,7 @@ def test_realtime_duplex_demo_speak_gate_rejects_text_channel():
{
"type": "response.speak",
"response_id": "resp-1",
"text": "must be emitted through response.audio_transcript.delta",
"text": "must be emitted through response.output_audio_transcript.delta",
}
)
@@ -1450,7 +1485,7 @@ def test_realtime_duplex_demo_gate_rejects_delta_done_mismatch():
demo = _load_demo_module()
state = demo.DemoState()
_add_response_transcript(state, "resp-1", transcript="delta文本")
done = next(event for event in state.events if event.get("type") == "response.audio_transcript.done")
done = next(event for event in state.events if event.get("type") == "response.output_audio_transcript.done")
done["transcript"] = "另一个done文本"
result = demo._evaluate_transcript_integrity(
@@ -2180,7 +2215,7 @@ def test_realtime_duplex_demo_listen_only_overlap_accepts_silence_unit_before_fi
state.add({"type": "response.created", "response": {"id": "resp-first"}})
state.add(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": "resp-first",
"delta": "YQ==",
}
@@ -2290,7 +2325,7 @@ def test_realtime_duplex_demo_writes_audio_per_response(tmp_path):
state.add({"type": "response.created", "response": {"id": response_id}})
state.add(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": response_id,
"delta": demo.base64.b64encode(payload).decode(),
"sample_rate_hz": 24000,
tests/engine/test_duplex_import_boundary.py
+2 -2
View File
@@ -122,8 +122,8 @@ from vllm_omni.clients.duplex import EventCollector
collector = EventCollector()
chunk = base64.b64encode(b"\\x00\\x00" * 2400).decode("ascii")
collector.add({"type": "response.created", "response_id": "r1"}, received_at_s=1.0)
collector.add({"type": "response.audio.delta", "response_id": "r1", "delta": chunk}, received_at_s=1.1)
collector.add({"type": "response.audio.delta", "response_id": "r1", "delta": chunk}, received_at_s=1.2)
collector.add({"type": "response.output_audio.delta", "response_id": "r1", "delta": chunk}, received_at_s=1.1)
collector.add({"type": "response.output_audio.delta", "response_id": "r1", "delta": chunk}, received_at_s=1.2)
summary = collector.timing_summary(after_s=0.0, response_id="r1")
if "audio_output" not in summary or "request_metrics" not in summary:
raise SystemExit("timing_summary produced no audio_output/request_metrics sections")
tests/entrypoints/duplex/test_websocket_actor.py
+3 -3
View File
@@ -122,12 +122,12 @@ async def test_stale_fence_payload_is_dropped_before_websocket_send():
actor = DuplexWebSocketActor(websocket, current_epoch=lambda: 2)
writer = asyncio.create_task(actor.writer_loop())
await actor.send_json({"type": "response.audio.delta", "epoch": 1})
await actor.send_json({"type": "response.audio.delta", "epoch": 2})
await actor.send_json({"type": "response.output_audio.delta", "epoch": 1})
await actor.send_json({"type": "response.output_audio.delta", "epoch": 2})
await actor.close_writer()
await writer
assert websocket.sent == [{"type": "response.audio.delta", "epoch": 2}]
assert websocket.sent == [{"type": "response.output_audio.delta", "epoch": 2}]
assert actor.stale_output_dropped == 1
tests/entrypoints/openai/test_duplex_session_attachment.py
+1 -1
View File
@@ -78,7 +78,7 @@ def test_event_journal_sequences_acknowledges_and_replays_exact_payloads() -> No
clock = _Clock()
journal = DuplexEventJournal(max_bytes=4096, ttl_s=60.0, clock=clock)
first = journal.record({"type": "response.audio.delta", "delta": "AAAA"})
first = journal.record({"type": "response.output_audio.delta", "delta": "AAAA"})
second = journal.record({"type": "response.done", "response_id": "resp-1"})
assert first.sequence == 1
tests/entrypoints/openai_api/test_duplex_handler.py
+22 -18
View File
@@ -2197,9 +2197,11 @@ def test_native_realtime_protocol_audio_delta_preserves_sample_rate_hz():
)
audio_events = [
payload for payload in payloads if payload["type"] in {"response.output_audio.delta", "response.audio.delta"}
payload
for payload in payloads
if payload["type"] in {"response.output_audio.delta", "response.output_audio.delta"}
]
assert {payload["type"] for payload in audio_events} == {"response.audio.delta"}
assert {payload["type"] for payload in audio_events} == {"response.output_audio.delta"}
assert {payload["format"] for payload in audio_events} == {"pcm16"}
assert {payload["sample_rate_hz"] for payload in audio_events} == {24000}
@@ -2227,14 +2229,14 @@ def test_native_realtime_protocol_ignores_removed_legacy_event_switches():
)
event_types = {payload["type"] for payload in payloads}
assert "response.audio.delta" in event_types
assert "response.audio.done" in event_types
assert "response.audio_transcript.delta" in event_types
assert "response.audio_transcript.done" in event_types
assert "response.output_audio.delta" not in event_types
assert "response.output_audio.done" not in event_types
assert "response.output_audio_transcript.delta" not in event_types
assert "response.output_audio_transcript.done" not in event_types
assert "response.output_audio.delta" in event_types
assert "response.output_audio.done" in event_types
assert "response.output_audio_transcript.delta" in event_types
assert "response.output_audio_transcript.done" in event_types
assert "response.audio.delta" not in event_types
assert "response.audio.done" not in event_types
assert "response.audio_transcript.delta" not in event_types
assert "response.audio_transcript.done" not in event_types
assert "response.output_item.created" not in event_types
assert "response.text.delta" not in event_types
@@ -2256,9 +2258,9 @@ def test_native_realtime_protocol_emits_terminal_audio_transcript_events():
)
by_type = {payload["type"]: payload for payload in payloads}
assert "response.audio.done" in by_type
assert "response.audio_transcript.delta" in by_type
assert "response.audio_transcript.done" in by_type
assert "response.output_audio.done" in by_type
assert "response.output_audio_transcript.delta" in by_type
assert "response.output_audio_transcript.done" in by_type
assert by_type["response.content_part.done"]["part"]["transcript"] == "hello"
assert by_type["response.output_item.done"]["item"]["object"] == "realtime.item"
assert by_type["response.done"]["response"]["output"][0]["object"] == "realtime.item"
@@ -2284,8 +2286,10 @@ def test_native_realtime_protocol_terminal_transcript_equals_joined_deltas():
)
)
transcript = "".join(str(event["delta"]) for event in events if event["type"] == "response.audio_transcript.delta")
done = [str(event["transcript"]) for event in events if event["type"] == "response.audio_transcript.done"]
transcript = "".join(
str(event["delta"]) for event in events if event["type"] == "response.output_audio_transcript.delta"
)
done = [str(event["transcript"]) for event in events if event["type"] == "response.output_audio_transcript.done"]
assert transcript == "It's Canberra. Next question."
assert done == [transcript]
@@ -2374,7 +2378,7 @@ def test_native_realtime_protocol_response_done_emits_audio_done_before_terminal
payloads = protocol._from_duplex_event({"type": "response.done", "response_id": "resp-done"})
event_types = [payload["type"] for payload in payloads]
assert event_types.index("response.audio.done") < event_types.index("response.content_part.done")
assert event_types.index("response.output_audio.done") < event_types.index("response.content_part.done")
assert event_types.index("response.content_part.done") < event_types.index("response.output_item.done")
assert event_types.index("conversation.item.done") < event_types.index("response.done")
assert event_types.index("response.output_item.done") < event_types.index("response.done")
@@ -3181,7 +3185,7 @@ async def test_auto_response_committed_overlap_does_not_precreate_empty_response
def commit_overlap_after_first_response(ws: TimedWebSocket, data: dict[str, Any]) -> None:
nonlocal overlap_sent
if data.get("type") != "response.audio.delta" or overlap_sent:
if data.get("type") != "response.output_audio.delta" or overlap_sent:
return
overlap_sent = True
chunk = {
@@ -8679,7 +8683,7 @@ async def test_minicpmo_native_duplex_uses_segment_text_metadata_for_transcript_
protocol = NativeRealtimeSessionProtocol(TimedWebSocket()) # type: ignore[arg-type]
realtime_events = [event for payload in sent for event in protocol._from_duplex_event(payload)]
transcript_deltas = [
m for m in realtime_events if m.get("type") == "response.audio_transcript.delta" and m.get("delta")
m for m in realtime_events if m.get("type") == "response.output_audio_transcript.delta" and m.get("delta")
]
assert len(created) == 2
assert len(done) == 2
tests/examples/test_minicpmo_realtime_duplex_simple_demo.py
+4 -4
View File
@@ -60,7 +60,7 @@ def test_post_commit_decision_accepts_drained_speak_after_commit_ack():
events = [
{"type": "input_audio_buffer.committed"},
{"type": "response.created", "response": {"id": "resp-a"}},
{"type": "response.audio.delta", "response_id": "resp-a", "delta": "AAAA"},
{"type": "response.output_audio.delta", "response_id": "resp-a", "delta": "AAAA"},
{"type": "response.done", "response_id": "resp-a"},
]
@@ -267,7 +267,7 @@ def test_streaming_output_writer_persists_audio_deltas_as_they_arrive(tmp_path,
writer.handle(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": "resp-a",
"delta": base64.b64encode(first_pcm).decode("ascii"),
"sample_rate_hz": 24_000,
@@ -275,14 +275,14 @@ def test_streaming_output_writer_persists_audio_deltas_as_they_arrive(tmp_path,
)
writer.handle(
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": "resp-a",
"delta": "你好",
}
)
writer.handle(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": "resp-a",
"delta": base64.b64encode(second_pcm).decode("ascii"),
"sample_rate_hz": 24_000,
tests/examples/test_minicpmo_realtime_web_static.py
+2 -2
View File
@@ -42,8 +42,8 @@ def test_client_uses_proxy_relative_realtime_url_and_model_policy_session():
assert "input_audio_buffer.commit" not in source
assert "playback.ack" in source
assert "event.event || event" in source
assert "response.audio.delta" in source
assert "response.audio_transcript.delta" in source
assert "response.output_audio.delta" in source
assert "response.output_audio_transcript.delta" in source
assert "conversation.item.input_audio_transcription" in source
assert "force_barge_in" not in source
assert "server_vad" not in source
tests/model_executor/models/personaplex/duplex/test_unified_runtime.py
+1 -1
View File
@@ -61,7 +61,7 @@ def _audio_client(
frames[:voiced_frames] = 1000
raw = frames.tobytes()
event = {
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": "response-1",
"sample_rate_hz": e2e_driver.SAMPLE_RATE_HZ,
"metadata": {
vllm_omni/benchmarks/duplex/omni_duplex_eval_clock.py
+5 -1
View File
@@ -49,7 +49,11 @@ def extract_timed_sentences(events: Iterable[dict[str, Any]], *, clock: str = "m
last: float | None = None
for event in events:
event_type = event.get("type")
if event_type not in {"response.output_text.delta", "response.audio_transcript.delta", "response.text.delta"}:
if event_type not in {
"response.output_text.delta",
"response.output_audio_transcript.delta",
"response.text.delta",
}:
continue
delta = event.get("delta")
if not isinstance(delta, str) or not delta.strip():
vllm_omni/benchmarks/omniinteract.py
+5 -5
View File
@@ -233,7 +233,7 @@ class _Playback:
if response_id:
self.completed.add(response_id)
continue
if event.get("type") != "response.audio.delta":
if event.get("type") != "response.output_audio.delta":
continue
encoded = event.get("delta") or event.get("audio")
if not response_id:
@@ -241,13 +241,13 @@ class _Playback:
if not isinstance(encoded, str):
raise ValueError("response audio payload is missing")
if event.get("format") is None:
self._warn_once("response.audio.delta omitted format; assumed pcm16")
self._warn_once("response.output_audio.delta omitted format; assumed pcm16")
elif event.get("format") != "pcm16":
raise ValueError("OmniInteract output must be pcm16")
rate = event.get("sample_rate_hz")
if rate is None:
rate = events.output_sample_rate_hz or OUTPUT_SAMPLE_RATE
self._warn_once(f"response.audio.delta omitted sample_rate_hz; assumed {rate}")
self._warn_once(f"response.output_audio.delta omitted sample_rate_hz; assumed {rate}")
if rate != OUTPUT_SAMPLE_RATE:
raise ValueError(f"OmniInteract output must use {OUTPUT_SAMPLE_RATE} Hz audio")
try:
@@ -683,7 +683,7 @@ def _collect_output(
timing = response_times.setdefault(response_id, [start_s, end_s])
timing[0], timing[1] = min(timing[0], start_s), max(timing[1], end_s)
text_event_types = {"response.audio_transcript.delta", "response.output_text.delta", "response.text.delta"}
text_event_types = {"response.output_audio_transcript.delta", "response.output_text.delta", "response.text.delta"}
text_ids = {collector.response_id(event) for event in collector.events if event.get("type") in text_event_types}
if None in text_ids or not text_ids <= created:
raise ValueError("response transcript has no matching response.created")
@@ -726,7 +726,7 @@ def _collect_output(
**{key: value for key, value in event.items() if key not in {"delta", "audio"}},
"audio_bytes": audio_sizes[index],
}
if event.get("type") == "response.audio.delta"
if event.get("type") == "response.output_audio.delta"
else dict(event)
for index, event in enumerate(collector.events)
]
vllm_omni/clients/duplex.py
+2 -4
View File
@@ -460,9 +460,7 @@ _EVENT_TYPES: dict[str, type[DuplexEvent]] = {
"session.expired": SessionExpired,
"response.created": ResponseCreated,
"response.done": ResponseDone,
"response.audio.delta": AudioDelta,
"response.output_audio.delta": AudioDelta,
"response.audio_transcript.delta": TranscriptDelta,
"response.output_audio_transcript.delta": TranscriptDelta,
"response.text.delta": TextDelta,
"response.output_text.delta": TextDelta,
@@ -1359,7 +1357,7 @@ class EventCollector:
if self.response_id(event) == response_id
and event.get("type")
in {
"response.audio_transcript.delta",
"response.output_audio_transcript.delta",
"response.output_text.delta",
"response.text.delta",
}
@@ -1406,7 +1404,7 @@ class EventCollector:
if (
event.get("type")
in {
"response.audio_transcript.delta",
"response.output_audio_transcript.delta",
"response.output_text.delta",
"response.text.delta",
}
vllm_omni/entrypoints/duplex/realtime_output.py
+4 -4
View File
@@ -168,7 +168,7 @@ class RealtimeOutputProjector:
text = ""
payloads.append(
{
"type": "response.audio_transcript.delta",
"type": "response.output_audio_transcript.delta",
"response_id": response_id,
"item_id": self._response_item_id(response_id),
"output_index": 0,
@@ -839,7 +839,7 @@ class RealtimeOutputProjector:
)
payloads.append(
{
"type": "response.audio.delta",
"type": "response.output_audio.delta",
"response_id": response_id,
"item_id": item_id,
"output_index": 0,
@@ -872,7 +872,7 @@ class RealtimeOutputProjector:
state.audio_done_emitted = True
payloads.append(
{
"type": "response.audio.done",
"type": "response.output_audio.done",
"response_id": response_id,
"item_id": item_id,
"output_index": 0,
@@ -882,7 +882,7 @@ class RealtimeOutputProjector:
if transcript:
payloads.append(
{
"type": "response.audio_transcript.done",
"type": "response.output_audio_transcript.done",
"response_id": response_id,
"item_id": item_id,
"output_index": 0,
vllm_omni/entrypoints/duplex/warmup.py
+1 -1
View File
@@ -108,7 +108,7 @@ async def _warmup_duplex_realtime(app, args, warmup_frames: int) -> None:
sent += 1
await asyncio.sleep(0.08)
# Wait for the pipeline's first audio output so every stage ran.
saw_audio = await _recv_until(lambda e: e.get("type") == "response.audio.delta", 30)
saw_audio = await _recv_until(lambda e: e.get("type") == "response.output_audio.delta", 30)
# Close the session EXPLICITLY: a bare websocket close parks the
# session in its disconnect grace window, where it would keep
# occupying the max_sessions=1 slot and block the first client.
vllm_omni/entrypoints/duplex/websocket.py
-2
View File
@@ -55,9 +55,7 @@ MODEL_OUTPUT_EVENTS = frozenset(
"response.output_item.added",
"response.content_part.added",
"response.output_audio.delta",
"response.audio.delta",
"response.output_audio.done",
"response.audio.done",
"response.output_text.delta",
"response.output_text.done",
"response.text.delta",
vllm_omni/entrypoints/openai/serving_speech.py
+1 -1
View File
@@ -1600,7 +1600,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
Field naming follows the OpenAI ``speech.audio.delta`` schema, which
carries the base64 chunk in ``audio`` (not ``delta`` — that is the
Realtime API ``response.audio.delta`` convention, a different event).
Realtime API ``response.output_audio.delta`` convention, a different event).
See https://platform.openai.com/docs/api-reference/audio-streaming.
The terminal ``speech.audio.done`` event carries a ``usage`` object
vllm_omni/experimental/fullduplex/client.py
+4 -4
View File
@@ -234,7 +234,7 @@ class RealtimeEventCollector:
event_type = stored_event.get("type")
if event_type == "response.created" and response_id and response_id not in self.response_ids:
self.response_ids.append(response_id)
if event_type == "response.audio.delta":
if event_type == "response.output_audio.delta":
delta = stored_event.get("delta") or stored_event.get("audio")
if isinstance(delta, str) and response_id:
try:
@@ -263,7 +263,7 @@ class RealtimeEventCollector:
if self.response_id(event) == response_id
and event.get("type")
in {
"response.audio_transcript.delta",
"response.output_audio_transcript.delta",
"response.output_text.delta",
"response.text.delta",
}
@@ -322,7 +322,7 @@ class RealtimeEventCollector:
if (
event.get("type")
in {
"response.audio_transcript.delta",
"response.output_audio_transcript.delta",
"response.output_text.delta",
"response.text.delta",
}
@@ -337,7 +337,7 @@ class RealtimeEventCollector:
if isinstance(stage0, dict):
stage0_metrics = stage0
if event.get("type") != "response.audio.delta" or (
if event.get("type") != "response.output_audio.delta" or (
response_id is not None and event_response_id != response_id
):
continue