Implement real streaming TTS for ElevenLabs to cut voice-assistant reply latency

The AudioSocket/media_runtime playback pipeline already supports chunked TTS streaming (voice_v2_streaming_tts), but every provider inherited the base TTSProvider.synthesize_chunks(), which just called the blocking synthesize() and yielded the entire finished audio as a single "chunk" - so the caller waited for full-utterance synthesis before any playback could start regardless of the flag. ElevenLabs is the production default (AI_VOICE_TTS_PROVIDER=elevenlabs in deployment/docker-compose.server.yml), so give it a real implementation that POSTs to the /stream endpoint and yields audio as network chunks arrive, instead of waiting for the whole response body. Chunk boundaries are re-aligned to whole 16-bit PCM samples so a split sample at a network read boundary can't corrupt playback. The full synthesized audio is still written to the on-disk cache afterwards so repeat phrases stay fast and skip the vendor call entirely, matching the existing synthesize() cache behavior. Added test_elevenlabs_tts_provider_streams_chunks_and_caches_full_audio to cover: chunk splitting mid-sample gets re-aligned, all yielded chunks are sample-aligned, the full audio round-trips through the cache, and a cached synthesis is replayed without invoking the streaming endpoint again. Verified via tests/test_ai_voice_tts_provider.py (9/9 pass) and a wider voice/tts-filtered run across the suite: the only failures present are the same pre-existing, already-documented ones (sales_service test cross-file isolation ordering, one known persona-prompt assertion) - identical set to before this change, no new failures. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-08-19 21:40:35 +05:00
parent cdabe61bc2
commit a9be976845
2 changed files with 134 additions and 0 deletions
@@ -777,6 +777,67 @@ class ElevenLabsTTSProvider(TTSProvider):
            self._write_cached_synthesis(synthesis, language=language)
        return synthesis

+    def synthesize_chunks(
+        self,
+        text: str,
+        *,
+        language: str | None = None,
+        style_hints: dict[str, object] | None = None,
+    ):
+        del style_hints
+        if not text:
+            return
+        cached = self._load_cached_synthesis(text, language=language)
+        if cached is not None:
+            if cached.audio_bytes:
+                yield cached
+            return
+        if not self._api_key:
+            raise RuntimeError("AI_VOICE_TTS_ELEVENLABS_API_KEY is required for ElevenLabs TTS")
+
+        # Raw PCM16 samples must stay 2-byte aligned across network chunk
+        # boundaries, or a split sample corrupts playback at that boundary.
+        collected = bytearray()
+        leftover = b""
+        with httpx.Client(timeout=self._timeout_seconds) as client:
+            with client.stream(
+                "POST",
+                f"{self._api_base}/v1/text-to-speech/{self._voice(language)}/stream",
+                headers={
+                    "xi-api-key": self._api_key,
+                    "Accept": "application/octet-stream",
+                    "Content-Type": "application/json",
+                },
+                params={"output_format": self._output_format},
+                json={
+                    "text": text,
+                    "model_id": self._model(language),
+                    "language_code": self._language_code(language),
+                },
+            ) as response:
+                response.raise_for_status()
+                for raw_chunk in response.iter_bytes():
+                    if not raw_chunk:
+                        continue
+                    data = leftover + raw_chunk
+                    if len(data) % 2:
+                        leftover = data[-1:]
+                        data = data[:-1]
+                    else:
+                        leftover = b""
+                    if not data:
+                        continue
+                    collected.extend(data)
+                    yield TTSSynthesis(text=text, audio_bytes=data, sample_rate_hz=self._sample_rate_hz)
+        if leftover:
+            collected.extend(leftover)
+
+        if collected:
+            full_synthesis = TTSSynthesis(text=text, audio_bytes=bytes(collected), sample_rate_hz=self._sample_rate_hz)
+            with self._cache_lock:
+                if self._load_cached_synthesis(text, language=language) is None:
+                    self._write_cached_synthesis(full_synthesis, language=language)
+

 def build_tts_provider(name: str) -> TTSProvider:
    normalized = str(name or "stub").strip().lower()
