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/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..4bf4975d0 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(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..9b7f2b915 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(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(exception=e) if self._receive_task: await self.cancel_task(self._receive_task) @@ -262,13 +260,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(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(exception=e) def _parse_message(self, message: Dict[str, Any]) -> BaseMessage: """Parse a raw message into the appropriate message type.""" diff --git a/src/pipecat/services/asyncai/tts.py b/src/pipecat/services/asyncai/tts.py index f916d9bdd..5507bccfc 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(exception=e) self._websocket = None await self._call_event_handler("on_connection_error", f"{e}") @@ -290,7 +289,7 @@ class AsyncAITTSService(InterruptibleTTSService): 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"{self} error: {msg['message']}", exception=e) else: logger.error(f"{self} error, unknown message type: {msg}") @@ -494,8 +493,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(exception=e) finally: await self.stop_ttfb_metrics() yield TTSStoppedFrame() diff --git a/src/pipecat/services/aws/stt.py b/src/pipecat/services/aws/stt.py index 8d89d3198..3ca782ed9 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(exception=e) retry_count += 1 if retry_count < max_retries: await asyncio.sleep(1) # Wait before retrying @@ -310,8 +309,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(exception=e) finally: self._ws_client = None await self._call_event_handler("on_disconnected") diff --git a/src/pipecat/services/azure/stt.py b/src/pipecat/services/azure/stt.py index 85a0508a0..4f9f95bb1 100644 --- a/src/pipecat/services/azure/stt.py +++ b/src/pipecat/services/azure/stt.py @@ -151,8 +151,7 @@ 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"{self} exception during initialization: {e}", exception=e) async def stop(self, frame: EndFrame): """Stop the speech recognition service. diff --git a/src/pipecat/services/cartesia/stt.py b/src/pipecat/services/cartesia/stt.py index a2ae9432f..7494ea909 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(exception=e) async def _disconnect_websocket(self): try: @@ -319,8 +318,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, exception=e) @traced_stt async def _handle_transcription( diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py index f8881200c..bf3747351 100644 --- a/src/pipecat/services/cartesia/tts.py +++ b/src/pipecat/services/cartesia/tts.py @@ -397,8 +397,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(exception=e) self._websocket = None await self._call_event_handler("on_connection_error", f"{e}") @@ -464,10 +463,9 @@ 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"{self} error: {msg}", exception=e) self._context_id = None else: logger.error(f"{self} error, unknown message type: {msg}") @@ -708,8 +706,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}")) + await self.push_error(error_msg=f"Cartesia API error: {error_text}") raise Exception(f"Cartesia API returned status {response.status}: {error_text}") audio_data = await response.read() @@ -725,8 +722,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}")) + await self.push_error(exception=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..2154d3ed7 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(exception=e) finally: # Reset state only after everything is cleaned up self._websocket = None @@ -280,8 +279,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"{self} error closing websocket: {e}", exception=e) finally: self._websocket = None await self._call_event_handler("on_disconnected") @@ -467,8 +465,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(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..a4162840a 100644 --- a/src/pipecat/services/deepgram/stt.py +++ b/src/pipecat/services/deepgram/stt.py @@ -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/elevenlabs/stt.py b/src/pipecat/services/elevenlabs/stt.py index 8cbb40d63..38b6e150e 100644 --- a/src/pipecat/services/elevenlabs/stt.py +++ b/src/pipecat/services/elevenlabs/stt.py @@ -645,8 +645,7 @@ 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"{self}: unable to connect to ElevenLabs Realtime STT: {e}", exception=e) async def _disconnect_websocket(self): """Disconnect from ElevenLabs Realtime STT WebSocket.""" @@ -714,13 +713,11 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService): elif message_type == "input_error": error_msg = data.get("error", "Unknown input error") - logger.error(f"ElevenLabs input error: {error_msg}") - await self.push_error(ErrorFrame(f"Input error: {error_msg}")) + await self.push_error(error_msg=f"ElevenLabs input error: {error_msg}") elif message_type in ["auth_error", "quota_exceeded", "transcriber_error", "error"]: error_msg = data.get("error", data.get("message", "Unknown error")) - logger.error(f"ElevenLabs error ({message_type}): {error_msg}") - await self.push_error(ErrorFrame(f"{message_type}: {error_msg}")) + await self.push_error(error_msg=f"ElevenLabs error ({message_type}): {error_msg}") else: logger.debug(f"Unknown message type: {message_type}") diff --git a/src/pipecat/services/fish/tts.py b/src/pipecat/services/fish/tts.py index db4f0f387..da71f3829 100644 --- a/src/pipecat/services/fish/tts.py +++ b/src/pipecat/services/fish/tts.py @@ -242,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(exception=e) finally: self._request_id = None self._started = False diff --git a/src/pipecat/services/xtts/tts.py b/src/pipecat/services/xtts/tts.py index 34ddecbe2..30b7b90d9 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"{self} error getting studio speakers (status: {r.status}, error: {text})" ) return self._studio_speakers = await r.json()