ouyangbo/hermes-agent
1
0
Fork 0
Code Issues Pull Requests Actions 9 Packages Projects Releases Wiki Activity

feat(tts): provider-agnostic streaming core with a shared sentence chunker

Browse Source 
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, <think> blocks stripped even when split across deltas, short
fragments merged forward so they never stall as tiny clips.
This commit is contained in:
Brooklyn Nicholson
2026-07-22 17:46:55 -05:00
parent 87088d1821
commit 592effcb2a
2 changed files with 421 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/tools/test_tts_streaming.py
+228
View File
@@ -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("<think>secret reason") == []
assert c.feed("ing</think>The 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()
tools/tts_streaming.py
+193
View File
@@ -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.<name>.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"<think[\s>].*?</think>", 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
``<think>`` 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 "<think" in self.buf and "</think>" 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()