From bc4dcb1b02b9aeb49308e01c2beeacb9154e4c66 Mon Sep 17 00:00:00 2001 From: Carlos Diosdado Date: Tue, 28 Jul 2026 21:45:46 -0700 Subject: [PATCH] feat(tts): Gemini SSE + xAI WebSocket streaming providers, tts.streaming.provider knob, docs + E2E tests Salvaged from PR #47588 and rebased onto the post-campaign streaming core: the StreamingTTSProvider ABC/registry and the ElevenLabs/OpenAI streamers already live on main (tools/tts_streaming.py), so this ports the pieces main lacked: - GeminiStreamer: streamGenerateContent?alt=sse -> base64 PCM chunks (24 kHz mono int16), reusing main's DEFAULT_GEMINI_TTS_* constants. - XAIStreamer: WebSocket wss://api.x.ai/v1/tts -> binary PCM frames, async->sync bridged via the _collect_async test seam. - tts.streaming.provider config knob: pin one streamer, or 'auto' to walk the priority list elevenlabs -> gemini -> openai -> xai. Unset keeps the never-swap-the-user's-voice default. - docs/streaming-tts.md: architecture, capability matrix, how to add a provider. - Unit tests for the knob, SSE parsing, and the WS bridge; key-gated E2E tests (skipped without credentials). Refs: #47588 --- docs/streaming-tts.md | 93 +++++++++++ tests/tools/test_tts_streaming.py | 105 ++++++++++++ tests/tools/test_tts_streaming_e2e.py | 109 +++++++++++++ tools/tts_streaming.py | 219 ++++++++++++++++++++++++-- 4 files changed, 515 insertions(+), 11 deletions(-) create mode 100644 docs/streaming-tts.md create mode 100644 tests/tools/test_tts_streaming_e2e.py diff --git a/docs/streaming-tts.md b/docs/streaming-tts.md new file mode 100644 index 000000000000..297a524d07c7 --- /dev/null +++ b/docs/streaming-tts.md @@ -0,0 +1,93 @@ +# Streaming TTS + +Hermes can stream TTS audio as it arrives from the provider, instead of waiting +for the full audio before playing. This is used by voice mode (CLI/TUI live +conversation), the dashboard speak-stream WebSocket, and — via the gateway +`StreamingTTSConsumer` — any platform adapter that opts into streaming audio. +Voice replies start speaking after the first clause instead of after full +generation + synthesis. + +## Architecture + +The streaming pipeline has four parts: + +1. **Producer** — the LLM emits text deltas as it generates a response +2. **Sentence chunker** — `tools.tts_streaming.SentenceChunker` accumulates + deltas, strips `` blocks (even split across deltas), and flushes + complete sentences +3. **TTS provider** — a registered `StreamingTTSProvider` turns each sentence + into raw PCM chunks (int16 mono at the provider's declared `sample_rate`) +4. **Audio sink** — `sounddevice.OutputStream` for local playback + (`tools.tts_tool.stream_tts_to_speaker`), or a gateway platform adapter's + `write_streaming_tts` seam (`gateway/streaming_tts_consumer.py`) + +Providers with no chunked API still get per-*sentence* playback via the proven +sync `text_to_speech_tool` path, so edge (the default) is conversational too. +All spoken text is cleaned by `tools.tts_text_normalize.prepare_spoken_text` +(one cleaner, all paths). + +## How to pick a provider + +By default the dispatcher streams with the provider you already configured +(`tts.provider`) when that provider has a chunked API — it never silently +swaps your voice for a different provider just to get streaming. + +To override, set `tts.streaming.provider` in your `config.yaml`: + +- a provider name (`elevenlabs`, `gemini`, `openai`, `xai`) pins that streamer +- `auto` walks the priority list `elevenlabs → gemini → openai → xai` and uses + the first one whose credentials resolve — an explicit opt-in to "best + chunked voice available" + +```yaml +tts: + provider: gemini + streaming: + provider: gemini # or "auto" + gemini: + model: gemini-2.5-flash-preview-tts + voice: Kore +``` + +## Capability matrix + +| Provider | Transport | Chunked PCM | Credentials | +|-------------|---------------------------------------|-------------|-------------| +| elevenlabs | chunked HTTP (`pcm_24000`) | yes | `ELEVENLABS_API_KEY` / `tts.elevenlabs` | +| openai | chunked HTTP (`with_streaming_response`, `pcm`) | yes | `tts.openai.api_key` → env → managed gateway | +| gemini | SSE (`streamGenerateContent?alt=sse`) | yes | `GEMINI_API_KEY` / `GOOGLE_API_KEY` | +| xai | WebSocket (`wss://api.x.ai/v1/tts`) | yes | xAI OAuth or `XAI_API_KEY` | +| edge, piper, kitten, neutts, mistral, minimax, deepinfra, … | — | no (per-sentence sync fallback) | as usual | + +All credential lookups go through `resolve_provider_secret()` +(config > env/.env > credential pool) — never bare env reads. Streamed bodies +are capped at 16 MiB per sentence, mirroring the sync providers' bounded +upstream-body invariant. + +## Adding a new streaming provider + +1. Subclass `StreamingTTSProvider` in `tools/tts_streaming.py` +2. Set `sample_rate` (and `channels` / `sample_width` if not int16 mono) +3. Implement `available()` (a pure probe — never install anything) and + `stream(self, text) -> Iterator[bytes]` yielding raw PCM chunks +4. Decorate with `@register("yourname")` +5. Add tests in `tests/tools/test_tts_streaming.py` + +The ABC enforces the contract; the registry makes the provider discoverable; +the dispatcher (`stream_tts_to_speaker`) and the gateway consumer handle the +sentence buffer, stop events, and audio sink for free. + +## Gateway streaming (platform adapters) + +`gateway/streaming_tts_consumer.py` bridges agent deltas to an adapter's +streaming-audio seam. Adapters opt in by overriding, on +`BasePlatformAdapter`: + +- `supports_streaming_tts(chat_id, audio_format) -> bool` +- `begin_streaming_tts / write_streaming_tts / finish_streaming_tts / + abort_streaming_tts` + +All default to unsupported/no-op, so existing adapters are untouched. When a +turn's streaming audio completes, the whole-file auto-TTS reply for that turn +is suppressed (no double playback); when streaming fails before any audio was +audible, the gateway falls back to the legacy whole-file voice reply. diff --git a/tests/tools/test_tts_streaming.py b/tests/tools/test_tts_streaming.py index c12e23e45ec5..61f1849cddaa 100644 --- a/tests/tools/test_tts_streaming.py +++ b/tests/tools/test_tts_streaming.py @@ -342,3 +342,108 @@ def test_display_callback_fires_without_audio(monkeypatch): assert seen, "display_callback should fire even on the sync path" assert done.is_set() + + +# ── tts.streaming.provider config knob (salvaged from PR #47588) ───────── + + +def test_streaming_provider_knob_pins_streamer(monkeypatch): + _register_fake(monkeypatch, "faketts") + prov = ts.resolve_streaming_provider( + {"provider": "edge", "streaming": {"provider": "faketts"}} + ) + assert isinstance(prov, ts.StreamingTTSProvider) + + +def test_streaming_provider_knob_pin_unusable_returns_none(monkeypatch): + _register_fake(monkeypatch, "faketts", available=False) + # A pinned-but-unusable streamer must NOT fall through to another + # provider — the user asked for that one specifically. + _register_fake(monkeypatch, "otherstreamer") + assert ts.resolve_streaming_provider( + {"provider": "otherstreamer", "streaming": {"provider": "faketts"}} + ) is None + + +def test_streaming_provider_auto_walks_priority(monkeypatch): + # elevenlabs unavailable, gemini available → auto resolves gemini. + _register_fake(monkeypatch, "elevenlabs", available=False) + fake_gemini = _register_fake(monkeypatch, "gemini") + _register_fake(monkeypatch, "openai") + prov = ts.resolve_streaming_provider( + {"provider": "edge", "streaming": {"provider": "auto"}} + ) + assert isinstance(prov, fake_gemini) + + +def test_streaming_provider_auto_none_when_nothing_usable(monkeypatch): + for name in ts._PROVIDER_PRIORITY: + _register_fake(monkeypatch, name, available=False) + assert ts.resolve_streaming_provider( + {"provider": "edge", "streaming": {"provider": "auto"}} + ) is None + + +# ── Gemini SSE parsing ──────────────────────────────────────────────────── + + +def test_gemini_streamer_decodes_sse_pcm_chunks(monkeypatch): + import base64 + import json as _json + import sys + import types + + pcm1, pcm2 = b"\x01\x00" * 40, b"\x02\x00" * 40 + + def _event(pcm): + return "data: " + _json.dumps({ + "candidates": [{"content": {"parts": [ + {"inlineData": {"data": base64.b64encode(pcm).decode()}} + ]}}] + }) + + class _Resp: + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def raise_for_status(self): + pass + + def iter_lines(self, decode_unicode=True): + yield _event(pcm1) + yield "" # SSE separator + yield ": heartbeat" # comment line + yield _event(pcm2) + + captured = {} + + def _post(url, **kwargs): + captured["url"] = url + captured["params"] = kwargs.get("params") + captured["stream"] = kwargs.get("stream") + return _Resp() + + fake_requests = types.ModuleType("requests") + fake_requests.post = _post + monkeypatch.setitem(sys.modules, "requests", fake_requests) + monkeypatch.setattr(ts, "get_env_value", lambda k, *a: "g-key" if k in ("GEMINI_API_KEY", "GOOGLE_API_KEY") else None) + + streamer = ts.GeminiStreamer({}, {"voice": "Kore"}) + assert list(streamer.stream("Hello there.")) == [pcm1, pcm2] + assert captured["params"]["alt"] == "sse" + assert captured["params"]["key"] == "g-key" + assert captured["stream"] is True, "Gemini SSE must use a bounded streamed body" + + +# ── xAI WebSocket bridge ───────────────────────────────────────────────── + + +def test_xai_streamer_yields_collected_frames(monkeypatch): + frames = [b"\x01\x00" * 30, b"\x02\x00" * 30] + streamer = ts.XAIStreamer.__new__(ts.XAIStreamer) + streamer.tts_config, streamer.section = {}, {} + monkeypatch.setattr(streamer, "_collect_async", lambda text: list(frames)) + assert list(streamer.stream("A sentence.")) == frames diff --git a/tests/tools/test_tts_streaming_e2e.py b/tests/tools/test_tts_streaming_e2e.py new file mode 100644 index 000000000000..88152e1f00f1 --- /dev/null +++ b/tests/tools/test_tts_streaming_e2e.py @@ -0,0 +1,109 @@ +"""End-to-end tests for streaming TTS providers. Gated on real API keys. + +Salvaged from PR #47588 (@Cdddo) and adapted to the current provider ABC +(``StreamingTTSProvider(tts_config, section)``). These tests are SKIPPED by +default — they only run when the relevant credential is present. They're +useful for catching provider-API drift and verifying the integration +end-to-end, but they shouldn't run in CI without secrets configured. +""" +from __future__ import annotations + +import os + +import pytest + + +def _has_xai_creds() -> bool: + # Plain os.environ, matching the other gates here: the repo's test + # conftest scrubs dotenv/pooled secrets, so gating on the richer + # resolver would collect a test whose runtime credentials are gone. + return bool(os.environ.get("XAI_API_KEY")) + + +# --- ElevenLabs --- + + +@pytest.mark.skipif( + not os.environ.get("ELEVENLABS_API_KEY"), + reason="ELEVENLABS_API_KEY not set", +) +def test_elevenlabs_streaming_real(): + """Generate audio from the real ElevenLabs API and verify non-empty chunks.""" + from tools.tts_streaming import ElevenLabsStreamer + + provider = ElevenLabsStreamer({}, {}) + chunks = list(provider.stream("Hello world, this is a test.")) + assert len(chunks) > 0 + total_bytes = sum(len(c) for c in chunks) + # 1s of PCM at 24kHz mono int16 = 48000 bytes; expect at least 1k for real audio + assert total_bytes > 1000 + + +# --- Gemini --- + + +@pytest.mark.skipif( + not (os.environ.get("GEMINI_API_KEY") or os.environ.get("GOOGLE_API_KEY")), + reason="GEMINI_API_KEY/GOOGLE_API_KEY not set", +) +def test_gemini_streaming_real(): + """Generate audio from the real Gemini SSE API and verify non-empty chunks.""" + from tools.tts_streaming import GeminiStreamer + + provider = GeminiStreamer({}, {}) + chunks = list(provider.stream("Hola, esto es una prueba.")) + assert len(chunks) > 0 + total_bytes = sum(len(c) for c in chunks) + assert total_bytes > 1000 + + +# --- OpenAI --- + + +@pytest.mark.skipif( + not os.environ.get("OPENAI_API_KEY"), + reason="OPENAI_API_KEY not set", +) +def test_openai_streaming_real(): + """Generate audio from the real OpenAI API and verify non-empty chunks.""" + from tools.tts_streaming import OpenAIStreamer + + provider = OpenAIStreamer({}, {}) + chunks = list(provider.stream("Hello, this is a test.")) + assert len(chunks) > 0 + total_bytes = sum(len(c) for c in chunks) + assert total_bytes > 1000 + + +# --- xAI --- + + +@pytest.mark.skipif(not _has_xai_creds(), reason="no xAI credentials") +def test_xai_streaming_real(): + """Generate audio from the real xAI WebSocket API and verify non-empty frames.""" + from tools.tts_streaming import XAIStreamer + + provider = XAIStreamer({}, {}) + chunks = list(provider.stream("Hello, this is a test.")) + assert len(chunks) > 0 + total_bytes = sum(len(c) for c in chunks) + assert total_bytes > 1000 + + +# --- Resolver integration (no network; requires at least one key) --- + + +@pytest.mark.skipif( + not ( + os.environ.get("ELEVENLABS_API_KEY") + or os.environ.get("GEMINI_API_KEY") + or os.environ.get("GOOGLE_API_KEY") + or os.environ.get("OPENAI_API_KEY") + ), + reason="no streaming-capable TTS key set", +) +def test_auto_resolver_finds_a_real_provider(): + from tools.tts_streaming import StreamingTTSProvider, resolve_streaming_provider + + provider = resolve_streaming_provider({"streaming": {"provider": "auto"}}) + assert isinstance(provider, StreamingTTSProvider) diff --git a/tools/tts_streaming.py b/tools/tts_streaming.py index 9d772da71d50..79a9a2a85290 100644 --- a/tools/tts_streaming.py +++ b/tools/tts_streaming.py @@ -139,17 +139,8 @@ def register(name: str) -> Callable[[type[StreamingTTSProvider]], type[Streaming return _wrap -def resolve_streaming_provider( - tts_config: Dict, - preferred: Optional[str] = None, -) -> Optional[StreamingTTSProvider]: - """Return a ready streamer for the *configured* provider, else ``None``. - - ``None`` means "no chunked API for this provider" — the dispatcher then - speaks per-sentence via the sync path, preserving the user's chosen voice. - We never silently swap to a different provider just to get streaming. - """ - name = (preferred or _get_provider(tts_config)).lower().strip() +def _try_instantiate(name: str, tts_config: Dict) -> Optional[StreamingTTSProvider]: + """Construct the registered streamer *name* if it's usable, else None.""" cls = _REGISTRY.get(name) if cls is None or not cls.available(): return None @@ -160,6 +151,47 @@ def resolve_streaming_provider( return None +# Fallback priority for ``tts.streaming.provider: auto`` — best chunked +# latency/quality first. Deliberately hard-coded (a UX decision, not a +# config knob); edge is absent because it has no chunked-PCM API — the +# dispatcher's per-sentence sync path keeps it conversational instead. +_PROVIDER_PRIORITY: List[str] = ["elevenlabs", "gemini", "openai", "xai"] + + +def resolve_streaming_provider( + tts_config: Dict, + preferred: Optional[str] = None, +) -> Optional[StreamingTTSProvider]: + """Return a ready streamer for the *configured* provider, else ``None``. + + Resolution order: + + 1. ``tts.streaming.provider`` (config knob) when set: + * a provider name pins that exact streamer (or ``None`` if unusable); + * ``auto`` walks the priority list (``elevenlabs → gemini → openai + → xai``) and returns the first usable streamer — an explicit + opt-in to "give me the best chunked voice available". + 2. Otherwise the *configured* TTS provider (or ``preferred`` override). + ``None`` means "no chunked API for this provider" — the dispatcher + then speaks per-sentence via the sync path, preserving the user's + chosen voice. We never silently swap to a different provider just + to get streaming. + """ + streaming_cfg = tts_config.get("streaming") or {} + pinned = str(streaming_cfg.get("provider") or "").lower().strip() + if pinned == "auto": + for name in _PROVIDER_PRIORITY: + inst = _try_instantiate(name, tts_config) + if inst is not None: + return inst + return None + if pinned: + return _try_instantiate(pinned, tts_config) + + name = (preferred or _get_provider(tts_config)).lower().strip() + return _try_instantiate(name, tts_config) + + # --------------------------------------------------------------------------- # Providers # --------------------------------------------------------------------------- @@ -238,3 +270,168 @@ class OpenAIStreamer(StreamingTTSProvider): response_format="pcm", ) as response: yield from response.iter_bytes() + + +@register("gemini") +class GeminiStreamer(StreamingTTSProvider): + """Gemini ``streamGenerateContent?alt=sse`` → base64 PCM chunks (24 kHz). + + Each SSE event carries one base64-encoded PCM chunk under + ``candidates[0].content.parts[*].inlineData.data``; we decode and yield + each one as it arrives. + """ + + sample_rate = 24000 + + @staticmethod + def available() -> bool: + return bool( + get_env_value("GEMINI_API_KEY") or get_env_value("GOOGLE_API_KEY") + ) + + def stream(self, text: str) -> Iterator[bytes]: + import base64 + import json as _json + + import requests + + from tools.tts_tool import ( + DEFAULT_GEMINI_TTS_BASE_URL, + DEFAULT_GEMINI_TTS_MODEL, + DEFAULT_GEMINI_TTS_VOICE, + ) + + api_key = ( + get_env_value("GEMINI_API_KEY") or get_env_value("GOOGLE_API_KEY") + ) + model = str(self.section.get("model", DEFAULT_GEMINI_TTS_MODEL)).strip() or DEFAULT_GEMINI_TTS_MODEL + voice = str(self.section.get("voice", DEFAULT_GEMINI_TTS_VOICE)).strip() or DEFAULT_GEMINI_TTS_VOICE + base_url = str( + self.section.get("base_url") + or get_env_value("GEMINI_BASE_URL") + or DEFAULT_GEMINI_TTS_BASE_URL + ).strip().rstrip("/") + + payload = { + "contents": [{"parts": [{"text": text}]}], + "generationConfig": { + "responseModalities": ["AUDIO"], + "speechConfig": { + "voiceConfig": { + "prebuiltVoiceConfig": {"voiceName": voice}, + }, + }, + }, + } + # ``?alt=sse`` flips the response from a single JSON blob to an SSE + # feed of base64 PCM chunks — the whole point of this provider. + url = f"{base_url}/models/{model}:streamGenerateContent" + + def _sse_chunks() -> Iterator[bytes]: + with requests.post( + url, + params={"alt": "sse", "key": api_key}, + json=payload, + timeout=60, + stream=True, + ) as response: + response.raise_for_status() + for line in response.iter_lines(decode_unicode=True): + if not line or not line.startswith("data: "): + continue + try: + event = _json.loads(line[len("data: "):]) + parts = event["candidates"][0]["content"]["parts"] + except (ValueError, KeyError, IndexError, TypeError): + continue + for part in parts: + inline = part.get("inlineData") or part.get("inline_data") or {} + b64 = inline.get("data", "") + if not b64: + continue + try: + yield base64.b64decode(b64) + except (ValueError, TypeError) as exc: + logger.warning("Gemini SSE: bad base64 audio: %s", exc) + + yield from _sse_chunks() + + +@register("xai") +class XAIStreamer(StreamingTTSProvider): + """xAI WebSocket TTS → binary PCM frames (24 kHz mono int16). + + xAI's chunked TTS API is WebSocket-only (``wss://api.x.ai/v1/tts``). + The async WS loop is bridged to the sync iterator contract via + ``_collect_async`` — the seam unit tests monkeypatch. + """ + + sample_rate = 24000 + + @staticmethod + def available() -> bool: + return bool(get_env_value("XAI_API_KEY")) + + def stream(self, text: str) -> Iterator[bytes]: + yield from self._collect_async(text) + + # -- async→sync bridge (test seam) ------------------------------------ + + def _collect_async(self, text: str) -> List[bytes]: + import asyncio + + return asyncio.run(self._drain_async(text)) + + async def _drain_async(self, text: str) -> List[bytes]: + frames: List[bytes] = [] + async for frame in self._async_frames(text): + frames.append(frame) + return frames + + async def _async_frames(self, text: str): + import json as _json + + import websockets + + from tools.tts_tool import DEFAULT_XAI_VOICE_ID + + api_key = (get_env_value("XAI_API_KEY") or "").strip() + if not api_key: + raise RuntimeError("XAI_API_KEY not set; cannot use xAI streaming TTS") + voice = str(self.section.get("voice_id", DEFAULT_XAI_VOICE_ID)).strip() or DEFAULT_XAI_VOICE_ID + ws_url = str( + self.section.get("streaming_url") or "wss://api.x.ai/v1/tts" + ).strip() + + async with websockets.connect( + ws_url, extra_headers={"Authorization": f"Bearer {api_key}"} + ) as ws: + await ws.send(_json.dumps({ + "text": text, + "voice_id": voice, + "response_format": "pcm", + })) + try: + while True: + message = await ws.recv() + if isinstance(message, (bytes, bytearray, memoryview)): + yield bytes(message) + continue + try: + envelope = _json.loads(message) + except (ValueError, TypeError): + if message == "done": + return + continue + etype = envelope.get("type") + if etype == "done": + return + if etype == "error": + logger.warning("xAI WS error envelope: %s", + envelope.get("error") or envelope.get("message") or envelope) + return + except Exception as exc: + if exc.__class__.__name__ == "ConnectionClosed": + return + logger.warning("xAI WS receive failed: %s", exc) + return