Motivation

Implements M1 of the RFC in #22474.

PR #22089 shipped chunked streaming output for Qwen3-ASR via POST /v1/audio/transcriptions?stream=true (SSE over an HTTP upload), which assumes the entire audio file is known up-front. Real-time use cases (live captioning, voice assistants, meeting transcription) need the opposite direction: the server accepts audio as it arrives and pushes partial transcripts back as the speaker talks.

Modifications

New WebSocket endpoint

• Endpoint: WS /v1/audio/transcriptions/stream (registered in http_server.py)
• Wire protocol (inspired by OpenAI Realtime API conventions, simplified session/delta/final event model):
  • Client → session.start (JSON) → binary PCM16 frames → session.end (JSON)
  • Server → session.started / transcript.delta (per word) / transcript.final / error
• Audio format (M1): fixed protocol — PCM16, 16 kHz, mono, little-endian. session.start accepts optional model and language fields; any other format will be rejected at PCM-frame validation (invalid_audio_format if frame byte length is not a multiple of 2, i.e. truncated int16). A future milestone can add an audio_format negotiation field without breaking the existing wire format.
• Backpressure: --asr-max-buffer-seconds CLI flag (default 60s, validated > 0 at startup). If accumulated server-side audio exceeds the cap, the server sends a buffer_overflow error and closes the socket. Below the cap, the single-task coroutine alternates receive and inference, so while a chunk is inferring the client experiences standard TCP-level backpressure (ws.send blocks on a full socket buffer). No silent drop.
• Concurrency model: single-task — one coroutine alternates receive + inference; session.end is therefore serialized after any in-flight chunk.

Note on HTTP SSE path: switching get_prefix_text() from confirmed_text to emitted_text (see streaming_asr.py row below) also incidentally fixes a latent prompt-prefix continuation issue in the HTTP SSE path from #22089, where confirmed_text could roll back mid-sentence and cause the model to re-emit from scratch on long English audio. Regression covered by the existing 8 HTTP tests.

Architecture

