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

refactor(tools): split tts speaker playback worker into per-sentence phase helpers

Browse Source
This commit is contained in:
Teknium
2026-09-02 22:43:19 -07:00
parent e2793ca72b
commit aa97ee5ae2
1 changed files with 101 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tools/tts_tool_speaker.py
+101 -141
View File
@@ -1,20 +1,12 @@
"""Speaker-side streaming pipeline for ``tools.tts_tool.stream_tts_to_speaker``.
Turns a queue of LLM text deltas into audio the moment each sentence is
complete. Two paths share the sentence cutter (``tools.tts_streaming``):
* :class:`_StreamerPlayback` — a registered chunked streamer (ElevenLabs,
OpenAI, …). Every sentence gets a prefetch thread that fires the HTTP
request immediately and buffers PCM into a per-sentence queue; one playback
worker drains those queues in FIFO order through a sounddevice OutputStream
(or a temp WAV + system player when PortAudio is unavailable).
* :class:`_SyncSentencePipeline` — every other provider (edge, piper,
plugins). Per-sentence ``text_to_speech_tool`` synthesis on a single-thread
executor, overlapped with playback so sentence n+1 synthesizes while n plays.
Seams tests monkeypatch on the origin module (``_load_tts_config``,
``_import_sounddevice``, ``text_to_speech_tool``, ``_strip_markdown_for_tts``)
are resolved through :func:`_origin` at call time.
Turns a queue of LLM text deltas into audio the moment each sentence is complete.
Two paths share the sentence cutter (``tools.tts_streaming``): :class:`_StreamerPlayback`
for a registered chunked streamer (prefetch thread per sentence, one FIFO playback worker
through a sounddevice OutputStream or temp WAV + system player) and
:class:`_SyncSentencePipeline` for every other provider (per-sentence ``text_to_speech_tool``
on a single-thread executor, overlapped with playback). Seams tests monkeypatch on the origin
module are resolved through :func:`_origin` at call time.
"""
from __future__ import annotations
@@ -74,9 +66,8 @@ def _play_via_tempfile(audio_iter: Iterable[bytes], stop_evt: threading.Event, s
wf.setframerate(sample_rate)
for aligned in _align_int16_chunks(audio_iter, stop_evt):
wf.writeframes(aligned)
# wave.open() on a file object does NOT close it. On Windows the open
# write handle blocks the player and the unlink below (WinError 32),
# so release it before playback.
# wave.open() on a file object does NOT close it; on Windows the open write
# handle blocks the player and the unlink below (WinError 32).
tmp.close()
from tools.voice_mode import play_audio_file
play_audio_file(tmp_path)
@@ -104,14 +95,11 @@ def _drain_chunks(chunk_queue: "queue.Queue[Optional[bytes]]") -> List[bytes]:
class _SyncSentencePipeline:
"""Overlap per-sentence synthesis with playback for non-streaming providers.
Serial synthesize-then-play added a full synthesis-time of dead air per
sentence — for a local model at real-time-factor ~1, as long silent as
speaking. One single-thread synthesis executor (sentences FIFO; providers
never see concurrent calls) feeds one playback worker through a small
bounded queue: while sentence n plays, n+1 is already synthesizing. The
bound keeps lookahead/temp files small and gives the caller backpressure.
``text_to_speech_tool`` / ``play_audio_file`` are resolved late so tests
Serial synthesize-then-play added a full synthesis-time of dead air per sentence. One
single-thread synthesis executor (sentences FIFO; providers never see concurrent calls)
feeds one playback worker through a small bounded queue: while sentence n plays, n+1 is
already synthesizing. The bound keeps lookahead/temp files small and gives the caller
backpressure. ``text_to_speech_tool`` / ``play_audio_file`` are resolved late so tests
that monkeypatch them keep working.
"""
@@ -126,8 +114,7 @@ class _SyncSentencePipeline:
"""Queue one sentence. Blocks only when the lookahead bound is full."""
if self._stop.is_set():
return
future = self._executor.submit(self._synthesize_to_tmp, cleaned)
self._queue.put((cleaned, future))
self._queue.put((cleaned, self._executor.submit(self._synthesize_to_tmp, cleaned)))
def close(self) -> None:
"""Flush queued sentences in order (skipped if stopped), then join."""
@@ -158,9 +145,7 @@ class _SyncSentencePipeline:
tmp_path = None
try:
tmp_path = future.result()
if (tmp_path and not self._stop.is_set()
and os.path.isfile(tmp_path)
and os.path.getsize(tmp_path) > 0):
if tmp_path and not self._stop.is_set() and os.path.isfile(tmp_path) and os.path.getsize(tmp_path) > 0:
from tools.voice_mode import play_audio_file
play_audio_file(tmp_path)
except Exception as exc:
@@ -172,13 +157,12 @@ class _SyncSentencePipeline:
class _StreamerPlayback:
"""Prefetch + FIFO playback for a chunked :class:`StreamingTTSProvider`.
``speak(text)`` calls ``streamer.stream()`` right away and hands the
iterator to a prefetch thread (at most 3 in flight) that buffers chunks
into a bounded per-sentence queue; the single playback worker plays those
queues in order, so sentence N+1 is already arriving while N plays.
Output goes to a PortAudio stream when one could be opened, otherwise via
temp WAV files. A failing PortAudio write is retried on a reinitialized
stream up to ``_MAX_REINIT`` times before falling back to temp files.
``speak(text)`` calls ``streamer.stream()`` right away and hands the iterator to a
prefetch thread (at most 3 in flight) that buffers chunks into a bounded per-sentence
queue; the single playback worker plays those queues in order, so sentence N+1 is
already arriving while N plays. Output goes to a PortAudio stream when one could be
opened, otherwise via temp WAV files. A failing PortAudio write is retried on a
reinitialized stream up to ``_MAX_REINIT`` times before falling back to temp files.
"""
_MAX_REINIT = 3
@@ -198,19 +182,14 @@ class _StreamerPlayback:
def _create_output_stream(self):
sd = _origin()._import_sounddevice()
stream = sd.OutputStream(
samplerate=self.streamer.sample_rate,
channels=self.streamer.channels,
dtype="int16",
)
stream = sd.OutputStream(samplerate=self.streamer.sample_rate, channels=self.streamer.channels, dtype="int16")
stream.start()
return stream
def _open_output_stream(self):
# On macOS skip sounddevice entirely: PortAudio/CoreAudio init triggers
# a kTCCServiceMediaLibrary permission prompt even though output needs
# no media-library access. None routes every sentence through the
# tempfile -> play_audio_file -> afplay path.
# On macOS skip sounddevice entirely: PortAudio/CoreAudio init triggers a
# kTCCServiceMediaLibrary permission prompt even though output needs no
# media-library access. None routes every sentence through tempfile -> afplay.
if platform.system() == "Darwin":
return None
try:
@@ -223,12 +202,7 @@ class _StreamerPlayback:
def _reinit_output_stream(self):
"""Close the broken PortAudio stream and try to create a fresh one."""
if self.output_stream is not None:
try:
self.output_stream.stop()
self.output_stream.close()
except Exception:
pass
self.close_output_stream()
try:
self.output_stream = self._create_output_stream()
logger.info("TTS: PortAudio output stream reinitialized after error")
@@ -266,18 +240,11 @@ class _StreamerPlayback:
try:
for chunk in audio_iter:
if self.stop_event.is_set():
logger.info(
"TTS CUT: prefetch cancelled (stop_event set "
"mid-sentence) — partial audio only"
)
logger.info("TTS CUT: prefetch cancelled (stop_event set mid-sentence) — partial audio only")
break
chunk_queue.put(chunk, timeout=30.0)
except Exception as exc:
logger.warning(
"TTS CUT: streaming TTS prefetch failed mid-sentence "
"(partial audio only): %s",
exc,
)
logger.warning("TTS CUT: streaming TTS prefetch failed mid-sentence (partial audio only): %s", exc)
finally:
chunk_queue.put(None) # sentinel: no more chunks
self._prefetch_sem.release()
@@ -287,16 +254,60 @@ class _StreamerPlayback:
def _play_sentence_via_tempfile(self, chunk_queue) -> None:
_play_via_tempfile(iter(_drain_chunks(chunk_queue)), self.stop_event, self.streamer.sample_rate)
def _for_each_sentence(self, play: Callable[[queue.Queue], None]) -> None:
"""Feed queued sentences to *play* in order until the end sentinel; stopped sentences are skipped."""
while True:
chunk_queue = self._audio_queue.get()
if chunk_queue is None:
return
if not self.stop_event.is_set():
play(chunk_queue)
def _write_pcm(self, buf: bytes) -> None:
self._current_stream.write(self._np.frombuffer(buf, dtype="<i2").reshape(-1, 1))
def _recover_stream(self) -> bool:
"""Reinit the PortAudio stream after a failed write; False once ``_MAX_REINIT`` is exhausted."""
if self._reinit_count < self._MAX_REINIT:
self._reinit_count += 1
self._current_stream = self._reinit_output_stream()
return self._current_stream is not None
logger.warning(
"TTS: PortAudio reinit exhausted after %d attempts, falling back to tempfile for remaining sentences",
self._MAX_REINIT,
)
self._current_stream = None
return False
def _play_sentence_via_stream(self, chunk_queue) -> None:
"""Write one sentence's PCM to PortAudio; after an unrecoverable write failure the rest is dropped."""
if self._current_stream is None:
self._play_sentence_via_tempfile(chunk_queue)
return
pcm_leftover = b""
while True:
chunk = chunk_queue.get()
if chunk is None or self.stop_event.is_set():
return
buf = pcm_leftover + chunk
aligned_len = len(buf) - (len(buf) % 2)
if aligned_len >= 2:
try:
self._write_pcm(buf[:aligned_len])
except Exception as write_exc:
logger.warning("PortAudio write failed, attempting stream reinit: %s", write_exc)
if not self._recover_stream():
return
try:
self._write_pcm(buf[:aligned_len])
except Exception:
pass
pcm_leftover = buf[aligned_len:] if aligned_len < len(buf) else b""
def _playback_worker(self) -> None:
"""Single consumer: play audio segments from the queue in order."""
if self.output_stream is None:
while True:
chunk_queue = self._audio_queue.get()
if chunk_queue is None:
break
if self.stop_event.is_set():
continue
self._play_sentence_via_tempfile(chunk_queue)
self._for_each_sentence(self._play_sentence_via_tempfile)
return
import numpy as _np
@@ -307,59 +318,12 @@ class _StreamerPlayback:
def mark_audio_output_active(_active):
return None
def write_pcm(stream, buf: bytes) -> None:
stream.write(_np.frombuffer(buf, dtype="<i2").reshape(-1, 1))
self._np = _np
self._reinit_count = 0
self._current_stream = self.output_stream
mark_audio_output_active(True)
try:
reinit_count = 0
current_stream = self.output_stream
while True:
chunk_queue = self._audio_queue.get()
if chunk_queue is None:
break
if self.stop_event.is_set():
continue
if current_stream is None:
self._play_sentence_via_tempfile(chunk_queue)
continue
pcm_leftover = b""
while True:
chunk = chunk_queue.get()
if chunk is None or self.stop_event.is_set():
break
buf = pcm_leftover + chunk
aligned_len = len(buf) - (len(buf) % 2)
if aligned_len >= 2:
try:
write_pcm(current_stream, buf[:aligned_len])
except Exception as write_exc:
logger.warning(
"PortAudio write failed, attempting "
"stream reinit: %s",
write_exc,
)
if reinit_count < self._MAX_REINIT:
reinit_count += 1
current_stream = self._reinit_output_stream()
if current_stream is not None:
try:
write_pcm(current_stream, buf[:aligned_len])
except Exception:
pass
pcm_leftover = buf[aligned_len:] if aligned_len < len(buf) else b""
continue
else:
logger.warning(
"TTS: PortAudio reinit exhausted "
"after %d attempts, falling back "
"to tempfile for remaining "
"sentences",
self._MAX_REINIT,
)
current_stream = None
break
pcm_leftover = buf[aligned_len:] if aligned_len < len(buf) else b""
self._for_each_sentence(self._play_sentence_via_stream)
finally:
mark_audio_output_active(False)
@@ -379,19 +343,17 @@ def stream_tts_to_speaker(
display_callback: Optional[Callable[[str], None]] = None,
provider: Optional[str] = None,
):
"""Consume text deltas from *text_queue*, cut them into sentences, and speak
each one the moment it's ready — the conversational path.
"""Consume text deltas from *text_queue*, cut them into sentences, and speak each one
the moment it's ready — the conversational path.
A registered streaming provider plays chunked PCM for the lowest latency;
every other provider (edge, the default) is spoken per-sentence via the
sync ``text_to_speech_tool`` path, so audio still starts on sentence one.
A registered streaming provider plays chunked PCM for the lowest latency; every other
provider (edge, the default) is spoken per-sentence via the sync ``text_to_speech_tool``
path, so audio still starts on sentence one.
Protocol:
* The producer puts ``str`` deltas onto *text_queue*.
* A ``None`` sentinel signals end-of-text (flush remaining buffer).
* *stop_event* aborts early (barge-in / user interrupt).
* *tts_done_event* is **set** in the ``finally`` block so callers
waiting on it (continuous voice mode) know playback is finished.
Protocol: the producer puts ``str`` deltas onto *text_queue*; a ``None`` sentinel
signals end-of-text (flush remaining buffer); *stop_event* aborts early (barge-in);
*tts_done_event* is **set** in the ``finally`` block so callers waiting on it
(continuous voice mode) know playback is finished.
"""
tts_done_event.clear()
origin = _origin()
@@ -401,8 +363,8 @@ def stream_tts_to_speaker(
try:
tts_config = origin._load_tts_config()
# Prefer a chunked streamer for low time-to-first-audio; otherwise
# per-sentence sync synthesis (universal — edge + every non-streamer).
# Prefer a chunked streamer for low time-to-first-audio; otherwise per-sentence
# sync synthesis (universal — edge + every non-streamer).
from tools.tts_streaming import SentenceChunker, resolve_streaming_provider
streamer = resolve_streaming_provider(tts_config, preferred=provider)
@@ -411,9 +373,7 @@ def stream_tts_to_speaker(
sync_pipeline = _SyncSentencePipeline(stop_event)
else:
try:
stream_max_len = origin._resolve_max_text_length(
provider or origin._get_provider(tts_config), tts_config
)
stream_max_len = origin._resolve_max_text_length(provider or origin._get_provider(tts_config), tts_config)
except Exception:
stream_max_len = 0
playback = _StreamerPlayback(streamer, stop_event)
@@ -469,16 +429,16 @@ def stream_tts_to_speaker(
except Exception as exc:
logger.warning("Streaming TTS pipeline error: %s", exc)
finally:
# Flush the sync pipeline first: queued sentences finish playing (or
# are skipped when stop_event is set) BEFORE tts_done_event fires, so
# continuous voice mode never reopens the mic over its own voice.
# Flush the sync pipeline first: queued sentences finish playing (or are skipped
# when stop_event is set) BEFORE tts_done_event fires, so continuous voice mode
# never reopens the mic over its own voice.
if sync_pipeline is not None:
try:
sync_pipeline.close()
except Exception:
pass
# The end sentinel lives in finally: so an exception in the text pump
# still lets the playback worker exit.
# The end sentinel lives in finally: so an exception in the text pump still lets
# the playback worker exit.
if playback is not None:
playback.finish()
tts_done_event.set()