diff --git a/python/sglang/srt/entrypoints/http_server.py b/python/sglang/srt/entrypoints/http_server.py index 759e6a832..9c76ae2cc 100644 --- a/python/sglang/srt/entrypoints/http_server.py +++ b/python/sglang/srt/entrypoints/http_server.py @@ -54,6 +54,7 @@ from fastapi import ( Query, Request, UploadFile, + WebSocket, ) from fastapi.exceptions import RequestValidationError from fastapi.middleware.cors import CORSMiddleware @@ -1605,6 +1606,17 @@ async def openai_v1_audio_transcriptions( ) +@app.websocket("/v1/realtime") +async def openai_v1_realtime_transcription(ws: WebSocket): + """OpenAI Realtime transcription WebSocket endpoint.""" + # /v1/realtime is OpenAI's unified Realtime URL covering transcription + + # chat modes. This handler implements the transcription subset only; + # chat-mode session.update payloads are rejected by the + # `Literal["transcription"]` constraint on TranscriptionSessionConfig.type + # (see realtime/protocol.py). + await ws.app.state.openai_serving_transcription.handle_websocket(ws) + + @app.get("/v1/models", response_class=ORJSONResponse) async def available_models(): """Show available models. OpenAI-compatible endpoint.""" diff --git a/python/sglang/srt/entrypoints/openai/realtime/__init__.py b/python/sglang/srt/entrypoints/openai/realtime/__init__.py new file mode 100644 index 000000000..33607defd --- /dev/null +++ b/python/sglang/srt/entrypoints/openai/realtime/__init__.py @@ -0,0 +1,5 @@ +"""Realtime transcription WebSocket package. Exposes the FastAPI WS entry.""" + +from sglang.srt.entrypoints.openai.realtime.handler import ( + handle_realtime_transcription as handle_realtime_transcription, +) diff --git a/python/sglang/srt/entrypoints/openai/realtime/handler.py b/python/sglang/srt/entrypoints/openai/realtime/handler.py new file mode 100644 index 000000000..30768ccda --- /dev/null +++ b/python/sglang/srt/entrypoints/openai/realtime/handler.py @@ -0,0 +1,119 @@ +"""WebSocket entry for realtime transcription. + +Handles accept, concurrency, cleanup. Event loop is in session.py. +""" + +from __future__ import annotations + +import asyncio +import logging + +from fastapi import WebSocket, WebSocketDisconnect +from openai.types.realtime import RealtimeErrorEvent +from openai.types.realtime.realtime_error import RealtimeError + +from sglang.srt.entrypoints.openai.realtime.session import RealtimeConnection +from sglang.srt.entrypoints.openai.transcription_adapters.base import ( + TranscriptionAdapter, +) +from sglang.srt.managers.tokenizer_manager import TokenizerManager +from sglang.srt.server_args import ServerArgs +from sglang.srt.utils import random_uuid + +logger = logging.getLogger(__name__) + + +async def _safe_send(websocket: WebSocket, text: str) -> None: + try: + await websocket.send_text(text) + except (WebSocketDisconnect, RuntimeError) as e: + logger.debug("[realtime] send failed (peer gone): %s", e) + + +async def _safe_close(websocket: WebSocket) -> None: + try: + await websocket.close() + except (WebSocketDisconnect, RuntimeError) as e: + logger.debug("[realtime] close failed (already closed): %s", e) + + +async def _reject_before_session( + websocket: WebSocket, + code: str, + message: str, + *, + error_type: str = "invalid_request_error", +) -> None: + """Reject path that runs before acquiring the session semaphore, so + unsupported / over-capacity peers don't hold a session slot.""" + try: + await websocket.accept() + except (WebSocketDisconnect, RuntimeError) as e: + logger.debug("[realtime] reject: accept failed: %s", e) + return + logger.info("[realtime] rejected (%s)", code) + envelope = RealtimeErrorEvent( + event_id=f"event_{random_uuid()}", + type="error", + error=RealtimeError(type=error_type, code=code, message=message), + ) + await _safe_send(websocket, envelope.model_dump_json()) + await _safe_close(websocket) + + +async def handle_realtime_transcription( + websocket: WebSocket, + tokenizer_manager: TokenizerManager, + adapter: TranscriptionAdapter, + server_args: ServerArgs, + session_semaphore: asyncio.Semaphore, +) -> None: + """WS endpoint for /v1/realtime. Pre-session validation runs before + the semaphore so rejects don't consume a session slot; the + ``async with`` then guarantees the slot is released even if + RealtimeConnection raises.""" + if not adapter.supports_chunked_streaming: + await _reject_before_session( + websocket, + "not_supported", + "Model does not support streaming ASR", + ) + return + + if session_semaphore.locked(): + await _reject_before_session( + websocket, + "too_many_sessions", + f"Maximum concurrent sessions reached " + f"({server_args.asr_max_concurrent_sessions}).", + error_type="rate_limit_exceeded", + ) + return + + async with session_semaphore: + try: + try: + await websocket.accept() + except (WebSocketDisconnect, RuntimeError) as e: + logger.debug("[realtime] accept failed: %s", e) + return + connection = RealtimeConnection( + websocket, tokenizer_manager, adapter, server_args + ) + await connection.run() + except WebSocketDisconnect: + logger.info("[realtime] client disconnected (normal)") + except Exception: + logger.exception("[realtime] unexpected error in session") + envelope = RealtimeErrorEvent( + event_id=f"event_{random_uuid()}", + type="error", + error=RealtimeError( + type="server_error", + code="inference_failed", + message="Internal server error", + ), + ) + await _safe_send(websocket, envelope.model_dump_json()) + finally: + await _safe_close(websocket) diff --git a/python/sglang/srt/entrypoints/openai/realtime/protocol.py b/python/sglang/srt/entrypoints/openai/realtime/protocol.py new file mode 100644 index 000000000..08a76cac3 --- /dev/null +++ b/python/sglang/srt/entrypoints/openai/realtime/protocol.py @@ -0,0 +1,78 @@ +"""Wire schema for Realtime WS transcription sessions.""" + +from __future__ import annotations + +from typing import Literal, Optional, Union + +from openai.types.realtime import SessionUpdateEvent as _SessionUpdateEvent +from openai.types.realtime.audio_transcription import ( + AudioTranscription as _AudioTranscription, +) +from openai.types.realtime.realtime_audio_formats import AudioPCM as _AudioPCM +from openai.types.realtime.realtime_audio_formats import AudioPCMA as _AudioPCMA +from openai.types.realtime.realtime_audio_formats import AudioPCMU as _AudioPCMU +from openai.types.realtime.realtime_transcription_session_audio import ( + RealtimeTranscriptionSessionAudio as _AudioCfg, +) +from openai.types.realtime.realtime_transcription_session_audio_input import ( + RealtimeTranscriptionSessionAudioInput as _AudioInputCfg, +) +from openai.types.realtime.realtime_transcription_session_create_request import ( + RealtimeTranscriptionSessionCreateRequest as _SessionCfg, +) +from pydantic import Field +from typing_extensions import Annotated + +# Fallback rate when the client omits `audio.input.format.rate`. SDK pins +# `AudioPCM.rate` to Literal[24000], so this matches the only value the SDK +# accepts when the field is present. +DEFAULT_INPUT_SAMPLE_RATE = 24000 + +# Wire rates we accept on `audio.input.format.rate` and resample to +# `adapter.model_sample_rate` server-side. 24000 matches the SDK pin; +# 16000 and 48000 widen it to cover common ASR-client and consumer-audio +# rates. Add a value here only after verifying transcription quality. +SUPPORTED_INPUT_SAMPLE_RATES = (16000, 24000, 48000) + + +class AudioPCM(_AudioPCM): + type: Literal["audio/pcm"] = "audio/pcm" + rate: Optional[int] = None + + +class AudioPCMU(_AudioPCMU): + type: Literal["audio/pcmu"] = "audio/pcmu" + + +class AudioPCMA(_AudioPCMA): + type: Literal["audio/pcma"] = "audio/pcma" + + +AudioInputFormat = Annotated[ + Union[AudioPCM, AudioPCMU, AudioPCMA], + Field(discriminator="type"), +] + + +class AudioTranscription(_AudioTranscription): + # SDK pins model to Literal["whisper-1", "gpt-4o-*-transcribe", ...]; + # sglang serves arbitrary ASR models (Qwen3-ASR, etc.) and treats the + # client-supplied name as echo-only. + model: Optional[str] = None + + +class TranscriptionSessionAudioInput(_AudioInputCfg): + format: Optional[AudioInputFormat] = None + transcription: Optional[AudioTranscription] = None + + +class TranscriptionSessionAudio(_AudioCfg): + input: Optional[TranscriptionSessionAudioInput] = None + + +class TranscriptionSessionConfig(_SessionCfg): + audio: Optional[TranscriptionSessionAudio] = None + + +class SessionUpdateEvent(_SessionUpdateEvent): + session: TranscriptionSessionConfig diff --git a/python/sglang/srt/entrypoints/openai/realtime/session.py b/python/sglang/srt/entrypoints/openai/realtime/session.py new file mode 100644 index 000000000..c5951993e --- /dev/null +++ b/python/sglang/srt/entrypoints/openai/realtime/session.py @@ -0,0 +1,740 @@ +"""WebSocket session for realtime ASR. + +Pre-commit deltas reference the reserved current_item_id that the +subsequent input_audio_buffer.committed and conversation.item.created +events will announce — sglang-specific, deviates from OpenAI's +commit-only delta emission. +""" + +from __future__ import annotations + +import asyncio +import io +import json +import logging +import math +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional + +import numpy as np +import pybase64 +import soundfile as sf +from fastapi import WebSocket, WebSocketDisconnect +from openai.types.realtime import ( + ConversationItemCreatedEvent, + InputAudioBufferAppendEvent, + InputAudioBufferClearedEvent, + InputAudioBufferClearEvent, + InputAudioBufferCommitEvent, + InputAudioBufferCommittedEvent, + RealtimeErrorEvent, + SessionCreatedEvent, + SessionUpdatedEvent, +) +from openai.types.realtime.conversation_item_input_audio_transcription_completed_event import ( + ConversationItemInputAudioTranscriptionCompletedEvent, + UsageTranscriptTextUsageDuration, +) +from openai.types.realtime.conversation_item_input_audio_transcription_delta_event import ( + ConversationItemInputAudioTranscriptionDeltaEvent, +) +from openai.types.realtime.conversation_item_input_audio_transcription_failed_event import ( + ConversationItemInputAudioTranscriptionFailedEvent, +) +from openai.types.realtime.conversation_item_input_audio_transcription_failed_event import ( + Error as TranscriptionFailedError, +) +from openai.types.realtime.realtime_conversation_item_user_message import ( + Content as InputAudioContent, +) +from openai.types.realtime.realtime_conversation_item_user_message import ( + RealtimeConversationItemUserMessage, +) +from openai.types.realtime.realtime_error import RealtimeError +from pydantic import BaseModel, ValidationError + +from sglang.srt.entrypoints.openai.protocol import TranscriptionRequest +from sglang.srt.entrypoints.openai.realtime.protocol import ( + DEFAULT_INPUT_SAMPLE_RATE, + SUPPORTED_INPUT_SAMPLE_RATES, + AudioPCM, + SessionUpdateEvent, + TranscriptionSessionAudioInput, + TranscriptionSessionConfig, +) +from sglang.srt.entrypoints.openai.streaming_asr import ( + StreamingASRState, + needs_space, + normalize_whitespace, + process_asr_chunk, +) +from sglang.srt.entrypoints.openai.transcription_adapters.base import ( + TranscriptionAdapter, +) +from sglang.srt.managers.tokenizer_manager import TokenizerManager +from sglang.srt.server_args import ServerArgs +from sglang.srt.utils import random_uuid + +logger = logging.getLogger(__name__) + + +# PCM16: 16-bit samples → 2 bytes each. Used for frame-length validation +# and bytes/sec arithmetic against `np.frombuffer(..., dtype=np.int16)` below. +_SAMPLE_WIDTH = 2 + + +def _resample_to_target_rate(pcm: bytes, src_rate: int, target_rate: int) -> bytes: + if src_rate == target_rate or not pcm: + return pcm + import torch + import torchaudio + + samples = np.frombuffer(pcm, dtype=np.int16).astype(np.float32) / 32768.0 + audio = torch.from_numpy(samples).unsqueeze(0) + audio = torchaudio.functional.resample( + audio, orig_freq=src_rate, new_freq=target_rate + ) + samples = audio.squeeze(0).numpy() + # Clip to int16 range via 2^15 - 1 so a clipped 1.0 stays representable. + return (np.clip(samples, -1.0, 1.0) * 32767.0).astype(np.int16).tobytes() + + +def _pcm_to_wav(pcm: bytes, sample_rate: int) -> bytes: + samples = np.frombuffer(pcm, dtype=np.int16) + buf = io.BytesIO() + sf.write(buf, samples, sample_rate, format="WAV") + return buf.getvalue() + + +_CLIENT_EVENT_TYPES: Dict[str, type] = { + "session.update": SessionUpdateEvent, + "input_audio_buffer.append": InputAudioBufferAppendEvent, + "input_audio_buffer.commit": InputAudioBufferCommitEvent, + "input_audio_buffer.clear": InputAudioBufferClearEvent, +} + + +def _parse_client_event(raw: Dict[str, Any]) -> Optional[BaseModel]: + """Parse, returning None if type is unknown. Raises ValidationError on + a malformed payload of a known type.""" + cls = _CLIENT_EVENT_TYPES.get(raw.get("type")) + if cls is None: + return None + return cls.model_validate(raw) + + +@dataclass +class _SessionConfig: + """Session-level configuration negotiated via session.update: audio + input format, requested language, sampling params. Persists until + the session ends; ``configured`` gates audio-frame handling so the + server doesn't run inference on PCM sent before session.update.""" + + input_sample_rate: int = DEFAULT_INPUT_SAMPLE_RATE + language: Optional[str] = None + client_model: Optional[str] = None + sampling_params: Optional[Dict[str, Any]] = None + configured: bool = False + + +@dataclass +class _AudioState: + """Per-item audio state: PCM buffer accumulated from + input_audio_buffer.append, the chunked ASR rollback state, and the + static buffer-size limits set at __init__. pcm_buffer / state / + last_inference_offset reset on commit-roll and clear; the size limits + stay constant for the session's lifetime.""" + + max_buffer_bytes: int + chunk_size_bytes: int + state: StreamingASRState + pcm_buffer: bytearray = field(default_factory=bytearray) + last_inference_offset: int = 0 + + +@dataclass +class _ItemState: + """Per-item conversation-item ids and the wire-formatted deltas + emitted so far for the current item. current_item_id is reserved at + __init__ and only announced to the client by + input_audio_buffer.committed.""" + + current_item_id: str + previous_item_id: Optional[str] = None + emitted_deltas: List[str] = field(default_factory=list) + + +class RealtimeConnection: + """One realtime transcription session. Drives the WS receive loop, + dispatches typed client events to the matching _on_* handler, and + triggers chunked ASR inference at audio buffer thresholds.""" + + def __init__( + self, + websocket: WebSocket, + tokenizer_manager: TokenizerManager, + adapter: TranscriptionAdapter, + server_args: ServerArgs, + ) -> None: + self.websocket = websocket + self.tokenizer_manager = tokenizer_manager + self.adapter = adapter + self.server_args = server_args + + self.session_id = f"sess_{random_uuid()}" + self._current_client_event_id: Optional[str] = None + + self.model_sample_rate = adapter.model_sample_rate + self.bytes_per_second = self.model_sample_rate * _SAMPLE_WIDTH + self.max_buffer_seconds = server_args.asr_max_buffer_seconds + + self.config = _SessionConfig() + + state = StreamingASRState(**adapter.chunked_streaming_config) + chunk_size_bytes = int(state.chunk_size_sec * self.bytes_per_second) + if chunk_size_bytes <= 0: + raise RuntimeError( + f"adapter.chunked_streaming_config produced non-positive " + f"chunk_size_sec; got {state.chunk_size_sec!r}" + ) + self.audio = _AudioState( + max_buffer_bytes=self.max_buffer_seconds * self.bytes_per_second, + chunk_size_bytes=chunk_size_bytes, + state=state, + ) + + self.item = _ItemState(current_item_id=f"item_{random_uuid()}") + + async def run(self) -> None: + await self._send( + SessionCreatedEvent( + event_id=f"event_{random_uuid()}", + type="session.created", + session=self._build_session_info(), + ) + ) + + try: + await self._run_loop() + except WebSocketDisconnect: + logger.info("[realtime] client disconnected: %s", self.session_id) + except Exception: + logger.exception("[realtime] unexpected error: %s", self.session_id) + try: + await self._send_error( + "inference_failed", + "Internal server error", + error_type="server_error", + ) + except (WebSocketDisconnect, RuntimeError) as e: + logger.debug( + "[realtime] failed to notify client of unexpected error: %s", + e, + ) + + async def _run_loop(self) -> None: + """Receive-and-dispatch loop. Validation errors emit an error event + and continue; fatal append-path errors (buffer overflow, append-time + inference failure) close the WebSocket and terminate the loop. + """ + while True: + self._current_client_event_id = None + message = await self.websocket.receive() + if message["type"] == "websocket.disconnect": + return + + text = message.get("text") + if not text: + if message.get("bytes") is not None: + # OpenAI Realtime is base64 PCM in JSON; binary frames aren't supported. + await self._send_error( + "invalid_payload", + "Binary frames are not supported on /v1/realtime; " + "use input_audio_buffer.append with base64 audio.", + ) + continue + + try: + raw = json.loads(text) + except json.JSONDecodeError: + await self._send_error("invalid_payload", "Invalid JSON") + continue + if not isinstance(raw, dict): + await self._send_error( + "invalid_payload", "Top-level event must be a JSON object" + ) + continue + + self._current_client_event_id = raw.get("event_id") + try: + event = _parse_client_event(raw) + except ValidationError as e: + # Report first error only; matches OpenAI server behavior. + err = e.errors()[0] + loc = ".".join(str(x) for x in err["loc"]) + await self._send_error( + "invalid_value", + err.get("msg") or "Invalid payload", + param=loc or None, + ) + continue + if event is None: + await self._send_error( + "unknown_event", + f"Unknown event type: {raw.get('type')!r}", + ) + continue + terminate = await self._dispatch(event) + if terminate: + return + + async def _dispatch(self, event: BaseModel) -> bool: + """Returns True if the session should terminate.""" + if isinstance(event, InputAudioBufferAppendEvent): + return await self._on_input_audio_buffer_append(event) + if isinstance(event, SessionUpdateEvent): + await self._on_session_update(event) + elif isinstance(event, InputAudioBufferCommitEvent): + await self._on_input_audio_buffer_commit(event) + elif isinstance(event, InputAudioBufferClearEvent): + await self._on_input_audio_buffer_clear(event) + return False + + async def _on_session_update(self, event: SessionUpdateEvent) -> None: + cfg = event.session + + # Normalize audio to an empty input cfg if absent so downstream + # `audio.X is not None` reads as a business rule, not an existence check. + # transcription stays nullable so partial-update can detect whether + # the client sent the block. + audio = ( + cfg.audio.input if cfg.audio else None + ) or TranscriptionSessionAudioInput() + transcription = audio.transcription + + # Validate first, then mutate config only after the whole update is accepted. + if audio.turn_detection is not None: + await self._send_error( + "not_supported", + "Server-side VAD is not implemented; " + "set audio.input.turn_detection: null and commit explicitly.", + param="session.audio.input.turn_detection", + ) + return + if audio.noise_reduction is not None: + await self._send_error( + "not_supported", + "audio.input.noise_reduction is not supported; set to null.", + param="session.audio.input.noise_reduction", + ) + return + if transcription is not None and transcription.prompt is not None: + await self._send_error( + "not_supported", + "audio.input.transcription.prompt is not supported.", + param="session.audio.input.transcription.prompt", + ) + return + if ( + transcription is not None + and transcription.model + and transcription.model != self.server_args.served_model_name + ): + await self._send_error( + "not_supported", + f"Model {transcription.model!r} is not served by this endpoint " + f"(serving {self.server_args.served_model_name!r}); set " + f"transcription.model to null or to the server's model name.", + param="session.audio.input.transcription.model", + ) + return + + new_rate = self.config.input_sample_rate # default: keep current + fmt = audio.format + if fmt is not None: + if not isinstance(fmt, AudioPCM): + # G.711 (pcmu / pcma): not implemented. + await self._send_error( + "not_supported", + f"audio.input.format.type must be 'audio/pcm'; " + f"{fmt.type!r} is not implemented", + param="session.audio.input.format.type", + ) + return + if fmt.rate is not None and fmt.rate not in SUPPORTED_INPUT_SAMPLE_RATES: + await self._send_error( + "invalid_value", + f"audio.input.format.rate must be one of " + f"{SUPPORTED_INPUT_SAMPLE_RATES}, got {fmt.rate}", + param="session.audio.input.format.rate", + ) + return + new_rate = fmt.rate or DEFAULT_INPUT_SAMPLE_RATE + # Changing the rate mid-item would leave already-buffered PCM + # at the old rate mixed with new audio at the new rate, so + # require the client to commit or clear before switching. + if new_rate != self.config.input_sample_rate and self.audio.pcm_buffer: + await self._send_error( + "invalid_state", + "Cannot change audio.input.format.rate while audio is " + "buffered; commit or clear the current item first.", + param="session.audio.input.format.rate", + ) + return + + # Mutation pass — no early returns past this point. + self.config.input_sample_rate = new_rate + if transcription is not None: + self.config.client_model = transcription.model + self.config.language = transcription.language + self.config.sampling_params = self.adapter.build_sampling_params( + TranscriptionRequest(language=self.config.language) + ) + self.config.configured = True + + # Side effects: log + ack. + if cfg.include: + logger.info( + "[realtime] %s: include[] received but not implemented; ignoring: %s", + self.session_id, + cfg.include, + ) + if self.config.input_sample_rate != self.model_sample_rate: + logger.info( + "[realtime] %s configured: resample %d→%d (ratio %.2f), language=%s", + self.session_id, + self.config.input_sample_rate, + self.model_sample_rate, + self.config.input_sample_rate / self.model_sample_rate, + self.config.language, + ) + await self._send( + SessionUpdatedEvent( + event_id=f"event_{random_uuid()}", + type="session.updated", + session=self._build_session_info(), + ) + ) + + async def _on_input_audio_buffer_append( + self, event: InputAudioBufferAppendEvent + ) -> bool: + """Returns True if the session should terminate (buffer overflow or + append-time inference failure).""" + if not self.config.configured: + await self._send_error( + "invalid_state", "Send session.update before audio frames" + ) + return False + + # Empty audio is a no-op (heartbeat frames); skip b64decode. + if not event.audio: + return False + + try: + data = pybase64.b64decode(event.audio, validate=True) + except (ValueError, TypeError): + await self._send_error( + "invalid_audio", "audio field is not valid base64", param="audio" + ) + return False + + if len(data) % _SAMPLE_WIDTH != 0: + await self._send_error( + "invalid_audio_format", + f"PCM16 frame length must be a multiple of {_SAMPLE_WIDTH} bytes", + ) + return False + + # Estimate post-resample size before resampling so oversized frames fail early. + src_samples = len(data) // _SAMPLE_WIDTH + target_samples = math.ceil( + src_samples * self.model_sample_rate / self.config.input_sample_rate + ) + if ( + len(self.audio.pcm_buffer) + target_samples * _SAMPLE_WIDTH + > self.audio.max_buffer_bytes + ): + # Close 1009 ("message too big") so clients can distinguish + # session-resource exhaustion from a normal close. + await self._send_error_and_close( + "buffer_overflow", + f"Accumulated audio exceeded {self.max_buffer_seconds}s; " + f"client is sending faster than inference can keep up", + close_code=1009, + ) + return True + + if self.config.input_sample_rate != self.model_sample_rate: + data = await asyncio.to_thread( + _resample_to_target_rate, + data, + self.config.input_sample_rate, + self.model_sample_rate, + ) + self.audio.pcm_buffer.extend(data) + + new_audio_bytes = len(self.audio.pcm_buffer) - self.audio.last_inference_offset + if new_audio_bytes >= self.audio.chunk_size_bytes: + ok = await self._run_inference(is_last=False) + if not ok: + # WS already closed inside _run_inference. + return True + return False + + async def _on_input_audio_buffer_commit( + self, event: InputAudioBufferCommitEvent + ) -> None: + if not self.config.configured: + await self._send_error("invalid_state", "Send session.update before commit") + return + if not self.audio.pcm_buffer and not self.audio.state.full_transcript: + await self._send_error( + "invalid_state", "Cannot commit an empty audio buffer" + ) + return + + has_new_audio = len(self.audio.pcm_buffer) > self.audio.last_inference_offset + item_id = self.item.current_item_id + prev_item_id = self.item.previous_item_id + + partial_transcript = normalize_whitespace("".join(self.item.emitted_deltas)) + + await self._send( + InputAudioBufferCommittedEvent( + event_id=f"event_{random_uuid()}", + type="input_audio_buffer.committed", + item_id=item_id, + previous_item_id=prev_item_id, + ) + ) + await self._send( + ConversationItemCreatedEvent( + event_id=f"event_{random_uuid()}", + type="conversation.item.created", + previous_item_id=prev_item_id, + item=RealtimeConversationItemUserMessage( + id=item_id, + type="message", + role="user", + status="completed", + content=[ + InputAudioContent( + type="input_audio", transcript=partial_transcript + ) + ], + ), + ) + ) + + # Capture pcm duration before `_start_next_item()` runs: starting + # the next item clears pcm_buffer, so reading it after gives 0. + pcm_duration_seconds = len(self.audio.pcm_buffer) / self.bytes_per_second + + if has_new_audio: + ok = await self._run_inference(is_last=True) + if not ok: + # _run_inference already emitted transcription.failed and + # rolled the item; don't also emit completed. + return + elif self.audio.state.full_transcript: + # Audio length was exactly a chunk_size_bytes multiple. Flush + # the tail tokens update() held back. + tail = self.audio.state.finalize() + await self._emit_transcription_delta(tail) + + # Build from emitted_deltas, not state.full_transcript: prefix injection + # means the last chunk's full_transcript is only the continuation tail. + transcript = normalize_whitespace("".join(self.item.emitted_deltas)) + + await self._send( + ConversationItemInputAudioTranscriptionCompletedEvent( + event_id=f"event_{random_uuid()}", + type="conversation.item.input_audio_transcription.completed", + item_id=item_id, + content_index=0, + transcript=transcript, + usage=UsageTranscriptTextUsageDuration( + type="duration", seconds=pcm_duration_seconds + ), + ) + ) + + self._start_next_item() + + async def _on_input_audio_buffer_clear( + self, event: InputAudioBufferClearEvent + ) -> None: + # Reserve a fresh current_item_id so post-clear pre-commit deltas + # don't share an item_id with deltas the client already received + # for the abandoned audio. previous_item_id is NOT touched — the + # cleared item was never committed, so the prior-commit chain + # shouldn't include it. + self._reset_inference_state() + self.item.current_item_id = f"item_{random_uuid()}" + await self._send( + InputAudioBufferClearedEvent( + event_id=f"event_{random_uuid()}", type="input_audio_buffer.cleared" + ) + ) + + async def _run_inference(self, is_last: bool) -> bool: + """Run ASR on the current cumulative buffer. Returns False on failure: + commit-time emits transcription.failed and rolls the item; append-time + emits a generic error envelope and closes the WebSocket.""" + wav_data = await asyncio.to_thread( + _pcm_to_wav, bytes(self.audio.pcm_buffer), self.model_sample_rate + ) + try: + delta = await process_asr_chunk( + tokenizer_manager=self.tokenizer_manager, + adapter=self.adapter, + state=self.audio.state, + audio_data=wav_data, + sampling_params=self.config.sampling_params, + is_last=is_last, + ) + except Exception: + logger.exception( + "[realtime] inference failed: session=%s item=%s buffer_bytes=%d", + self.session_id, + self.item.current_item_id, + len(self.audio.pcm_buffer), + ) + if is_last: + # Commit-time failure: committed + created already emitted, + # so the item exists client-side and transcription.failed + # can reference it. Wire message is hardcoded "Transcription + # failed" — don't leak backend traces to the client; full + # error is in the logger.exception above. + await self._send( + ConversationItemInputAudioTranscriptionFailedEvent( + event_id=f"event_{random_uuid()}", + type="conversation.item.input_audio_transcription.failed", + item_id=self.item.current_item_id, + content_index=0, + error=TranscriptionFailedError( + type="server_error", + code="inference_failed", + message="Transcription failed", + ), + ) + ) + self._start_next_item() + else: + # Append-time failure: the item isn't visible client-side + # yet (committed/created fire at commit), so + # transcription.failed would reference a ghost id. + await self._send_error_and_close( + "inference_failed", + "Transcription failed", + close_code=1011, + ) + return False + + self.audio.last_inference_offset = len(self.audio.pcm_buffer) + await self._emit_transcription_delta(delta) + return True + + async def _emit_transcription_delta(self, delta: str) -> None: + """emitted_deltas stores wire-formatted text (with leading + boundary spaces baked in), so "".join(...) reconstructs the + cumulative transcript verbatim.""" + if not delta: + return + for word in delta.split(" "): + if not word: + continue + prev = self.item.emitted_deltas[-1] if self.item.emitted_deltas else "" + formatted = f" {word}" if needs_space(prev, word) else word + self.item.emitted_deltas.append(formatted) + await self._send( + ConversationItemInputAudioTranscriptionDeltaEvent( + event_id=f"event_{random_uuid()}", + type="conversation.item.input_audio_transcription.delta", + item_id=self.item.current_item_id, + content_index=0, + delta=formatted, + ) + ) + + def _start_next_item(self) -> None: + self.item.previous_item_id = self.item.current_item_id + self.item.current_item_id = f"item_{random_uuid()}" + self._reset_inference_state() + + def _reset_inference_state(self) -> None: + """Missing any of these resets leaks state across items.""" + self.audio.state = StreamingASRState(**self.adapter.chunked_streaming_config) + self.audio.pcm_buffer.clear() # in-place; reuses the buffer's allocation + self.item.emitted_deltas.clear() + self.audio.last_inference_offset = 0 + + def _build_session_info(self) -> TranscriptionSessionConfig: + # id / object aren't SDK fields; round-trip via extra='allow' so + # dumps emit them like the real server. + return TranscriptionSessionConfig.model_validate( + { + "type": "transcription", + "id": self.session_id, + "object": "realtime.transcription_session", + "audio": { + "input": { + "format": { + "type": "audio/pcm", + "rate": self.config.input_sample_rate, + }, + "transcription": { + "model": self.config.client_model, + "language": self.config.language, + }, + "noise_reduction": None, + "turn_detection": None, + } + }, + } + ) + + async def _send(self, event: BaseModel) -> None: + await self.websocket.send_text(event.model_dump_json()) + + async def _send_error( + self, + code: str, + message: str, + *, + error_type: str = "invalid_request_error", + param: Optional[str] = None, + ) -> None: + envelope = RealtimeErrorEvent( + event_id=f"event_{random_uuid()}", + type="error", + error=RealtimeError( + type=error_type, + code=code, + message=message, + param=param, + event_id=self._current_client_event_id, + ), + ) + await self.websocket.send_text(envelope.model_dump_json()) + + async def _send_error_and_close( + self, + code: str, + message: str, + *, + close_code: int, + error_type: str = "server_error", + ) -> None: + # Independent try-blocks: a failed send must not skip the close. + # We still need to release local starlette socket state even when + # the wire send doesn't reach the peer. + try: + await self._send_error(code, message, error_type=error_type) + except (WebSocketDisconnect, RuntimeError) as e: + logger.debug("[realtime] send error %s before close failed: %s", code, e) + try: + await self.websocket.close(code=close_code) + except (WebSocketDisconnect, RuntimeError) as e: + logger.debug("[realtime] close %d after %s failed: %s", close_code, code, e) diff --git a/python/sglang/srt/entrypoints/openai/serving_transcription.py b/python/sglang/srt/entrypoints/openai/serving_transcription.py index 7bdd98097..debb95b1b 100644 --- a/python/sglang/srt/entrypoints/openai/serving_transcription.py +++ b/python/sglang/srt/entrypoints/openai/serving_transcription.py @@ -29,7 +29,7 @@ import time import uuid from typing import TYPE_CHECKING, AsyncGenerator, List, Optional, Union -from fastapi import Request +from fastapi import Request, WebSocket from fastapi.responses import ORJSONResponse, Response, StreamingResponse from sglang.srt.entrypoints.openai.protocol import ( @@ -42,9 +42,14 @@ from sglang.srt.entrypoints.openai.protocol import ( TranscriptionUsage, TranscriptionVerboseResponse, ) +from sglang.srt.entrypoints.openai.realtime import ( + handle_realtime_transcription, +) from sglang.srt.entrypoints.openai.serving_base import OpenAIServingBase from sglang.srt.entrypoints.openai.streaming_asr import ( StreamingASRState, + needs_space, + process_asr_chunk, split_audio_chunks, ) from sglang.srt.entrypoints.openai.transcription_adapters import resolve_adapter @@ -65,6 +70,11 @@ class OpenAIServingTranscription(OpenAIServingBase): self._adapter = resolve_adapter( getattr(model_config.hf_config, "architectures", []) ) + # Cap concurrent /v1/realtime sessions. The Semaphore is bound to the + # event loop on first acquire (uvicorn's loop in normal serving). + self._session_semaphore = asyncio.Semaphore( + tokenizer_manager.server_args.asr_max_concurrent_sessions + ) def _request_id_prefix(self) -> str: return "trsc-" @@ -190,7 +200,7 @@ class OpenAIServingTranscription(OpenAIServingBase): # For fused auto-detect, parse_fused_output returns the scrubbed # user-visible text. On parse failure (FSM abort, truncation) it - # returns (None, None) and we fall back to a best-effort scrub — + # returns (None, None) and we fall back to strip_special_tokens — # the language stays unset rather than reporting a bogus detection. if getattr(request, "_fused_autodetect", False): lang, visible = self._adapter.parse_fused_output( @@ -375,13 +385,14 @@ class OpenAIServingTranscription(OpenAIServingBase): - Token-level streaming within chunks (stream=True) - Encoder window caching across chunks - Cross-chunk KV cache reuse - - WebSocket endpoint for real-time audio input """ created_time = int(time.time()) request_id = f"{self._request_id_prefix()}{uuid.uuid4().hex}" model = request.model state = StreamingASRState(**self._adapter.chunked_streaming_config) - first_word = True + # Track only the trailing char of the cumulative emit; `needs_space` + # uses prev[-1] / cur[0] so we don't need to keep the full buffer. + last_char = "" try: chunks = split_audio_chunks(request.audio_data, state.chunk_size_sec) @@ -391,49 +402,24 @@ class OpenAIServingTranscription(OpenAIServingBase): logger.info("[streaming_asr] client disconnected, stopping") break is_last = i == len(chunks) - 1 - prompt = self._adapter.prompt_template + state.get_prefix_text() - chunk_request = GenerateReqInput( - text=prompt, + delta = await process_asr_chunk( + tokenizer_manager=self.tokenizer_manager, + adapter=self._adapter, + state=state, audio_data=chunk_audio, sampling_params=adapted_request.sampling_params, - stream=False, - modalities=["audio"], + is_last=is_last, + raw_request=raw_request, routing_key=self.extract_routing_key(raw_request), ) - try: - ret = None - async for ret in self.tokenizer_manager.generate_request( - chunk_request, raw_request - ): - break - except asyncio.CancelledError: - raise - except ValueError as e: - logger.warning( - "[streaming_asr] chunk %d failed with ValueError: %s", i, e - ) - continue - - if ret is None: - logger.warning("[streaming_asr] empty response for chunk %d", i) - continue - - text = self._adapter.postprocess_text(ret.get("text", "")) - - if is_last: - state.full_transcript = text - delta = state.finalize() - else: - delta = state.update(text) - if delta: for word in delta.split(" "): if not word: continue - content = word if first_word else " " + word - first_word = False + content = f" {word}" if needs_space(last_char, word) else word + last_char = content[-1] chunk_resp = TranscriptionStreamResponse( id=request_id, created=created_time, @@ -469,3 +455,12 @@ class OpenAIServingTranscription(OpenAIServingBase): yield f"data: {error}\n\n" yield "data: [DONE]\n\n" + + async def handle_websocket(self, websocket: WebSocket) -> None: + await handle_realtime_transcription( + websocket, + tokenizer_manager=self.tokenizer_manager, + adapter=self._adapter, + server_args=self.tokenizer_manager.server_args, + session_semaphore=self._session_semaphore, + ) diff --git a/python/sglang/srt/entrypoints/openai/streaming_asr.py b/python/sglang/srt/entrypoints/openai/streaming_asr.py index 77a808b23..a347cc8f3 100644 --- a/python/sglang/srt/entrypoints/openai/streaming_asr.py +++ b/python/sglang/srt/entrypoints/openai/streaming_asr.py @@ -1,8 +1,26 @@ +import asyncio import io +import logging +import re from dataclasses import dataclass -from typing import List +from typing import Any, Dict, List, Optional import soundfile as sf +from fastapi import Request + +from sglang.srt.entrypoints.openai.transcription_adapters.base import ( + TranscriptionAdapter, +) +from sglang.srt.managers.io_struct import GenerateReqInput +from sglang.srt.managers.tokenizer_manager import TokenizerManager + +logger = logging.getLogger(__name__) + + +# Collapse whitespace before punctuation so batched-inference token +# boundary jitter (" ," vs ",") doesn't leak into deltas. Covers both +# ASCII punctuation and the CJK / fullwidth equivalents. +_PUNCT_WS_RE = re.compile(r"\s+([,.;:!?，。！？；：、])") @dataclass @@ -22,13 +40,23 @@ class StreamingASRState: unfixed_chunk_num: int unfixed_token_num: int confirmed_text: str = "" + # Monotonic accumulator; used as prompt prefix so the model sees a + # natural continuation point, not the rolled-back ``confirmed_text``. + emitted_text: str = "" full_transcript: str = "" chunk_index: int = 0 def get_prefix_text(self) -> str: - if self.chunk_index < self.unfixed_chunk_num or not self.confirmed_text: + if self.chunk_index < self.unfixed_chunk_num or not self.emitted_text: return "" - return self.confirmed_text + return self.emitted_text + + def _record_emit(self, delta: str) -> str: + if delta: + self.emitted_text = ( + f"{self.emitted_text} {delta}".strip() if self.emitted_text else delta + ) + return delta def update(self, new_transcript: str) -> str: old_confirmed = self.confirmed_text @@ -40,7 +68,7 @@ class StreamingASRState: self.full_transcript = new_transcript self.chunk_index += 1 if self.confirmed_text.startswith(old_confirmed): - return self.confirmed_text[len(old_confirmed) :].strip() + return self._record_emit(self.confirmed_text[len(old_confirmed) :].strip()) # Model revised earlier text, use word level common prefix to avoid # re-emitting already-sent content and cutting mid-word. old_words = old_confirmed.split() @@ -50,7 +78,7 @@ class StreamingASRState: if ow != nw: break common_count += 1 - return " ".join(new_words[common_count:]) + return self._record_emit(" ".join(new_words[common_count:])) def finalize(self) -> str: confirmed_words = self.confirmed_text.split() @@ -64,8 +92,8 @@ class StreamingASRState: common_count += 1 self.confirmed_text = self.full_transcript if common_count == 0 and confirmed_words and all_words: - return self.full_transcript - return " ".join(all_words[common_count:]) + return self._record_emit(self.full_transcript) + return self._record_emit(" ".join(all_words[common_count:])) def split_audio_chunks(audio_data: bytes, chunk_size_sec: float) -> List[bytes]: @@ -91,3 +119,91 @@ def split_audio_chunks(audio_data: bytes, chunk_size_sec: float) -> List[bytes]: sf.write(buf, data[:end], sample_rate, format="WAV") chunks.append(buf.getvalue()) return chunks + + +def normalize_whitespace(text: str) -> str: + return _PUNCT_WS_RE.sub(r"\1", text) + + +_NO_SPACE_BEFORE = frozenset(".,!?;:%)]}，。！？；：、）】》」』") +_NO_SPACE_AFTER = frozenset("([{（【《「『") + + +def _is_cjk(c: str) -> bool: + """Whether char is a CJK-context glyph that doesn't take inter-word + spaces — ideographs, Japanese kana, CJK punctuation, fullwidth forms. + Excludes Hangul / Devanagari / Arabic etc., which are non-ASCII but + space-separated and need the normal boundary space.""" + cp = ord(c) + return ( + 0x3000 <= cp <= 0x303F # CJK Symbols and Punctuation (，。、《》「」…) + or 0x3040 <= cp <= 0x309F # Hiragana + or 0x30A0 <= cp <= 0x30FF # Katakana + or 0x3400 <= cp <= 0x4DBF # CJK Unified Ideographs Ext A + or 0x4E00 <= cp <= 0x9FFF # CJK Unified Ideographs + or 0xFF00 <= cp <= 0xFFEF # Halfwidth & Fullwidth Forms (fullwidth ASCII) + ) + + +def needs_space(prev: str, cur: str) -> bool: + """Return whether a boundary space is needed between emitted deltas. + + Avoid spaces around punctuation and between adjacent CJK-context glyphs. + Shared by the realtime WS and HTTP SSE chunked streaming paths. + """ + if not prev or not cur: + return False + if prev[-1].isspace() or cur[0].isspace(): + return False + if cur[0] in _NO_SPACE_BEFORE or prev[-1] in _NO_SPACE_AFTER: + return False + if _is_cjk(prev[-1]) and _is_cjk(cur[0]): + return False + return True + + +async def process_asr_chunk( + tokenizer_manager: TokenizerManager, + adapter: TranscriptionAdapter, + state: StreamingASRState, + audio_data: bytes, + sampling_params: Dict[str, Any], + is_last: bool, + raw_request: Optional[Request] = None, + routing_key: Optional[str] = None, +) -> str: + """Run inference on one audio chunk. Shared by the HTTP and WebSocket paths.""" + prompt = adapter.prompt_template + state.get_prefix_text() + + chunk_request = GenerateReqInput( + text=prompt, + audio_data=audio_data, + sampling_params=sampling_params, + stream=False, + modalities=["audio"], + ) + if routing_key is not None: + chunk_request.routing_key = routing_key + + try: + ret = None + async for ret in tokenizer_manager.generate_request(chunk_request, raw_request): + break + except asyncio.CancelledError: + raise + except ValueError: + logger.warning( + "[streaming_asr] chunk %d failed", state.chunk_index, exc_info=True + ) + raise + + if ret is None: + logger.warning("[streaming_asr] empty response for chunk %d", state.chunk_index) + return "" + + text = normalize_whitespace(adapter.postprocess_text(ret.get("text", ""))) + + if is_last: + state.full_transcript = text + return state.finalize() + return state.update(text) diff --git a/python/sglang/srt/entrypoints/openai/transcription_adapters/base.py b/python/sglang/srt/entrypoints/openai/transcription_adapters/base.py index 4c3b7123d..cd97b4299 100644 --- a/python/sglang/srt/entrypoints/openai/transcription_adapters/base.py +++ b/python/sglang/srt/entrypoints/openai/transcription_adapters/base.py @@ -81,6 +81,14 @@ class TranscriptionAdapter(ABC): """Whether this model uses chunk-based streaming instead of token-level streaming.""" return False + @property + def model_sample_rate(self) -> int: + """Target sample rate in Hz the model expects. Realtime WS path + resamples client PCM to this rate before chunking. Default 16000 + matches Whisper / Qwen3-ASR; override for models expecting other rates. + """ + return 16000 + @property def prompt_template(self) -> str: """Prompt template for chunked streaming requests. diff --git a/python/sglang/srt/server_args.py b/python/sglang/srt/server_args.py index 86fbba470..eab4ee773 100644 --- a/python/sglang/srt/server_args.py +++ b/python/sglang/srt/server_args.py @@ -512,6 +512,8 @@ class ServerArgs: tool_call_parser: Optional[str] = None tool_server: Optional[str] = None sampling_defaults: str = "model" + asr_max_buffer_seconds: int = 60 + asr_max_concurrent_sessions: int = 32 # Data parallelism dp_size: int = 1 @@ -864,6 +866,8 @@ class ServerArgs: self._handle_multimodal() # Validate SSL arguments early (before dummy-model short-circuit). self._handle_ssl_validation() + # Validate transcription/ASR-specific server args (model-independent). + self._handle_asr_validation() # Validate PD disaggregation flags early (before dummy-model short-circuit). from sglang.srt.arg_groups.pd_disaggregation_hook import ( @@ -4143,6 +4147,19 @@ class ServerArgs: ) self.enable_mixed_chunk = False + def _handle_asr_validation(self): + """Validate transcription/ASR-specific server args.""" + if self.asr_max_buffer_seconds <= 0: + raise ValueError( + f"--asr-max-buffer-seconds must be positive " + f"(got {self.asr_max_buffer_seconds})." + ) + if self.asr_max_concurrent_sessions <= 0: + raise ValueError( + f"--asr-max-concurrent-sessions must be positive " + f"(got {self.asr_max_concurrent_sessions})." + ) + def _handle_other_validations(self): # Handle model inference tensor dump. if self.debug_tensor_dump_output_folder is not None: @@ -5192,6 +5209,24 @@ class ServerArgs: "'model' uses the model's generation_config.json to get the recommended " "sampling parameters if available. Default is 'model'.", ) + parser.add_argument( + "--asr-max-buffer-seconds", + type=int, + default=ServerArgs.asr_max_buffer_seconds, + help="Maximum seconds of PCM audio the streaming ASR WebSocket handler " + "will accumulate before closing the session with a buffer_overflow " + "error. Guards against OOM when a client streams audio faster than " + "inference can consume it. Default 60s.", + ) + parser.add_argument( + "--asr-max-concurrent-sessions", + type=int, + default=ServerArgs.asr_max_concurrent_sessions, + help="Maximum number of concurrent realtime ASR WebSocket sessions " + "served by /v1/realtime. New connections beyond this cap are " + "accepted, sent an error{code:too_many_sessions} frame, and closed. " + "Default 32.", + ) # Data parallelism parser.add_argument( diff --git a/test/manual/models/test_qwen3_asr.py b/test/manual/models/test_qwen3_asr.py index c0b772bf5..c2e7b4278 100644 --- a/test/manual/models/test_qwen3_asr.py +++ b/test/manual/models/test_qwen3_asr.py @@ -1,17 +1,31 @@ """ Test Qwen3-ASR model support in SGLang. -Tests /v1/audio/transcriptions endpoint (OpenAI-compatible). +Tests /v1/audio/transcriptions (HTTP) and /v1/realtime (OpenAI Realtime +transcription WebSocket). Usage: python test/manual/models/test_qwen3_asr.py """ +import asyncio +import base64 import io +import json import os +import re import unittest +import numpy as np import requests +import soundfile as sf + +try: + import websockets + + HAS_WEBSOCKETS = True +except ImportError: + HAS_WEBSOCKETS = False from sglang.srt.utils import kill_process_tree from sglang.test.test_utils import ( @@ -29,8 +43,83 @@ TEST_AUDIO_EN_URL = ( TEST_AUDIO_ZH_URL = ( "https://qianwen-res.oss-cn-beijing.aliyuncs.com/Qwen3-ASR-Repo/asr_zh.wav" ) +TEST_AUDIO_MLK_URL = ( + "https://huggingface.co/datasets/Narsil/asr_dummy/resolve/main/mlk.flac" +) +TEST_AUDIO_LIBRI_URL = ( + "https://huggingface.co/datasets/Narsil/asr_dummy/resolve/main/1.flac" +) +TEST_AUDIO_SPANISH_URL = ( + "https://huggingface.co/datasets/Narsil/asr_dummy/resolve/main/4.flac" +) +TEST_AUDIO_HINDI_URL = ( + "https://huggingface.co/datasets/Narsil/asr_dummy/resolve/main/hindi.ogg" +) +TEST_AUDIO_MP3_URL = ( + "https://huggingface.co/datasets/Narsil/asr_dummy/resolve/main/i-know-kung-fu.mp3" +) TEST_AUDIO_EN_LOCAL = "/tmp/test_qwen3_asr_en.wav" TEST_AUDIO_ZH_LOCAL = "/tmp/test_qwen3_asr_zh.wav" +TEST_AUDIO_MLK_LOCAL = "/tmp/test_qwen3_asr_mlk.flac" +TEST_AUDIO_LIBRI_LOCAL = "/tmp/test_qwen3_asr_libri.flac" +TEST_AUDIO_SPANISH_LOCAL = "/tmp/test_qwen3_asr_spanish.flac" +TEST_AUDIO_HINDI_LOCAL = "/tmp/test_qwen3_asr_hindi.ogg" +TEST_AUDIO_MP3_LOCAL = "/tmp/test_qwen3_asr_kungfu.mp3" + +# Captured from Qwen3-ASR-0.6B non-streaming inference (2026-04-14). +# Refresh if model weights or sampling params change. +EXPECTED_TRANSCRIPTS = { + "en": ( + "Oh yeah, yeah. He wasn't even that big when I started listening to him." + " But and his solo music didn't do overly well, but he did very well" + " when he started writing for other people." + ), + "zh": "甚至出现交易几乎停滞的情况。", + "mlk": ( + "I have a dream that one day this nation will rise up and live out" + " the true meaning of its creed." + ), + "libri": ( + "He hoped there would be stew for dinner—turnips and carrots and" + " bruised potatoes and fat mutton pieces—to be ladled out in thick" + " peppered flour-fatted sauce." + ), + "spanish": ( + "y en las ramas medio sumergidas revoloteaban algunos pájaros" + " de químico y legendario plumaje" + ), + "hindi": "मिर्ची में कितने विभिन्न प्रजातियाँ हैं", + "mp3": "I know kung fu.", +} + + +def _normalize_for_wer(text: str) -> list: + text = text.lower() + text = re.sub(r"[^\w\s\u0900-\u097f\u4e00-\u9fff]+", " ", text) + return text.split() + + +def _wer(hypothesis: str, reference: str) -> float: + hyp = _normalize_for_wer(hypothesis) + ref = _normalize_for_wer(reference) + if len(ref) <= 1 and not any(" " in w for w in ref): + # CJK fallback: str.split() degenerates, compare at char level. + hyp = list(hypothesis.replace(" ", "")) + ref = list(reference.replace(" ", "")) + if not ref: + return 0.0 if not hyp else float("inf") + n, m = len(hyp), len(ref) + dp = list(range(m + 1)) + for i in range(1, n + 1): + prev, dp[0] = dp[0], i + for j in range(1, m + 1): + cur = dp[j] + if hyp[i - 1] == ref[j - 1]: + dp[j] = prev + else: + dp[j] = 1 + min(prev, dp[j - 1], dp[j]) + prev = cur + return dp[m] / len(ref) def download_audio(url, local_path): @@ -45,8 +134,127 @@ def download_audio(url, local_path): return resp.content +def _pcm16_from_audio_bytes(audio_bytes, target_sr=16000): + data, sr = sf.read(io.BytesIO(audio_bytes), dtype="float32") + if len(data.shape) > 1: + data = data.mean(axis=1) + if sr != target_sr: + num_samples = int(len(data) / sr * target_sr) + indices = np.linspace(0, len(data) - 1, num_samples) + data = np.interp(indices, np.arange(len(data)), data) + sr = target_sr + pcm = (data * 32767).astype(np.int16).tobytes() + return pcm, sr + + +async def _stream_websocket_async( + websocket_url, pcm_bytes, sample_rate, language=None, realtime=False +): + chunk_duration = 0.5 + chunk_bytes = int(chunk_duration * sample_rate * 2) + duration_sec = round(len(pcm_bytes) / (sample_rate * 2), 2) + + async with websockets.connect(websocket_url) as websocket: + created = json.loads(await websocket.recv()) + assert ( + created.get("type") == "session.created" + ), f"expected session.created, got {created!r}" + session_id = created["session"]["id"] + + transcription_cfg = {"model": "qwen3-asr"} + if language: + transcription_cfg["language"] = language + await websocket.send( + json.dumps( + { + "type": "session.update", + "session": { + "type": "transcription", + "audio": { + "input": { + "format": {"type": "audio/pcm", "rate": sample_rate}, + "transcription": transcription_cfg, + "noise_reduction": None, + "turn_detection": None, + } + }, + }, + } + ) + ) + + while True: + evt = json.loads(await websocket.recv()) + if evt.get("type") == "session.updated": + break + if evt.get("type") == "error": + raise RuntimeError(f"websocket error during update: {evt!r}") + + deltas = [] + completed_msg = {} + + async def receive_loop(): + async for raw in websocket: + resp = json.loads(raw) + t = resp.get("type") + if t == "conversation.item.input_audio_transcription.delta": + deltas.append(resp["delta"]) + elif t == "conversation.item.input_audio_transcription.completed": + assert ( + "usage" in resp + ), f"transcription.completed missing required usage field: {resp!r}" + assert resp["usage"].get("type") == "duration", resp["usage"] + completed_msg.update(resp) + return + elif t in ( + "input_audio_buffer.committed", + "conversation.item.created", + ): + continue + elif t == "error": + err = resp.get("error", {}) + raise RuntimeError( + f"websocket error [{err.get('code', '?')}]: " + f"{err.get('message', '')}" + ) + elif t == "conversation.item.input_audio_transcription.failed": + raise RuntimeError(f"transcription failed: {resp!r}") + + receiver = asyncio.create_task(receive_loop()) + + for offset in range(0, len(pcm_bytes), chunk_bytes): + chunk = pcm_bytes[offset : offset + chunk_bytes] + await websocket.send( + json.dumps( + { + "type": "input_audio_buffer.append", + "audio": base64.b64encode(chunk).decode("ascii"), + } + ) + ) + if realtime: + await asyncio.sleep(chunk_duration) + + await websocket.send(json.dumps({"type": "input_audio_buffer.commit"})) + try: + await asyncio.wait_for(receiver, timeout=60) + except asyncio.TimeoutError as e: + raise RuntimeError( + f"timed out waiting for transcription.completed; " + f"got {len(deltas)} deltas, last={deltas[-1] if deltas else None!r}" + ) from e + + assert completed_msg, "no transcription.completed received" + return { + "text": completed_msg.get("transcript", ""), + "deltas": deltas, + "session_id": session_id, + "duration_sec": duration_sec, + } + + class TestQwen3ASRTranscription(CustomTestCase): - """Test Qwen3-ASR via /v1/audio/transcriptions endpoint.""" + """Test Qwen3-ASR via HTTP /v1/audio/transcriptions and OpenAI Realtime WebSocket /v1/realtime.""" @classmethod def setUpClass(cls): @@ -68,7 +276,7 @@ class TestQwen3ASRTranscription(CustomTestCase): kill_process_tree(cls.process.pid) def _transcribe(self, audio_url, local_path, language=None): - """Send a transcription request.""" + """Send an HTTP transcription request.""" audio_bytes = download_audio(audio_url, local_path) data = {"model": "qwen3-asr"} if language: @@ -113,6 +321,342 @@ class TestQwen3ASRTranscription(CustomTestCase): ) print(f"[Consistency] All 3 requests match: {results[0][:80]}...") + def _websocket_url(self): + return ( + self.base_url.replace("http://", "ws://").replace("https://", "wss://") + + "/v1/realtime" + ) + + def _stream_websocket( + self, + audio_url, + local_path, + language=None, + realtime=False, + target_sr=16000, + ): + audio_bytes = download_audio(audio_url, local_path) + pcm, sr = _pcm16_from_audio_bytes(audio_bytes, target_sr=target_sr) + return asyncio.run( + _stream_websocket_async( + self._websocket_url(), pcm, sr, language=language, realtime=realtime + ) + ) + + def _assert_close_to_ref( + self, hypothesis: str, ref_key: str, max_wer: float = 0.15 + ): + # 15% tolerates chunked-streaming artifacts without hiding regressions. + reference = EXPECTED_TRANSCRIPTS[ref_key] + wer = _wer(hypothesis, reference) + self.assertLessEqual( + wer, + max_wer, + f"WER {wer:.3f} > {max_wer} for {ref_key!r}\n" + f" hyp: {hypothesis!r}\n ref: {reference!r}", + ) + + @unittest.skipUnless(HAS_WEBSOCKETS, "websockets package not installed") + def test_english_websocket_streaming(self): + result = self._stream_websocket(TEST_AUDIO_EN_URL, TEST_AUDIO_EN_LOCAL) + self._assert_close_to_ref(result["text"], "en") + self.assertGreater(len(result["deltas"]), 0) + print( + f"[EN WS] final={result['text']} " + f"({len(result['deltas'])} deltas, {result['duration_sec']}s)" + ) + + @unittest.skipUnless(HAS_WEBSOCKETS, "websockets package not installed") + def test_chinese_websocket_streaming(self): + result = self._stream_websocket( + TEST_AUDIO_ZH_URL, TEST_AUDIO_ZH_LOCAL, language="zh" + ) + self._assert_close_to_ref(result["text"], "zh") + print( + f"[ZH WS] final={result['text']} " + f"({len(result['deltas'])} deltas, {result['duration_sec']}s)" + ) + + @unittest.skipUnless(HAS_WEBSOCKETS, "websockets package not installed") + def test_websocket_streaming_realtime(self): + # Pace appends at wall-clock so multiple deltas land before commit. + result = self._stream_websocket( + TEST_AUDIO_EN_URL, TEST_AUDIO_EN_LOCAL, realtime=True + ) + self._assert_close_to_ref(result["text"], "en") + self.assertGreaterEqual(len(result["deltas"]), 2, result["deltas"]) + print( + f"[Realtime WS] final={result['text']} " + f"({len(result['deltas'])} deltas, {result['duration_sec']}s)" + ) + + @unittest.skipUnless(HAS_WEBSOCKETS, "websockets package not installed") + def test_mlk_speech_websocket_streaming(self): + # FLAC 22050 Hz — exercises client-side resample to 16 kHz. + result = self._stream_websocket(TEST_AUDIO_MLK_URL, TEST_AUDIO_MLK_LOCAL) + self._assert_close_to_ref(result["text"], "mlk") + print( + f"[MLK WS] final={result['text']} " + f"({len(result['deltas'])} deltas, {result['duration_sec']}s)" + ) + + @unittest.skipUnless(HAS_WEBSOCKETS, "websockets package not installed") + def test_websocket_concurrent_sessions(self): + # Verify state isolation: 3 concurrent sessions on identical audio + # must yield identical finals + 3 distinct session ids. + audio_bytes = download_audio(TEST_AUDIO_EN_URL, TEST_AUDIO_EN_LOCAL) + pcm, sr = _pcm16_from_audio_bytes(audio_bytes) + + async def run_n_concurrent(n): + return await asyncio.gather( + *[ + _stream_websocket_async(self._websocket_url(), pcm, sr) + for _ in range(n) + ] + ) + + results = asyncio.run(run_n_concurrent(3)) + + session_ids = {r["session_id"] for r in results} + self.assertEqual(len(session_ids), 3) + for r in results: + self.assertTrue(len(r["text"]) > 0) + finals = [r["text"] for r in results] + self.assertEqual(len(set(finals)), 1, f"finals diverged: {finals}") + print( + f"[Concurrent x3 WS] all finals match: {finals[0]} " + f"(session_ids={sorted(session_ids)})" + ) + + @unittest.skipUnless(HAS_WEBSOCKETS, "websockets package not installed") + def test_spanish_websocket_streaming(self): + # FLAC 48 kHz PCM_24 — keep native rate so server-side resample runs. + result = self._stream_websocket( + TEST_AUDIO_SPANISH_URL, + TEST_AUDIO_SPANISH_LOCAL, + language="es", + target_sr=48000, + ) + self._assert_close_to_ref(result["text"], "spanish") + print( + f"[Spanish WS] final={result['text']} " + f"({len(result['deltas'])} deltas, {result['duration_sec']}s)" + ) + + @unittest.skipUnless(HAS_WEBSOCKETS, "websockets package not installed") + def test_websocket_short_clip(self): + # 3s clip exercises the mid-chunk tail flush at commit. + audio_bytes = download_audio(TEST_AUDIO_MP3_URL, TEST_AUDIO_MP3_LOCAL) + full_pcm, sr = _pcm16_from_audio_bytes(audio_bytes) + short_pcm = full_pcm[: sr * 2 * 3] + result = asyncio.run( + _stream_websocket_async(self._websocket_url(), short_pcm, sr) + ) + self._assert_close_to_ref(result["text"], "mp3") + print( + f"[Short clip WS] final={result['text']} " + f"({len(result['deltas'])} deltas, {result['duration_sec']}s)" + ) + + @unittest.skipUnless(HAS_WEBSOCKETS, "websockets package not installed") + def test_websocket_chunk_boundary_flush(self): + # Exact 4s = 2 × chunk_size_sec to hit the exact-boundary tail-flush + # path at commit. EN clip (not mp3) because mp3 starts with silence. + audio_bytes = download_audio(TEST_AUDIO_EN_URL, TEST_AUDIO_EN_LOCAL) + full_pcm, sr = _pcm16_from_audio_bytes(audio_bytes) + boundary_bytes = int(4.0 * sr * 2) # 2 × chunk_size_sec at 16 kHz int16 mono + boundary_pcm = full_pcm[:boundary_bytes] + assert len(boundary_pcm) == boundary_bytes, "audio shorter than 4s" + result = asyncio.run( + _stream_websocket_async(self._websocket_url(), boundary_pcm, sr) + ) + self.assertTrue(len(result["text"]) > 0, result) + print( + f"[Chunk boundary WS] final={result['text']} " + f"({len(result['deltas'])} deltas, {result['duration_sec']}s)" + ) + + @unittest.skipUnless(HAS_WEBSOCKETS, "websockets package not installed") + def test_websocket_rejects_unsupported_sample_rate(self): + async def run(): + async with websockets.connect(self._websocket_url()) as ws: + created = json.loads(await ws.recv()) + self.assertEqual(created.get("type"), "session.created", created) + await ws.send( + json.dumps( + { + "type": "session.update", + "session": { + "type": "transcription", + "audio": { + "input": { + "format": { + "type": "audio/pcm", + "rate": 22050, + }, + "transcription": {"model": "qwen3-asr"}, + "noise_reduction": None, + "turn_detection": None, + } + }, + }, + } + ) + ) + evt = json.loads(await ws.recv()) + self.assertEqual(evt.get("type"), "error", evt) + err = evt.get("error", {}) + self.assertEqual(err.get("code"), "invalid_value", err) + self.assertEqual( + err.get("param"), "session.audio.input.format.rate", err + ) + + asyncio.run(run()) + print("[Unsupported rate WS] 22050 rejected with invalid_value") + + @unittest.skipUnless(HAS_WEBSOCKETS, "websockets package not installed") + def test_websocket_rejects_non_dict_transcription(self): + # Use a valid nested format so Pydantic surfaces the transcription + # error rather than the format error first. + async def run(): + async with websockets.connect(self._websocket_url()) as ws: + created = json.loads(await ws.recv()) + self.assertEqual(created.get("type"), "session.created", created) + await ws.send( + json.dumps( + { + "type": "session.update", + "session": { + "type": "transcription", + "audio": { + "input": { + "format": { + "type": "audio/pcm", + "rate": 16000, + }, + "transcription": "qwen3-asr", + "noise_reduction": None, + "turn_detection": None, + } + }, + }, + } + ) + ) + evt = json.loads(await ws.recv()) + self.assertEqual(evt.get("type"), "error", evt) + err = evt.get("error", {}) + self.assertEqual(err.get("code"), "invalid_value", err) + self.assertEqual( + err.get("param"), "session.audio.input.transcription", err + ) + + asyncio.run(run()) + print("[Non-dict transcription WS] string rejected with invalid_value") + + @unittest.skipUnless(HAS_WEBSOCKETS, "websockets package not installed") + def test_websocket_two_commits_propagates_previous_item_id(self): + # Two commits in one session must (a) emit `previous_item_id: null` on + # the first committed event, (b) emit `previous_item_id` equal to the + # first item's id on the second committed event, (c) produce two + # distinct item_ids, (d) reset per-item state between commits so the + # second transcript reflects only the second audio (no leak). + audio_zh = download_audio(TEST_AUDIO_ZH_URL, TEST_AUDIO_ZH_LOCAL) + pcm_zh, sr = _pcm16_from_audio_bytes(audio_zh) + audio_kungfu = download_audio(TEST_AUDIO_MP3_URL, TEST_AUDIO_MP3_LOCAL) + pcm_kungfu, _ = _pcm16_from_audio_bytes(audio_kungfu) + + async def run_one_cycle(ws, pcm, sample_rate): + """Send `pcm` as 0.5s base64 appends, commit, drain until completed.""" + chunk_bytes = int(0.5 * sample_rate * 2) + for offset in range(0, len(pcm), chunk_bytes): + await ws.send( + json.dumps( + { + "type": "input_audio_buffer.append", + "audio": base64.b64encode( + pcm[offset : offset + chunk_bytes] + ).decode("ascii"), + } + ) + ) + await ws.send(json.dumps({"type": "input_audio_buffer.commit"})) + committed = None + while True: + evt = json.loads(await ws.recv()) + t = evt.get("type") + if t == "input_audio_buffer.committed": + committed = evt + elif t == "conversation.item.input_audio_transcription.completed": + return committed, evt["transcript"] + elif t in ( + "error", + "conversation.item.input_audio_transcription.failed", + ): + raise RuntimeError(f"unexpected event: {evt!r}") + + async def run(): + async with websockets.connect(self._websocket_url()) as ws: + created = json.loads(await ws.recv()) + self.assertEqual(created.get("type"), "session.created", created) + await ws.send( + json.dumps( + { + "type": "session.update", + "session": { + "type": "transcription", + "audio": { + "input": { + "format": {"type": "audio/pcm", "rate": sr}, + "transcription": {"model": "qwen3-asr"}, + "noise_reduction": None, + "turn_detection": None, + } + }, + }, + } + ) + ) + while True: + evt = json.loads(await ws.recv()) + if evt.get("type") == "session.updated": + break + if evt.get("type") == "error": + raise RuntimeError(f"session.update failed: {evt!r}") + + committed_1, transcript_1 = await run_one_cycle(ws, pcm_zh, sr) + self.assertIsNone( + committed_1["previous_item_id"], + f"first commit's previous_item_id must be JSON null, got {committed_1!r}", + ) + first_item_id = committed_1["item_id"] + self.assertTrue(len(transcript_1) > 0, transcript_1) + + committed_2, transcript_2 = await run_one_cycle(ws, pcm_kungfu, sr) + self.assertEqual( + committed_2["previous_item_id"], + first_item_id, + f"second commit's previous_item_id must equal first item_id; " + f"got prev={committed_2['previous_item_id']!r} " + f"vs first={first_item_id!r}", + ) + self.assertNotEqual( + committed_2["item_id"], + first_item_id, + "item_ids must be distinct across commits", + ) + # State reset: second transcript must reflect only the second + # audio, not leak from the first. + wer = _wer(transcript_2, EXPECTED_TRANSCRIPTS["mp3"]) + self.assertLess( + wer, + 0.15, + f"second transcript leaked first audio's content; " + f"got {transcript_2!r} (WER {wer:.3f} vs canonical {EXPECTED_TRANSCRIPTS['mp3']!r})", + ) + + asyncio.run(run()) + if __name__ == "__main__": unittest.main(verbosity=3) diff --git a/test/registered/unit/entrypoints/openai/test_serving_transcription.py b/test/registered/unit/entrypoints/openai/test_serving_transcription.py index 49dbafc47..de3926be8 100644 --- a/test/registered/unit/entrypoints/openai/test_serving_transcription.py +++ b/test/registered/unit/entrypoints/openai/test_serving_transcription.py @@ -51,7 +51,10 @@ class _MockTokenizerManager: # Not a real ServerArgs, so base class sets allowed_custom_labels=None. # Default tests assume cumulative-text streaming (the sglang upstream # default); tests for incremental_streaming_output=True override this. - self.server_args = Mock(incremental_streaming_output=False) + self.server_args = Mock( + incremental_streaming_output=False, + asr_max_concurrent_sessions=32, + ) self.tokenizer = Mock() self._stream_chunks = stream_chunks @@ -255,7 +258,10 @@ class TestStreamingIncrementalOutputMode(CustomTestCase): for i, d in enumerate(chunk_deltas) ] tm = _MockTokenizerManager(chunks) - tm.server_args = Mock(incremental_streaming_output=True) + tm.server_args = Mock( + incremental_streaming_output=True, + asr_max_concurrent_sessions=32, + ) serving = OpenAIServingTranscription(tm) request = TranscriptionRequest(model="whisper", stream=True)