The WS transport uses an OO RealtimeConnection class (mirroring vLLM's RealtimeConnection in vllm/entrypoints/openai/realtime/connection.py) — one instance per WebSocket connection, holding session state and exposing lifecycle methods. Wire messages are typed Pydantic models so schema validation, type-checking, and JSON serialization flow from a single source of truth.

The HTTP SSE and WebSocket paths both route through a shared inference driver, keeping streaming state in StreamingASRState at the adapter layer rather than lifting it into the transport layer.

flowchart TB
    HTTP["HTTP<br/>(file upload)"]
    WS["WebSocket<br/>(live PCM frames)"]

    subgraph serving["OpenAIServingTranscription"]
        SSE["_generate_chunked_asr_stream"]
        WSH["handle_websocket<br/>(delegator)"]
    end

    PAC["process_asr_chunk<br/>(shared inference driver)"]

    subgraph state["StreamingASRState"]
        CT["confirmed_text<br/>chunk-local rollback, delta diff basis"]
        ET["emitted_text<br/>monotonic accumulator, prompt prefix source"]
        UF["update() / finalize()"]
    end

    HTTP --> SSE
    WS --> WSH
    SSE --> PAC
    WSH --> PAC
    PAC --> state

WebSocket session lifecycle

Inside serving_transcription_websocket.py, one coroutine alternates receive and inference. Each PCM batch of chunk_size_bytes triggers exactly one inference pass through the shared driver.

sequenceDiagram
    autonumber
    participant C as Client
    participant H as RealtimeConnection
    participant I as process_asr_chunk
    participant S as StreamingASRState

    C->>H: session.start (JSON)
    H->>H: _init<br/>accept + adapter capability check
    H-->>C: session.started

    loop per chunk_size_bytes of new audio
        C->>H: binary PCM16 frames
        H->>H: _on_audio_frame<br/>accumulate into pcm_buffer
        H->>I: _run_inference(_pcm_to_wav(buffer))
        I->>S: update() → delta
        I-->>H: delta str
        H-->>C: transcript.delta (per word)
    end

    C->>H: session.end (JSON)
    H->>I: _run_inference(is_last=True)
    I->>S: finalize() → tail
    I-->>H: tail delta
    H-->>C: transcript.final
    H->>C: close socket

Files touched

Accuracy Tests

Manual end-to-end verification against HTTP non-streaming (one-shot) ground truth across 7 audio fixtures × 3 paths (HTTP JSON / HTTP SSE / WebSocket), all captured against a single live sglang.launch_server running Qwen/Qwen3-ASR-0.6B on a GH200:

Cross-path consistency: on every fixture the HTTP-SSE output and the WebSocket output are byte-for-byte identical — you can read the SSE and WS columns and see this for yourself. This is the observable consequence of the two transports routing through the same process_asr_chunk driver over the same StreamingASRState; the transport layer introduces no additional drift beyond what chunked inference itself produces vs. one-shot.

WER threshold rationale: WS assertions use WER ≤ 0.15. This tolerates the chunked-inference boundary artifacts inherited from #22089 (e.g. the Uh huh. prefix on the EN clip) while still catching real regressions — the maximum observed WER across these 7 fixtures is 0.054, so the 0.15 threshold leaves ~3× headroom for flakiness without hiding regressions.

5 consecutive stability runs of the 19-test suite passed (all 19 tests green each run), with byte-identical WS transcripts across runs on a GH200. Median wall-clock 52.3s per run once the model weights are cached.

Streaming verification: on the 15s EN audio in realtime-pacing mode (client sends PCM at 0.5s/frame = wall-clock rate), ~30 out of 37 transcript.delta events arrive at the client before the client sends session.end, confirming true incremental server push (not batch-at-end).

M1 completion checklist (from RFC)

• ✅ WebSocket endpoint /v1/audio/transcriptions/stream
• ✅ Protocol: session.start / binary PCM / session.end ⇄ session.started / transcript.delta / transcript.final / error
• ✅ 8 error codes: invalid_json, invalid_payload, invalid_state, invalid_audio_format, unknown_message, buffer_overflow, unsupported_model, internal_error
• ✅ Single-task concurrency, --asr-max-buffer-seconds backpressure
• ✅ No scheduler or engine changes
• ✅ Reuse StreamingASRState, process_asr_chunk, TranscriptionAdapter
• ⬜ M2 (cross-chunk RadixCache prefill reuse): out of scope for this PR
• ⬜ M3 (token-level streaming within chunks via GenerateReqInput.stream=True, forced-alignment timestamps): out of scope

Known limitations

Listed here so reviewers don't have to re-discover them:

Test plan

Tests live under test/manual/ because they require downloading the Qwen/Qwen3-ASR-0.6B checkpoint (~1.2GB) and a GPU. They are runnable locally with a single command and complete in ~52s on one H100.

Automated (single command)

cd /path/to/sglang
python test/manual/models/test_qwen3_asr.py

The file uses popen_launch_server to spin up its own sglang.launch_server, runs 19 tests, and tears the server down.

Test breakdown:

WS assertions use _assert_close_to_ref(WER ≤ 0.15) against EXPECTED_TRANSCRIPTS (a dict of canonical transcripts captured from one-shot non-streaming inference). _wer normalizes case/punctuation and falls back to character-level comparison for CJK.

Manual (single audio, step-by-step)

Launch the server:

python -m sglang.launch_server \
  --model-path Qwen/Qwen3-ASR-0.6B \
  --served-model-name qwen3-asr \
  --trust-remote-code \
  --host 127.0.0.1 --port 30000

HTTP non-streaming (baseline / ground truth):

curl -s http://127.0.0.1:30000/v1/audio/transcriptions \
  -F file=@audio.wav -F model=qwen3-asr | jq -r .text

HTTP SSE streaming:

curl -Ns http://127.0.0.1:30000/v1/audio/transcriptions \
  -F file=@audio.wav -F model=qwen3-asr -F stream=true

WebSocket streaming — save as wsasr.py and run python wsasr.py audio.wav [language]. Sends PCM in 0.5s frames at wall-clock realtime pacing while concurrently reading transcript.delta events; each line is prefixed with t=<sec> from session start so you can see deltas arriving while audio is still uploading.

import asyncio
import json
import sys
import time

import numpy as np
import soundfile as sf
import websockets


async def main(path, lang=None):
    data, sr = sf.read(path, dtype="float32")
    if data.ndim > 1:
        data = data.mean(axis=1)
    if sr != 16000:
        n = int(len(data) / sr * 16000)
        data = np.interp(np.linspace(0, len(data) - 1, n), np.arange(len(data)), data)
        sr = 16000
    pcm = (data * 32767).astype(np.int16).tobytes()

    url = "ws://127.0.0.1:30000/v1/audio/transcriptions/stream"
    async with websockets.connect(url) as ws:
        start = {"type": "session.start"}
        if lang:
            start["language"] = lang

        t0 = time.perf_counter()
        await ws.send(json.dumps(start))

        def stamp():
            return time.perf_counter() - t0

        ack_raw = await ws.recv()
        ack = json.loads(ack_raw)
        print(f"[+{stamp():5.2f}] << {ack_raw}")
        if ack.get("type") != "session.started":
            print(f"[+{stamp():5.2f}] !! expected session.started, got {ack.get('type')}; aborting")
            return

        async def receive_loop():
            try:
                async for raw in ws:
                    print(f"[+{stamp():5.2f}] << {raw}")
                    msg = json.loads(raw)
                    if msg["type"] == "transcript.final":
                        return
                    if msg["type"] == "error":
                        print(f"[+{stamp():5.2f}] !! server error, closing")
                        return
            except websockets.ConnectionClosed as e:
                print(f"[+{stamp():5.2f}] !! connection closed: {e}")

        receiver = asyncio.create_task(receive_loop())

        chunk = int(0.5 * sr) * 2  # 0.5s of int16 PCM
        try:
            for i in range(0, len(pcm), chunk):
                if receiver.done():
                    print(f"[+{stamp():5.2f}] !! receiver ended early, stopping send")
                    break
                await ws.send(pcm[i : i + chunk])
                await asyncio.sleep(0.5)  # realtime pacing; drop for fast mode

            if not receiver.done():
                print(f"[+{stamp():5.2f}] >> session.end")
                await ws.send(json.dumps({"type": "session.end"}))
        except websockets.ConnectionClosed as e:
            print(f"[+{stamp():5.2f}] !! send failed, connection closed: {e}")

        await receiver


if __name__ == "__main__":
    asyncio.run(main(sys.argv[1], sys.argv[2] if len(sys.argv) > 2 else None))

Error-path verification

Manually tested all 8 error codes against a running server: invalid_json, invalid_payload, invalid_state × 3 variants, invalid_audio_format, unknown_message, buffer_overflow, unsupported_model, internal_error. All return the documented error event and close the socket.

Speed Tests and Profiling

No impact on inference speed — this PR is a thin WebSocket transport layer on top of unchanged chunked inference. Per-chunk latency is bound by the existing chunk_size_sec (2s) + model inference time (~0.5–1.5s on H100 for Qwen3-ASR-0.6B). No new CUDA kernels, no new memory patterns, no scheduler changes.

Checklist

• Format your code according to the Format code with pre-commit (black, isort, ruff, codespell all passing)
• Add unit tests according to the Run and add unit tests (19 tests in test/manual/models/test_qwen3_asr.py)
• Documentation: wire protocol documented in the Modifications section above
• Provide accuracy benchmark results — see "Accuracy Tests" section above
• Follow the SGLang code style guidance

Review and Merge Process

1. Ping Merge Oncalls to start the process.
2. Get approvals from CODEOWNERS and other reviewers.
3. Trigger CI tests with comments.
4. After green CI and required approvals, merge.

Related

• RFC: [RFC]: Real-Time Streaming Audio Input for ASR Models #22474
• Parent issue: [model] support qwen3-asr #22025
• Prerequisite: PR [Feature] Add chunk-based streaming ASR for Qwen3-ASR #22089 (chunked streaming output, merged)
• Prerequisite: PR [refactor] [asr] Add transcription adapter for extensible ASR models support #22181 (TranscriptionAdapter refactor, merged)
• Related: PR [Feature] Adding Qwen3-asr Model Support #22073 (Qwen3-ASR model support, merged)