diff --git a/changelog/3504.fixed.md b/changelog/3504.fixed.md new file mode 100644 index 000000000..2d675547a --- /dev/null +++ b/changelog/3504.fixed.md @@ -0,0 +1 @@ +- Moved `NVIDIATTSService` and `NVIDIASTTService` client initialization from constructor to `start()` for better error handling. diff --git a/src/pipecat/services/nvidia/stt.py b/src/pipecat/services/nvidia/stt.py index 0d671571f..45c59de7a 100644 --- a/src/pipecat/services/nvidia/stt.py +++ b/src/pipecat/services/nvidia/stt.py @@ -134,6 +134,7 @@ class NvidiaSTTService(STTService): params = params or NvidiaSTTService.InputParams() + self._server = server self._api_key = api_key self._use_ssl = use_ssl self._profanity_filter = False @@ -162,19 +163,55 @@ class NvidiaSTTService(STTService): self.set_model_name(model_function_map.get("model_name")) - metadata = [ - ["function-id", self._function_id], - ["authorization", f"Bearer {api_key}"], - ] - auth = riva.client.Auth(None, self._use_ssl, server, metadata) - - self._asr_service = riva.client.ASRService(auth) - + self._asr_service = None self._queue = None self._config = None self._thread_task = None self._response_task = None + def _initialize_client(self): + metadata = [ + ["function-id", self._function_id], + ["authorization", f"Bearer {self._api_key}"], + ] + auth = riva.client.Auth(None, self._use_ssl, self._server, metadata) + + self._asr_service = riva.client.ASRService(auth) + + def _create_recognition_config(self): + """Create the NVIDIA Riva ASR recognition configuration.""" + config = riva.client.StreamingRecognitionConfig( + config=riva.client.RecognitionConfig( + encoding=riva.client.AudioEncoding.LINEAR_PCM, + language_code=self._language_code, + model="", + max_alternatives=1, + profanity_filter=self._profanity_filter, + enable_automatic_punctuation=self._automatic_punctuation, + verbatim_transcripts=not self._no_verbatim_transcripts, + sample_rate_hertz=self.sample_rate, + audio_channel_count=1, + ), + interim_results=True, + ) + + riva.client.add_word_boosting_to_config( + config, self._boosted_lm_words, self._boosted_lm_score + ) + + riva.client.add_endpoint_parameters_to_config( + config, + self._start_history, + self._start_threshold, + self._stop_history, + self._stop_history_eou, + self._stop_threshold, + self._stop_threshold_eou, + ) + riva.client.add_custom_configuration_to_config(config, self._custom_configuration) + + return config + def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -206,41 +243,9 @@ class NvidiaSTTService(STTService): frame: StartFrame indicating pipeline start. """ await super().start(frame) + self._initialize_client() + self._config = self._create_recognition_config() - if self._config: - return - - config = riva.client.StreamingRecognitionConfig( - config=riva.client.RecognitionConfig( - encoding=riva.client.AudioEncoding.LINEAR_PCM, - language_code=self._language_code, - model="", - max_alternatives=1, - profanity_filter=self._profanity_filter, - enable_automatic_punctuation=self._automatic_punctuation, - verbatim_transcripts=not self._no_verbatim_transcripts, - sample_rate_hertz=self.sample_rate, - audio_channel_count=1, - ), - interim_results=True, - ) - - riva.client.add_word_boosting_to_config( - config, self._boosted_lm_words, self._boosted_lm_score - ) - - riva.client.add_endpoint_parameters_to_config( - config, - self._start_history, - self._start_threshold, - self._stop_history, - self._stop_history_eou, - self._stop_threshold, - self._stop_threshold_eou, - ) - riva.client.add_custom_configuration_to_config(config, self._custom_configuration) - - self._config = config self._queue = asyncio.Queue() if not self._thread_task: @@ -250,6 +255,8 @@ class NvidiaSTTService(STTService): self._response_queue = asyncio.Queue() self._response_task = self.create_task(self._response_task_handler()) + logger.debug(f"Initialized NvidiaSTTService with model: {self.model_name}") + async def stop(self, frame: EndFrame): """Stop the NVIDIA Riva STT service and clean up resources. @@ -503,8 +510,6 @@ class NvidiaSegmentedSTTService(SegmentedSTTService): auth = riva.client.Auth(None, self._use_ssl, self._server, metadata) self._asr_service = riva.client.ASRService(auth) - logger.info(f"Initialized NvidiaSegmentedSTTService with model: {self.model_name}") - def _create_recognition_config(self): """Create the NVIDIA Riva ASR recognition configuration.""" # Create base configuration @@ -572,6 +577,7 @@ class NvidiaSegmentedSTTService(SegmentedSTTService): await super().start(frame) self._initialize_client() self._config = self._create_recognition_config() + logger.debug(f"Initialized NvidiaSegmentedSTTService with model: {self.model_name}") async def set_language(self, language: Language): """Set the language for the STT service. @@ -605,21 +611,12 @@ class NvidiaSegmentedSTTService(SegmentedSTTService): Frame: TranscriptionFrame containing the transcribed text. """ try: - await self.start_processing_metrics() - await self.start_ttfb_metrics() - - # Make sure the client is initialized - if self._asr_service is None: - self._initialize_client() - - # Make sure the config is created - if self._config is None: - self._config = self._create_recognition_config() - - # Type assertion to satisfy the IDE assert self._asr_service is not None, "ASR service not initialized" assert self._config is not None, "Recognition config not created" + await self.start_processing_metrics() + await self.start_ttfb_metrics() + # Process audio with NVIDIA Riva ASR - explicitly request non-future response raw_response = self._asr_service.offline_recognize(audio, self._config, future=False) @@ -627,43 +624,40 @@ class NvidiaSegmentedSTTService(SegmentedSTTService): await self.stop_processing_metrics() # Process the response - handle different possible return types - try: - # If it's a future-like object, get the result - if hasattr(raw_response, "result"): - response = raw_response.result() - else: - response = raw_response + # If it's a future-like object, get the result + if hasattr(raw_response, "result"): + response = raw_response.result() + else: + response = raw_response - # Process transcription results - transcription_found = False + # Process transcription results + transcription_found = False - # Now we can safely check results - # Type hint for the IDE - results = getattr(response, "results", []) + # Now we can safely check results + # Type hint for the IDE + results = getattr(response, "results", []) - for result in results: - alternatives = getattr(result, "alternatives", []) - if alternatives: - text = alternatives[0].transcript.strip() - if text: - logger.debug(f"Transcription: [{text}]") - yield TranscriptionFrame( - text, - self._user_id, - time_now_iso8601(), - self._language_enum, - ) - transcription_found = True + for result in results: + alternatives = getattr(result, "alternatives", []) + if alternatives: + text = alternatives[0].transcript.strip() + if text: + logger.debug(f"Transcription: [{text}]") + yield TranscriptionFrame( + text, + self._user_id, + time_now_iso8601(), + self._language_enum, + ) + transcription_found = True - await self._handle_transcription(text, True, self._language_enum) - - if not transcription_found: - logger.debug("No transcription results found in NVIDIA Riva response") - - except AttributeError as ae: - logger.error(f"Unexpected response structure from NVIDIA Riva: {ae}") - yield ErrorFrame(f"Unexpected NVIDIA Riva response format: {str(ae)}") + await self._handle_transcription(text, True, self._language_enum) + if not transcription_found: + logger.debug(f"{self}: No transcription results found in NVIDIA Riva response") + except AttributeError as ae: + logger.error(f"{self}: Unexpected response structure from NVIDIA Riva: {ae}") + yield ErrorFrame(f"{self}: Unexpected NVIDIA Riva response format: {str(ae)}") except Exception as e: logger.error(f"{self} exception: {e}") yield ErrorFrame(error=f"{self} error: {e}") diff --git a/src/pipecat/services/nvidia/tts.py b/src/pipecat/services/nvidia/tts.py index 70be24a8f..d62b7e500 100644 --- a/src/pipecat/services/nvidia/tts.py +++ b/src/pipecat/services/nvidia/tts.py @@ -25,6 +25,7 @@ from pydantic import BaseModel from pipecat.frames.frames import ( ErrorFrame, Frame, + StartFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -93,6 +94,7 @@ class NvidiaTTSService(TTSService): params = params or NvidiaTTSService.InputParams() + self._server = server self._api_key = api_key self._voice_id = voice_id self._language_code = params.language @@ -102,18 +104,8 @@ class NvidiaTTSService(TTSService): self.set_model_name(model_function_map.get("model_name")) self.set_voice(voice_id) - metadata = [ - ["function-id", self._function_id], - ["authorization", f"Bearer {api_key}"], - ] - auth = riva.client.Auth(None, self._use_ssl, server, metadata) - - self._service = riva.client.SpeechSynthesisService(auth) - - # warm up the service - config_response = self._service.stub.GetRivaSynthesisConfig( - riva.client.proto.riva_tts_pb2.RivaSynthesisConfigRequest() - ) + self._service = None + self._config = None async def set_model(self, model: str): """Attempt to set the TTS model. @@ -129,6 +121,39 @@ class NvidiaTTSService(TTSService): f"{self.__class__.__name__}(api_key=, model_function_map={example})" ) + def _initialize_client(self): + if self._service is not None: + return + + metadata = [ + ["function-id", self._function_id], + ["authorization", f"Bearer {self._api_key}"], + ] + auth = riva.client.Auth(None, self._use_ssl, self._server, metadata) + + self._service = riva.client.SpeechSynthesisService(auth) + + def _create_synthesis_config(self): + if not self._service: + return + + # warm up the service + config = self._service.stub.GetRivaSynthesisConfig( + riva.client.proto.riva_tts_pb2.RivaSynthesisConfigRequest() + ) + return config + + async def start(self, frame: StartFrame): + """Start the Cartesia TTS service. + + Args: + frame: The start frame containing initialization parameters. + """ + await super().start(frame) + self._initialize_client() + self._config = self._create_synthesis_config() + logger.debug(f"Initialized NvidiaTTSService with model: {self.model_name}") + @traced_tts async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: """Generate speech from text using NVIDIA Riva TTS. @@ -161,12 +186,15 @@ class NvidiaTTSService(TTSService): logger.error(f"{self} exception: {e}") add_response(None) - await self.start_ttfb_metrics() - yield TTSStartedFrame() - - logger.debug(f"{self}: Generating TTS [{text}]") - try: + assert self._service is not None, "TTS service not initialized" + assert self._config is not None, "Synthesis configuration not created" + + await self.start_ttfb_metrics() + yield TTSStartedFrame() + + logger.debug(f"{self}: Generating TTS [{text}]") + queue = asyncio.Queue() await asyncio.to_thread(read_audio_responses, queue) @@ -181,9 +209,12 @@ class NvidiaTTSService(TTSService): ) yield frame resp = await asyncio.wait_for(queue.get(), timeout=NVIDIA_TTS_TIMEOUT_SECS) + + await self.start_tts_usage_metrics(text) + yield TTSStoppedFrame() except asyncio.TimeoutError: logger.error(f"{self} timeout waiting for audio response") yield ErrorFrame(error=f"{self} error: {e}") - - await self.start_tts_usage_metrics(text) - yield TTSStoppedFrame() + except Exception as e: + logger.error(f"{self} exception: {e}") + yield ErrorFrame(error=f"{self} error: {e}")