From bc009d8f983c36f0dd6c3c7db0b95137ad84c038 Mon Sep 17 00:00:00 2001 From: sathwika Date: Tue, 7 Apr 2026 14:49:45 +0530 Subject: [PATCH 1/6] Add stitching support and enhancements for NvidiaTTSService --- src/pipecat/services/nvidia/tts.py | 376 ++++++++++++++++++++++++----- 1 file changed, 309 insertions(+), 67 deletions(-) diff --git a/src/pipecat/services/nvidia/tts.py b/src/pipecat/services/nvidia/tts.py index c42acc02e..558793593 100644 --- a/src/pipecat/services/nvidia/tts.py +++ b/src/pipecat/services/nvidia/tts.py @@ -4,16 +4,29 @@ # SPDX-License-Identifier: BSD 2-Clause License # -"""NVIDIA Riva text-to-speech service implementation. +"""NVIDIA Nemotron Speech text-to-speech service implementation. -This module provides integration with NVIDIA Riva's TTS services through +This module provides integration with NVIDIA Nemotron Speech's TTS services through gRPC API for high-quality speech synthesis. + +Refer to the NVIDIA TTS NIM documentation for usage, customization, +and local deployment steps: +https://docs.nvidia.com/nim/speech/latest/tts/ + +For zero-shot voice cloning, request access to the Magpie TTS Zero-Shot model: +https://developer.nvidia.com/riva-tts-zeroshot-models + +Local or private cloud deployment is recommended for best zero-shot performance. """ import asyncio import os +import queue +import textwrap +import threading from dataclasses import dataclass, field -from typing import Any, AsyncGenerator, AsyncIterator, Generator, Mapping, Optional +from pathlib import Path +from typing import Any, AsyncGenerator, Mapping, Optional from pipecat.utils.tracing.service_decorators import traced_tts @@ -24,21 +37,28 @@ from loguru import logger from pydantic import BaseModel from pipecat.frames.frames import ( + CancelFrame, + EndFrame, ErrorFrame, Frame, StartFrame, TTSAudioRawFrame, + TTSStartedFrame, ) from pipecat.services.settings import NOT_GIVEN, TTSSettings, _NotGiven from pipecat.services.tts_service import TTSService from pipecat.transcriptions.language import Language try: + import grpc import riva.client import riva.client.proto.riva_tts_pb2 as rtts + from riva.client.proto.riva_audio_pb2 import AudioEncoding except ModuleNotFoundError as e: logger.error(f"Exception: {e}") - logger.error("In order to use NVIDIA Riva TTS, you need to `pip install pipecat-ai[nvidia]`.") + logger.error( + "In order to use NVIDIA Nemotron Speech TTS, you need to `pip install pipecat-ai[nvidia]`." + ) raise Exception(f"Missing module: {e}") @@ -54,18 +74,19 @@ class NvidiaTTSSettings(TTSSettings): class NvidiaTTSService(TTSService): - """NVIDIA Riva text-to-speech service. + """NVIDIA Nemotron Speech text-to-speech service. - Provides high-quality text-to-speech synthesis using NVIDIA Riva's + Provides high-quality text-to-speech synthesis using NVIDIA Nemotron Speech's cloud-based TTS models. Supports multiple voices, languages, and configurable quality settings. """ Settings = NvidiaTTSSettings _settings: Settings + _MAX_CHUNK_LEN = 200 class InputParams(BaseModel): - """Input parameters for Riva TTS configuration. + """Input parameters for Nemotron Speech TTS configuration. .. deprecated:: 0.0.105 Use ``NvidiaTTSService.Settings`` directly via the ``settings`` parameter instead. @@ -81,7 +102,7 @@ class NvidiaTTSService(TTSService): def __init__( self, *, - api_key: str, + api_key: Optional[str] = None, server: str = "grpc.nvcf.nvidia.com:443", voice_id: Optional[str] = None, sample_rate: Optional[int] = None, @@ -92,13 +113,19 @@ class NvidiaTTSService(TTSService): params: Optional[InputParams] = None, settings: Optional[Settings] = None, use_ssl: bool = True, + custom_dictionary: Optional[dict] = None, + encoding: Optional[AudioEncoding] = AudioEncoding.LINEAR_PCM, + zero_shot_audio_prompt_file: Optional[Path] = None, + audio_prompt_encoding: Optional[AudioEncoding] = AudioEncoding.ENCODING_UNSPECIFIED, **kwargs, ): - """Initialize the NVIDIA Riva TTS service. + """Initialize the NVIDIA Nemotron Speech TTS service. Args: - api_key: NVIDIA API key for authentication. + api_key: NVIDIA API key for authentication. Required when using the + cloud endpoint. Not needed for local deployments. server: gRPC server endpoint. Defaults to NVIDIA's cloud endpoint. + For local deployments, pass the local address (e.g. ``localhost:50051``). voice_id: Voice model identifier. Defaults to multilingual Aria voice. .. deprecated:: 0.0.105 @@ -113,7 +140,21 @@ class NvidiaTTSService(TTSService): settings: Runtime-updatable settings. When provided alongside deprecated parameters, ``settings`` values take precedence. - use_ssl: Whether to use SSL for the NVIDIA Riva server. Defaults to True. + use_ssl: Whether to use SSL for the gRPC connection. Defaults to True + for the NVIDIA cloud endpoint. Set to False for local deployments. + custom_dictionary: Custom pronunciation dictionary mapping words + (graphemes) to IPA phonetic representations (phonemes), + e.g. ``{"NVIDIA": "ɛn.vɪ.diː.ʌ"}``. See + https://docs.nvidia.com/nim/speech/latest/tts/phoneme-support.html + for the list of supported IPA phonemes. + encoding: Output audio encoding format. Defaults to ``AudioEncoding.LINEAR_PCM``. + zero_shot_audio_prompt_file: Path to audio prompt file for zero-shot voice + cloning. Audio length should be between 3-10 seconds. The file + is read once at init time and cached in memory. Requires the + Magpie TTS Zero-Shot model. See + https://docs.nvidia.com/nim/speech/latest/tts/voice-cloning.html + audio_prompt_encoding: Encoding of the zero-shot audio prompt file. + Defaults to ``AudioEncoding.ENCODING_UNSPECIFIED``. **kwargs: Additional arguments passed to parent TTSService. """ # 1. Initialize default_settings with hardcoded defaults @@ -144,7 +185,7 @@ class NvidiaTTSService(TTSService): super().__init__( sample_rate=sample_rate, - push_start_frame=True, + push_start_frame=False, push_stop_frames=True, settings=default_settings, **kwargs, @@ -155,18 +196,55 @@ class NvidiaTTSService(TTSService): self._function_id = model_function_map.get("function_id") self._use_ssl = use_ssl + self._custom_dictionary: Optional[str] = None + if custom_dictionary: + entries = [f"{k} {v}" for k, v in custom_dictionary.items()] + self._custom_dictionary = ",".join(entries) + self._encoding = encoding + self._audio_prompt_encoding = audio_prompt_encoding + + self._zero_shot_audio_prompt_file = zero_shot_audio_prompt_file + self._zero_shot_audio_prompt: Optional[bytes] = None + if self._zero_shot_audio_prompt_file is not None: + if not self._zero_shot_audio_prompt_file.exists(): + raise FileNotFoundError( + f"Zero-shot audio prompt file not found: {self._zero_shot_audio_prompt_file}" + ) + with self._zero_shot_audio_prompt_file.open("rb") as f: + self._zero_shot_audio_prompt = f.read() + logger.debug( + f"Loaded zero-shot audio prompt from {self._zero_shot_audio_prompt_file} " + f"({len(self._zero_shot_audio_prompt)} bytes)" + ) + self._service = None self._config = None + # Persistent gRPC stream state for cross-sentence stitching + self._text_queue: Optional[queue.Queue] = None + self._synth_thread: Optional[threading.Thread] = None + self._response_task: Optional[asyncio.Task] = None + self._response_queue: asyncio.Queue = asyncio.Queue() + self._active_context_id: Optional[str] = None + + def can_generate_metrics(self) -> bool: + """Check if this service can generate metrics. + + Returns: + True as this service supports metric generation. + """ + return True + async def set_model(self, model: str): """Set the TTS model. .. deprecated:: 0.0.104 - Model cannot be changed after initialization for NVIDIA Riva TTS. + Model cannot be changed after initialization for NVIDIA Nemotron Speech TTS. Set model and function id in the constructor instead. Example:: + NvidiaTTSService( api_key=..., model_function_map={"function_id": "", "model_name": ""}, @@ -181,7 +259,7 @@ class NvidiaTTSService(TTSService): warnings.simplefilter("always") warnings.warn( "'set_model' is deprecated. Model cannot be changed after initialization" - " for NVIDIA Riva TTS. Set model and function id in the constructor" + " for NVIDIA Nemotron Speech TTS. Set model and function id in the constructor" " instead, e.g.: NvidiaTTSService(api_key=..., model_function_map=" "{'function_id': '', 'model_name': ''})", DeprecationWarning, @@ -191,13 +269,13 @@ class NvidiaTTSService(TTSService): async def _update_settings(self, delta: Settings) -> dict[str, Any]: """Apply a settings delta. - Settings are stored but not applied to the active connection. + Settings are stored and will take effect on the next synthesis turn. + Mid-stream changes cannot be applied to the active gRPC connection. """ changed = await super()._update_settings(delta) - if not changed: - return changed - # TODO: reconnect gRPC client to apply changed settings. - self._warn_unhandled_updated_settings(changed) + if changed: + fields = ", ".join(sorted(changed)) + logger.debug(f"{self.name}: settings updated [{fields}], will apply on next turn") return changed def _initialize_client(self): @@ -216,14 +294,21 @@ class NvidiaTTSService(TTSService): if not self._service: return - # warm up the service - config = self._service.stub.GetRivaSynthesisConfig( - riva.client.proto.riva_tts_pb2.RivaSynthesisConfigRequest() - ) - return config + try: + config = self._service.stub.GetRivaSynthesisConfig( + riva.client.proto.riva_tts_pb2.RivaSynthesisConfigRequest() + ) + return config + except grpc.RpcError as e: + status = e.code().name if hasattr(e, "code") else "UNKNOWN" + details = e.details() if hasattr(e, "details") else str(e) + logger.error( + f"{self} failed to get synthesis config from server (gRPC {status}): {details}" + ) + return None async def start(self, frame: StartFrame): - """Start the Cartesia TTS service. + """Start the NVIDIA Nemotron Speech TTS service. Args: frame: The start frame containing initialization parameters. @@ -233,65 +318,222 @@ class NvidiaTTSService(TTSService): self._config = self._create_synthesis_config() logger.debug(f"Initialized NvidiaTTSService with model: {self._settings.model}") + async def stop(self, frame: EndFrame): + """Stop the NVIDIA Nemotron Speech TTS service and clean up resources. + + Args: + frame: EndFrame indicating pipeline stop. + """ + await super().stop(frame) + await self._close_synthesis_stream() + + async def cancel(self, frame: CancelFrame): + """Cancel the NVIDIA Nemotron Speech TTS service operation. + + Args: + frame: CancelFrame indicating operation cancellation. + """ + await super().cancel(frame) + await self._close_synthesis_stream() + + def _start_synthesis_stream(self, context_id: str): + """Start a persistent gRPC synthesis stream for the current turn. + + Creates a queue-backed generator that feeds text to + ``synthesize_online``. The gRPC stream stays open until a ``None`` + sentinel is pushed into the queue. + """ + self._text_queue = queue.Queue() + self._active_context_id = context_id + self._response_queue = asyncio.Queue() + + self._synth_thread = threading.Thread( + target=self._synthesis_thread_handler, + daemon=True, + name="nvidia-tts-synth", + ) + self._synth_thread.start() + self._response_task = self.create_task( + self._response_consumer(), name="nvidia-tts-response" + ) + + def _build_base_request(self) -> rtts.SynthesizeSpeechRequest: + """Build a reusable ``SynthesizeSpeechRequest`` with current settings.""" + req = rtts.SynthesizeSpeechRequest( + text="", + language_code=str(self._settings.language or "en-US"), + sample_rate_hz=self.sample_rate, + encoding=self._encoding, + ) + voice = self._settings.voice + if voice: + req.voice_name = voice + if self._zero_shot_audio_prompt is not None: + req.zero_shot_data.audio_prompt = self._zero_shot_audio_prompt + req.zero_shot_data.encoding = self._audio_prompt_encoding + req.zero_shot_data.quality = self._settings.quality + if self._custom_dictionary: + req.custom_dictionary = self._custom_dictionary + return req + + def _synthesis_thread_handler(self): + """Run ``SynthesizeOnline`` gRPC stream in a background thread. + + Builds request objects directly to avoid a Python 3.12 compatibility + bug in ``riva.client.SpeechSynthesisService.synthesize_online``. + """ + base_req = self._build_base_request() + + def request_generator(): + while True: + text = self._text_queue.get() + if text is None: + break + base_req.text = text + yield base_req + + try: + responses = self._service.stub.SynthesizeOnline( + request_generator(), + metadata=self._service.auth.get_auth_metadata(), + ) + for resp in responses: + asyncio.run_coroutine_threadsafe( + self._response_queue.put(resp), self.get_event_loop() + ) + except Exception as e: + logger.error(f"{self} gRPC synthesis stream error: {e}") + asyncio.run_coroutine_threadsafe(self._response_queue.put(e), self.get_event_loop()) + finally: + asyncio.run_coroutine_threadsafe(self._response_queue.put(None), self.get_event_loop()) + + async def _response_consumer(self): + """Consume gRPC responses and append audio to the active audio context.""" + while True: + item = await self._response_queue.get() + if item is None: + break + if isinstance(item, Exception): + await self.push_error(f"{self} synthesis error: {item}") + break + await self.stop_ttfb_metrics() + frame = TTSAudioRawFrame( + audio=item.audio, + sample_rate=self.sample_rate, + num_channels=1, + context_id=self._active_context_id, + ) + await self.append_to_audio_context(self._active_context_id, frame) + + async def _close_synthesis_stream(self): + """Close the active gRPC synthesis stream. + + Sends a sentinel to end the request generator, waits for the gRPC + thread to finish producing all remaining audio, then lets the + response consumer drain naturally before cleaning up. + """ + if self._text_queue is not None: + self._text_queue.put(None) + + if self._synth_thread is not None: + await asyncio.to_thread(self._synth_thread.join) + self._synth_thread = None + + self._text_queue = None + + if self._response_task is not None: + try: + await self._response_task + except asyncio.CancelledError: + pass + self._response_task = None + + self._active_context_id = None + + async def flush_audio(self, context_id: Optional[str] = None): + """Flush the synthesis stream at the end of an LLM turn. + + Sends a sentinel to the gRPC stream, waits for remaining audio, + then delegates to the base class for audio context cleanup. + + Args: + context_id: The audio context to flush. + """ + await self._close_synthesis_stream() + await super().flush_audio(context_id) + + async def on_audio_context_interrupted(self, context_id: str): + """Handle interruption by closing the active synthesis stream.""" + await self.stop_all_metrics() + await self._close_synthesis_stream() + + @staticmethod + def _split_text_into_chunks(text: str) -> list[str]: + """Split text into <=200-character chunks at whitespace boundaries. + + Magpie stitches chunks seamlessly in the gRPC stream, so splitting + conservatively at 200 chars avoids max char limits without affecting audio + quality. + + Args: + text: Input text to split. + + Returns: + List of text chunks, each at most 200 characters. + """ + text = text.strip() + if not text: + return [] + return textwrap.wrap( + text, + width=NvidiaTTSService._MAX_CHUNK_LEN, + break_long_words=True, + break_on_hyphens=False, + ) + @traced_tts async def run_tts(self, text: str, context_id: str) -> AsyncGenerator[Frame, None]: - """Generate speech from text using NVIDIA Riva TTS. + """Generate speech from text using NVIDIA Nemotron Speech TTS. + + On the first call for a turn, starts a persistent ``synthesize_online`` + gRPC stream. Subsequent calls within the same turn feed text into the + existing stream, enabling Magpie's cross-sentence stitching. + + Text is split into chunks respecting Magpie's per-request limits. Each chunk becomes + a separate request in the gRPC stream, stitched seamlessly by Magpie. Args: text: The text to synthesize into speech. context_id: The context ID for tracking audio frames. Yields: - Frame: Audio frames containing the synthesized speech data. + None on success. Audio is delivered asynchronously via the + response consumer. ErrorFrame on failure. """ - - def read_audio_responses() -> Generator[rtts.SynthesizeSpeechResponse, None, None]: - responses = self._service.synthesize_online( - text, - self._settings.voice, - self._settings.language, - sample_rate_hz=self.sample_rate, - zero_shot_audio_prompt_file=None, - zero_shot_quality=self._settings.quality, - custom_dictionary={}, - ) - return responses - - def async_next(it): - try: - return next(it) - except StopIteration: - return None - - async def async_iterator(iterator) -> AsyncIterator[rtts.SynthesizeSpeechResponse]: - while True: - item = await asyncio.to_thread(async_next, iterator) - if item is None: - return - yield item + text = text.strip() + if not text or not any(c.isalnum() for c in text): + return try: assert self._service is not None, "TTS service not initialized" assert self._config is not None, "Synthesis configuration not created" + # First call for this turn: create audio context and start gRPC stream + if not self.audio_context_available(context_id): + await self.create_audio_context(context_id) + await self.start_ttfb_metrics() + yield TTSStartedFrame(context_id=context_id) + self._start_synthesis_stream(context_id) + logger.trace(f"{self}: Started synthesis stream for context {context_id}") + logger.debug(f"{self}: Generating TTS [{text}]") - responses = await asyncio.to_thread(read_audio_responses) - - async for resp in async_iterator(responses): - await self.stop_ttfb_metrics() - frame = TTSAudioRawFrame( - audio=resp.audio, - sample_rate=self.sample_rate, - num_channels=1, - context_id=context_id, - ) - yield frame + for chunk in self._split_text_into_chunks(text): + if any(c.isalnum() for c in chunk): + self._text_queue.put(chunk) await self.start_tts_usage_metrics(text) - except asyncio.TimeoutError as e: - logger.error(f"{self} timeout waiting for audio response") - yield ErrorFrame(error=f"{self} error: {e}") + yield None except Exception as e: logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"{self} error: {e}") \ No newline at end of file From 9f0b18b03d8e442c90c29ec27ef987b0878de8db Mon Sep 17 00:00:00 2001 From: sathwika Date: Tue, 7 Apr 2026 18:18:55 +0530 Subject: [PATCH 2/6] Add changelog fragments for PR #4249 --- changelog/4249.added.md | 9 +++++++++ changelog/4249.changed.md | 5 +++++ 2 files changed, 14 insertions(+) create mode 100644 changelog/4249.added.md create mode 100644 changelog/4249.changed.md diff --git a/changelog/4249.added.md b/changelog/4249.added.md new file mode 100644 index 000000000..516e2e3a8 --- /dev/null +++ b/changelog/4249.added.md @@ -0,0 +1,9 @@ +- Added cross-sentence stitching support to NvidiaTTSService. Multiple sentences within an LLM turn are fed into a single `SynthesizeOnline` gRPC stream, enabling Magpie's seamless audio across sentence boundaries (requires Magpie TTS model v1.7.0+). +- Added new parameters to NvidiaTTSService: + - `custom_dictionary` for IPA-based custom pronunciation + - `encoding` for output audio encoding format + - `zero_shot_audio_prompt_file` for zero-shot voice cloning with file caching + - `audio_prompt_encoding` for zero-shot audio prompt format +- Added `can_generate_metrics` returning True for NvidiaTTSService. +- Added `stop_all_metrics()` call on audio context interruption. +- Added gRPC error handling for synthesis config retrieval. \ No newline at end of file diff --git a/changelog/4249.changed.md b/changelog/4249.changed.md new file mode 100644 index 000000000..460a06afa --- /dev/null +++ b/changelog/4249.changed.md @@ -0,0 +1,5 @@ +- Updated NvidiaTTSService: + - Made `api_key` optional for local NIM deployments + - Fixed `_update_settings` to correctly indicate that runtime settings updates (voice, language, quality) take effect on the next turn + - Replaced per-sentence synchronous `synthesize_online` calls with async queue-backed gRPC streaming + - Renamed Riva references to Nemotron Speech \ No newline at end of file From 8abda808ca7bb23cf8d8b1b050c6e64510b54355 Mon Sep 17 00:00:00 2001 From: sathwika Date: Tue, 7 Apr 2026 18:41:50 +0530 Subject: [PATCH 3/6] Add Nvidia copyright header --- src/pipecat/services/nvidia/tts.py | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/src/pipecat/services/nvidia/tts.py b/src/pipecat/services/nvidia/tts.py index 558793593..e6a4e2e74 100644 --- a/src/pipecat/services/nvidia/tts.py +++ b/src/pipecat/services/nvidia/tts.py @@ -1,5 +1,6 @@ # # Copyright (c) 2024-2026, Daily +# Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. # # SPDX-License-Identifier: BSD 2-Clause License # @@ -379,8 +380,9 @@ class NvidiaTTSService(TTSService): def _synthesis_thread_handler(self): """Run ``SynthesizeOnline`` gRPC stream in a background thread. - Builds request objects directly to avoid a Python 3.12 compatibility - bug in ``riva.client.SpeechSynthesisService.synthesize_online``. + Uses a queue-backed generator to feed text chunks into a single + ``SynthesizeOnline`` call, enabling Magpie's cross-sentence stitching. + Audio responses are forwarded to the async response queue. """ base_req = self._build_base_request() From 746fadc2b53ea54a5d9f3041e415a160cfe5b29f Mon Sep 17 00:00:00 2001 From: sathwika Date: Fri, 10 Apr 2026 17:18:22 +0530 Subject: [PATCH 4/6] thread simplification + handling interuption --- changelog/4249.added.md | 15 +- changelog/4249.changed.md | 13 +- src/pipecat/services/nvidia/tts.py | 289 +++++++++++++++++++---------- 3 files changed, 204 insertions(+), 113 deletions(-) diff --git a/changelog/4249.added.md b/changelog/4249.added.md index 516e2e3a8..bc2ae6e74 100644 --- a/changelog/4249.added.md +++ b/changelog/4249.added.md @@ -1,9 +1,6 @@ -- Added cross-sentence stitching support to NvidiaTTSService. Multiple sentences within an LLM turn are fed into a single `SynthesizeOnline` gRPC stream, enabling Magpie's seamless audio across sentence boundaries (requires Magpie TTS model v1.7.0+). -- Added new parameters to NvidiaTTSService: - - `custom_dictionary` for IPA-based custom pronunciation - - `encoding` for output audio encoding format - - `zero_shot_audio_prompt_file` for zero-shot voice cloning with file caching - - `audio_prompt_encoding` for zero-shot audio prompt format -- Added `can_generate_metrics` returning True for NvidiaTTSService. -- Added `stop_all_metrics()` call on audio context interruption. -- Added gRPC error handling for synthesis config retrieval. \ No newline at end of file +- Added enhancements to `NvidiaTTSService`: + + - Cross-sentence stitching: multiple sentences within an LLM turn are fed into a single `SynthesizeOnline` gRPC stream for seamless audio across sentence boundaries (requires Magpie TTS model v1.7.0+). + - `custom_dictionary` and `encoding` parameters for IPA-based custom pronunciation and output audio encoding. + - Metrics generation (`can_generate_metrics` returns true) and `stop_all_metrics()` when an audio context is interrupted. + - gRPC error handling around synthesis config retrieval (`GetRivaSynthesisConfig`). \ No newline at end of file diff --git a/changelog/4249.changed.md b/changelog/4249.changed.md index 460a06afa..e5bd207b0 100644 --- a/changelog/4249.changed.md +++ b/changelog/4249.changed.md @@ -1,5 +1,8 @@ -- Updated NvidiaTTSService: - - Made `api_key` optional for local NIM deployments - - Fixed `_update_settings` to correctly indicate that runtime settings updates (voice, language, quality) take effect on the next turn - - Replaced per-sentence synchronous `synthesize_online` calls with async queue-backed gRPC streaming - - Renamed Riva references to Nemotron Speech \ No newline at end of file +- Updated `NvidiaTTSService`: + + - Made `api_key` optional for local NIM deployments. + - Voice, language, and quality can be updated without reconnecting the gRPC client; new values take effect on the next synthesis turn, not for the current turn's in-flight requests. + - Replaced per-sentence synchronous `synthesize_online` calls with async queue-backed gRPC streaming. + - Streaming now uses asyncio tasks with explicit gRPC cancellation on interruption and stale-response filtering when a stream is aborted or replaced. + - Renamed Riva references to Nemotron Speech in docs and messages. + - Disabled automatic TTS start frames at the service level (`push_start_frame=False`) and emit `TTSStartedFrame` when a stitched synthesis stream is started for a context. \ No newline at end of file diff --git a/src/pipecat/services/nvidia/tts.py b/src/pipecat/services/nvidia/tts.py index e6a4e2e74..add805f6e 100644 --- a/src/pipecat/services/nvidia/tts.py +++ b/src/pipecat/services/nvidia/tts.py @@ -13,11 +13,6 @@ gRPC API for high-quality speech synthesis. Refer to the NVIDIA TTS NIM documentation for usage, customization, and local deployment steps: https://docs.nvidia.com/nim/speech/latest/tts/ - -For zero-shot voice cloning, request access to the Magpie TTS Zero-Shot model: -https://developer.nvidia.com/riva-tts-zeroshot-models - -Local or private cloud deployment is recommended for best zero-shot performance. """ import asyncio @@ -26,7 +21,6 @@ import queue import textwrap import threading from dataclasses import dataclass, field -from pathlib import Path from typing import Any, AsyncGenerator, Mapping, Optional from pipecat.utils.tracing.service_decorators import traced_tts @@ -74,6 +68,19 @@ class NvidiaTTSSettings(TTSSettings): quality: int | _NotGiven = field(default_factory=lambda: NOT_GIVEN) +@dataclass +class _SynthesisStreamState: + """Runtime state for one active synthesis stream.""" + + context_id: str + text_queue: queue.Queue + response_queue: asyncio.Queue + stop_event: threading.Event + rpc_call: Any = None + synth_task: Optional[asyncio.Task] = None + response_task: Optional[asyncio.Task] = None + + class NvidiaTTSService(TTSService): """NVIDIA Nemotron Speech text-to-speech service. @@ -116,8 +123,6 @@ class NvidiaTTSService(TTSService): use_ssl: bool = True, custom_dictionary: Optional[dict] = None, encoding: Optional[AudioEncoding] = AudioEncoding.LINEAR_PCM, - zero_shot_audio_prompt_file: Optional[Path] = None, - audio_prompt_encoding: Optional[AudioEncoding] = AudioEncoding.ENCODING_UNSPECIFIED, **kwargs, ): """Initialize the NVIDIA Nemotron Speech TTS service. @@ -149,13 +154,6 @@ class NvidiaTTSService(TTSService): https://docs.nvidia.com/nim/speech/latest/tts/phoneme-support.html for the list of supported IPA phonemes. encoding: Output audio encoding format. Defaults to ``AudioEncoding.LINEAR_PCM``. - zero_shot_audio_prompt_file: Path to audio prompt file for zero-shot voice - cloning. Audio length should be between 3-10 seconds. The file - is read once at init time and cached in memory. Requires the - Magpie TTS Zero-Shot model. See - https://docs.nvidia.com/nim/speech/latest/tts/voice-cloning.html - audio_prompt_encoding: Encoding of the zero-shot audio prompt file. - Defaults to ``AudioEncoding.ENCODING_UNSPECIFIED``. **kwargs: Additional arguments passed to parent TTSService. """ # 1. Initialize default_settings with hardcoded defaults @@ -202,31 +200,12 @@ class NvidiaTTSService(TTSService): entries = [f"{k} {v}" for k, v in custom_dictionary.items()] self._custom_dictionary = ",".join(entries) self._encoding = encoding - self._audio_prompt_encoding = audio_prompt_encoding - - self._zero_shot_audio_prompt_file = zero_shot_audio_prompt_file - self._zero_shot_audio_prompt: Optional[bytes] = None - if self._zero_shot_audio_prompt_file is not None: - if not self._zero_shot_audio_prompt_file.exists(): - raise FileNotFoundError( - f"Zero-shot audio prompt file not found: {self._zero_shot_audio_prompt_file}" - ) - with self._zero_shot_audio_prompt_file.open("rb") as f: - self._zero_shot_audio_prompt = f.read() - logger.debug( - f"Loaded zero-shot audio prompt from {self._zero_shot_audio_prompt_file} " - f"({len(self._zero_shot_audio_prompt)} bytes)" - ) self._service = None self._config = None - # Persistent gRPC stream state for cross-sentence stitching - self._text_queue: Optional[queue.Queue] = None - self._synth_thread: Optional[threading.Thread] = None - self._response_task: Optional[asyncio.Task] = None - self._response_queue: asyncio.Queue = asyncio.Queue() - self._active_context_id: Optional[str] = None + # Runtime state for the active streaming turn. + self._stream_state: Optional[_SynthesisStreamState] = None def can_generate_metrics(self) -> bool: """Check if this service can generate metrics. @@ -245,7 +224,6 @@ class NvidiaTTSService(TTSService): Example:: - NvidiaTTSService( api_key=..., model_function_map={"function_id": "", "model_name": ""}, @@ -320,19 +298,19 @@ class NvidiaTTSService(TTSService): logger.debug(f"Initialized NvidiaTTSService with model: {self._settings.model}") async def stop(self, frame: EndFrame): - """Stop the NVIDIA Nemotron Speech TTS service and clean up resources. + """Stop the NVIDIA Nemotron Speech TTS service. Args: - frame: EndFrame indicating pipeline stop. + frame: The end frame. """ await super().stop(frame) await self._close_synthesis_stream() async def cancel(self, frame: CancelFrame): - """Cancel the NVIDIA Nemotron Speech TTS service operation. + """Cancel the NVIDIA Nemotron Speech TTS service. Args: - frame: CancelFrame indicating operation cancellation. + frame: The cancel frame. """ await super().cancel(frame) await self._close_synthesis_stream() @@ -344,18 +322,20 @@ class NvidiaTTSService(TTSService): ``synthesize_online``. The gRPC stream stays open until a ``None`` sentinel is pushed into the queue. """ - self._text_queue = queue.Queue() - self._active_context_id = context_id - self._response_queue = asyncio.Queue() - - self._synth_thread = threading.Thread( - target=self._synthesis_thread_handler, - daemon=True, - name="nvidia-tts-synth", + state = _SynthesisStreamState( + context_id=context_id, + text_queue=queue.Queue(), + response_queue=asyncio.Queue(), + stop_event=threading.Event(), ) - self._synth_thread.start() - self._response_task = self.create_task( - self._response_consumer(), name="nvidia-tts-response" + self._stream_state = state + logger.debug(f"{self}: starting synthesis stream") + + state.synth_task = self.create_task( + self._synth_task_handler(state), name="nvidia-tts-synth" + ) + state.response_task = self.create_task( + self._process_responses(state), name="nvidia-tts-response" ) def _build_base_request(self) -> rtts.SynthesizeSpeechRequest: @@ -369,105 +349,212 @@ class NvidiaTTSService(TTSService): voice = self._settings.voice if voice: req.voice_name = voice - if self._zero_shot_audio_prompt is not None: - req.zero_shot_data.audio_prompt = self._zero_shot_audio_prompt - req.zero_shot_data.encoding = self._audio_prompt_encoding - req.zero_shot_data.quality = self._settings.quality if self._custom_dictionary: req.custom_dictionary = self._custom_dictionary return req - def _synthesis_thread_handler(self): - """Run ``SynthesizeOnline`` gRPC stream in a background thread. + def _synthesis_handler(self, state: _SynthesisStreamState): + """Run ``SynthesizeOnline`` gRPC stream in a blocking call. Uses a queue-backed generator to feed text chunks into a single ``SynthesizeOnline`` call, enabling Magpie's cross-sentence stitching. Audio responses are forwarded to the async response queue. """ + event_loop = self.get_event_loop() base_req = self._build_base_request() def request_generator(): while True: - text = self._text_queue.get() - if text is None: + if state.stop_event.is_set(): + break + text = state.text_queue.get() + if text is None or state.stop_event.is_set(): break base_req.text = text yield base_req try: - responses = self._service.stub.SynthesizeOnline( + call = self._service.stub.SynthesizeOnline( request_generator(), metadata=self._service.auth.get_auth_metadata(), ) - for resp in responses: - asyncio.run_coroutine_threadsafe( - self._response_queue.put(resp), self.get_event_loop() - ) - except Exception as e: - logger.error(f"{self} gRPC synthesis stream error: {e}") - asyncio.run_coroutine_threadsafe(self._response_queue.put(e), self.get_event_loop()) - finally: - asyncio.run_coroutine_threadsafe(self._response_queue.put(None), self.get_event_loop()) + state.rpc_call = call - async def _response_consumer(self): + for resp in call: + if state.stop_event.is_set(): + break + asyncio.run_coroutine_threadsafe(state.response_queue.put(resp), event_loop) + except Exception as e: + if not state.stop_event.is_set(): + logger.error(f"{self} gRPC synthesis stream error: {e}") + asyncio.run_coroutine_threadsafe(state.response_queue.put(e), event_loop) + finally: + state.rpc_call = None + asyncio.run_coroutine_threadsafe(state.response_queue.put(None), event_loop) + + async def _synth_task_handler(self, state: _SynthesisStreamState): + """Wrap ``_synthesis_handler`` as an asyncio-managed task.""" + await asyncio.to_thread(self._synthesis_handler, state) + + def _cancel_stream_call(self, state: _SynthesisStreamState): + """Best-effort cancellation of in-flight gRPC call.""" + call = state.rpc_call + if call is not None and hasattr(call, "cancel"): + try: + call.cancel() + except Exception as e: + logger.debug(f"{self}: failed to cancel gRPC stream call: {e}") + + async def _process_responses(self, state: _SynthesisStreamState): """Consume gRPC responses and append audio to the active audio context.""" while True: - item = await self._response_queue.get() + item = await state.response_queue.get() if item is None: break if isinstance(item, Exception): - await self.push_error(f"{self} synthesis error: {item}") + # Ignore stale exceptions from interrupted streams. + if self._stream_state is state: + await self.push_error(f"{self} synthesis error: {item}") break + + # Stale stream responses must never leak into a newer stream context. + if self._stream_state is not state: + continue + await self.stop_ttfb_metrics() frame = TTSAudioRawFrame( audio=item.audio, sample_rate=self.sample_rate, num_channels=1, - context_id=self._active_context_id, + context_id=state.context_id, ) - await self.append_to_audio_context(self._active_context_id, frame) + await self.append_to_audio_context(state.context_id, frame) + + # Finalize ownership once the stream drains naturally. + if self._stream_state is state and not state.stop_event.is_set(): + self._stream_state = None + + def _signal_synthesis_close(self, state: _SynthesisStreamState): + """Signal the active synthesis request generator to close.""" + state.text_queue.put(None) async def _close_synthesis_stream(self): - """Close the active gRPC synthesis stream. + """Close the active gRPC synthesis stream gracefully. - Sends a sentinel to end the request generator, waits for the gRPC - thread to finish producing all remaining audio, then lets the - response consumer drain naturally before cleaning up. + Sends a sentinel to end the request generator, waits for the + synthesis task to finish producing all remaining audio, then lets + the response task drain naturally before cleaning up. """ - if self._text_queue is not None: - self._text_queue.put(None) + state = self._stream_state + if state is None: + return - if self._synth_thread is not None: - await asyncio.to_thread(self._synth_thread.join) - self._synth_thread = None + self._signal_synthesis_close(state) - self._text_queue = None - - if self._response_task is not None: + if state.synth_task is not None: try: - await self._response_task + await state.synth_task except asyncio.CancelledError: pass - self._response_task = None + state.synth_task = None - self._active_context_id = None + if state.response_task is not None: + try: + await state.response_task + except asyncio.CancelledError: + pass + state.response_task = None + + if self._stream_state is state: + self._stream_state = None + + async def _wait_for_synthesis_close_interruptibly(self, state: _SynthesisStreamState): + """Wait for synthesis close unless interruption preempts this stream.""" + while True: + if self._stream_state is not state or state.stop_event.is_set(): + # Interruption took ownership of stream shutdown. + return + + synth_done = state.synth_task is None or state.synth_task.done() + response_done = state.response_task is None or state.response_task.done() + if synth_done and response_done: + break + + # Poll in short intervals to keep this wait interruptible. + await asyncio.sleep(0.05) + + if state.synth_task is not None: + try: + await state.synth_task + except asyncio.CancelledError: + pass + state.synth_task = None + + if state.response_task is not None: + try: + await state.response_task + except asyncio.CancelledError: + pass + state.response_task = None + + if self._stream_state is state: + self._stream_state = None + + async def _abort_synthesis_stream(self): + """Abort the active gRPC synthesis stream immediately. + + Cancels the response task first to stop delivering audio, then + drains the text queue and signals the synthesis handler to stop. + Unlike ``_close_synthesis_stream``, pending audio is discarded. + """ + state = self._stream_state + if state is None: + return + + state.stop_event.set() + + if state.response_task is not None: + await self.cancel_task(state.response_task) + state.response_task = None + + while not state.text_queue.empty(): + try: + state.text_queue.get_nowait() + except queue.Empty: + break + self._signal_synthesis_close(state) + + self._cancel_stream_call(state) + + if state.synth_task is not None: + await self.cancel_task(state.synth_task) + state.synth_task = None + + if self._stream_state is state: + self._stream_state = None async def flush_audio(self, context_id: Optional[str] = None): - """Flush the synthesis stream at the end of an LLM turn. - - Sends a sentinel to the gRPC stream, waits for remaining audio, - then delegates to the base class for audio context cleanup. + """Flush any pending audio and finalize the current context. Args: - context_id: The audio context to flush. + context_id: The specific context to flush. If None, falls back to the + currently active context. """ - await self._close_synthesis_stream() + state = self._stream_state + if state is not None: + self._signal_synthesis_close(state) + await self._wait_for_synthesis_close_interruptibly(state) await super().flush_audio(context_id) async def on_audio_context_interrupted(self, context_id: str): - """Handle interruption by closing the active synthesis stream.""" + """Cancel the active gRPC synthesis stream when the bot is interrupted. + + Args: + context_id: The ID of the audio context that was interrupted. + """ await self.stop_all_metrics() - await self._close_synthesis_stream() + await self._abort_synthesis_stream() + await super().on_audio_context_interrupted(context_id) @staticmethod def _split_text_into_chunks(text: str) -> list[str]: @@ -530,9 +617,13 @@ class NvidiaTTSService(TTSService): logger.debug(f"{self}: Generating TTS [{text}]") + state = self._stream_state + if state is None: + raise RuntimeError("Synthesis stream not started") + for chunk in self._split_text_into_chunks(text): if any(c.isalnum() for c in chunk): - self._text_queue.put(chunk) + state.text_queue.put(chunk) await self.start_tts_usage_metrics(text) yield None From f45a410f56ffe2deee33f8042735d061283861d5 Mon Sep 17 00:00:00 2001 From: sathwika Date: Mon, 13 Apr 2026 14:35:17 +0530 Subject: [PATCH 5/6] refactor/simplify NvidiaTTSService synthesis stream shutdown --- src/pipecat/services/nvidia/tts.py | 88 +++++------------------------- 1 file changed, 15 insertions(+), 73 deletions(-) diff --git a/src/pipecat/services/nvidia/tts.py b/src/pipecat/services/nvidia/tts.py index add805f6e..be468efcc 100644 --- a/src/pipecat/services/nvidia/tts.py +++ b/src/pipecat/services/nvidia/tts.py @@ -269,9 +269,10 @@ class NvidiaTTSService(TTSService): self._service = riva.client.SpeechSynthesisService(auth) - def _create_synthesis_config(self): + def _create_synthesis_config(self) -> rtts.RivaSynthesisConfigResponse: + """Fetch and validate synthesis configuration from the server.""" if not self._service: - return + raise RuntimeError("TTS service not initialized") try: config = self._service.stub.GetRivaSynthesisConfig( @@ -284,7 +285,9 @@ class NvidiaTTSService(TTSService): logger.error( f"{self} failed to get synthesis config from server (gRPC {status}): {details}" ) - return None + raise RuntimeError( + f"{self}: startup failed while fetching synthesis config (gRPC {status})" + ) from e async def start(self, frame: StartFrame): """Start the NVIDIA Nemotron Speech TTS service. @@ -304,7 +307,7 @@ class NvidiaTTSService(TTSService): frame: The end frame. """ await super().stop(frame) - await self._close_synthesis_stream() + await self._abort_synthesis_stream() async def cancel(self, frame: CancelFrame): """Cancel the NVIDIA Nemotron Speech TTS service. @@ -313,7 +316,7 @@ class NvidiaTTSService(TTSService): frame: The cancel frame. """ await super().cancel(frame) - await self._close_synthesis_stream() + await self._abort_synthesis_stream() def _start_synthesis_stream(self, context_id: str): """Start a persistent gRPC synthesis stream for the current turn. @@ -412,8 +415,11 @@ class NvidiaTTSService(TTSService): if item is None: break if isinstance(item, Exception): - # Ignore stale exceptions from interrupted streams. - if self._stream_state is state: + # Treat stream exceptions as terminal for this stream. Once + # SynthesizeOnline raises, no further reliable audio is expected. + # Ignore stale or interruption-driven exceptions to avoid noisy + # errors during handoff to a new stream. + if self._stream_state is state and not state.stop_event.is_set(): await self.push_error(f"{self} synthesis error: {item}") break @@ -438,74 +444,12 @@ class NvidiaTTSService(TTSService): """Signal the active synthesis request generator to close.""" state.text_queue.put(None) - async def _close_synthesis_stream(self): - """Close the active gRPC synthesis stream gracefully. - - Sends a sentinel to end the request generator, waits for the - synthesis task to finish producing all remaining audio, then lets - the response task drain naturally before cleaning up. - """ - state = self._stream_state - if state is None: - return - - self._signal_synthesis_close(state) - - if state.synth_task is not None: - try: - await state.synth_task - except asyncio.CancelledError: - pass - state.synth_task = None - - if state.response_task is not None: - try: - await state.response_task - except asyncio.CancelledError: - pass - state.response_task = None - - if self._stream_state is state: - self._stream_state = None - - async def _wait_for_synthesis_close_interruptibly(self, state: _SynthesisStreamState): - """Wait for synthesis close unless interruption preempts this stream.""" - while True: - if self._stream_state is not state or state.stop_event.is_set(): - # Interruption took ownership of stream shutdown. - return - - synth_done = state.synth_task is None or state.synth_task.done() - response_done = state.response_task is None or state.response_task.done() - if synth_done and response_done: - break - - # Poll in short intervals to keep this wait interruptible. - await asyncio.sleep(0.05) - - if state.synth_task is not None: - try: - await state.synth_task - except asyncio.CancelledError: - pass - state.synth_task = None - - if state.response_task is not None: - try: - await state.response_task - except asyncio.CancelledError: - pass - state.response_task = None - - if self._stream_state is state: - self._stream_state = None - async def _abort_synthesis_stream(self): """Abort the active gRPC synthesis stream immediately. Cancels the response task first to stop delivering audio, then drains the text queue and signals the synthesis handler to stop. - Unlike ``_close_synthesis_stream``, pending audio is discarded. + Pending audio is discarded. """ state = self._stream_state if state is None: @@ -543,7 +487,6 @@ class NvidiaTTSService(TTSService): state = self._stream_state if state is not None: self._signal_synthesis_close(state) - await self._wait_for_synthesis_close_interruptibly(state) await super().flush_audio(context_id) async def on_audio_context_interrupted(self, context_id: str): @@ -605,7 +548,6 @@ class NvidiaTTSService(TTSService): try: assert self._service is not None, "TTS service not initialized" - assert self._config is not None, "Synthesis configuration not created" # First call for this turn: create audio context and start gRPC stream if not self.audio_context_available(context_id): @@ -629,4 +571,4 @@ class NvidiaTTSService(TTSService): yield None except Exception as e: logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") \ No newline at end of file + yield ErrorFrame(error=f"{self} error: {e}") From 1d5dcf1698f75b4d1d8b19c2f489dd1708bf0b98 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Mon, 13 Apr 2026 09:34:13 -0300 Subject: [PATCH 6/6] Invoking to remove the audio context when there is no more audio to receive. --- src/pipecat/services/nvidia/tts.py | 1 + 1 file changed, 1 insertion(+) diff --git a/src/pipecat/services/nvidia/tts.py b/src/pipecat/services/nvidia/tts.py index be468efcc..daa4e0bd7 100644 --- a/src/pipecat/services/nvidia/tts.py +++ b/src/pipecat/services/nvidia/tts.py @@ -413,6 +413,7 @@ class NvidiaTTSService(TTSService): while True: item = await state.response_queue.get() if item is None: + await self.remove_audio_context(state.context_id) break if isinstance(item, Exception): # Treat stream exceptions as terminal for this stream. Once