[Frontend][Voxtral] Add opt-in segment timestamps to /v1/realtime

docs/serving/online_serving/speech_to_text.md

@@ -181,7 +181,8 @@ Audio must be sent as base64-encoded PCM16 audio at 16kHz sample rate, mono chan
 
 1. Client connects to `ws://host/v1/realtime`
 2. Server sends `session.created` event
-3. Client optionally sends `session.update` with model/params
+3. Client optionally sends `session.update` with model/params, and the server
+    acknowledges with `session.updated`
 4. Client sends `input_audio_buffer.commit` when ready
 5. Client sends `input_audio_buffer.append` events with base64 PCM16 chunks
 6. Server sends `transcription.delta` events with incremental text
@@ -196,17 +197,61 @@ Audio must be sent as base64-encoded PCM16 audio at 16kHz sample rate, mono chan
 | ----- | ----------- |
 | `input_audio_buffer.append` | Send base64-encoded audio chunk: `{"type": "input_audio_buffer.append", "audio": "<base64>"}` |
 | `input_audio_buffer.commit` | Trigger transcription processing or end: `{"type": "input_audio_buffer.commit", "final": bool}` |
-| `session.update` | Configure session: `{"type": "session.update", "model": "model-name"}` |
+| `session.update` | Configure session: `{"type": "session.update", "model": "model-name", "timestamp_granularities": ["segment"]}` |
 
 ### Server → Client Events
 
 | Event | Description |
 | ----- | ----------- |
 | `session.created` | Connection established with session ID and timestamp |
+| `session.updated` | Acknowledgement of `session.update`, echoing the configuration that took effect |
 | `transcription.delta` | Incremental transcription text: `{"type": "transcription.delta", "delta": "text"}` |
 | `transcription.done` | Final transcription with usage stats |
 | `error` | Error notification with message and optional code |
 
+#### Segment-level timestamps
+
+Models that emit one output token per audio frame can report when each piece
+of transcribed audio ended. Opt in per connection:
+
+```json
+{"type": "session.update", "model": "mistralai/Voxtral-Mini-4B-Realtime-2602", "timestamp_granularities": ["segment"]}
+```
+
+If the request was honoured the server replies with `session.updated`, echoing
+`"timestamp_granularities": ["segment"]`. If it cannot be, the server replies
+with `error` instead and the session stays unconfigured, so the opt-in is
+never silently dropped. Once enabled, `transcription.delta` and
+`transcription.done` carry a `segments` array:
+
+```json
+{"type": "transcription.done", "text": " Mary had a little lamb", "segments": [{"text": " Mary had", "end": 1.36}, {"text": " a little lamb", "end": 2.08}]}
+```
+
+Points worth knowing before building on this:
+
+- **End only.** The model marks where a group of audio ends, not where it
+  starts. Deriving a start from the previous `end` is wrong across pauses.
+- **Segments are emission groups, not words.** The model may commit several
+  words at once, and then they share one entry. Expect at most one entry per
+  spoken word.
+- **Granularity is one audio frame**, 80 ms for Voxtral realtime.
+- **The clock is relative to the utterance**, and restarts on every non-final
+  `input_audio_buffer.commit`.
+- **Timestamps trail their text by one event.** The boundary marker itself
+  decodes to no text, so a delta carrying `segments` has an empty `delta`. A
+  client that skips events with empty text will drop them.
+- `transcription.done` repeats every segment of the utterance, including the
+  trailing one, which no delta can carry because generation ends first.
+- Requires `--stream-interval 1` (the default), and the opt-in is rejected
+  otherwise. Batched streaming stalls realtime transcription entirely - the
+  server withholds an output until `stream-interval` tokens accumulate, but
+  the next token cannot be generated until the previous one is fed back - so
+  omitting `timestamp_granularities` is not a way to keep such a server
+  usable.
+- Clients that do not opt in see byte-identical payloads: the `segments` key
+  is omitted entirely.
+
 #### Example Clients
 
 - [openai_realtime_client.py](https://github.com/vllm-project/vllm/tree/main/examples/speech_to_text/realtime/openai_realtime_client.py) - Upload and transcribe an audio file

examples/speech_to_text/realtime/openai_realtime_client.py

@@ -45,7 +45,13 @@ def audio_to_pcm16_base64(audio_path: str) -> str:
     return base64.b64encode(pcm16.tobytes()).decode("utf-8")
 
 
-async def realtime_transcribe(audio_path: str, host: str, port: int, model: str):
+async def realtime_transcribe(
+    audio_path: str,
+    host: str,
+    port: int,
+    model: str,
+    segment_timestamps: bool = False,
+):
     """
     Connect to the Realtime API and transcribe an audio file.
     """
@@ -60,8 +66,21 @@ async def realtime_transcribe(audio_path: str, host: str, port: int, model: str)
             print(f"Unexpected response: {response}")
             return
 
-        # Validate model
-        await ws.send(json.dumps({"type": "session.update", "model": model}))
+        # Validate model, and opt in to segment timestamps if asked
+        session_update = {"type": "session.update", "model": model}
+        if segment_timestamps:
+            session_update["timestamp_granularities"] = ["segment"]
+        await ws.send(json.dumps(session_update))
+
+        # The server echoes the configuration that took effect, so a request
+        # the model cannot honour is never silently dropped.
+        response = json.loads(await ws.recv())
+        if response["type"] == "error":
+            print(f"Session update rejected: {response['error']}")
+            return
+        if segment_timestamps and not response.get("timestamp_granularities"):
+            print("Server did not enable segment timestamps.")
+            return
 
         # Signal ready to start
         await ws.send(json.dumps({"type": "input_audio_buffer.commit"}))
@@ -99,6 +118,10 @@ async def realtime_transcribe(audio_path: str, host: str, port: int, model: str)
                 print(response["delta"], end="", flush=True)
             elif response["type"] == "transcription.done":
                 print(f"\n\nFinal transcription: {response['text']}")
+                for segment in response.get("segments") or []:
+                    # End times only: the model marks where a group of audio
+                    # ends, not where it starts.
+                    print(f"  ends at {segment['end']:>7.2f}s  {segment['text']!r}")
                 if response.get("usage"):
                     print(f"Usage: {response['usage']}")
                 break
@@ -115,7 +138,11 @@ def main(args):
         audio_path = str(AudioAsset("mary_had_lamb").get_local_path())
         print(f"No audio path provided, using default: {audio_path}")
 
-    asyncio.run(realtime_transcribe(audio_path, args.host, args.port, args.model))
+    asyncio.run(
+        realtime_transcribe(
+            audio_path, args.host, args.port, args.model, args.segment_timestamps
+        )
+    )
 
 
 if __name__ == "__main__":
@@ -134,6 +161,11 @@ def main(args):
         default=None,
         help="Path to the audio file to transcribe.",
     )
