fix(openai): read TTS response by content-type, not model name

livekit-plugins/livekit-plugins-openai/livekit/plugins/openai/tts.py

@@ -47,6 +47,29 @@
 # Models that use audio stream format (character-based billing)
 AUDIO_STREAM_MODELS = {"tts-1", "tts-1-hd"}
 
+SSE_CONTENT_TYPE = "text/event-stream"
+
+# Content types `AudioEmitter` knows how to decode, mirroring the codecs decoder table.
+DECODABLE_CONTENT_TYPES = frozenset(
+    {
+        "audio/mpeg",
+        "audio/mp3",
+        "audio/x-mpeg",
+        "audio/aac",
+        "audio/x-aac",
+        "audio/flac",
+        "audio/x-flac",
+        "audio/wav",
+        "audio/wave",
+        "audio/x-wav",
+        "audio/opus",
+        "audio/ogg",
+        "audio/webm",
+        "audio/mp4",
+        "audio/pcm",
+    }
+)
+
 
 @dataclass
 class _TTSOptions:
@@ -198,11 +221,7 @@ def with_azure(
     def synthesize(
         self, text: str, *, conn_options: APIConnectOptions = DEFAULT_API_CONNECT_OPTIONS
     ) -> tts.ChunkedStream:
-        # Use audio stream format for tts-1/tts-1-hd (character-based billing)
-        # Use SSE stream format for newer models like gpt-4o-mini-tts (token-based billing)
-        if self._opts.model in AUDIO_STREAM_MODELS:
-            return AudioChunkedStream(tts=self, input_text=text, conn_options=conn_options)
-        return SSEChunkedStream(tts=self, input_text=text, conn_options=conn_options)
+        return ChunkedStream(tts=self, input_text=text, conn_options=conn_options)
 
     def prewarm(self) -> None:
         async def _prewarm() -> None:
@@ -221,52 +240,13 @@ async def aclose(self) -> None:
             await self._client.close()
 
 
-class AudioChunkedStream(tts.ChunkedStream):
-    """ChunkedStream for tts-1 and tts-1-hd models using audio stream format."""
-
-    def __init__(self, *, tts: TTS, input_text: str, conn_options: APIConnectOptions) -> None:
-        super().__init__(tts=tts, input_text=input_text, conn_options=conn_options)
-        self._tts: TTS = tts
-        self._opts = replace(tts._opts)
-
-    async def _run(self, output_emitter: tts.AudioEmitter) -> None:
-        oai_stream = self._tts._client.audio.speech.with_streaming_response.create(
-            input=self.input_text,
-            model=self._opts.model,
-            voice=self._opts.voice,
-            response_format=self._opts.response_format,  # type: ignore
-            speed=self._opts.speed,
-            instructions=self._opts.instructions or openai.omit,
-            stream_format="audio",
-            timeout=httpx.Timeout(30, connect=self._conn_options.timeout),
-        )
-
-        try:
-            async with oai_stream as stream:
-                output_emitter.initialize(
-                    request_id=stream.request_id or "",
-                    sample_rate=SAMPLE_RATE,
-                    num_channels=NUM_CHANNELS,
-                    mime_type=f"audio/{self._opts.response_format}",
-                )
-
-                async for data in stream.iter_bytes():
-                    output_emitter.push(data)
-
-            output_emitter.flush()
-
-        except openai.APITimeoutError:
-            raise APITimeoutError() from None
-        except openai.APIStatusError as e:
-            raise APIStatusError(
-                e.message, status_code=e.status_code, request_id=e.request_id, body=e.body
-            ) from None
-        except Exception as e:
-            raise APIConnectionError() from e
+class ChunkedStream(tts.ChunkedStream):
+    """ChunkedStream that reads the body according to what the server actually returned.
 
-
-class SSEChunkedStream(tts.ChunkedStream):
-    """ChunkedStream for gpt-4o-mini-tts and newer models using SSE stream format."""
+    `stream_format` is an OpenAI-specific extension. OpenAI-compatible servers ignore it and
+    answer with the audio bytes of `response_format` rather than an SSE stream, so the response
+    `Content-Type` -- not the requested `stream_format` -- decides how the body is parsed.
+    """
 
     def __init__(self, *, tts: TTS, input_text: str, conn_options: APIConnectOptions) -> None:
         super().__init__(tts=tts, input_text=input_text, conn_options=conn_options)
@@ -281,52 +261,65 @@ async def _run(self, output_emitter: tts.AudioEmitter) -> None:
             response_format=self._opts.response_format,  # type: ignore
             speed=self._opts.speed,
             instructions=self._opts.instructions or openai.omit,
-            stream_format="sse",
+            # `sse` is not supported for tts-1/tts-1-hd (character-based billing)
+            stream_format="audio" if self._opts.model in AUDIO_STREAM_MODELS else "sse",
             timeout=httpx.Timeout(30, connect=self._conn_options.timeout),
         )
 
         try:
             async with oai_stream as stream:
+                media_type = stream.headers.get("content-type", "").split(";")[0].strip().lower()
+                # a server that ignored response_format still declares what it sent
+                mime_type = (
+                    media_type
+                    if media_type in DECODABLE_CONTENT_TYPES
+                    else f"audio/{self._opts.response_format}"
+                )
                 output_emitter.initialize(
                     request_id=stream.request_id or "",
                     sample_rate=SAMPLE_RATE,
                     num_channels=NUM_CHANNELS,
-                    mime_type=f"audio/{self._opts.response_format}",
+                    mime_type=mime_type,
                 )
 