@@ -328,6 +328,79 @@ def test_elevenlabs_tts_provider_posts_voice_id_model_and_language_code(tmp_path
    assert calls[0]["json"]["language_code"] == "kk"


+def test_elevenlabs_tts_provider_streams_chunks_and_caches_full_audio(tmp_path, monkeypatch):
+    calls: list[dict] = []
+
+    class _DummyStreamResponse:
+        def __init__(self, chunks: list[bytes]) -> None:
+            self._chunks = chunks
+
+        def raise_for_status(self) -> None:
+            return None
+
+        def iter_bytes(self):
+            yield from self._chunks
+
+    class _DummyStreamContext:
+        def __init__(self, response: _DummyStreamResponse) -> None:
+            self._response = response
+
+        def __enter__(self) -> _DummyStreamResponse:
+            return self._response
+
+        def __exit__(self, exc_type, exc, tb) -> None:
+            return None
+
+    class _DummyClient:
+        def __init__(self, *, timeout: float) -> None:
+            self.timeout = timeout
+
+        def __enter__(self) -> _DummyClient:
+            return self
+
+        def __exit__(self, exc_type, exc, tb) -> None:
+            return None
+
+        def stream(self, method: str, url: str, *, headers: dict[str, str], params: dict[str, str], json: dict):
+            calls.append({"method": method, "url": url, "headers": headers, "params": params, "json": json})
+            # Split mid-sample on purpose to exercise the 2-byte alignment guard.
+            return _DummyStreamContext(_DummyStreamResponse([b"\x01\x00\x02", b"\x00\x03\x00\x04\x00"]))
+
+    monkeypatch.setenv("AI_VOICE_TTS_ELEVENLABS_API_KEY", "elevenlabs-key")
+    monkeypatch.setenv("AI_VOICE_TTS_CACHE_ENABLED", "1")
+    monkeypatch.setenv("AI_VOICE_TTS_CACHE_DIR", str(tmp_path))
+    monkeypatch.setattr(tts_module.httpx, "Client", _DummyClient)
+
+    provider = tts_module.ElevenLabsTTSProvider(
+        api_base="https://api.elevenlabs.example",
+        ru_voice="nPczCjzI2devNBz1zQrb",
+        output_format="pcm_16000",
+    )
+
+    chunks = list(provider.synthesize_chunks("Привет из ElevenLabs", language="ru"))
+
+    assert b"".join(chunk.audio_bytes for chunk in chunks) == b"\x01\x00\x02\x00\x03\x00\x04\x00"
+    assert all(len(chunk.audio_bytes) % 2 == 0 for chunk in chunks)
+    assert len(calls) == 1
+    assert calls[0]["url"] == "https://api.elevenlabs.example/v1/text-to-speech/nPczCjzI2devNBz1zQrb/stream"
+    assert list(Path(tmp_path).rglob("*.pcm"))
+
+    def _unexpected_stream(*args, **kwargs):
+        raise AssertionError("ElevenLabs streaming TTS should not be called again once cached")
+
+    monkeypatch.setattr(_DummyClient, "stream", _unexpected_stream)
+    replay_provider = tts_module.ElevenLabsTTSProvider(
+        api_base="https://api.elevenlabs.example",
+        ru_voice="nPczCjzI2devNBz1zQrb",
+        output_format="pcm_16000",
+    )
+    replay_chunks = list(replay_provider.synthesize_chunks("Привет из ElevenLabs", language="ru"))
+
+    assert len(replay_chunks) == 1
+    assert replay_chunks[0].audio_bytes == b"\x01\x00\x02\x00\x03\x00\x04\x00"
+    assert len(calls) == 1
+
+
 def test_runtime_configured_tts_provider_uses_database_selected_provider(monkeypatch):
    calls: list[dict] = []