Add SSE stream_format support for audio speech streaming (#4490)

Signed-off-by: syd520zy <529477025@qq.com>
Signed-off-by: faceless void <100186390+syd520zy@users.noreply.github.com>
Signed-off-by: Nick Cao <ncao@redhat.com>
Signed-off-by: Chendi Xue <chendi.xue@intel.com>
Signed-off-by: kTorp <Kristoffer.Torp@amd.com>
Signed-off-by: Junhong Liu <98734602+LJH-LBJ@users.noreply.github.com>
Signed-off-by: Hongsheng Liu <liuhongsheng4@huawei.com>
Signed-off-by: Hyoseop Song <crad_on25@naver.com>
Signed-off-by: Hyoseop Song  <crad_on25@naver.com>
Signed-off-by: amy-why-3459 <wuhaiyan17@huawei.com>
Co-authored-by: Nick Cao <ncao@redhat.com>
Co-authored-by: Chendi.Xue <chendi.xue@intel.com>
Co-authored-by: Hongsheng Liu <liuhongsheng4@huawei.com>
Co-authored-by: Kristoffer <kristoffer.torp@amd.com>
Co-authored-by: Junhong Liu <ljh_lbj@163.com>
Co-authored-by: lishunyang <lishunyang12@163.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>
Co-authored-by: Hyoseop Song <crad_on25@naver.com>
Co-authored-by: Zhou Taichang <tzhouam@connect.ust.hk>
Co-authored-by: Alicia <115451386+congw729@users.noreply.github.com>
Co-authored-by: amy-why-3459 <wuhaiyan17@huawei.com>
This commit is contained in:
9 files changed
+320 -51

No files matched your search

@@ -122,9 +122,6 @@ def test_speech_missing_required_fields(omni_server: OmniServer, openai_client:
{"response_format": "mpeg"}, ("response_format", "literal_error", "wav"), id="response_format_invalid"
),
pytest.param({"stream_format": 1}, ("stream_format", "literal_error", "audio"), id="stream_format_wrong_type"),
pytest.param(
{"stream_format": "sse"}, ("stream_format", "value_error", "audio"), id="stream_format_sse_blocked"
),
pytest.param({"stream": "wrong_type"}, ("stream", "bool_parsing", "validation error"), id="stream_wrong_type"),
pytest.param(
{
@@ -1826,15 +1826,27 @@ class TestFileValidationFunctions:
class TestStreamingProtocolValidation:
"""Unit tests for the stream field validators in OpenAICreateSpeechRequest."""
"""Unit tests for streaming validators in OpenAICreateSpeechRequest."""
def test_default_is_non_streaming(self):
req = OpenAICreateSpeechRequest(input="Hello")
assert req.stream is False
assert req.stream_format is None
assert req.is_streaming() is False
def test_stream_validation_errors(self):
"""stream=True requires response_format not in ('pcm', 'wav') and speed=1.0."""
"""stream=True requires response_format in ('pcm', 'wav') and speed=1.0."""
with pytest.raises(ValidationError, match="requires response_format='pcm' or 'wav'"):
OpenAICreateSpeechRequest(input="Hello", stream=True, response_format="mp3")
with pytest.raises(ValidationError, match="Speed adjustment is not supported"):
OpenAICreateSpeechRequest(input="Hello", stream=True, response_format="pcm", speed=2.0)
def test_stream_format_audio_validation_errors(self):
with pytest.raises(ValidationError, match="requires response_format='pcm' or 'wav'"):
OpenAICreateSpeechRequest(input="Hello", stream_format="audio", response_format="mp3")
with pytest.raises(ValidationError, match="Speed adjustment is not supported"):
OpenAICreateSpeechRequest(input="Hello", stream_format="audio", response_format="pcm", speed=2.0)
def test_stream_valid(self):
"""stream=True + response_format in ('pcm', 'wav') + speed=1.0 is accepted."""
req = OpenAICreateSpeechRequest(input="Hello", stream=True, response_format="pcm")
@@ -1843,10 +1855,17 @@ class TestStreamingProtocolValidation:
req = OpenAICreateSpeechRequest(input="Hello", stream=True, response_format="wav")
assert req.stream is True
def test_sse_stream_format_is_blocked(self):
"""stream_format='sse' is blocked."""
with pytest.raises(ValidationError, match="sse"):
OpenAICreateSpeechRequest(input="Hello", stream_format="sse")
def test_stream_format_audio_is_valid(self):
req = OpenAICreateSpeechRequest(input="Hello", stream_format="audio", response_format="pcm")
assert req.stream_format == "audio"
assert req.is_raw_audio_stream() is True
def test_sse_stream_format_is_valid(self):
"""stream_format='sse' is accepted for /audio/speech."""
req = OpenAICreateSpeechRequest(input="Hello", stream_format="sse")
assert req.stream_format == "sse"
assert req.is_sse_stream() is True
class TestStreamingResponse:
@@ -1927,6 +1946,146 @@ class TestStreamingResponse:
assert "audio/pcm" in response.headers["content-type"]
assert len(response.content) > 0
def test_stream_format_audio_streaming(self, streaming_app):
"""stream_format=audio without stream=True returns raw audio/pcm chunks."""
client = TestClient(streaming_app)
response = client.post(
"/v1/audio/speech",
json={"input": "Hello", "stream_format": "audio", "response_format": "pcm"},
)
assert response.status_code == 200
assert "audio/pcm" in response.headers["content-type"]
assert "text/event-stream" not in response.headers["content-type"]
assert len(response.content) > 0
def test_sse_streaming(self, streaming_app):
"""stream_format=sse without stream=True returns audio deltas as SSE."""
client = TestClient(streaming_app)
response = client.post(
"/v1/audio/speech",
json={"input": "Hello", "stream_format": "sse", "response_format": "pcm"},
)
assert response.status_code == 200
assert "text/event-stream" in response.headers["content-type"]
body = response.text
assert "event: speech.audio.delta" in body
assert "event: speech.audio.done" in body
data_line = next(line for line in body.splitlines() if line.startswith("data: "))
payload = json.loads(data_line.removeprefix("data: "))
assert payload["type"] == "speech.audio.delta"
assert payload["response_format"] == "pcm"
assert base64.b64decode(payload["audio"])
def test_stream_true_prefers_raw_audio_over_sse(self, streaming_app):
"""stream=True keeps the existing raw audio stream behavior even with stream_format=sse."""
client = TestClient(streaming_app)
response = client.post(
"/v1/audio/speech",
json={"input": "Hello", "stream": True, "stream_format": "sse", "response_format": "pcm"},
)
assert response.status_code == 200
assert "audio/pcm" in response.headers["content-type"]
assert "text/event-stream" not in response.headers["content-type"]
assert len(response.content) > 0
def test_sse_rejects_unsupported_response_format(self, streaming_app):
"""stream_format=sse with a non-pcm/wav format must fail before streaming starts."""
client = TestClient(streaming_app)
response = client.post(
"/v1/audio/speech",
json={"input": "Hello", "stream_format": "sse", "response_format": "mp3"},
)
assert response.status_code in (400, 422)
assert "text/event-stream" not in response.headers.get("content-type", "")
def test_sse_rejects_speed_adjustment(self, streaming_app):
"""stream_format=sse with speed != 1.0 must fail before streaming starts."""
client = TestClient(streaming_app)
response = client.post(
"/v1/audio/speech",
json={"input": "Hello", "stream_format": "sse", "response_format": "pcm", "speed": 2.0},
)
assert response.status_code in (400, 422)
assert "text/event-stream" not in response.headers.get("content-type", "")
def test_stream_format_audio_rejects_unsupported_response_format(self, streaming_app):
client = TestClient(streaming_app)
response = client.post(
"/v1/audio/speech",
json={"input": "Hello", "stream_format": "audio", "response_format": "mp3"},
)
assert response.status_code in (400, 422)
assert "audio/" not in response.headers.get("content-type", "")
def test_stream_format_audio_rejects_speed_adjustment(self, streaming_app):
client = TestClient(streaming_app)
response = client.post(
"/v1/audio/speech",
json={"input": "Hello", "stream_format": "audio", "response_format": "pcm", "speed": 2.0},
)
assert response.status_code in (400, 422)
assert "audio/" not in response.headers.get("content-type", "")
@pytest.fixture
def erroring_streaming_app(self, mocker: MockerFixture):
"""Test app whose mock engine raises mid-stream, to exercise the SSE error event."""
async def mock_generate_streaming(*args, **kwargs):
raise RuntimeError("boom: simulated engine failure")
yield # pragma: no cover - generator marker, unreachable
mock_engine_client = mocker.MagicMock()
mock_engine_client.errored = False
mock_engine_client.generate = mocker.MagicMock(side_effect=mock_generate_streaming)
mock_engine_client.default_sampling_params_list = [{}]
mock_models = mocker.MagicMock()
mock_models.is_base_model.return_value = True
speech_server = OmniOpenAIServingSpeech(
engine_client=mock_engine_client,
models=mock_models,
request_logger=mocker.MagicMock(),
)
original_create_speech = speech_server.create_speech
sig = signature(original_create_speech)
new_parameters = [p for name, p in sig.parameters.items() if name != "raw_request"]
new_sig = Signature(parameters=new_parameters, return_annotation=sig.return_annotation)
async def awaitable_create_speech(*args, **kwargs):
return await original_create_speech(*args, **kwargs)
awaitable_create_speech.__signature__ = new_sig
speech_server.create_speech = awaitable_create_speech
app = FastAPI()
app.add_api_route("/v1/audio/speech", speech_server.create_speech, methods=["POST"], response_model=None)
return app
def test_sse_emits_error_event_on_generator_failure(self, erroring_streaming_app):
"""An exception inside the SSE generator must surface as a speech.audio.error event."""
client = TestClient(erroring_streaming_app)
response = client.post(
"/v1/audio/speech",
json={"input": "Hello", "stream_format": "sse", "response_format": "pcm"},
)
assert response.status_code == 200
assert "text/event-stream" in response.headers["content-type"]
body = response.text
assert "event: speech.audio.error" in body
error_line = next(line for line in body.splitlines() if line.startswith("data: "))
payload = json.loads(error_line.removeprefix("data: "))
assert payload["type"] == "speech.audio.error"
assert "error" in payload
assert payload["error"]["message"]
def test_non_streaming_unchanged(self, streaming_app):
"""Non-streaming path must still return audio/wav."""
client = TestClient(streaming_app)
@@ -2883,6 +3042,40 @@ class TestTTSAsyncOffloading:
qwen3_tts_server._build_tts_params.assert_called_once()
qwen3_tts_server._estimate_prompt_len_async.assert_awaited_once()
def test_prepare_speech_generation_treats_sse_as_streaming(self, qwen3_tts_server, mocker: MockerFixture):
"""stream_format=sse should request delta-style multimodal outputs."""
qwen3_tts_server._validate_tts_request = mocker.MagicMock(return_value=None)
qwen3_tts_server._build_tts_params = mocker.MagicMock(
return_value={"text": ["hello"], "task_type": ["CustomVoice"], "speaker": ["Vivian"]}
)
qwen3_tts_server._estimate_prompt_len_async = mocker.AsyncMock(return_value=512)
mock_coerce = mocker.patch(
"vllm_omni.entrypoints.openai.serving_speech.coerce_param_message_types",
return_value=qwen3_tts_server.engine_client.default_sampling_params_list,
)
request = OpenAICreateSpeechRequest(input="hello", stream_format="sse")
asyncio.run(qwen3_tts_server._prepare_speech_generation(request))
mock_coerce.assert_called_once_with(qwen3_tts_server.engine_client.default_sampling_params_list, True)
def test_prepare_speech_generation_treats_audio_as_streaming(self, qwen3_tts_server, mocker: MockerFixture):
"""stream_format=audio should request delta-style multimodal outputs."""
qwen3_tts_server._validate_tts_request = mocker.MagicMock(return_value=None)
qwen3_tts_server._build_tts_params = mocker.MagicMock(
return_value={"text": ["hello"], "task_type": ["CustomVoice"], "speaker": ["Vivian"]}
)
qwen3_tts_server._estimate_prompt_len_async = mocker.AsyncMock(return_value=512)
mock_coerce = mocker.patch(
"vllm_omni.entrypoints.openai.serving_speech.coerce_param_message_types",
return_value=qwen3_tts_server.engine_client.default_sampling_params_list,
)
request = OpenAICreateSpeechRequest(input="hello", stream_format="audio", response_format="pcm")
asyncio.run(qwen3_tts_server._prepare_speech_generation(request))
mock_coerce.assert_called_once_with(qwen3_tts_server.engine_client.default_sampling_params_list, True)
def test_prepare_speech_generation_qwen3_voicedesign_non_streaming_mode_false(
self, qwen3_tts_server, mocker: MockerFixture
):
+1 -1
View File
@@ -1200,7 +1200,7 @@ _remove_route_from_router(router, "/v1/audio/speech", {"POST"})
"/v1/audio/speech",
dependencies=[Depends(validate_json_request)],
responses={
HTTPStatus.OK.value: {"content": {"audio/*": {}}},
HTTPStatus.OK.value: {"content": {"audio/*": {}, "text/event-stream": {}}},
HTTPStatus.BAD_REQUEST.value: {"model": ErrorResponse},
HTTPStatus.NOT_FOUND.value: {"model": ErrorResponse},
HTTPStatus.INTERNAL_SERVER_ERROR.value: {"model": ErrorResponse},
@@ -24,13 +24,9 @@ class AudioMixin:
audio_tensor = audio_obj.audio_tensor
sample_rate = audio_obj.sample_rate
response_format = audio_obj.response_format.lower()
stream_format = audio_obj.stream_format
base64_encode = audio_obj.base64_encode
speed = audio_obj.speed
if stream_format != "audio":
raise ValueError(f"Unsupported stream format: {stream_format}")
if soundfile is None:
raise ImportError(
"soundfile is required for audio generation. Please install it with: pip install soundfile"
+22 -11
View File
@@ -65,12 +65,18 @@ class OpenAICreateSpeechRequest(BaseModel):
ge=0.25,
le=4.0,
)
stream_format: Literal["sse", "audio"] | None = "audio"
stream_format: Literal["sse", "audio"] | None = Field(
default=None,
description=(
"Streaming output format. 'audio' streams raw pcm/wav bytes; "
"'sse' streams OpenAI speech.audio.* SSE events. Omit for non-streaming."
),
)
stream: bool = Field(
default=False,
description=(
"If true, stream raw PCM audio chunks as they are decoded. "
"Requires response_format='pcm'. Speed adjustment is not supported when streaming."
"Legacy streaming switch; equivalent to stream_format='audio'. "
"Requires response_format='pcm' or 'wav'. Speed adjustment is not supported when streaming."
),
)
@@ -158,8 +164,6 @@ class OpenAICreateSpeechRequest(BaseModel):
@field_validator("stream_format")
@classmethod
def validate_stream_format(cls, v: str) -> str:
if v == "sse":
raise ValueError("'sse' is not a supported stream_format yet. Please use 'audio'.")
return v
@field_validator("ref_audio", mode="before")
@@ -287,20 +291,28 @@ class OpenAICreateSpeechRequest(BaseModel):
raise ValueError("'speaker_embedding' and 'ref_audio' are mutually exclusive")
return self
def is_raw_audio_stream(self) -> bool:
return self.stream or self.stream_format == "audio"
def is_sse_stream(self) -> bool:
return self.stream_format == "sse" and not self.is_raw_audio_stream()
def is_streaming(self) -> bool:
return self.is_raw_audio_stream() or self.stream_format == "sse"
@model_validator(mode="after")
def validate_streaming_constraints(self) -> "OpenAICreateSpeechRequest":
if self.stream:
if self.is_streaming():
if self.response_format not in ("pcm", "wav"):
raise ValueError(
"Streaming (stream=true) requires response_format='pcm' or 'wav'. "
"Streaming (stream=true, stream_format='audio', or stream_format='sse') "
"requires response_format='pcm' or 'wav'. "
f"Got response_format='{self.response_format}'."
)
if self.speed is None:
self.speed = 1.0
elif self.speed != 1.0:
raise ValueError(
"Speed adjustment is not supported when streaming (stream=true). Set speed=1.0 or omit it."
)
raise ValueError("Speed adjustment is not supported when streaming. Set speed=1.0 or omit it.")
return self
@@ -356,7 +368,6 @@ class CreateAudio(BaseModel):
sample_rate: int = 24000
response_format: str = "wav"
speed: float = 1.0
stream_format: Literal["sse", "audio"] | None = "audio"
base64_encode: bool = True
class Config:
@@ -152,7 +152,6 @@ class OmniOpenAIServingAudioGenerate(OpenAIServing, AudioMixin):
sample_rate=int(sample_rate),
response_format=request.response_format or "wav",
speed=request.speed or 1.0,
stream_format=request.stream_format,
base64_encode=False,
)
@@ -2520,7 +2520,6 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin):
sample_rate=sample_rate,
response_format="wav",
speed=1.0,
stream_format="audio",
base64_encode=True,
)
@@ -3475,7 +3474,6 @@ class OmniOpenAIServingChat(OpenAIServingChat, AudioMixin):
sample_rate=sample_rate,
response_format="wav",
speed=1.0,
stream_format="audio",
base64_encode=True,
)
audio_response: AudioResponse = self.create_audio(audio_obj)
+98 -22
View File
@@ -2659,7 +2659,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
sample_rate=sample_rate_val,
response_format="pcm",
speed=1.0,
stream_format="audio",
base64_encode=False,
)
if first_audio_chunk_s is None:
@@ -2728,6 +2727,53 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
if not artifact_ready:
self._discard_ref_audio_artifact_warmup(request_id)
async def _generate_audio_sse_events(
self,
generator,
request_id: str,
response_format: str = "pcm",
raw_request: Request | None = None,
request_start_s: float | None = None,
):
"""Generate OpenAI-style SSE events with base64 audio deltas.
Field naming follows the OpenAI ``speech.audio.delta`` schema, which
carries the base64 chunk in ``audio`` (not ``delta`` — that is the
Realtime API ``response.audio.delta`` convention, a different event).
See https://platform.openai.com/docs/api-reference/audio-streaming.
"""
try:
async for chunk in self._generate_audio_chunks(
generator,
request_id,
response_format,
raw_request=raw_request,
request_start_s=request_start_s,
):
payload = {
"type": "speech.audio.delta",
"audio": base64.b64encode(chunk).decode("ascii"),
"response_format": response_format,
}
data = json.dumps(payload, separators=(",", ":"))
yield f"event: speech.audio.delta\ndata: {data}\n\n"
done = json.dumps({"type": "speech.audio.done"}, separators=(",", ":"))
yield f"event: speech.audio.done\ndata: {done}\n\n"
except asyncio.CancelledError:
raise
except Exception as e:
payload = {
"type": "speech.audio.error",
"error": {
"message": str(e),
"type": "server_error",
"param": None,
"code": HTTPStatus.INTERNAL_SERVER_ERROR.value,
},
}
data = json.dumps(payload, separators=(",", ":"))
yield f"event: speech.audio.error\ndata: {data}\n\n"
@staticmethod
def _extract_audio_output(res) -> tuple[dict | None, str | None]:
"""Return (audio_output dict, audio key) or (None, None).
@@ -3305,7 +3351,8 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
# we don't emit redundant MM data & drain after emitting.
# list() makes a copy to avoid mutating the params.
sampling_params_list = list(self.engine_client.default_sampling_params_list)
sampling_params_list = coerce_param_message_types(sampling_params_list, request.stream)
is_streaming_request = request.is_streaming()
sampling_params_list = coerce_param_message_types(sampling_params_list, is_streaming_request)
# Build prompt + tts_params via the per-model adapter (RFC #4327). Every
# dedicated TTS model resolves to an adapter that owns its validation,
@@ -3636,7 +3683,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
sample_rate=sample_rate,
response_format=request.response_format or "wav",
speed=request.speed or 1.0,
stream_format=request.stream_format,
base64_encode=base64_encode,
)
audio_response: AudioResponse = self.create_audio(audio_obj)
@@ -3749,7 +3795,6 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
sample_rate=sample_rate,
response_format=request.response_format or "wav",
speed=request.speed or 1.0,
stream_format=request.stream_format,
base64_encode=False,
)
audio_response: AudioResponse = self.create_audio(audio_obj)
@@ -3779,6 +3824,25 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
error_body = json.dumps({"error": {"message": message, "type": err_type, "param": None, "code": status_code}})
return Response(content=error_body, media_type="application/json", status_code=status_code)
def _validate_speech_streaming_request(
self,
request: OpenAICreateSpeechRequest,
*,
mode_label: str,
) -> tuple[str, Response | None]:
"""Validate pcm/wav + speed constraints for streaming speech responses."""
response_format = (request.response_format or "wav").lower()
if response_format not in ("pcm", "wav"):
return response_format, self.create_error_response(
f"{mode_label} is only supported for 'pcm' and 'wav' formats. Got '{response_format}'."
)
if request.speed is not None and request.speed != 1.0:
return response_format, self.create_error_response(
f"{mode_label} is not supported with speed adjustment. "
"Use a non-streaming request or remove the speed parameter."
)
return response_format, None
async def create_speech(
self,
request: OpenAICreateSpeechRequest,
@@ -3800,7 +3864,9 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
- ref_text: Transcript of reference audio (Base task)
- x_vector_only_mode: Use speaker embedding only (Base task)
Streaming is supported via stream=True with response_format='pcm' or 'wav'.
Streaming is supported via ``stream_format='audio'`` or the legacy
``stream=True`` switch, with ``response_format='pcm'`` or ``'wav'``.
``stream_format='sse'`` returns OpenAI ``speech.audio.*`` SSE events instead.
Each Code2Wav chunk is yielded as raw audio bytes as soon as it is decoded.
For WAV format, a header with placeholder size values is emitted first.
"""
@@ -3820,7 +3886,7 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
)
try:
if request.stream:
if request.is_raw_audio_stream():
if request.word_timestamps:
return self.create_error_response(
"word_timestamps=true is currently supported by the WebSocket "
@@ -3828,22 +3894,12 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
"stream_audio=true and response_format='pcm'."
)
# Determine response format and media type for streaming
response_format = (request.response_format or "wav").lower()
# Only pcm and wav support streaming without post-processing
if response_format not in ["pcm", "wav"]:
return self.create_error_response(
f"Streaming is only supported for 'pcm' and 'wav' formats. "
f"Got '{response_format}'. For other formats, use stream=False."
)
# Check if speed adjustment is requested (not compatible with streaming)
if request.speed is not None and request.speed != 1.0:
return self.create_error_response(
"Streaming is not supported with speed adjustment. "
"Use stream=False or remove the speed parameter."
)
response_format, error = self._validate_speech_streaming_request(
request,
mode_label="Streaming",
)
if error is not None:
return error
media_type = "audio/wav" if response_format == "wav" else "audio/pcm"
_, generator, _ = await self._prepare_speech_generation(request, request_id=request_id)
@@ -3858,6 +3914,26 @@ class OmniOpenAIServingSpeech(OpenAIServing, AudioMixin):
media_type=media_type,
)
if request.is_sse_stream():
response_format, error = self._validate_speech_streaming_request(
request,
mode_label="SSE streaming",
)
if error is not None:
return error
_, generator, _ = await self._prepare_speech_generation(request, request_id=request_id)
return StreamingResponse(
self._generate_audio_sse_events(
generator,
request_id,
response_format,
raw_request=raw_request,
request_start_s=request_start_s,
),
media_type="text/event-stream",
)
audio_bytes, media_type = await self._generate_audio_bytes(request, request_id=request_id)
total_ms = (time.perf_counter() - request_start_s) * 1000.0
logger.info(
@@ -907,7 +907,6 @@ class OmniStreamingVideoHandler:
sample_rate=24000,
response_format="wav",
speed=1.0,
stream_format="audio",
base64_encode=True,
)
mixin = AudioMixin()