From b57a9d254655822ad8a62511152db58f3bdd609d Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 26 Nov 2025 12:15:23 -0500 Subject: [PATCH] Review of services (AsyncGenerator, Fatal=True) --- src/pipecat/services/asyncai/tts.py | 4 ++-- src/pipecat/services/cartesia/tts.py | 4 ++-- src/pipecat/services/gladia/stt.py | 3 +-- src/pipecat/services/google/gemini_live/llm.py | 4 ++-- src/pipecat/services/google/tts.py | 2 +- src/pipecat/services/hume/tts.py | 2 +- src/pipecat/services/minimax/tts.py | 2 +- src/pipecat/services/sarvam/stt.py | 2 +- src/pipecat/services/sarvam/tts.py | 6 +++--- src/pipecat/services/speechmatics/tts.py | 9 +++------ 10 files changed, 17 insertions(+), 21 deletions(-) diff --git a/src/pipecat/services/asyncai/tts.py b/src/pipecat/services/asyncai/tts.py index 4c9c0c96c..2f93f451c 100644 --- a/src/pipecat/services/asyncai/tts.py +++ b/src/pipecat/services/asyncai/tts.py @@ -472,7 +472,7 @@ class AsyncAIHttpTTSService(TTSService): async with self._session.post(url, json=payload, headers=headers) as response: if response.status != 200: error_text = await response.text() - await self.push_error(error_msg=f"Async API error: {error_text}") + yield ErrorFrame(error=f"Async API error: {error_text}") raise Exception(f"Async API returned status {response.status}: {error_text}") audio_data = await response.read() @@ -488,7 +488,7 @@ class AsyncAIHttpTTSService(TTSService): yield frame except Exception as e: - await self.push_error(exception=e) + yield ErrorFrame(error=f"{self} error: {e}") finally: await self.stop_ttfb_metrics() yield TTSStoppedFrame() diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py index 70ce91f30..40bdcc286 100644 --- a/src/pipecat/services/cartesia/tts.py +++ b/src/pipecat/services/cartesia/tts.py @@ -703,7 +703,7 @@ class CartesiaHttpTTSService(TTSService): async with session.post(url, json=payload, headers=headers) as response: if response.status != 200: error_text = await response.text() - await self.push_error(error_msg=f"Cartesia API error: {error_text}") + yield ErrorFrame(error=f"Cartesia API error: {error_text}") raise Exception(f"Cartesia API returned status {response.status}: {error_text}") audio_data = await response.read() @@ -719,7 +719,7 @@ class CartesiaHttpTTSService(TTSService): yield frame except Exception as e: - await self.push_error(exception=e) + yield ErrorFrame(error=f"{self} error: {e}") finally: await self.stop_ttfb_metrics() yield TTSStoppedFrame() diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py index 9b3fb11f7..2a84fac0f 100644 --- a/src/pipecat/services/gladia/stt.py +++ b/src/pipecat/services/gladia/stt.py @@ -630,8 +630,7 @@ class GladiaSTTService(STTService): self._reconnection_attempts += 1 if self._reconnection_attempts > self._max_reconnection_attempts: await self.push_error( - error_msg=f"Max reconnection attempts ({self._max_reconnection_attempts}) reached", - fatal=True, + error_msg=f"Max reconnection attempts ({self._max_reconnection_attempts}) reached" ) self._should_reconnect = False return False diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index fc0fc069a..078f2ac1a 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/src/pipecat/services/google/gemini_live/llm.py @@ -1255,7 +1255,7 @@ class GeminiLiveLLMService(LLMService): f"Max consecutive failures ({MAX_CONSECUTIVE_FAILURES}) reached, " "treating as fatal error" ) - await self.push_error(error_msg=error_msg, exception=error, fatal=True) + await self.push_error(error_msg=error_msg, exception=error) return False else: logger.info( @@ -1742,7 +1742,7 @@ class GeminiLiveLLMService(LLMService): # state management, and that exponential backoff for retries can have # cost/stability implications for a service cluster, let's just treat a # send-side error as fatal. - await self.push_error(ErrorFrame(error=f"{self} Send error: {error}", fatal=True)) + await self.push_error(ErrorFrame(error=f"{self} Send error: {error}")) def create_context_aggregator( self, diff --git a/src/pipecat/services/google/tts.py b/src/pipecat/services/google/tts.py index 8395c68aa..4c48b3c97 100644 --- a/src/pipecat/services/google/tts.py +++ b/src/pipecat/services/google/tts.py @@ -995,7 +995,7 @@ class GoogleTTSService(GoogleBaseTTSService): yield frame except Exception as e: - await self.push_error(exception=e) + yield ErrorFrame(error=f"{self} error: {e}") class GeminiTTSService(GoogleBaseTTSService): diff --git a/src/pipecat/services/hume/tts.py b/src/pipecat/services/hume/tts.py index 999cd460c..6d1590bc9 100644 --- a/src/pipecat/services/hume/tts.py +++ b/src/pipecat/services/hume/tts.py @@ -287,7 +287,7 @@ class HumeTTSService(WordTTSService): self._cumulative_time = utterance_duration except Exception as e: - await self.push_error(exception=e) + yield ErrorFrame(error=f"{self} error: {e}") finally: # Ensure TTFB timer is stopped even on early failures await self.stop_ttfb_metrics() diff --git a/src/pipecat/services/minimax/tts.py b/src/pipecat/services/minimax/tts.py index 992809a83..32d7deb00 100644 --- a/src/pipecat/services/minimax/tts.py +++ b/src/pipecat/services/minimax/tts.py @@ -337,7 +337,7 @@ class MiniMaxHttpTTSService(TTSService): continue except Exception as e: - await self.push_error(exception=e) + yield ErrorFrame(error=f"{self} error: {e}") finally: await self.stop_ttfb_metrics() yield TTSStoppedFrame() diff --git a/src/pipecat/services/sarvam/stt.py b/src/pipecat/services/sarvam/stt.py index 29b6e9a90..19bb7e13c 100644 --- a/src/pipecat/services/sarvam/stt.py +++ b/src/pipecat/services/sarvam/stt.py @@ -275,7 +275,7 @@ class SarvamSTTService(STTService): await self._socket_client.translate(**method_kwargs) except Exception as e: - await self.push_error(error_msg=f"Error sending audio to Sarvam: {e}", exception=e) + yield ErrorFrame(error=f"{self} error: {e}") yield None diff --git a/src/pipecat/services/sarvam/tts.py b/src/pipecat/services/sarvam/tts.py index 3add2c989..3b519e6f6 100644 --- a/src/pipecat/services/sarvam/tts.py +++ b/src/pipecat/services/sarvam/tts.py @@ -254,7 +254,7 @@ class SarvamHttpTTSService(TTSService): async with self._session.post(url, json=payload, headers=headers) as response: if response.status != 200: error_text = await response.text() - await self.push_error(error_msg=f"Sarvam API error: {error_text}") + yield ErrorFrame(error=f"Sarvam API error: {error_text}") return response_data = await response.json() @@ -263,7 +263,7 @@ class SarvamHttpTTSService(TTSService): # Decode base64 audio data if "audios" not in response_data or not response_data["audios"]: - await self.push_error(error_msg="No audio data received") + yield ErrorFrame(error="No audio data received") return # Get the first audio (there should be only one for single text input) @@ -284,7 +284,7 @@ class SarvamHttpTTSService(TTSService): yield frame except Exception as e: - await self.push_error(error_msg=f"Error generating TTS: {e}", exception=e) + yield ErrorFrame(error=f"{self} error: {e}") finally: await self.stop_ttfb_metrics() yield TTSStoppedFrame() diff --git a/src/pipecat/services/speechmatics/tts.py b/src/pipecat/services/speechmatics/tts.py index b8fe172e7..ebba2a030 100644 --- a/src/pipecat/services/speechmatics/tts.py +++ b/src/pipecat/services/speechmatics/tts.py @@ -174,16 +174,13 @@ class SpeechmaticsTTSService(TTSService): except (ValueError, ArithmeticError): yield ErrorFrame( - error=f"{self} Service unavailable [503] (attempts {attempt})", - fatal=True, + error=f"{self} Service unavailable [503] (attempts {attempt})" ) return # != 200 : Error if response.status != 200: - yield ErrorFrame( - error=f"{self} Service unavailable [{response.status}]", fatal=True - ) + yield ErrorFrame(error=f"{self} Service unavailable [{response.status}]") return # Update Pipecat metrics @@ -225,7 +222,7 @@ class SpeechmaticsTTSService(TTSService): break except Exception as e: - yield ErrorFrame(error=f"{self}: Error generating TTS: {e}", fatal=True) + yield ErrorFrame(error=f"{self}: Error generating TTS: {e}") finally: # Emit the TTS stopped frame yield TTSStoppedFrame()