From 1330ef3ad60fe10108b8b345869a50fd7b1e8ee2 Mon Sep 17 00:00:00 2001 From: Filipi Fuchter Date: Wed, 26 Nov 2025 18:34:25 -0300 Subject: [PATCH] Enhanced error handling across the framework. Co-authored-by: Mark Backman --- CHANGELOG.md | 11 ++++ src/pipecat/frames/frames.py | 2 + .../processors/filters/wake_check_filter.py | 4 +- src/pipecat/processors/frame_processor.py | 54 ++++++++++++++++--- .../processors/frameworks/langchain.py | 4 +- .../processors/frameworks/strands_agents.py | 4 +- src/pipecat/serializers/plivo.py | 2 +- src/pipecat/serializers/telnyx.py | 2 +- src/pipecat/serializers/twilio.py | 2 +- src/pipecat/services/ai_service.py | 2 +- src/pipecat/services/anthropic/llm.py | 3 +- src/pipecat/services/assemblyai/stt.py | 18 +++---- src/pipecat/services/asyncai/tts.py | 23 +++----- src/pipecat/services/aws/llm.py | 2 +- src/pipecat/services/aws/nova_sonic/llm.py | 6 +-- src/pipecat/services/aws/stt.py | 28 ++++------ src/pipecat/services/aws/tts.py | 1 - src/pipecat/services/azure/image.py | 2 - src/pipecat/services/azure/realtime/llm.py | 2 +- src/pipecat/services/azure/stt.py | 8 +-- src/pipecat/services/azure/tts.py | 12 ++--- src/pipecat/services/cartesia/stt.py | 9 ++-- src/pipecat/services/cartesia/tts.py | 23 +++----- src/pipecat/services/deepgram/flux/stt.py | 16 ++---- src/pipecat/services/deepgram/stt.py | 4 +- src/pipecat/services/deepgram/tts.py | 4 +- src/pipecat/services/elevenlabs/stt.py | 17 +++--- src/pipecat/services/elevenlabs/tts.py | 25 +++------ src/pipecat/services/fal/image.py | 1 - src/pipecat/services/fal/stt.py | 3 +- src/pipecat/services/fish/tts.py | 15 ++---- src/pipecat/services/gladia/stt.py | 13 +++-- .../services/google/gemini_live/llm.py | 10 ++-- src/pipecat/services/google/image.py | 2 - src/pipecat/services/google/llm.py | 4 +- src/pipecat/services/google/stt.py | 12 ++--- src/pipecat/services/google/tts.py | 6 +-- src/pipecat/services/groq/tts.py | 3 +- src/pipecat/services/heygen/client.py | 2 +- src/pipecat/services/hume/tts.py | 3 +- src/pipecat/services/inworld/tts.py | 5 +- src/pipecat/services/lmnt/tts.py | 15 ++---- src/pipecat/services/mcp_service.py | 4 -- src/pipecat/services/mem0/memory.py | 5 +- src/pipecat/services/minimax/tts.py | 4 +- src/pipecat/services/moondream/vision.py | 1 - src/pipecat/services/neuphonic/tts.py | 19 +++---- src/pipecat/services/openai/image.py | 1 - src/pipecat/services/openai/realtime/llm.py | 11 ++-- src/pipecat/services/openai/tts.py | 3 +- .../services/openai_realtime_beta/azure.py | 2 +- .../services/openai_realtime_beta/openai.py | 9 ++-- src/pipecat/services/piper/tts.py | 5 +- src/pipecat/services/playht/tts.py | 18 +++---- src/pipecat/services/rime/tts.py | 19 +++---- src/pipecat/services/riva/stt.py | 4 +- src/pipecat/services/riva/tts.py | 3 +- src/pipecat/services/sarvam/stt.py | 19 +++---- src/pipecat/services/sarvam/tts.py | 31 +++++------ src/pipecat/services/simli/video.py | 6 +-- src/pipecat/services/soniox/stt.py | 15 ++---- src/pipecat/services/speechmatics/stt.py | 14 +++-- src/pipecat/services/speechmatics/tts.py | 11 ++-- src/pipecat/services/stt_service.py | 2 +- src/pipecat/services/tts_service.py | 4 +- src/pipecat/services/ultravox/stt.py | 13 ++--- src/pipecat/services/whisper/base_stt.py | 3 +- src/pipecat/services/whisper/stt.py | 4 +- src/pipecat/services/xtts/tts.py | 8 +-- src/pipecat/transports/daily/transport.py | 9 ++-- .../transports/smallwebrtc/connection.py | 2 +- src/pipecat/transports/tavus/transport.py | 2 +- src/pipecat/utils/asyncio/task_manager.py | 4 +- src/pipecat/utils/base_object.py | 2 +- 74 files changed, 268 insertions(+), 373 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a3adbdc38..9b32f1884 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,17 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Enhanced error handling across the framework: + - Added `on_error` callback to `FrameProcessor` for centralized error handling. + - Renamed `push_error(error: ErrorFrame)` to `push_error_frame(error: ErrorFrame)` + for clarity. + - Added new `push_error` method for simplified error reporting: + ``` + push_error(error_msg: str, exception: Optional[Exception] = None, fatal: bool = False) + ``` + - Standardized error logging by replacing `logger.exception` calls with + `logger.error` throughout the codebase. + - Added `cache_read_input_tokens`, `cache_creation_input_tokens` and `reasoning_tokens` to OTel spans for LLM call diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e437d48a1..60530e39a 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -835,11 +835,13 @@ class ErrorFrame(SystemFrame): error: Description of the error that occurred. fatal: Whether the error is fatal and requires bot shutdown. processor: The frame processor that generated the error. + exception: The exception that occurred. """ error: str fatal: bool = False processor: Optional["FrameProcessor"] = None + exception: Optional[Exception] = None def __str__(self): return f"{self.name}(error: {self.error}, fatal: {self.fatal})" diff --git a/src/pipecat/processors/filters/wake_check_filter.py b/src/pipecat/processors/filters/wake_check_filter.py index 1f01d0a1b..3fe960b55 100644 --- a/src/pipecat/processors/filters/wake_check_filter.py +++ b/src/pipecat/processors/filters/wake_check_filter.py @@ -126,6 +126,4 @@ class WakeCheckFilter(FrameProcessor): else: await self.push_frame(frame, direction) except Exception as e: - error_msg = f"Error in wake word filter: {e}" - logger.exception(error_msg) - await self.push_error(ErrorFrame(error_msg)) + await self.push_error(error_msg=f"Error in wake word filter: {e}", exception=e) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 69aee0acd..63390cf94 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -142,6 +142,7 @@ class FrameProcessor(BaseObject): - on_after_process_frame: Called after a frame is processed - on_before_push_frame: Called before a frame is pushed - on_after_push_frame: Called after a frame is pushed + - on_error: Called when an error is raised in the frame processing. """ def __init__( @@ -234,6 +235,7 @@ class FrameProcessor(BaseObject): self._register_event_handler("on_after_process_frame", sync=True) self._register_event_handler("on_before_push_frame", sync=True) self._register_event_handler("on_after_push_frame", sync=True) + self._register_event_handler("on_error", sync=True) @property def id(self) -> int: @@ -630,7 +632,43 @@ class FrameProcessor(BaseObject): elif isinstance(frame, (FrameProcessorResumeFrame, FrameProcessorResumeUrgentFrame)): await self.__resume(frame) - async def push_error(self, error: ErrorFrame): + async def push_error( + self, + error_msg: str, + exception: Optional[Exception] = None, + fatal: bool = False, + ): + """Creates and pushes an ErrorFrame upstream. + + Creates and pushes an ErrorFrame upstream to notify other processors in the + pipeline about an error condition. The error frame will include context about + which processor generated the error. + + Args: + error_msg: Descriptive message explaining the error condition. + exception: Optional exception object that caused the error, if available. + This provides additional context for debugging and error handling. + fatal: Whether this error should be considered fatal to the pipeline. + Fatal errors typically cause the entire pipeline to stop processing. + Defaults to False for non-fatal errors. + + Example:: + + ```python + # Non-fatal error + await self.push_error("Failed to process audio chunk, skipping") + + # Fatal error with exception context + try: + result = some_critical_operation() + except Exception as e: + await self.push_error("Critical operation failed", exception=e, fatal=True) + ``` + """ + error_frame = ErrorFrame(error=error_msg, fatal=fatal, exception=exception, processor=self) + await self.push_error_frame(error=error_frame) + + async def push_error_frame(self, error: ErrorFrame): """Push an error frame upstream. Args: @@ -638,6 +676,8 @@ class FrameProcessor(BaseObject): """ if not error.processor: error.processor = self + await self._call_event_handler("on_error", error) + logger.error(f"{error.processor} error: {error.error}") await self.push_frame(error, FrameDirection.UPSTREAM) async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): @@ -759,8 +799,10 @@ class FrameProcessor(BaseObject): await self.__cancel_process_task() self.__create_process_task() except Exception as e: - logger.exception(f"Uncaught exception in {self} when handling _start_interruption: {e}") - await self.push_error(ErrorFrame(str(e))) + await self.push_error( + error_msg=f"Uncaught exception handling _start_interruption: {e}", + exception=e, + ) async def __internal_push_frame(self, frame: Frame, direction: FrameDirection): """Internal method to push frames to adjacent processors. @@ -797,8 +839,7 @@ class FrameProcessor(BaseObject): await self._observer.on_push_frame(data) await self._prev.queue_frame(frame, direction) except Exception as e: - logger.exception(f"Uncaught exception in {self}: {e}") - await self.push_error(ErrorFrame(str(e))) + await self.push_error(error_msg=f"Uncaught exception: {e}", exception=e) def _check_started(self, frame: Frame): """Check if the processor has been started. @@ -874,8 +915,7 @@ class FrameProcessor(BaseObject): await self._call_event_handler("on_after_process_frame", frame) except Exception as e: - logger.exception(f"{self}: error processing frame: {e}") - await self.push_error(ErrorFrame(str(e))) + await self.push_error(error_msg=f"Error processing frame: {e}", exception=e) async def __input_frame_task_handler(self): """Handle frames from the input queue. diff --git a/src/pipecat/processors/frameworks/langchain.py b/src/pipecat/processors/frameworks/langchain.py index b8a472a3d..08a84760f 100644 --- a/src/pipecat/processors/frameworks/langchain.py +++ b/src/pipecat/processors/frameworks/langchain.py @@ -24,7 +24,7 @@ try: from langchain_core.messages import AIMessageChunk from langchain_core.runnables import Runnable except ModuleNotFoundError as e: - logger.exception("In order to use Langchain, you need to `pip install pipecat-ai[langchain]`. ") + logger.error("In order to use Langchain, you need to `pip install pipecat-ai[langchain]`. ") raise Exception(f"Missing module: {e}") @@ -113,6 +113,6 @@ class LangchainProcessor(FrameProcessor): except GeneratorExit: logger.warning(f"{self} generator was closed prematurely") except Exception as e: - logger.exception(f"{self} an unknown error occurred: {e}") + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: await self.push_frame(LLMFullResponseEndFrame()) diff --git a/src/pipecat/processors/frameworks/strands_agents.py b/src/pipecat/processors/frameworks/strands_agents.py index 323a4e921..8022f5387 100644 --- a/src/pipecat/processors/frameworks/strands_agents.py +++ b/src/pipecat/processors/frameworks/strands_agents.py @@ -23,7 +23,7 @@ try: from strands import Agent from strands.multiagent.graph import Graph except ModuleNotFoundError as e: - logger.exception("In order to use Strands Agents, you need to `pip install strands-agents`.") + logger.error("In order to use Strands Agents, you need to `pip install strands-agents`.") raise Exception(f"Missing module: {e}") @@ -143,7 +143,7 @@ class StrandsAgentsProcessor(FrameProcessor): except GeneratorExit: logger.warning(f"{self} generator was closed prematurely") except Exception as e: - logger.exception(f"{self} an unknown error occurred: {e}") + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: if ttfb_tracking: await self.stop_ttfb_metrics() diff --git a/src/pipecat/serializers/plivo.py b/src/pipecat/serializers/plivo.py index 51ae11151..b49acc664 100644 --- a/src/pipecat/serializers/plivo.py +++ b/src/pipecat/serializers/plivo.py @@ -199,7 +199,7 @@ class PlivoFrameSerializer(FrameSerializer): ) except Exception as e: - logger.exception(f"Failed to hang up Plivo call: {e}") + logger.error(f"Failed to hang up Plivo call: {e}") async def deserialize(self, data: str | bytes) -> Frame | None: """Deserializes Plivo WebSocket data to Pipecat frames. diff --git a/src/pipecat/serializers/telnyx.py b/src/pipecat/serializers/telnyx.py index 7603c9bf7..26dca7dac 100644 --- a/src/pipecat/serializers/telnyx.py +++ b/src/pipecat/serializers/telnyx.py @@ -225,7 +225,7 @@ class TelnyxFrameSerializer(FrameSerializer): ) except Exception as e: - logger.exception(f"Failed to hang up Telnyx call: {e}") + logger.error(f"Failed to hang up Telnyx call: {e}") async def deserialize(self, data: str | bytes) -> Frame | None: """Deserializes Telnyx WebSocket data to Pipecat frames. diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py index d61b29295..240fb2853 100644 --- a/src/pipecat/serializers/twilio.py +++ b/src/pipecat/serializers/twilio.py @@ -236,7 +236,7 @@ class TwilioFrameSerializer(FrameSerializer): ) except Exception as e: - logger.exception(f"Failed to hang up Twilio call: {e}") + logger.error(f"Failed to hang up Twilio call: {e}") async def deserialize(self, data: str | bytes) -> Frame | None: """Deserializes Twilio WebSocket data to Pipecat frames. diff --git a/src/pipecat/services/ai_service.py b/src/pipecat/services/ai_service.py index 759efcf00..70d53822e 100644 --- a/src/pipecat/services/ai_service.py +++ b/src/pipecat/services/ai_service.py @@ -166,6 +166,6 @@ class AIService(FrameProcessor): async for f in generator: if f: if isinstance(f, ErrorFrame): - await self.push_error(f) + await self.push_error_frame(f) else: await self.push_frame(f) diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 2e3e0272d..a5c67e90e 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -458,8 +458,7 @@ class AnthropicLLMService(LLMService): except httpx.TimeoutException: await self._call_event_handler("on_completion_timeout") except Exception as e: - logger.exception(f"{self} exception: {e}") - await self.push_error(ErrorFrame(f"{e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: await self.stop_processing_metrics() await self.push_frame(LLMFullResponseEndFrame()) diff --git a/src/pipecat/services/assemblyai/stt.py b/src/pipecat/services/assemblyai/stt.py index d78a42841..24251319f 100644 --- a/src/pipecat/services/assemblyai/stt.py +++ b/src/pipecat/services/assemblyai/stt.py @@ -206,9 +206,8 @@ class AssemblyAISTTService(STTService): await self._call_event_handler("on_connected") except Exception as e: - logger.error(f"{self} exception: {e}") self._connected = False - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) raise async def _disconnect(self): @@ -233,8 +232,7 @@ class AssemblyAISTTService(STTService): logger.warning("Timed out waiting for termination message from server") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) if self._receive_task: await self.cancel_task(self._receive_task) @@ -242,8 +240,7 @@ class AssemblyAISTTService(STTService): await self._websocket.close() except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: self._websocket = None @@ -262,13 +259,11 @@ class AssemblyAISTTService(STTService): except websockets.exceptions.ConnectionClosedOK: break except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) break except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) def _parse_message(self, message: Dict[str, Any]) -> BaseMessage: """Parse a raw message into the appropriate message type.""" @@ -297,8 +292,7 @@ class AssemblyAISTTService(STTService): elif isinstance(parsed_message, TerminationMessage): await self._handle_termination(parsed_message) except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) async def _handle_termination(self, message: TerminationMessage): """Handle termination message.""" diff --git a/src/pipecat/services/asyncai/tts.py b/src/pipecat/services/asyncai/tts.py index f916d9bdd..3b3faca57 100644 --- a/src/pipecat/services/asyncai/tts.py +++ b/src/pipecat/services/asyncai/tts.py @@ -228,8 +228,7 @@ class AsyncAITTSService(InterruptibleTTSService): await self._call_event_handler("on_connected") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) self._websocket = None await self._call_event_handler("on_connection_error", f"{e}") @@ -241,8 +240,7 @@ class AsyncAITTSService(InterruptibleTTSService): logger.debug("Disconnecting from Async") await self._websocket.close() except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: self._websocket = None self._started = False @@ -287,12 +285,11 @@ class AsyncAITTSService(InterruptibleTTSService): ) await self.push_frame(frame) elif msg.get("error_code"): - logger.error(f"{self} error: {msg}") await self.push_frame(TTSStoppedFrame()) await self.stop_all_metrics() - await self.push_error(ErrorFrame(error=f"{self} error: {msg['message']}")) + await self.push_error(error_msg=f"Error: {msg['message']}") else: - logger.error(f"{self} error, unknown message type: {msg}") + await self.push_error(error_msg=f"Unknown message type: {msg}") async def _keepalive_task_handler(self): """Send periodic keepalive messages to maintain WebSocket connection.""" @@ -335,16 +332,14 @@ class AsyncAITTSService(InterruptibleTTSService): await self._get_websocket().send(msg) await self.start_tts_usage_metrics(text) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") yield TTSStoppedFrame() await self._disconnect() await self._connect() return yield None except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") class AsyncAIHttpTTSService(TTSService): @@ -477,8 +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() - logger.error(f"Async API error: {error_text}") - await self.push_error(ErrorFrame(error=f"Async API error: {error_text}")) + await self.push_error(error_msg=f"Async API error: {error_text}") raise Exception(f"Async API returned status {response.status}: {error_text}") audio_data = await response.read() @@ -494,8 +488,7 @@ class AsyncAIHttpTTSService(TTSService): yield frame except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: await self.stop_ttfb_metrics() yield TTSStoppedFrame() diff --git a/src/pipecat/services/aws/llm.py b/src/pipecat/services/aws/llm.py index ccbac43b7..a9d9af716 100644 --- a/src/pipecat/services/aws/llm.py +++ b/src/pipecat/services/aws/llm.py @@ -1136,7 +1136,7 @@ class AWSBedrockLLMService(LLMService): except (ReadTimeoutError, asyncio.TimeoutError): await self._call_event_handler("on_completion_timeout") except Exception as e: - logger.exception(f"{self} exception: {e}") + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: await self.stop_processing_metrics() await self.push_frame(LLMFullResponseEndFrame()) diff --git a/src/pipecat/services/aws/nova_sonic/llm.py b/src/pipecat/services/aws/nova_sonic/llm.py index 95240f748..d08d2603f 100644 --- a/src/pipecat/services/aws/nova_sonic/llm.py +++ b/src/pipecat/services/aws/nova_sonic/llm.py @@ -453,7 +453,7 @@ class AWSNovaSonicLLMService(LLMService): self._ready_to_send_context = True await self._finish_connecting_if_context_available() except Exception as e: - logger.error(f"{self} initialization error: {e}") + await self.push_error(error_msg=f"Initialization error: {e}", exception=e) await self._disconnect() async def _process_completed_function_calls(self, send_new_results: bool): @@ -577,7 +577,7 @@ class AWSNovaSonicLLMService(LLMService): logger.info("Finished disconnecting") except Exception as e: - logger.error(f"{self} error disconnecting: {e}") + await self.push_error(error_msg=f"Error disconnecting: {e}", exception=e) def _create_client(self) -> BedrockRuntimeClient: config = Config( @@ -885,7 +885,7 @@ class AWSNovaSonicLLMService(LLMService): # Errors are kind of expected while disconnecting, so just # ignore them and do nothing return - logger.error(f"{self} error processing responses: {e}") + await self.push_error(error_msg=f"Error processing responses: {e}", exception=e) if self._wants_connection: await self.reset_conversation() diff --git a/src/pipecat/services/aws/stt.py b/src/pipecat/services/aws/stt.py index 8d89d3198..50a8f161a 100644 --- a/src/pipecat/services/aws/stt.py +++ b/src/pipecat/services/aws/stt.py @@ -140,8 +140,7 @@ class AWSTranscribeSTTService(STTService): return logger.warning("WebSocket connection not established after connect") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) retry_count += 1 if retry_count < max_retries: await asyncio.sleep(1) # Wait before retrying @@ -182,8 +181,7 @@ class AWSTranscribeSTTService(STTService): try: await self._connect() except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") return # Format the audio data according to AWS event stream format @@ -200,13 +198,11 @@ class AWSTranscribeSTTService(STTService): await self._disconnect() # Don't yield error here - we'll retry on next frame except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") await self._disconnect() except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") await self._disconnect() async def _connect(self): @@ -289,8 +285,7 @@ class AWSTranscribeSTTService(STTService): await self._call_event_handler("on_connected") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) await self._disconnect() raise @@ -310,8 +305,7 @@ class AWSTranscribeSTTService(STTService): await self._ws_client.send(json.dumps(end_stream)) await self._ws_client.close() except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: self._ws_client = None await self._call_event_handler("on_disconnected") @@ -529,15 +523,15 @@ class AWSTranscribeSTTService(STTService): ) elif headers.get(":message-type") == "exception": error_msg = payload.get("Message", "Unknown error") - logger.error(f"{self} Exception from AWS: {error_msg}") - await self.push_frame(ErrorFrame(f"AWS Transcribe error: {error_msg}")) + await self.push_error(error_msg=f"AWS Transcribe error: {error_msg}") else: logger.debug(f"{self} Other message type received: {headers}") logger.debug(f"{self} Payload: {payload}") except websockets.exceptions.ConnectionClosed as e: - logger.error(f"{self} WebSocket connection closed in receive loop: {e}") + await self.push_error( + error_msg=f"WebSocket connection closed in receive loop", exception=e + ) break except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) break diff --git a/src/pipecat/services/aws/tts.py b/src/pipecat/services/aws/tts.py index f22c42399..7e319abea 100644 --- a/src/pipecat/services/aws/tts.py +++ b/src/pipecat/services/aws/tts.py @@ -312,7 +312,6 @@ class AWSPollyTTSService(TTSService): yield TTSStoppedFrame() except (BotoCoreError, ClientError) as error: - logger.exception(f"{self} error generating TTS: {error}") error_message = f"AWS Polly TTS error: {str(error)}" yield ErrorFrame(error=error_message) diff --git a/src/pipecat/services/azure/image.py b/src/pipecat/services/azure/image.py index f07b86f83..6b0b140e0 100644 --- a/src/pipecat/services/azure/image.py +++ b/src/pipecat/services/azure/image.py @@ -91,7 +91,6 @@ class AzureImageGenServiceREST(ImageGenService): while status != "succeeded": attempts_left -= 1 if attempts_left == 0: - logger.error(f"{self} error: image generation timed out") yield ErrorFrame("Image generation timed out") return @@ -104,7 +103,6 @@ class AzureImageGenServiceREST(ImageGenService): image_url = json_response["result"]["data"][0]["url"] if json_response else None if not image_url: - logger.error(f"{self} error: image generation failed") yield ErrorFrame("Image generation failed") return diff --git a/src/pipecat/services/azure/realtime/llm.py b/src/pipecat/services/azure/realtime/llm.py index 66ba95eea..b06a05c64 100644 --- a/src/pipecat/services/azure/realtime/llm.py +++ b/src/pipecat/services/azure/realtime/llm.py @@ -61,5 +61,5 @@ class AzureRealtimeLLMService(OpenAIRealtimeLLMService): ) self._receive_task = self.create_task(self._receive_task_handler()) except Exception as e: - logger.error(f"{self} initialization error: {e}") + await self.push_error(error_msg=f"initialization error: {e}", exception=e) self._websocket = None diff --git a/src/pipecat/services/azure/stt.py b/src/pipecat/services/azure/stt.py index 85a0508a0..d8186b354 100644 --- a/src/pipecat/services/azure/stt.py +++ b/src/pipecat/services/azure/stt.py @@ -121,8 +121,7 @@ class AzureSTTService(STTService): self._audio_stream.write(audio) yield None except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") async def start(self, frame: StartFrame): """Start the speech recognition service. @@ -151,8 +150,9 @@ class AzureSTTService(STTService): self._speech_recognizer.recognized.connect(self._on_handle_recognized) self._speech_recognizer.start_continuous_recognition_async() except Exception as e: - logger.error(f"{self} exception during initialization: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error( + error_msg=f"Uncaught exception during initialization: {e}", exception=e + ) async def stop(self, frame: EndFrame): """Stop the speech recognition service. diff --git a/src/pipecat/services/azure/tts.py b/src/pipecat/services/azure/tts.py index a1040f312..154930fac 100644 --- a/src/pipecat/services/azure/tts.py +++ b/src/pipecat/services/azure/tts.py @@ -327,7 +327,6 @@ class AzureTTSService(AzureBaseTTSService): try: if self._speech_synthesizer is None: error_msg = "Speech synthesizer not initialized." - logger.error(error_msg) yield ErrorFrame(error=error_msg) return @@ -355,15 +354,13 @@ class AzureTTSService(AzureBaseTTSService): yield TTSStoppedFrame() except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") yield TTSStoppedFrame() # Could add reconnection logic here if needed return except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") class AzureHttpTTSService(AzureBaseTTSService): @@ -440,5 +437,6 @@ class AzureHttpTTSService(AzureBaseTTSService): cancellation_details = result.cancellation_details logger.warning(f"Speech synthesis canceled: {cancellation_details.reason}") if cancellation_details.reason == CancellationReason.Error: - logger.error(f"{self} error: {cancellation_details.error_details}") - yield ErrorFrame(error=f"{self} error: {cancellation_details.error_details}") + yield ErrorFrame( + error=f"Unknown error occurred: {cancellation_details.error_details}" + ) diff --git a/src/pipecat/services/cartesia/stt.py b/src/pipecat/services/cartesia/stt.py index a2ae9432f..135f4a347 100644 --- a/src/pipecat/services/cartesia/stt.py +++ b/src/pipecat/services/cartesia/stt.py @@ -276,8 +276,7 @@ class CartesiaSTTService(WebsocketSTTService): self._websocket = await websocket_connect(ws_url, additional_headers=headers) await self._call_event_handler("on_connected") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) async def _disconnect_websocket(self): try: @@ -285,8 +284,7 @@ class CartesiaSTTService(WebsocketSTTService): logger.debug("Disconnecting from Cartesia STT") await self._websocket.close() except Exception as e: - logger.error(f"{self} error closing websocket: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Error closing websocket: {e}", exception=e) finally: self._websocket = None await self._call_event_handler("on_disconnected") @@ -319,8 +317,7 @@ class CartesiaSTTService(WebsocketSTTService): elif data["type"] == "error": error_msg = data.get("message", "Unknown error") - logger.error(f"Cartesia error: {error_msg}") - await self.push_error(ErrorFrame(error=error_msg)) + await self.push_error(error_msg=error_msg) @traced_stt async def _handle_transcription( diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py index d42802cf2..9a47d9237 100644 --- a/src/pipecat/services/cartesia/tts.py +++ b/src/pipecat/services/cartesia/tts.py @@ -497,8 +497,7 @@ class CartesiaTTSService(AudioContextWordTTSService): ) await self._call_event_handler("on_connected") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) self._websocket = None await self._call_event_handler("on_connection_error", f"{e}") @@ -510,8 +509,7 @@ class CartesiaTTSService(AudioContextWordTTSService): logger.debug("Disconnecting from Cartesia") await self._websocket.close() except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: self._context_id = None self._websocket = None @@ -564,13 +562,12 @@ class CartesiaTTSService(AudioContextWordTTSService): ) await self.append_to_audio_context(msg["context_id"], frame) elif msg["type"] == "error": - logger.error(f"{self} error: {msg}") await self.push_frame(TTSStoppedFrame()) await self.stop_all_metrics() - await self.push_error(ErrorFrame(error=f"{self} error: {msg['error']}")) + await self.push_error(error_msg=f"Error: {msg}") self._context_id = None else: - logger.error(f"{self} error, unknown message type: {msg}") + await self.push_error(error_msg=f"Error, unknown message type: {msg}") async def _receive_messages(self): while True: @@ -608,16 +605,14 @@ class CartesiaTTSService(AudioContextWordTTSService): await self._get_websocket().send(msg) await self.start_tts_usage_metrics(text) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") yield TTSStoppedFrame() await self._disconnect() await self._connect() return yield None except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") class CartesiaHttpTTSService(TTSService): @@ -808,8 +803,7 @@ class CartesiaHttpTTSService(TTSService): async with session.post(url, json=payload, headers=headers) as response: if response.status != 200: error_text = await response.text() - logger.error(f"Cartesia API error: {error_text}") - await self.push_error(ErrorFrame(error=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() @@ -825,8 +819,7 @@ class CartesiaHttpTTSService(TTSService): yield frame except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + yield ErrorFrame(error=f"Unknown error occurred: {e}") finally: await self.stop_ttfb_metrics() yield TTSStoppedFrame() diff --git a/src/pipecat/services/deepgram/flux/stt.py b/src/pipecat/services/deepgram/flux/stt.py index a0d45afe4..536ffdd77 100644 --- a/src/pipecat/services/deepgram/flux/stt.py +++ b/src/pipecat/services/deepgram/flux/stt.py @@ -192,8 +192,7 @@ class DeepgramFluxSTTService(WebsocketSTTService): try: await self._disconnect_websocket() except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: # Reset state only after everything is cleaned up self._websocket = None @@ -251,8 +250,7 @@ class DeepgramFluxSTTService(WebsocketSTTService): logger.debug("Connected to Deepgram Flux Websocket") await self._call_event_handler("on_connected") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) self._websocket = None await self._call_event_handler("on_connection_error", f"{e}") @@ -280,8 +278,7 @@ class DeepgramFluxSTTService(WebsocketSTTService): logger.debug("Disconnecting from Deepgram Flux Websocket") await self._websocket.close() except Exception as e: - logger.error(f"{self} error closing websocket: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Error closing websocket: {e}", exception=e) finally: self._websocket = None await self._call_event_handler("on_disconnected") @@ -381,7 +378,6 @@ class DeepgramFluxSTTService(WebsocketSTTService): are issues sending the audio data. """ if not self._websocket: - logger.error("Not connected to Deepgram Flux.") yield ErrorFrame("Not connected to Deepgram Flux.") return @@ -389,8 +385,7 @@ class DeepgramFluxSTTService(WebsocketSTTService): self._last_stt_time = time.monotonic() await self.send_with_retry(audio, self._report_error) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") return yield None @@ -467,8 +462,7 @@ class DeepgramFluxSTTService(WebsocketSTTService): # Skip malformed messages continue except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) # Error will be handled inside WebsocketService->_receive_task_handler raise else: diff --git a/src/pipecat/services/deepgram/stt.py b/src/pipecat/services/deepgram/stt.py index 0a4cd7e6e..6513fb186 100644 --- a/src/pipecat/services/deepgram/stt.py +++ b/src/pipecat/services/deepgram/stt.py @@ -233,7 +233,7 @@ class DeepgramSTTService(STTService): ) if not await self._connection.start(options=self._settings, addons=self._addons): - logger.error(f"{self}: unable to connect to Deepgram") + await self.push_error(error_msg=f"Unable to connect to Deepgram") async def _disconnect(self): if await self._connection.is_connected(): @@ -256,7 +256,7 @@ class DeepgramSTTService(STTService): async def _on_error(self, *args, **kwargs): error: ErrorResponse = kwargs["error"] logger.warning(f"{self} connection error, will retry: {error}") - await self.push_error(ErrorFrame(error=f"{error}")) + await self.push_error(error_msg=f"{error}") await self.stop_all_metrics() # NOTE(aleix): we don't disconnect (i.e. call finish on the connection) # because this triggers more errors internally in the Deepgram SDK. So, diff --git a/src/pipecat/services/deepgram/tts.py b/src/pipecat/services/deepgram/tts.py index 08e0ab2c9..d52ea6799 100644 --- a/src/pipecat/services/deepgram/tts.py +++ b/src/pipecat/services/deepgram/tts.py @@ -290,8 +290,7 @@ class DeepgramTTSService(WebsocketTTSService): yield None except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") class DeepgramHttpTTSService(TTSService): @@ -401,5 +400,4 @@ class DeepgramHttpTTSService(TTSService): yield TTSStoppedFrame() except Exception as e: - logger.exception(f"{self} exception: {e}") yield ErrorFrame(f"Error getting audio: {str(e)}") diff --git a/src/pipecat/services/elevenlabs/stt.py b/src/pipecat/services/elevenlabs/stt.py index 95a802edd..5fa04d1c1 100644 --- a/src/pipecat/services/elevenlabs/stt.py +++ b/src/pipecat/services/elevenlabs/stt.py @@ -351,8 +351,7 @@ class ElevenLabsSTTService(SegmentedSTTService): ) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") def audio_format_from_sample_rate(sample_rate: int) -> str: @@ -598,7 +597,6 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService): } await self._websocket.send(json.dumps(message)) except Exception as e: - logger.error(f"Error sending audio: {e}") yield ErrorFrame(f"ElevenLabs Realtime STT error: {str(e)}") yield None @@ -663,8 +661,9 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService): await self._call_event_handler("on_connected") logger.debug("Connected to ElevenLabs Realtime STT") except Exception as e: - logger.error(f"{self}: unable to connect to ElevenLabs Realtime STT: {e}") - await self.push_error(ErrorFrame(f"Connection error: {str(e)}")) + await self.push_error( + error_msg=f"Unable to connect to ElevenLabs Realtime STT: {e}", exception=e + ) async def _disconnect_websocket(self): """Disconnect from ElevenLabs Realtime STT WebSocket.""" @@ -673,7 +672,7 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService): logger.debug("Disconnecting from ElevenLabs Realtime STT") await self._websocket.close() except Exception as e: - logger.error(f"{self} error closing websocket: {e}") + await self.push_error(error_msg=f"Error closing websocket: {e}", exception=e) finally: self._websocket = None await self._call_event_handler("on_disconnected") @@ -733,17 +732,17 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService): elif message_type == "error": error_msg = data.get("error", "Unknown error") logger.error(f"ElevenLabs error: {error_msg}") - await self.push_error(ErrorFrame(f"Error: {error_msg}")) + await self.push_error(error_msg=f"Error: {error_msg}") elif message_type == "auth_error": error_msg = data.get("error", "Authentication error") logger.error(f"ElevenLabs auth error: {error_msg}") - await self.push_error(ErrorFrame(f"Auth error: {error_msg}")) + await self.push_error(error_msg=f"Auth error: {error_msg}") elif message_type == "quota_exceeded_error": error_msg = data.get("error", "Quota exceeded") logger.error(f"ElevenLabs quota exceeded: {error_msg}") - await self.push_error(ErrorFrame(f"Quota exceeded: {error_msg}")) + await self.push_error(error_msg=f"Quota exceeded: {error_msg}") else: logger.debug(f"Unknown message type: {message_type}") diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index bbe05f9dc..4a7dd0a21 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -424,8 +424,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService): json.dumps({"context_id": self._context_id, "close_context": True}) ) except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) self._context_id = None self._started = False @@ -536,9 +535,8 @@ class ElevenLabsTTSService(AudioContextWordTTSService): await self._call_event_handler("on_connected") except Exception as e: - logger.error(f"{self} exception: {e}") self._websocket = None - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) await self._call_event_handler("on_connection_error", f"{e}") async def _disconnect_websocket(self): @@ -553,8 +551,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService): await self._websocket.close() logger.debug("Disconnected from ElevenLabs") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: self._started = False self._context_id = None @@ -584,8 +581,7 @@ class ElevenLabsTTSService(AudioContextWordTTSService): json.dumps({"context_id": self._context_id, "close_context": True}) ) except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) self._context_id = None self._started = False self._partial_word = "" @@ -740,15 +736,13 @@ class ElevenLabsTTSService(AudioContextWordTTSService): else: await self._send_text(text) except Exception as e: - logger.error(f"{self} exception: {e}") yield TTSStoppedFrame() - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") self._started = False return yield None except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") class ElevenLabsHttpTTSService(WordTTSService): @@ -1043,7 +1037,6 @@ class ElevenLabsHttpTTSService(WordTTSService): ) as response: if response.status != 200: error_text = await response.text() - logger.error(f"{self} error: {error_text}") yield ErrorFrame(error=f"ElevenLabs API error: {error_text}") return @@ -1091,8 +1084,7 @@ class ElevenLabsHttpTTSService(WordTTSService): logger.warning(f"Failed to parse JSON from stream: {e}") continue except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") continue # After processing all chunks, emit any remaining partial word @@ -1116,8 +1108,7 @@ class ElevenLabsHttpTTSService(WordTTSService): self._previous_text = text except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") finally: await self.stop_ttfb_metrics() # Let the parent class handle TTSStoppedFrame diff --git a/src/pipecat/services/fal/image.py b/src/pipecat/services/fal/image.py index c110ec3be..6bca6701a 100644 --- a/src/pipecat/services/fal/image.py +++ b/src/pipecat/services/fal/image.py @@ -110,7 +110,6 @@ class FalImageGenService(ImageGenService): image_url = response["images"][0]["url"] if response else None if not image_url: - logger.error(f"{self} error: image generation failed") yield ErrorFrame("Image generation failed") return diff --git a/src/pipecat/services/fal/stt.py b/src/pipecat/services/fal/stt.py index 8b84aaeeb..544f6e128 100644 --- a/src/pipecat/services/fal/stt.py +++ b/src/pipecat/services/fal/stt.py @@ -290,5 +290,4 @@ class FalSTTService(SegmentedSTTService): ) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") diff --git a/src/pipecat/services/fish/tts.py b/src/pipecat/services/fish/tts.py index 3aa4da80d..7cc9a3611 100644 --- a/src/pipecat/services/fish/tts.py +++ b/src/pipecat/services/fish/tts.py @@ -228,8 +228,7 @@ class FishAudioTTSService(InterruptibleTTSService): await self._call_event_handler("on_connected") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) self._websocket = None await self._call_event_handler("on_connection_error", f"{e}") @@ -243,8 +242,7 @@ class FishAudioTTSService(InterruptibleTTSService): await self._websocket.send(ormsgpack.packb(stop_message)) await self._websocket.close() except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: self._request_id = None self._started = False @@ -286,8 +284,7 @@ class FishAudioTTSService(InterruptibleTTSService): continue except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) @traced_tts async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: @@ -323,8 +320,7 @@ class FishAudioTTSService(InterruptibleTTSService): flush_message = {"event": "flush"} await self._get_websocket().send(ormsgpack.packb(flush_message)) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") yield TTSStoppedFrame() await self._disconnect() await self._connect() @@ -332,5 +328,4 @@ class FishAudioTTSService(InterruptibleTTSService): yield None except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py index 624745293..27b037f36 100644 --- a/src/pipecat/services/gladia/stt.py +++ b/src/pipecat/services/gladia/stt.py @@ -468,8 +468,7 @@ class GladiaSTTService(STTService): break except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) self._connection_active = False if not self._should_reconnect: @@ -559,8 +558,7 @@ class GladiaSTTService(STTService): except websockets.exceptions.ConnectionClosed: logger.debug("Connection closed during keepalive") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) async def _receive_task_handler(self): try: @@ -623,8 +621,7 @@ class GladiaSTTService(STTService): # Expected when closing the connection pass except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) async def _maybe_reconnect(self) -> bool: """Handle exponential backoff reconnection logic.""" @@ -632,7 +629,9 @@ class GladiaSTTService(STTService): return False self._reconnection_attempts += 1 if self._reconnection_attempts > self._max_reconnection_attempts: - logger.error(f"Max reconnection attempts ({self._max_reconnection_attempts}) reached") + await self.push_error( + error_msg=f"Max reconnection attempts ({self._max_reconnection_attempts}) reached", + ) self._should_reconnect = False return False delay = self._reconnection_delay * (2 ** (self._reconnection_attempts - 1)) diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index 6fd195c0d..19bcda730 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/src/pipecat/services/google/gemini_live/llm.py @@ -1175,7 +1175,7 @@ class GeminiLiveLLMService(LLMService): self._connection_task = self.create_task(self._connection_task_handler(config=config)) except Exception as e: - await self.push_error(ErrorFrame(error=f"{self} Initialization error: {e}")) + await self.push_error(error_msg=f"Initialization error: {e}", exception=e) async def _connection_task_handler(self, config: LiveConnectConfig): async with self._client.aio.live.connect(model=self._model_name, config=config) as session: @@ -1252,11 +1252,11 @@ class GeminiLiveLLMService(LLMService): ) if self._consecutive_failures >= MAX_CONSECUTIVE_FAILURES: - logger.error( + error_msg = ( f"Max consecutive failures ({MAX_CONSECUTIVE_FAILURES}) reached, " "treating as fatal error" ) - await self.push_error(ErrorFrame(error=f"{self} Error in receive loop: {error}")) + await self.push_error(error_msg=error_msg, exception=error) return False else: logger.info( @@ -1284,7 +1284,7 @@ class GeminiLiveLLMService(LLMService): self._completed_tool_calls = set() self._disconnecting = False except Exception as e: - logger.error(f"{self} error disconnecting: {e}") + await self.push_error(error_msg=f"Error disconnecting: {e}", exception=e) async def _send_user_audio(self, frame): """Send user audio frame to Gemini Live API.""" @@ -1745,7 +1745,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(error_msg=f"Send error: {error}") def create_context_aggregator( self, diff --git a/src/pipecat/services/google/image.py b/src/pipecat/services/google/image.py index 5d7a461b3..f92c7bb2b 100644 --- a/src/pipecat/services/google/image.py +++ b/src/pipecat/services/google/image.py @@ -110,7 +110,6 @@ class GoogleImageGenService(ImageGenService): await self.stop_ttfb_metrics() if not response or not response.generated_images: - logger.error(f"{self} error: image generation failed") yield ErrorFrame("Image generation failed") return @@ -128,5 +127,4 @@ class GoogleImageGenService(ImageGenService): yield frame except Exception as e: - logger.error(f"{self} error generating image: {e}") yield ErrorFrame(f"Image generation error: {str(e)}") diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 883932b76..840b473b2 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -793,7 +793,7 @@ class GoogleLLMService(LLMService): return generation_params.setdefault("thinking_config", {})["thinking_budget"] = 0 except Exception as e: - logger.exception(f"Failed to unset thinking budget: {e}") + logger.error(f"Failed to unset thinking budget: {e}") async def _stream_content( self, params_from_context: GeminiLLMInvocationParams @@ -983,7 +983,7 @@ class GoogleLLMService(LLMService): except DeadlineExceeded: await self._call_event_handler("on_completion_timeout") except Exception as e: - logger.exception(f"{self} exception: {e}") + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: if grounding_metadata and isinstance(grounding_metadata, dict): llm_search_frame = LLMSearchResponseFrame( diff --git a/src/pipecat/services/google/stt.py b/src/pipecat/services/google/stt.py index beb5db740..7f07ed2f0 100644 --- a/src/pipecat/services/google/stt.py +++ b/src/pipecat/services/google/stt.py @@ -774,8 +774,7 @@ class GoogleSTTService(STTService): yield cloud_speech.StreamingRecognizeRequest(audio=audio_data) except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) raise async def _stream_audio(self): @@ -806,15 +805,13 @@ class GoogleSTTService(STTService): break except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) await asyncio.sleep(1) # Brief delay before reconnecting self._stream_start_time = int(time.time() * 1000) except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: """Process an audio chunk for STT transcription. @@ -902,8 +899,7 @@ class GoogleSTTService(STTService): ) raise except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) # Re-raise the exception to let it propagate (e.g. in the case of a # timeout, propagate to _stream_audio to reconnect) raise diff --git a/src/pipecat/services/google/tts.py b/src/pipecat/services/google/tts.py index cf03e2d52..22b15bb56 100644 --- a/src/pipecat/services/google/tts.py +++ b/src/pipecat/services/google/tts.py @@ -737,7 +737,6 @@ class GoogleHttpTTSService(TTSService): yield TTSStoppedFrame() except Exception as e: - logger.error(f"{self} exception: {e}") error_message = f"TTS generation error: {str(e)}" yield ErrorFrame(error=error_message) @@ -996,9 +995,7 @@ class GoogleTTSService(GoogleBaseTTSService): yield frame except Exception as e: - logger.error(f"{self} exception: {e}") - error_message = f"TTS generation error: {str(e)}" - yield ErrorFrame(error=error_message) + await self.push_error(error_msg=f"TTS generation error: {str(e)}", exception=e) class GeminiTTSService(GoogleBaseTTSService): @@ -1248,6 +1245,5 @@ class GeminiTTSService(GoogleBaseTTSService): yield frame except Exception as e: - logger.error(f"{self} exception: {e}") error_message = f"Gemini TTS generation error: {str(e)}" yield ErrorFrame(error=error_message) diff --git a/src/pipecat/services/groq/tts.py b/src/pipecat/services/groq/tts.py index 9026c4c4c..2966137cb 100644 --- a/src/pipecat/services/groq/tts.py +++ b/src/pipecat/services/groq/tts.py @@ -146,7 +146,6 @@ class GroqTTSService(TTSService): bytes = w.readframes(num_frames) yield TTSAudioRawFrame(bytes, frame_rate, channels) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") yield TTSStoppedFrame() diff --git a/src/pipecat/services/heygen/client.py b/src/pipecat/services/heygen/client.py index e142aab04..ba517090c 100644 --- a/src/pipecat/services/heygen/client.py +++ b/src/pipecat/services/heygen/client.py @@ -179,7 +179,7 @@ class HeyGenClient: await self._task_manager.cancel_task(self._event_task) self._event_task = None except Exception as e: - logger.exception(f"Exception during cleanup: {e}") + logger.error(f"Exception during cleanup: {e}") async def start(self, frame: StartFrame, audio_chunk_size: int) -> None: """Start the client and establish all necessary connections. diff --git a/src/pipecat/services/hume/tts.py b/src/pipecat/services/hume/tts.py index 278626748..a2472c6d3 100644 --- a/src/pipecat/services/hume/tts.py +++ b/src/pipecat/services/hume/tts.py @@ -287,8 +287,7 @@ class HumeTTSService(WordTTSService): self._cumulative_time = utterance_duration except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: # Ensure TTFB timer is stopped even on early failures await self.stop_ttfb_metrics() diff --git a/src/pipecat/services/inworld/tts.py b/src/pipecat/services/inworld/tts.py index ab218b3c0..6652fab72 100644 --- a/src/pipecat/services/inworld/tts.py +++ b/src/pipecat/services/inworld/tts.py @@ -397,8 +397,7 @@ class InworldTTSService(TTSService): # STEP 7: ERROR HANDLING # ================================================================================ # Log any unexpected errors and notify the pipeline - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: # ================================================================================ # STEP 8: CLEANUP AND COMPLETION @@ -513,7 +512,7 @@ class InworldTTSService(TTSService): # Extract the base64-encoded audio content from response if "audioContent" not in response_data: logger.error("No audioContent in Inworld API response") - await self.push_error(ErrorFrame(error="No audioContent in response")) + yield ErrorFrame(error="No audioContent in response") return # ================================================================================ diff --git a/src/pipecat/services/lmnt/tts.py b/src/pipecat/services/lmnt/tts.py index ebcad0f20..25de40c4b 100644 --- a/src/pipecat/services/lmnt/tts.py +++ b/src/pipecat/services/lmnt/tts.py @@ -214,8 +214,7 @@ class LmntTTSService(InterruptibleTTSService): await self._call_event_handler("on_connected") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) self._websocket = None await self._call_event_handler("on_connection_error", f"{e}") @@ -231,8 +230,7 @@ class LmntTTSService(InterruptibleTTSService): # await self._websocket.send(json.dumps({"eof": True})) await self._websocket.close() except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Error disconnecting from LMNT: {e}", exception=e) finally: self._started = False self._websocket = None @@ -266,10 +264,9 @@ class LmntTTSService(InterruptibleTTSService): try: msg = json.loads(message) if "error" in msg: - logger.error(f"{self} error: {msg['error']}") await self.push_frame(TTSStoppedFrame()) await self.stop_all_metrics() - await self.push_error(ErrorFrame(error=f"{self} error: {msg['error']}")) + await self.push_error(error_msg=f"Error: {msg['error']}") return except json.JSONDecodeError: logger.error(f"Invalid JSON message: {message}") @@ -302,13 +299,11 @@ class LmntTTSService(InterruptibleTTSService): await self._get_websocket().send(json.dumps({"flush": True})) await self.start_tts_usage_metrics(text) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") yield TTSStoppedFrame() await self._disconnect() await self._connect() return yield None except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") diff --git a/src/pipecat/services/mcp_service.py b/src/pipecat/services/mcp_service.py index ccfb47adb..2c1a53768 100644 --- a/src/pipecat/services/mcp_service.py +++ b/src/pipecat/services/mcp_service.py @@ -176,7 +176,6 @@ class MCPClient(BaseObject): except Exception as e: error_msg = f"Error calling mcp tool {params.function_name}: {str(e)}" logger.error(error_msg) - logger.exception("Full exception details:") await params.result_callback(error_msg) async def _stdio_list_tools(self) -> ToolsSchema: @@ -207,7 +206,6 @@ class MCPClient(BaseObject): except Exception as e: error_msg = f"Error calling mcp tool {params.function_name}: {str(e)}" logger.error(error_msg) - logger.exception("Full exception details:") await params.result_callback(error_msg) async def _streamable_http_list_tools(self) -> ToolsSchema: @@ -246,7 +244,6 @@ class MCPClient(BaseObject): except Exception as e: error_msg = f"Error calling mcp tool {params.function_name}: {str(e)}" logger.error(error_msg) - logger.exception("Full exception details:") await params.result_callback(error_msg) async def _call_tool(self, session, function_name, arguments, result_callback): @@ -302,7 +299,6 @@ class MCPClient(BaseObject): except Exception as e: logger.error(f"Failed to read tool '{tool_name}': {str(e)}") - logger.exception("Full exception details:") continue logger.debug(f"Completed reading {len(tool_schemas)} tools") diff --git a/src/pipecat/services/mem0/memory.py b/src/pipecat/services/mem0/memory.py index a80398728..66256496c 100644 --- a/src/pipecat/services/mem0/memory.py +++ b/src/pipecat/services/mem0/memory.py @@ -253,8 +253,9 @@ class Mem0MemoryService(FrameProcessor): # Otherwise, pass the enhanced context frame downstream await self.push_frame(frame) except Exception as e: - logger.error(f"Error processing with Mem0: {str(e)}") - await self.push_frame(ErrorFrame(f"Error processing with Mem0: {str(e)}")) + await self.push_error( + error_msg=f"Error processing with Mem0: {str(e)}", exception=e + ) await self.push_frame(frame) # Still pass the original frame through else: # For non-context frames, just pass them through diff --git a/src/pipecat/services/minimax/tts.py b/src/pipecat/services/minimax/tts.py index 4cd86fac3..446a343e7 100644 --- a/src/pipecat/services/minimax/tts.py +++ b/src/pipecat/services/minimax/tts.py @@ -314,7 +314,6 @@ class MiniMaxHttpTTSService(TTSService): ) as response: if response.status != 200: error_message = f"MiniMax TTS error: HTTP {response.status}" - logger.error(error_message) yield ErrorFrame(error=error_message) return @@ -392,8 +391,7 @@ class MiniMaxHttpTTSService(TTSService): continue except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}", exception=e) finally: await self.stop_ttfb_metrics() yield TTSStoppedFrame() diff --git a/src/pipecat/services/moondream/vision.py b/src/pipecat/services/moondream/vision.py index b7527f76c..e9ce86383 100644 --- a/src/pipecat/services/moondream/vision.py +++ b/src/pipecat/services/moondream/vision.py @@ -110,7 +110,6 @@ class MoondreamService(VisionService): if analysis fails. """ if not self._model: - logger.error(f"{self} error: Moondream model not available ({self.model_name})") yield ErrorFrame("Moondream model not available") return diff --git a/src/pipecat/services/neuphonic/tts.py b/src/pipecat/services/neuphonic/tts.py index 60b0ebcb1..413691ccf 100644 --- a/src/pipecat/services/neuphonic/tts.py +++ b/src/pipecat/services/neuphonic/tts.py @@ -285,8 +285,7 @@ class NeuphonicTTSService(InterruptibleTTSService): await self._call_event_handler("on_connected") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) self._websocket = None await self._call_event_handler("on_connection_error", f"{e}") @@ -299,8 +298,7 @@ class NeuphonicTTSService(InterruptibleTTSService): logger.debug("Disconnecting from Neuphonic") await self._websocket.close() except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: self._started = False self._websocket = None @@ -365,16 +363,14 @@ class NeuphonicTTSService(InterruptibleTTSService): await self._send_text(text) await self.start_tts_usage_metrics(text) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") yield TTSStoppedFrame() await self._disconnect() await self._connect() return yield None except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") class NeuphonicHttpTTSService(TTSService): @@ -538,7 +534,6 @@ class NeuphonicHttpTTSService(TTSService): if response.status != 200: error_text = await response.text() error_message = f"Neuphonic API error: HTTP {response.status} - {error_text}" - logger.error(error_message) yield ErrorFrame(error=error_message) return @@ -568,8 +563,7 @@ class NeuphonicHttpTTSService(TTSService): yield TTSAudioRawFrame(audio_bytes, self.sample_rate, 1) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") # Don't yield error frame for individual message failures continue @@ -577,8 +571,7 @@ class NeuphonicHttpTTSService(TTSService): logger.debug("TTS generation cancelled") raise except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") finally: await self.stop_ttfb_metrics() yield TTSStoppedFrame() diff --git a/src/pipecat/services/openai/image.py b/src/pipecat/services/openai/image.py index 9501aace9..9daa8f2b9 100644 --- a/src/pipecat/services/openai/image.py +++ b/src/pipecat/services/openai/image.py @@ -76,7 +76,6 @@ class OpenAIImageGenService(ImageGenService): image_url = image.data[0].url if not image_url: - logger.error(f"{self} No image provided in response: {image}") yield ErrorFrame("Image generation failed") return diff --git a/src/pipecat/services/openai/realtime/llm.py b/src/pipecat/services/openai/realtime/llm.py index bea9986f3..7a57bf88c 100644 --- a/src/pipecat/services/openai/realtime/llm.py +++ b/src/pipecat/services/openai/realtime/llm.py @@ -443,7 +443,7 @@ class OpenAIRealtimeLLMService(LLMService): ) self._receive_task = self.create_task(self._receive_task_handler()) except Exception as e: - logger.error(f"{self} initialization error: {e}") + await self.push_error(error_msg=f"Error connecting: {e}", exception=e) self._websocket = None async def _disconnect(self): @@ -460,7 +460,7 @@ class OpenAIRealtimeLLMService(LLMService): self._completed_tool_calls = set() self._disconnecting = False except Exception as e: - logger.error(f"{self} error disconnecting: {e}") + await self.push_error(error_msg=f"Error disconnecting: {e}", exception=e) async def _ws_send(self, realtime_message): try: @@ -473,12 +473,11 @@ class OpenAIRealtimeLLMService(LLMService): # somehow *started* the websocket send attempt while we still # had a connection) return - logger.error(f"Error sending message to websocket: {e}") # In server-to-server contexts, a WebSocket error should be quite rare. Given how hard # it is to recover from a send-side error with proper 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"Error sending client event: {e}")) + await self.push_error(error_msg=f"Error sending client event: {e}", exception=e) async def _update_settings(self): settings = self._session_properties @@ -674,7 +673,7 @@ class OpenAIRealtimeLLMService(LLMService): self._current_assistant_response = None # error handling if evt.response.status == "failed": - await self.push_error(ErrorFrame(error=evt.response.status_details["error"]["message"])) + await self.push_error(error_msg=evt.response.status_details["error"]["message"]) return # response content for item in evt.response.output: @@ -766,7 +765,7 @@ class OpenAIRealtimeLLMService(LLMService): async def _handle_evt_error(self, evt): # Errors are fatal to this connection. Send an ErrorFrame. - await self.push_error(ErrorFrame(error=f"Error: {evt}")) + await self.push_error(error_msg=f"Error: {evt}") # # state and client events for the current conversation diff --git a/src/pipecat/services/openai/tts.py b/src/pipecat/services/openai/tts.py index 23cb75324..aab1c8313 100644 --- a/src/pipecat/services/openai/tts.py +++ b/src/pipecat/services/openai/tts.py @@ -206,5 +206,4 @@ class OpenAITTSService(TTSService): yield frame yield TTSStoppedFrame() except BadRequestError as e: - logger.exception(f"{self} error generating TTS: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") diff --git a/src/pipecat/services/openai_realtime_beta/azure.py b/src/pipecat/services/openai_realtime_beta/azure.py index 2ec556ee3..e55dd075f 100644 --- a/src/pipecat/services/openai_realtime_beta/azure.py +++ b/src/pipecat/services/openai_realtime_beta/azure.py @@ -79,5 +79,5 @@ class AzureRealtimeBetaLLMService(OpenAIRealtimeBetaLLMService): ) self._receive_task = self.create_task(self._receive_task_handler()) except Exception as e: - logger.error(f"{self} initialization error: {e}") + await self.push_error(error_msg=f"Error connecting: {e}", exception=e) self._websocket = None diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index d0cb39bf6..16a00d08f 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -425,7 +425,7 @@ class OpenAIRealtimeBetaLLMService(LLMService): ) self._receive_task = self.create_task(self._receive_task_handler()) except Exception as e: - logger.error(f"{self} initialization error: {e}") + await self.push_error(error_msg=f"Error connecting: {e}", exception=e) self._websocket = None async def _disconnect(self): @@ -441,7 +441,7 @@ class OpenAIRealtimeBetaLLMService(LLMService): self._receive_task = None self._disconnecting = False except Exception as e: - logger.error(f"{self} error disconnecting: {e}") + await self.push_error(error_msg=f"Error disconnecting: {e}", exception=e) async def _ws_send(self, realtime_message): try: @@ -450,12 +450,11 @@ class OpenAIRealtimeBetaLLMService(LLMService): except Exception as e: if self._disconnecting: return - logger.error(f"Error sending message to websocket: {e}") # In server-to-server contexts, a WebSocket error should be quite rare. Given how hard # it is to recover from a send-side error with proper 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"Error sending client event: {e}")) + await self.push_error(error_msg=f"Error sending client event: {e}", exception=e) async def _update_settings(self): settings = self._session_properties @@ -686,7 +685,7 @@ class OpenAIRealtimeBetaLLMService(LLMService): async def _handle_evt_error(self, evt): # Errors are fatal to this connection. Send an ErrorFrame. - await self.push_error(ErrorFrame(error=f"Error: {evt}")) + await self.push_error(error_msg=f"Error: {evt}") async def _handle_assistant_output(self, output): # We haven't seen intermixed audio and function_call items in the same response. But let's diff --git a/src/pipecat/services/piper/tts.py b/src/pipecat/services/piper/tts.py index dd842ff11..6bd9374bc 100644 --- a/src/pipecat/services/piper/tts.py +++ b/src/pipecat/services/piper/tts.py @@ -88,9 +88,6 @@ class PiperTTSService(TTSService): ) as response: if response.status != 200: error = await response.text() - logger.error( - f"{self} error getting audio (status: {response.status}, error: {error})" - ) yield ErrorFrame( error=f"Error getting audio (status: {response.status}, error: {error})" ) @@ -109,7 +106,7 @@ class PiperTTSService(TTSService): yield frame except Exception as e: logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") finally: logger.debug(f"{self}: Finished TTS [{text}]") await self.stop_ttfb_metrics() diff --git a/src/pipecat/services/playht/tts.py b/src/pipecat/services/playht/tts.py index 2c40065aa..5af54a9aa 100644 --- a/src/pipecat/services/playht/tts.py +++ b/src/pipecat/services/playht/tts.py @@ -266,8 +266,7 @@ class PlayHTTTSService(InterruptibleTTSService): self._websocket = None await self._call_event_handler("on_connection_error", f"{e}") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Error connecting: {e}", exception=e) self._websocket = None await self._call_event_handler("on_connection_error", f"{e}") @@ -280,8 +279,7 @@ class PlayHTTTSService(InterruptibleTTSService): logger.debug("Disconnecting from PlayHT") await self._websocket.close() except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Error disconnecting: {e}", exception=e) finally: self._request_id = None self._websocket = None @@ -351,8 +349,7 @@ class PlayHTTTSService(InterruptibleTTSService): await self.push_frame(TTSStoppedFrame()) self._request_id = None elif "error" in msg: - logger.error(f"{self} error: {msg}") - await self.push_error(ErrorFrame(error=f"{self} error: {msg['error']}")) + await self.push_error(error_msg=f"Error: {msg['error']}") except json.JSONDecodeError: logger.error(f"Invalid JSON message: {message}") @@ -394,8 +391,7 @@ class PlayHTTTSService(InterruptibleTTSService): await self._get_websocket().send(json.dumps(tts_command)) await self.start_tts_usage_metrics(text) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") yield TTSStoppedFrame() await self._disconnect() await self._connect() @@ -405,8 +401,7 @@ class PlayHTTTSService(InterruptibleTTSService): yield None except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") class PlayHTHttpTTSService(TTSService): @@ -626,8 +621,7 @@ class PlayHTHttpTTSService(TTSService): yield frame except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") finally: await self.stop_ttfb_metrics() yield TTSStoppedFrame() diff --git a/src/pipecat/services/rime/tts.py b/src/pipecat/services/rime/tts.py index c9f461350..fb64ded20 100644 --- a/src/pipecat/services/rime/tts.py +++ b/src/pipecat/services/rime/tts.py @@ -300,8 +300,7 @@ class RimeTTSService(AudioContextWordTTSService): await self._call_event_handler("on_connected") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Error connecting: {e}", exception=e) self._websocket = None await self._call_event_handler("on_connection_error", f"{e}") @@ -313,8 +312,7 @@ class RimeTTSService(AudioContextWordTTSService): await self._websocket.send(json.dumps(self._build_eos_msg())) await self._websocket.close() except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Error disconnecting: {e}", exception=e) finally: self._context_id = None self._websocket = None @@ -407,10 +405,9 @@ class RimeTTSService(AudioContextWordTTSService): logger.debug(f"Updated cumulative time to: {self._cumulative_time}") elif msg["type"] == "error": - logger.error(f"{self} error: {msg}") await self.push_frame(TTSStoppedFrame()) await self.stop_all_metrics() - await self.push_error(ErrorFrame(error=f"{self} error: {msg['message']}")) + await self.push_error(error_msg=f"Error: {msg['message']}") self._context_id = None async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): @@ -452,16 +449,14 @@ class RimeTTSService(AudioContextWordTTSService): await self._get_websocket().send(json.dumps(msg)) await self.start_tts_usage_metrics(text) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") yield TTSStoppedFrame() await self._disconnect() await self._connect() return yield None except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") class RimeHttpTTSService(TTSService): @@ -592,7 +587,6 @@ class RimeHttpTTSService(TTSService): ) as response: if response.status != 200: error_message = f"Rime TTS error: HTTP {response.status}" - logger.error(error_message) yield ErrorFrame(error=error_message) return @@ -610,8 +604,7 @@ class RimeHttpTTSService(TTSService): yield frame except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") finally: await self.stop_ttfb_metrics() yield TTSStoppedFrame() diff --git a/src/pipecat/services/riva/stt.py b/src/pipecat/services/riva/stt.py index 25cc78c7a..4dba62bcb 100644 --- a/src/pipecat/services/riva/stt.py +++ b/src/pipecat/services/riva/stt.py @@ -655,12 +655,10 @@ class RivaSegmentedSTTService(SegmentedSTTService): logger.debug("No transcription results found in Riva response") except AttributeError as ae: - logger.error(f"Unexpected response structure from Riva: {ae}") yield ErrorFrame(f"Unexpected Riva response format: {str(ae)}") except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") class ParakeetSTTService(RivaSTTService): diff --git a/src/pipecat/services/riva/tts.py b/src/pipecat/services/riva/tts.py index 8b9effbba..370971068 100644 --- a/src/pipecat/services/riva/tts.py +++ b/src/pipecat/services/riva/tts.py @@ -180,8 +180,7 @@ class RivaTTSService(TTSService): yield frame resp = await asyncio.wait_for(queue.get(), timeout=RIVA_TTS_TIMEOUT_SECS) except asyncio.TimeoutError: - logger.error(f"{self} timeout waiting for audio response") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") await self.start_tts_usage_metrics(text) yield TTSStoppedFrame() diff --git a/src/pipecat/services/sarvam/stt.py b/src/pipecat/services/sarvam/stt.py index 98e643fde..6f9a02642 100644 --- a/src/pipecat/services/sarvam/stt.py +++ b/src/pipecat/services/sarvam/stt.py @@ -275,8 +275,7 @@ class SarvamSTTService(STTService): await self._socket_client.translate(**method_kwargs) except Exception as e: - logger.error(f"Error sending audio to Sarvam: {e}") - await self.push_error(ErrorFrame(f"Failed to send audio: {e}")) + yield ErrorFrame(error=f"Error sending audio to Sarvam: {e}", exception=e) yield None @@ -332,13 +331,11 @@ class SarvamSTTService(STTService): logger.info("Connected to Sarvam successfully") except ApiError as e: - logger.error(f"Sarvam API error: {e}") - await self.push_error(ErrorFrame(f"Sarvam API error: {e}")) + await self.push_error(error_msg=f"Sarvam API error: {e}", exception=e) except Exception as e: - logger.error(f"Failed to connect to Sarvam: {e}") self._socket_client = None self._websocket_context = None - await self.push_error(ErrorFrame(f"Failed to connect to Sarvam: {e}")) + await self.push_error(error_msg=f"Failed to connect to Sarvam: {e}", exception=e) async def _disconnect(self): """Disconnect from Sarvam WebSocket API using SDK.""" @@ -351,7 +348,9 @@ class SarvamSTTService(STTService): # Exit the async context manager await self._websocket_context.__aexit__(None, None, None) except Exception as e: - logger.error(f"Error closing WebSocket connection: {e}") + await self.push_error( + error_msg=f"Error closing WebSocket connection: {e}", exception=e + ) finally: logger.debug("Disconnected from Sarvam WebSocket") self._socket_client = None @@ -371,8 +370,7 @@ class SarvamSTTService(STTService): # Messages will be handled via the _message_handler callback await self._socket_client.start_listening() except Exception as e: - logger.error(f"Error in Sarvam receive task: {e}") - await self.push_error(ErrorFrame(f"Sarvam receive task error: {e}")) + await self.push_error(error_msg=f"Sarvam receive task error: {e}", exception=e) async def _handle_message(self, message): """Handle incoming WebSocket message from Sarvam SDK. @@ -427,8 +425,7 @@ class SarvamSTTService(STTService): await self.stop_processing_metrics() except Exception as e: - logger.error(f"Error handling Sarvam message: {e}") - await self.push_error(ErrorFrame(f"Failed to handle message: {e}")) + await self.push_error(error_msg=f"Failed to handle message: {e}", exception=e) await self.stop_all_metrics() @traced_stt diff --git a/src/pipecat/services/sarvam/tts.py b/src/pipecat/services/sarvam/tts.py index 127a0d589..197eb1548 100644 --- a/src/pipecat/services/sarvam/tts.py +++ b/src/pipecat/services/sarvam/tts.py @@ -254,8 +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() - logger.error(f"Sarvam API error: {error_text}") - await self.push_error(ErrorFrame(error=f"Sarvam API error: {error_text}")) + yield ErrorFrame(error=f"Sarvam API error: {error_text}") return response_data = await response.json() @@ -264,8 +263,7 @@ class SarvamHttpTTSService(TTSService): # Decode base64 audio data if "audios" not in response_data or not response_data["audios"]: - logger.error("No audio data received from Sarvam API") - await self.push_error(ErrorFrame(error="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) @@ -286,8 +284,7 @@ class SarvamHttpTTSService(TTSService): yield frame except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + yield ErrorFrame(error=f"Error generating TTS: {e}", exception=e) finally: await self.stop_ttfb_metrics() yield TTSStoppedFrame() @@ -560,8 +557,7 @@ class SarvamTTSService(InterruptibleTTSService): await self._disconnect_websocket() except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: # Reset state only after everything is cleaned up self._started = False @@ -585,8 +581,9 @@ class SarvamTTSService(InterruptibleTTSService): await self._call_event_handler("on_connected") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error( + error_msg=f"Error connecting to Sarvam TTS Websocket: {e}", exception=e + ) self._websocket = None await self._call_event_handler("on_connection_error", f"{e}") @@ -602,8 +599,7 @@ class SarvamTTSService(InterruptibleTTSService): await self._websocket.send(json.dumps(config_message)) logger.debug("Configuration sent successfully") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) raise async def _disconnect_websocket(self): @@ -615,8 +611,7 @@ class SarvamTTSService(InterruptibleTTSService): logger.debug("Disconnecting from Sarvam") await self._websocket.close() except Exception as e: - logger.error(f"{self} error closing websocket: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Error closing websocket: {e}", exception=e) finally: self._started = False self._websocket = None @@ -640,7 +635,7 @@ class SarvamTTSService(InterruptibleTTSService): await self.push_frame(frame) elif msg.get("type") == "error": error_msg = msg["data"]["message"] - logger.error(f"TTS Error: {error_msg}") + await self.push_error(error_msg=f"TTS Error: {error_msg}") # If it's a timeout error, the connection might need to be reset if "too long" in error_msg.lower() or "timeout" in error_msg.lower(): @@ -702,13 +697,11 @@ class SarvamTTSService(InterruptibleTTSService): await self._send_text(text) await self.start_tts_usage_metrics(text) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") yield TTSStoppedFrame() await self._disconnect() await self._connect() return yield None except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") diff --git a/src/pipecat/services/simli/video.py b/src/pipecat/services/simli/video.py index e6514ca7b..35961cff1 100644 --- a/src/pipecat/services/simli/video.py +++ b/src/pipecat/services/simli/video.py @@ -178,7 +178,7 @@ class SimliVideoService(FrameProcessor): self._audio_task = self.create_task(self._consume_and_process_audio()) self._video_task = self.create_task(self._consume_and_process_video()) except Exception as e: - logger.error(f"{self}: unable to start connection: {e}") + await self.push_error(error_msg=f"Unable to start connection: {e}", exception=e) async def _consume_and_process_audio(self): """Consume audio frames from Simli and push them downstream.""" @@ -256,7 +256,7 @@ class SimliVideoService(FrameProcessor): await self._simli_client.send(audioBytes) return except Exception as e: - logger.exception(f"{self} exception: {e}") + await self.push_error(error_msg=f"Error sending audio: {e}", exception=e) elif isinstance(frame, TTSStoppedFrame): try: if self._previously_interrupted and len(self._audio_buffer) > 0: @@ -264,7 +264,7 @@ class SimliVideoService(FrameProcessor): self._previously_interrupted = False self._audio_buffer = bytearray() except Exception as e: - logger.exception(f"{self} exception: {e}") + await self.push_error(error_msg=f"Error stopping TTS: {e}", exception=e) return elif isinstance(frame, (EndFrame, CancelFrame)): await self._stop() diff --git a/src/pipecat/services/soniox/stt.py b/src/pipecat/services/soniox/stt.py index b4bcb7ba1..4afcaa7bb 100644 --- a/src/pipecat/services/soniox/stt.py +++ b/src/pipecat/services/soniox/stt.py @@ -194,7 +194,7 @@ class SonioxSTTService(STTService): self._websocket = await websocket_connect(self._url) if not self._websocket: - logger.error(f"Unable to connect to Soniox API at {self._url}") + await self.push_error(error_msg=f"Unable to connect to Soniox API at {self._url}") # If vad_force_turn_endpoint is not enabled, we need to enable endpoint detection. # Either one or the other is required. @@ -327,8 +327,7 @@ class SonioxSTTService(STTService): # Expected when closing the connection logger.debug("WebSocket connection closed, keepalive task stopped.") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) async def _receive_task_handler(self): if not self._websocket: @@ -404,13 +403,8 @@ class SonioxSTTService(STTService): if error_code or error_message: # In case of error, still send the final transcript (if any remaining in the buffer). await send_endpoint_transcript() - logger.error( - f"{self} error: {error_code} (_receive_task_handler) - {error_message}" - ) await self.push_error( - ErrorFrame( - error=f"{self} error: {error_code} (_receive_task_handler) - {error_message}" - ) + error_msg=f"Error: {error_code} (_receive_task_handler) - {error_message}" ) finished = content.get("finished") @@ -425,5 +419,4 @@ class SonioxSTTService(STTService): # Expected when closing the connection. pass except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Error receiving message: {e}", exception=e) diff --git a/src/pipecat/services/speechmatics/stt.py b/src/pipecat/services/speechmatics/stt.py index 2d95bd69e..f18314d8d 100644 --- a/src/pipecat/services/speechmatics/stt.py +++ b/src/pipecat/services/speechmatics/stt.py @@ -467,8 +467,7 @@ class SpeechmaticsSTTService(STTService): await self._client.send_audio(audio) yield None except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") await self._disconnect() def update_params( @@ -514,8 +513,7 @@ class SpeechmaticsSTTService(STTService): self._client.send_message(payload), self.get_event_loop() ) except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) raise RuntimeError(f"error sending message to STT: {e}") async def _connect(self) -> None: @@ -581,8 +579,7 @@ class SpeechmaticsSTTService(STTService): logger.debug(f"{self} Connected to Speechmatics STT service") await self._call_event_handler("on_connected") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Error connecting to Speechmatics: {e}", exception=e) self._client = None async def _disconnect(self) -> None: @@ -596,8 +593,9 @@ class SpeechmaticsSTTService(STTService): except asyncio.TimeoutError: logger.warning(f"{self} Timeout while closing Speechmatics client connection") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error( + error_msg=f"Error disconnecting from Speechmatics: {e}", exception=e + ) finally: self._client = None await self._call_event_handler("on_disconnected") diff --git a/src/pipecat/services/speechmatics/tts.py b/src/pipecat/services/speechmatics/tts.py index b8fe172e7..417540663 100644 --- a/src/pipecat/services/speechmatics/tts.py +++ b/src/pipecat/services/speechmatics/tts.py @@ -163,7 +163,7 @@ class SpeechmaticsTTSService(TTSService): # Report error frame yield ErrorFrame( - error=f"{self} Service unavailable [503] (attempt {attempt}, retry in {backoff_time:.2f}s)" + error=f"Service unavailable [503] (attempt {attempt}, retry in {backoff_time:.2f}s)" ) # Wait before retrying @@ -174,16 +174,13 @@ class SpeechmaticsTTSService(TTSService): except (ValueError, ArithmeticError): yield ErrorFrame( - error=f"{self} Service unavailable [503] (attempts {attempt})", - fatal=True, + error=f"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"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"Error generating TTS: {e}") finally: # Emit the TTS stopped frame yield TTSStoppedFrame() diff --git a/src/pipecat/services/stt_service.py b/src/pipecat/services/stt_service.py index 7c8bae6bf..f81848415 100644 --- a/src/pipecat/services/stt_service.py +++ b/src/pipecat/services/stt_service.py @@ -329,4 +329,4 @@ class WebsocketSTTService(STTService, WebsocketService): async def _report_error(self, error: ErrorFrame): await self._call_event_handler("on_connection_error", error.error) - await self.push_error(error) + await self.push_error_frame(error) diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index 8d7c4e6bb..977d6957b 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -781,7 +781,7 @@ class WebsocketTTSService(TTSService, WebsocketService): async def _report_error(self, error: ErrorFrame): await self._call_event_handler("on_connection_error", error.error) - await self.push_error(error) + await self.push_error_frame(error) class InterruptibleTTSService(WebsocketTTSService): @@ -843,7 +843,7 @@ class WebsocketWordTTSService(WordTTSService, WebsocketService): async def _report_error(self, error: ErrorFrame): await self._call_event_handler("on_connection_error", error.error) - await self.push_error(error) + await self.push_error_frame(error) class InterruptibleWordTTSService(WebsocketWordTTSService): diff --git a/src/pipecat/services/ultravox/stt.py b/src/pipecat/services/ultravox/stt.py index 14eaebf6a..bef74f346 100644 --- a/src/pipecat/services/ultravox/stt.py +++ b/src/pipecat/services/ultravox/stt.py @@ -246,8 +246,7 @@ class UltravoxSTTService(AIService): logger.info("Model warm-up completed successfully") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) def _generate_silent_audio(self, sample_rate=16000, duration_sec=1.0): """Generate silent audio as a numpy array. @@ -377,7 +376,7 @@ class UltravoxSTTService(AIService): if arr.size > 0: # Check if array is not empty audio_arrays.append(arr) except Exception as e: - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") # Handle numpy array data elif isinstance(f.audio, np.ndarray): if f.audio.size > 0: # Check if array is not empty @@ -437,17 +436,11 @@ class UltravoxSTTService(AIService): yield LLMFullResponseEndFrame() except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") else: - logger.error("No model available for text generation") yield ErrorFrame("No model available for text generation") except Exception as e: - logger.error(f"{self} exception: {e}") - import traceback - - logger.error(traceback.format_exc()) yield ErrorFrame(f"Error processing audio: {str(e)}") finally: self._buffer.is_processing = False diff --git a/src/pipecat/services/whisper/base_stt.py b/src/pipecat/services/whisper/base_stt.py index 743895253..c6e6b63da 100644 --- a/src/pipecat/services/whisper/base_stt.py +++ b/src/pipecat/services/whisper/base_stt.py @@ -226,8 +226,7 @@ class BaseWhisperSTTService(SegmentedSTTService): logger.warning("Received empty transcription from API") except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") async def _transcribe(self, audio: bytes) -> Transcription: """Transcribe audio data to text. diff --git a/src/pipecat/services/whisper/stt.py b/src/pipecat/services/whisper/stt.py index 4e326e2ca..ec69b63ea 100644 --- a/src/pipecat/services/whisper/stt.py +++ b/src/pipecat/services/whisper/stt.py @@ -285,7 +285,6 @@ class WhisperSTTService(SegmentedSTTService): The service will normalize it to float32 in the range [-1, 1]. """ if not self._model: - logger.error(f"{self} error: Whisper model not available") yield ErrorFrame("Whisper model not available") return @@ -428,5 +427,4 @@ class WhisperSTTServiceMLX(WhisperSTTService): ) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") diff --git a/src/pipecat/services/xtts/tts.py b/src/pipecat/services/xtts/tts.py index 34ddecbe2..cf98b2bdc 100644 --- a/src/pipecat/services/xtts/tts.py +++ b/src/pipecat/services/xtts/tts.py @@ -141,13 +141,8 @@ class XTTSService(TTSService): async with self._aiohttp_session.get(self._settings["base_url"] + "/studio_speakers") as r: if r.status != 200: text = await r.text() - logger.error( - f"{self} error getting studio speakers (status: {r.status}, error: {text})" - ) await self.push_error( - ErrorFrame( - error=f"Error getting studio speakers (status: {r.status}, error: {text})" - ) + error_msg=f"Error getting studio speakers (status: {r.status}, error: {text})" ) return self._studio_speakers = await r.json() @@ -186,7 +181,6 @@ class XTTSService(TTSService): async with self._aiohttp_session.post(url, json=payload) as r: if r.status != 200: text = await r.text() - logger.error(f"{self} error getting audio (status: {r.status}, error: {text})") yield ErrorFrame(error=f"Error getting audio (status: {r.status}, error: {text})") return diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index 51ef637b8..b3cc8acf7 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/src/pipecat/transports/daily/transport.py @@ -2506,13 +2506,10 @@ class DailyTransport(BaseTransport): async def _on_error(self, error): """Handle error events and push error frames.""" await self._call_event_handler("on_error", error) - # Push error frame to notify the pipeline - error_frame = ErrorFrame(error) - if self._input: - await self._input.push_error(error_frame) + await self._input.push_error(error_msg=error) elif self._output: - await self._output.push_error(error_frame) + await self._output.push_error(error_msg=error) else: logger.error("Both input and output are None while trying to push error") raise Exception("No valid input or output channel to push error") @@ -2568,7 +2565,7 @@ class DailyTransport(BaseTransport): except asyncio.TimeoutError: logger.error(f"Timeout handling dialin-ready event ({url})") except Exception as e: - logger.exception(f"Error handling dialin-ready event ({url}): {e}") + logger.error(f"Error handling dialin-ready event ({url}): {e}") async def _on_dialin_connected(self, data): """Handle dial-in connected events.""" diff --git a/src/pipecat/transports/smallwebrtc/connection.py b/src/pipecat/transports/smallwebrtc/connection.py index 60dd7798c..ef353417f 100644 --- a/src/pipecat/transports/smallwebrtc/connection.py +++ b/src/pipecat/transports/smallwebrtc/connection.py @@ -316,7 +316,7 @@ class SmallWebRTCConnection(BaseObject): logger.debug("Client not connected. Queuing app-message.") self._pending_app_messages.append(json_message) except Exception as e: - logger.exception(f"Error parsing JSON message {message}, {e}") + logger.error(f"Error parsing JSON message {message}, {e}") # Despite the fact that aiortc provides this listener, they don't have a status for "disconnected" # So, in case we loose connection, this event will not be triggered diff --git a/src/pipecat/transports/tavus/transport.py b/src/pipecat/transports/tavus/transport.py index 40ffeeed7..be1ff9afe 100644 --- a/src/pipecat/transports/tavus/transport.py +++ b/src/pipecat/transports/tavus/transport.py @@ -265,7 +265,7 @@ class TavusTransportClient: try: await self._client.cleanup() except Exception as e: - logger.exception(f"Exception during cleanup: {e}") + logger.error(f"Exception during cleanup: {e}") async def _on_joined(self, data): """Handle joined event.""" diff --git a/src/pipecat/utils/asyncio/task_manager.py b/src/pipecat/utils/asyncio/task_manager.py index 78c2ff0f1..cf433edb1 100644 --- a/src/pipecat/utils/asyncio/task_manager.py +++ b/src/pipecat/utils/asyncio/task_manager.py @@ -162,7 +162,7 @@ class TaskManager(BaseTaskManager): # Re-raise the exception to ensure the task is cancelled. raise except Exception as e: - logger.exception(f"{name}: unexpected exception: {e}") + logger.error(f"{name}: unexpected exception: {e}") if not self._params: raise Exception("TaskManager is not setup: unable to get event loop") @@ -197,7 +197,7 @@ class TaskManager(BaseTaskManager): # Here are sure the task is cancelled properly. pass except Exception as e: - logger.exception(f"{name}: unexpected exception while cancelling task: {e}") + logger.error(f"{name}: unexpected exception while cancelling task: {e}") except BaseException as e: logger.critical(f"{name}: fatal base exception while cancelling task: {e}") raise diff --git a/src/pipecat/utils/base_object.py b/src/pipecat/utils/base_object.py index 4a62e5e4d..bace5e64b 100644 --- a/src/pipecat/utils/base_object.py +++ b/src/pipecat/utils/base_object.py @@ -187,7 +187,7 @@ class BaseObject(ABC): else: handler(self, *args, **kwargs) except Exception as e: - logger.exception(f"Exception in event handler {event_name}: {e}") + logger.error(f"Exception in event handler {event_name}: {e}") def _event_task_finished(self, task: asyncio.Task): """Clean up completed event handler tasks.