+    parser.add_argument(
+        "--segment-timestamps",
+        action="store_true",
+        help="Ask the server for the end time of each transcribed segment.",
+    )
     parser.add_argument(
         "--host",
         type=str,

tests/entrypoints/speech_to_text/realtime/test_realtime_validation.py

@@ -3,7 +3,6 @@
 
 import asyncio
 import json
-import warnings
 
 import numpy as np
 import pybase64 as base64
@@ -70,6 +69,138 @@ def mary_had_lamb_audio_chunks() -> list[str]:
     return chunks
 
 
+async def _start_session(
+    ws, model_name: str, timestamp_granularities: list[str] | None = None
+) -> dict:
+    """Open a session and return the session.updated acknowledgement."""
+    event = await receive_event(ws, timeout=30.0)
+    assert event["type"] == "session.created"
+
+    update: dict = {"type": "session.update", "model": model_name}
+    if timestamp_granularities is not None:
+        update["timestamp_granularities"] = timestamp_granularities
+    await send_event(ws, update)
+
+    event = await receive_event(ws, timeout=10.0)
+    assert event["type"] == "session.updated"
+    return event
+
+
+async def _stream_utterance(ws, chunks: list[str], timeout: float = 60.0) -> list[dict]:
+    """Stream one utterance and return every event up to transcription.done."""
+    await send_event(ws, {"type": "input_audio_buffer.commit"})
+    for chunk in chunks:
+        await send_event(ws, {"type": "input_audio_buffer.append", "audio": chunk})
+    await send_event(ws, {"type": "input_audio_buffer.commit", "final": True})
+
+    events = []
+    while True:
+        event = await receive_event(ws, timeout=timeout)
+        if event["type"] == "error":
+            pytest.fail(f"Received error: {event}")
+        events.append(event)
+        if event["type"] == "transcription.done":
+            return events
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("model_name", [MODEL_NAME])
+async def test_segment_timestamps(
+    model_name, mary_had_lamb_audio_chunks, rocm_aiter_fa_attention
+):
+    """Segment timestamps are opt-in, aligned, and invisible when not asked for.
+
+    The alignment bound is the point of this test: the token index of a
+    generated token leads the audio it describes by the streaming prefix
+    minus the left pad minus the transcription delay. Subtracting only the
+    delay - as https://github.com/vllm-project/vllm/issues/39735 proposes -
+    leaves the 32-frame left pad in and puts every timestamp 2.56 s late,
+    which the upper bound catches.
+    """
+    server_args = ["--enforce-eager", "--max-model-len", "2048"]
+
+    if model_name.startswith("mistralai"):
+        server_args += MISTRAL_FORMAT_ARGS
+
+    add_attention_backend(server_args, rocm_aiter_fa_attention)
+
+    chunk_duration_s = 1600 / 16000
+    duration_s = len(mary_had_lamb_audio_chunks) * chunk_duration_s
+
+    with RemoteOpenAIServer(
+        model_name, server_args, env_dict=REALTIME_ENV_OVERRIDES
+    ) as remote_server:
+        ws_url = _get_websocket_url(remote_server)
+
+        # --- Not opted in: the wire must be byte-identical to before ------
+        async with websockets.connect(ws_url) as ws:
+            ack = await _start_session(ws, model_name)
+            assert ack["timestamp_granularities"] == []
+
+            # (ROCm) generous timeout: first use triggers aiter JIT.
+            events = await _stream_utterance(
+                ws, mary_had_lamb_audio_chunks, timeout=600.0
+            )
+            done = events[-1]
+            assert set(done) == {"type", "text", "usage"}
+            assert all("segments" not in event for event in events)
+            baseline_text = done["text"]
+
+        # --- Opted in: same text, plus timestamps -------------------------
+        async with websockets.connect(ws_url) as ws:
+            ack = await _start_session(ws, model_name, ["segment"])
+            assert ack["timestamp_granularities"] == ["segment"]
+
+            events = await _stream_utterance(ws, mary_had_lamb_audio_chunks)
+            done = events[-1]
+            segments = done["segments"]
+
+            # Opting in must not change what was transcribed.
+            assert done["text"] == baseline_text
+            assert segments
+
+            # done repeats the deltas' segments, plus the trailing segment
+            # that generation ended before any delta could carry.
+            streamed = [
+                segment
+                for event in events
+                if event["type"] == "transcription.delta"
+                for segment in event["segments"]
+            ]
+            assert segments[: len(streamed)] == streamed
+            assert 0 <= len(segments) - len(streamed) <= 1
+
+            ends = [segment["end"] for segment in segments]
+            assert ends == sorted(ends)
+            assert all(end >= 0.08 for end in ends)
+            assert all(abs(end / 0.08 - round(end / 0.08)) < 1e-6 for end in ends)
+
+            # Bounded on both sides: +32 frames of left pad would overshoot,
+            # a negative offset would undershoot.
+            assert 0.5 * duration_s <= ends[-1] <= duration_s + 0.5
+
+            # Entries are emission groups, so there are at most as many as
+            # there are words, and together they reconstruct the transcript.
+            assert len(segments) <= len(baseline_text.split())
+            reconstructed = "".join(segment["text"] for segment in segments)
+            assert baseline_text.endswith(reconstructed)
+            assert len(reconstructed) >= 0.9 * len(baseline_text)
+
+        # --- The clock restarts on every utterance ------------------------
+        async with websockets.connect(ws_url) as ws:
+            await _start_session(ws, model_name, ["segment"])
+            short_chunks = mary_had_lamb_audio_chunks[:40]
+
+            first = await _stream_utterance(ws, short_chunks)
+            second = await _stream_utterance(ws, short_chunks)
+
+            assert first[-1]["segments"]
+            assert second[-1]["segments"]
+            # Not "greater than the first utterance's last end": each commit
+            # is a new engine request with a fresh prompt and left pad.
+            assert second[-1]["segments"][0]["end"] < 1.0
+
+
 @pytest.mark.asyncio
 @pytest.mark.parametrize("model_name", [MODEL_NAME])
 async def test_multi_chunk_streaming(
@@ -95,17 +226,8 @@ async def test_multi_chunk_streaming(
             await send_event(ws, {"type": "session.update", "model": model_name})
 
             # Wait for the server to acknowledge the session update.
-            try:
-                while True:
-                    event = await receive_event(ws, timeout=5.0)
-                    if event["type"] == "session.updated":
-                        break
-            except TimeoutError:
-                warnings.warn(
-                    f"session.updated not received within {5.0}s after "
-                    "session.update. The server may not implement this event.",
-                    stacklevel=2,
-                )
+            event = await receive_event(ws, timeout=10.0)
+            assert event["type"] == "session.updated"
 
             # (ROCm) Warm-up: send a non-final commit (required to start
             # transcription) with a small audio chunk to trigger aiter
@@ -203,17 +325,8 @@ async def test_empty_commit_does_not_crash_engine(
 
             await send_event(ws, {"type": "session.update", "model": model_name})
 
-            try:
-                while True:
-                    event = await receive_event(ws, timeout=5.0)
-                    if event["type"] == "session.updated":
-                        break
-            except TimeoutError:
-                warnings.warn(
-                    f"session.updated not received within {5.0}s after "
-                    "session.update. The server may not implement this event.",
-                    stacklevel=2,
-                )
+            event = await receive_event(ws, timeout=10.0)
+            assert event["type"] == "session.updated"
 
             # Start generation without sending any audio
             await send_event(ws, {"type": "input_audio_buffer.commit"})
@@ -239,17 +352,8 @@ async def test_empty_commit_does_not_crash_engine(
 
             await send_event(ws, {"type": "session.update", "model": model_name})
 
-            try:
-                while True:
-                    event = await receive_event(ws, timeout=5.0)
-                    if event["type"] == "session.updated":
-                        break
-            except TimeoutError:
-                warnings.warn(
-                    f"session.updated not received within {5.0}s after "
-                    "session.update. The server may not implement this event.",
-                    stacklevel=2,
-                )
+            event = await receive_event(ws, timeout=10.0)
+            assert event["type"] == "session.updated"
 
             # Start transcription
             await send_event(ws, {"type": "input_audio_buffer.commit"})