From a9be976845cac779d97114ae0b57ccd2f9898039 Mon Sep 17 00:00:00 2001 From: Didar Kozhikov Date: Wed, 19 Aug 2026 21:40:35 +0500 Subject: [PATCH] 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 --- .../ai_voice_runtime_service/providers/tts.py | 61 ++++++++++++++++ tests/test_ai_voice_tts_provider.py | 73 +++++++++++++++++++ 2 files changed, 134 insertions(+) diff --git a/services/ai_voice_runtime_service/providers/tts.py b/services/ai_voice_runtime_service/providers/tts.py index c62c3e9..2c96234 100644 --- a/services/ai_voice_runtime_service/providers/tts.py +++ b/services/ai_voice_runtime_service/providers/tts.py @@ -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() diff --git a/tests/test_ai_voice_tts_provider.py b/tests/test_ai_voice_tts_provider.py index d7fdabb..288583d 100644 --- a/tests/test_ai_voice_tts_provider.py +++ b/tests/test_ai_voice_tts_provider.py @@ -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] = []