-                # Parse SSE events from the stream
-                async for line in stream.iter_lines():
-                    if not line or not line.startswith("data: "):
-                        continue
-
-                    data = line[6:]  # Remove "data: " prefix
-                    if data == "[DONE]":
-                        break
-
-                    try:
-                        event = json.loads(data)
-                    except json.JSONDecodeError:
-                        continue
-
-                    event_type = event.get("type", "")
-
-                    if event_type == "speech.audio.delta":
-                        # Decode base64 audio and push to emitter
-                        audio_b64 = event.get("delta", "") or event.get("audio", "")
-                        if audio_b64:
-                            audio_data = base64.b64decode(audio_b64)
-                            output_emitter.push(audio_data)
-
-                    elif event_type == "speech.audio.done":
-                        # Extract token usage from the done event
-                        usage = event.get("usage", {})
-                        input_tokens = usage.get("input_tokens", 0)
-                        output_tokens = usage.get("output_tokens", 0)
-                        if input_tokens or output_tokens:
-                            self._set_token_usage(
-                                input_tokens=input_tokens,
-                                output_tokens=output_tokens,
-                            )
+                if media_type != SSE_CONTENT_TYPE:
+                    # An OpenAI-compatible server that ignored stream_format and returned the
+                    # audio bytes of response_format directly.
+                    async for chunk in stream.iter_bytes():
+                        output_emitter.push(chunk)
+                else:
+                    async for line in stream.iter_lines():
+                        if not line or not line.startswith("data: "):
+                            continue
+
+                        data = line[6:]  # Remove "data: " prefix
+                        if data == "[DONE]":
+                            break
+
+                        try:
+                            event = json.loads(data)
+                        except json.JSONDecodeError:
+                            continue
+
+                        event_type = event.get("type", "")
+
+                        if event_type == "speech.audio.delta":
+                            # Decode base64 audio and push to emitter
+                            audio_b64 = event.get("delta", "") or event.get("audio", "")
+                            if audio_b64:
+                                audio_data = base64.b64decode(audio_b64)
+                                output_emitter.push(audio_data)
+
+                        elif event_type == "speech.audio.done":
+                            # Extract token usage from the done event
+                            usage = event.get("usage", {})
+                            input_tokens = usage.get("input_tokens", 0)
+                            output_tokens = usage.get("output_tokens", 0)
+                            if input_tokens or output_tokens:
+                                self._set_token_usage(
+                                    input_tokens=input_tokens,
+                                    output_tokens=output_tokens,
+                                )
 
             output_emitter.flush()
 

tests/test_plugin_openai_tts.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+import base64
+import io
+import json
+import wave
+
+import httpx
+import openai
+import pytest
+
+from livekit.plugins.openai import TTS
+
+pytestmark = pytest.mark.unit
+
+PCM = b"\x00\x01" * 4800  # 200ms of 16-bit mono at 24kHz
+
+
+def _tts(handler, *, model: str, response_format: str = "pcm") -> TTS:
+    client = openai.AsyncClient(
+        api_key="test",
+        base_url="https://compatible.example.com/v1",
+        max_retries=0,
+        http_client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
+    )
+    return TTS(model=model, client=client, response_format=response_format)
+
+
+async def _synthesize(tts: TTS) -> bytes:
+    audio = b""
+    async with tts.synthesize("hello") as stream:
+        async for ev in stream:
+            audio += ev.frame.data.tobytes()
+    return audio
+
+
+@pytest.mark.parametrize("model", ["hexgrad/Kokoro-82M", "gpt-4o-mini-tts"])
+async def test_audio_body_is_decoded_whatever_the_model(model: str) -> None:
+    """A compatible server ignores stream_format and answers with plain audio bytes."""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, content=PCM, headers={"content-type": "audio/pcm"})
+
+    audio = await _synthesize(_tts(handler, model=model))
+
+    assert audio[: len(PCM)] == PCM
+
+
+async def test_declared_content_type_wins_over_requested_format() -> None:
+    """A server may ignore response_format; decode what it says it sent, not what we asked for."""
+    buf = io.BytesIO()
+    with wave.open(buf, "wb") as w:
+        w.setnchannels(1)
+        w.setsampwidth(2)
+        w.setframerate(24000)
+        w.writeframes(PCM)
+    wav = buf.getvalue()
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        return httpx.Response(200, content=wav, headers={"content-type": "audio/wav"})
+
+    # mp3 is requested, but the server answers with wav and says so
+    audio = await _synthesize(_tts(handler, model="kokoro", response_format="mp3"))
+
+    assert audio
+
+
+async def test_sse_body_is_parsed() -> None:
+    """OpenAI answers stream_format="sse" with an event stream, which is still parsed."""
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        events = [
+            json.dumps({"type": "speech.audio.delta", "delta": base64.b64encode(PCM).decode()}),
+            json.dumps({"type": "speech.audio.done", "usage": {"input_tokens": 1}}),
+        ]
+        body = "".join(f"data: {e}\n\n" for e in events) + "data: [DONE]\n\n"
+        return httpx.Response(
+            200, content=body.encode(), headers={"content-type": "text/event-stream"}
+        )
+
+    audio = await _synthesize(_tts(handler, model="gpt-4o-mini-tts"))
+
+    assert audio[: len(PCM)] == PCM
+
+
+@pytest.mark.parametrize(
+    ("model", "expected"),
+    [
+        ("tts-1", "audio"),
+        ("tts-1-hd", "audio"),
+        ("gpt-4o-mini-tts", "sse"),
+    ],
+)
+async def test_stream_format_requested_per_model(model: str, expected: str) -> None:
+    """`sse` is not supported for tts-1/tts-1-hd, so those must keep requesting `audio`."""
+    requests: list[dict] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        requests.append(json.loads(request.content))
+        return httpx.Response(200, content=PCM, headers={"content-type": "audio/pcm"})
+
+    await _synthesize(_tts(handler, model=model))
+
+    assert requests[0]["stream_format"] == expected