From 592effcb2a42dbdf0b51ac9fef7176066ba86ac6 Mon Sep 17 00:00:00 2001 From: Brooklyn Nicholson Date: Wed, 22 Jul 2026 17:46:55 -0500 Subject: [PATCH] feat(tts): provider-agnostic streaming core with a shared sentence chunker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit StreamingTTSProvider ABC + registry + resolve_streaming_provider() — ElevenLabs (pcm_24000) and OpenAI (response_format=pcm) stream chunked PCM; every other provider keeps its configured voice and falls back to per-sentence sync synthesis. SentenceChunker is the one incremental cutter every surface shares: sentence-boundary cuts on the delta stream, blocks stripped even when split across deltas, short fragments merged forward so they never stall as tiny clips. --- tests/tools/test_tts_streaming.py | 228 ++++++++++++++++++++++++++++++ tools/tts_streaming.py | 193 +++++++++++++++++++++++++ 2 files changed, 421 insertions(+) create mode 100644 tests/tools/test_tts_streaming.py create mode 100644 tools/tts_streaming.py diff --git a/tests/tools/test_tts_streaming.py b/tests/tools/test_tts_streaming.py new file mode 100644 index 000000000000..77241ea485b7 --- /dev/null +++ b/tests/tools/test_tts_streaming.py @@ -0,0 +1,228 @@ +"""Tests for the provider-agnostic streaming TTS backend (tools.tts_streaming) +and its dispatch through tools.tts_tool.stream_tts_to_speaker. + +No live audio or network: the ElevenLabs/OpenAI SDKs, sounddevice, and the sync +synth path are all mocked. Covers the registry/resolver, provider availability, +the chunked-streamer playback path, and the universal per-sentence sync fallback. +""" + +import queue +import threading +from unittest.mock import MagicMock, patch + +import pytest + +import tools.tts_streaming as ts + +pytest.importorskip("numpy") + + +# ── SentenceChunker ────────────────────────────────────────────────────── + + +class TestSentenceChunker: + def test_cuts_sentence_the_moment_its_boundary_arrives(self): + c = ts.SentenceChunker() + assert c.feed("This is the first full") == [] + assert c.feed(" sentence of it all. And") == ["This is the first full sentence of it all. "] + assert c.flush() == ["And"] + + def test_short_fragment_rides_with_the_next_sentence(self): + c = ts.SentenceChunker() + # "Ha! " alone is under min_len — it must not become its own clip. + assert c.feed("Ha! ") == [] + assert c.feed("That was a good one, honestly. ") == [ + "Ha! That was a good one, honestly. " + ] + + def test_think_blocks_are_stripped_even_across_deltas(self): + c = ts.SentenceChunker() + assert c.feed("secret reason") == [] + assert c.feed("ingThe actual spoken answer. ") == ["The actual spoken answer. "] + + def test_flush_drains_the_tail(self): + c = ts.SentenceChunker() + c.feed("no boundary here") + assert c.flush() == ["no boundary here"] + assert c.flush() == [] + + def test_paragraph_break_is_a_boundary(self): + c = ts.SentenceChunker() + assert c.feed("A paragraph without punctuation\n\nnext one") == [ + "A paragraph without punctuation\n\n" + ] + + +# ── Registry + resolver ────────────────────────────────────────────────── + + +def _register_fake(monkeypatch, name, available=True, chunks=(b"\x00\x00",)): + class _Fake(ts.StreamingTTSProvider): + sample_rate = 24000 + + @staticmethod + def available(): + return available + + def stream(self, text): + yield from chunks + + monkeypatch.setitem(ts._REGISTRY, name, _Fake) + return _Fake + + +def test_resolve_returns_configured_streamer(monkeypatch): + _register_fake(monkeypatch, "faketts") + prov = ts.resolve_streaming_provider({"provider": "faketts"}) + assert isinstance(prov, ts.StreamingTTSProvider) + + +def test_resolve_none_for_unregistered_provider(monkeypatch): + # edge is a sync provider — not registered — so the dispatcher keeps its voice. + assert ts.resolve_streaming_provider({"provider": "edge"}) is None + + +def test_resolve_none_when_provider_unavailable(monkeypatch): + _register_fake(monkeypatch, "faketts", available=False) + assert ts.resolve_streaming_provider({"provider": "faketts"}) is None + + +def test_resolve_honors_preferred_override(monkeypatch): + _register_fake(monkeypatch, "faketts") + prov = ts.resolve_streaming_provider({"provider": "edge"}, preferred="faketts") + assert isinstance(prov, ts.StreamingTTSProvider) + + +def test_never_swaps_provider_for_streaming(monkeypatch): + # A registered streamer must NOT be substituted when the user picked another + # (non-streaming) provider — that would silently change their voice. + _register_fake(monkeypatch, "elevenlabs") + assert ts.resolve_streaming_provider({"provider": "edge"}) is None + + +# ── Built-in provider availability ─────────────────────────────────────── + + +def test_elevenlabs_available_reflects_key(monkeypatch): + monkeypatch.setattr(ts, "get_env_value", lambda k, *a: "key" if k == "ELEVENLABS_API_KEY" else None) + assert ts.ElevenLabsStreamer.available() is True + monkeypatch.setattr(ts, "get_env_value", lambda k, *a: None) + assert ts.ElevenLabsStreamer.available() is False + + +def test_openai_available_reflects_key(monkeypatch): + monkeypatch.setattr(ts, "get_env_value", lambda k, *a: "key" if k == "OPENAI_API_KEY" else None) + assert ts.OpenAIStreamer.available() is True + + +# ── Dispatch: chunked streamer path ────────────────────────────────────── + + +def _drain_queue(sentences): + q = queue.Queue() + for s in sentences: + q.put(s) + q.put(None) + return q + + +def _sd_mock(): + sd = MagicMock() + out = MagicMock() + sd.OutputStream.return_value = out + return sd, out + + +def test_streamer_path_writes_pcm_to_output(monkeypatch): + from tools import tts_tool + + class _Fake(ts.StreamingTTSProvider): + sample_rate = 24000 + + @staticmethod + def available(): + return True + + def stream(self, text): + yield b"\x01\x00" * 50 + yield b"\x02\x00" * 50 + + sd, out = _sd_mock() + q = _drain_queue(["Hello there, this is a full sentence."]) + stop, done = threading.Event(), threading.Event() + + with patch("tools.tts_streaming.resolve_streaming_provider", return_value=_Fake({}, {})), \ + patch.object(tts_tool, "_import_sounddevice", return_value=sd): + tts_tool.stream_tts_to_speaker(q, stop, done) + + assert out.write.called, "expected PCM chunks written to the output stream" + assert done.is_set() + + +def test_stop_event_aborts_streaming(monkeypatch): + from tools import tts_tool + + class _Fake(ts.StreamingTTSProvider): + sample_rate = 24000 + + @staticmethod + def available(): + return True + + def stream(self, text): + for _ in range(1000): + yield b"\x00\x00" * 50 + + sd, out = _sd_mock() + stop, done = threading.Event(), threading.Event() + stop.set() # pre-set: no audio should be written + q = _drain_queue(["A complete sentence here."]) + + with patch("tools.tts_streaming.resolve_streaming_provider", return_value=_Fake({}, {})), \ + patch.object(tts_tool, "_import_sounddevice", return_value=sd): + tts_tool.stream_tts_to_speaker(q, stop, done) + + assert not out.write.called + assert done.is_set() + + +# ── Dispatch: universal per-sentence sync fallback ─────────────────────── + + +def test_sync_fallback_speaks_each_sentence(monkeypatch): + from tools import tts_tool + + spoken = [] + monkeypatch.setattr(tts_tool, "text_to_speech_tool", + lambda text, output_path: spoken.append(text)) + played = [] + fake_vm = MagicMock() + fake_vm.play_audio_file.side_effect = lambda p: played.append(p) + monkeypatch.setitem(__import__("sys").modules, "tools.voice_mode", fake_vm) + monkeypatch.setattr("os.path.getsize", lambda p: 100) + monkeypatch.setattr("os.path.isfile", lambda p: True) + + q = _drain_queue(["First full sentence here. ", "Second full sentence here. "]) + stop, done = threading.Event(), threading.Event() + + with patch("tools.tts_streaming.resolve_streaming_provider", return_value=None): + tts_tool.stream_tts_to_speaker(q, stop, done) + + assert len(spoken) == 2, f"expected both sentences synthesized, got {spoken}" + assert len(played) == 2 + assert done.is_set() + + +def test_display_callback_fires_without_audio(monkeypatch): + from tools import tts_tool + + seen = [] + monkeypatch.setattr(tts_tool, "text_to_speech_tool", lambda text, output_path: None) + q = _drain_queue(["A sentence to display aloud."]) + stop, done = threading.Event(), threading.Event() + + with patch("tools.tts_streaming.resolve_streaming_provider", return_value=None): + tts_tool.stream_tts_to_speaker(q, stop, done, display_callback=seen.append) + + assert seen, "display_callback should fire even on the sync path" + assert done.is_set() diff --git a/tools/tts_streaming.py b/tools/tts_streaming.py new file mode 100644 index 000000000000..8d773d954f59 --- /dev/null +++ b/tools/tts_streaming.py @@ -0,0 +1,193 @@ +"""Provider-agnostic streaming TTS: sentence text → int16 PCM chunk iterator. + +The keystone of Hermes' conversational voice UX. `stream_tts_to_speaker` +(``tools.tts_tool``) owns the sentence buffer, sounddevice output, and +stop/queue protocol; this module owns the *provider* half — turning one +sentence into audio the moment it's ready, so playback starts on sentence one +instead of after the whole reply. + +Two provider shapes, one contract (int16 mono PCM at ``sample_rate``): + +* **True streamers** (`StreamingTTSProvider.stream`) — chunked APIs + (ElevenLabs pcm_24000, OpenAI pcm, …) that yield audio as it synthesizes. + Lowest time-to-first-audio. +* **Everyone else** — providers with no chunked API still get per-*sentence* + playback via the proven sync `text_to_speech_tool` path (handled by the + dispatcher, not here), so edge (the default) is conversational too. + +Adding a streamer is `@register("name")` on a `StreamingTTSProvider` subclass; +the dispatcher, config gate (`tts..streaming`), and resolver come free. +""" + +from __future__ import annotations + +import logging +import re +from abc import ABC, abstractmethod +from typing import Callable, Dict, Iterator, List, Optional + +from tools.tts_tool import _get_provider, get_env_value + +logger = logging.getLogger(__name__) + +# Sentence boundary: after .!? followed by whitespace, or a blank line. +SENTENCE_BOUNDARY_RE = re.compile(r"(?<=[.!?])(?:\s|\n)|(?:\n\n)") +_THINK_BLOCK_RE = re.compile(r"].*?", flags=re.DOTALL) + + +class SentenceChunker: + """Incremental sentence cutter for LLM token deltas. + + Shared by the speaker pipeline (`stream_tts_to_speaker`) and the + speak-stream WebSocket so every surface cuts speech identically. Strips + ```` blocks (even split across deltas) and merges fragments shorter + than *min_len* into the following sentence, so "Ha!" rides along with the + sentence after it instead of stalling as a tiny clip. + """ + + def __init__(self, min_len: int = 20): + self.min_len = min_len + self.buf = "" + + def feed(self, delta: str) -> List[str]: + """Absorb *delta*; return every complete sentence now ready to speak.""" + self.buf = _THINK_BLOCK_RE.sub("", self.buf + delta) + if "" not in self.buf: + return [] # open think tag — the closing tag may arrive next delta + out: List[str] = [] + start = 0 # skip boundaries that would leave the head too short + while m := SENTENCE_BOUNDARY_RE.search(self.buf, start): + head = self.buf[: m.end()] + if len(head.strip()) < self.min_len: + start = m.end() + continue + out.append(head) + self.buf = self.buf[m.end():] + start = 0 + return out + + def flush(self) -> List[str]: + """Drain the tail (end-of-text or long-idle flush).""" + tail = _THINK_BLOCK_RE.sub("", self.buf).strip() + self.buf = "" + return [tail] if tail else [] + + +# --------------------------------------------------------------------------- +# ABC + registry +# --------------------------------------------------------------------------- + +class StreamingTTSProvider(ABC): + """Yields raw int16, little-endian, mono PCM chunks at ``sample_rate``.""" + + sample_rate: int = 24000 + channels: int = 1 + sample_width: int = 2 # bytes/sample (int16) + + def __init__(self, tts_config: Dict, section: Dict): + self.tts_config = tts_config + self.section = section + + @staticmethod + @abstractmethod + def available() -> bool: + """True when this provider's credentials/SDK are usable right now.""" + + @abstractmethod + def stream(self, text: str) -> Iterator[bytes]: + """Yield PCM chunks for ``text``. Raise on failure (caller logs).""" + + +_REGISTRY: Dict[str, type[StreamingTTSProvider]] = {} + + +def register(name: str) -> Callable[[type[StreamingTTSProvider]], type[StreamingTTSProvider]]: + def _wrap(cls: type[StreamingTTSProvider]) -> type[StreamingTTSProvider]: + _REGISTRY[name] = cls + return cls + + 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() + cls = _REGISTRY.get(name) + if cls is None or not cls.available(): + return None + try: + return cls(tts_config, tts_config.get(name) or {}) + except Exception as exc: # pragma: no cover - defensive + logger.debug("streaming provider %s init failed: %s", name, exc) + return None + + +# --------------------------------------------------------------------------- +# Providers +# --------------------------------------------------------------------------- + +@register("elevenlabs") +class ElevenLabsStreamer(StreamingTTSProvider): + """ElevenLabs chunked HTTP → pcm_24000 (the original reference path).""" + + sample_rate = 24000 + + @staticmethod + def available() -> bool: + return bool(get_env_value("ELEVENLABS_API_KEY")) + + def stream(self, text: str) -> Iterator[bytes]: + from tools.tts_tool import ( + DEFAULT_ELEVENLABS_STREAMING_MODEL_ID, + DEFAULT_ELEVENLABS_VOICE_ID, + _import_elevenlabs, + ) + + client = _import_elevenlabs()(api_key=get_env_value("ELEVENLABS_API_KEY")) + voice_id = self.section.get("voice_id", DEFAULT_ELEVENLABS_VOICE_ID) + model_id = self.section.get( + "streaming_model_id", + self.section.get("model_id", DEFAULT_ELEVENLABS_STREAMING_MODEL_ID), + ) + yield from client.text_to_speech.convert( + text=text, + voice_id=voice_id, + model_id=model_id, + output_format="pcm_24000", + ) + + +@register("openai") +class OpenAIStreamer(StreamingTTSProvider): + """OpenAI speech with ``response_format=pcm`` (24 kHz mono int16).""" + + sample_rate = 24000 + + @staticmethod + def available() -> bool: + return bool(get_env_value("OPENAI_API_KEY")) + + def stream(self, text: str) -> Iterator[bytes]: + from openai import OpenAI + + client = OpenAI( + api_key=get_env_value("OPENAI_API_KEY"), + base_url=get_env_value("OPENAI_BASE_URL") or None, + ) + model = self.section.get("model", "gpt-4o-mini-tts") + voice = self.section.get("voice", "alloy") + with client.audio.speech.with_streaming_response.create( + model=model, + voice=voice, + input=text, + response_format="pcm", + ) as response: + yield from response.iter_bytes()