improve error handling and don't swallow exceptions

CHANGELOG.md

@@ -15,6 +15,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed an issue where exceptions that occurred inside frame processors were
+  being swallowed and not displayed.
+
 - Fixed an issue in `FastAPIWebsocketTransport` where it would still try to send
   data to the websocket after being closed.
 

src/pipecat/processors/filters/wake_check_filter.py

@@ -82,5 +82,5 @@ class WakeCheckFilter(FrameProcessor):
                 await self.push_frame(frame, direction)
         except Exception as e:
             error_msg = f"Error in wake word filter: {e}"
-            logger.error(error_msg)
+            logger.exception(error_msg)
             await self.push_error(ErrorFrame(error_msg))

src/pipecat/processors/frame_processor.py

@@ -147,12 +147,15 @@ class FrameProcessor:
         await self.push_frame(error, FrameDirection.UPSTREAM)
 
     async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):
-        if direction == FrameDirection.DOWNSTREAM and self._next:
+        try:
-            logger.trace(f"Pushing {frame} from {self} to {self._next}")
+            if direction == FrameDirection.DOWNSTREAM and self._next:
-            await self._next.process_frame(frame, direction)
+                logger.trace(f"Pushing {frame} from {self} to {self._next}")
-        elif direction == FrameDirection.UPSTREAM and self._prev:
+                await self._next.process_frame(frame, direction)
-            logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}")
+            elif direction == FrameDirection.UPSTREAM and self._prev:
-            await self._prev.process_frame(frame, direction)
+                logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}")
+                await self._prev.process_frame(frame, direction)
+        except Exception as e:
+            logger.exception(f"Uncaught exception in {self}: {e}")
 
     def __str__(self):
         return self.name

src/pipecat/processors/frameworks/langchain.py

@@ -75,5 +75,6 @@ class LangchainProcessor(FrameProcessor):
         except GeneratorExit:
             logger.warning(f"{self} generator was closed prematurely")
         except Exception as e:
-            logger.error(f"{self} an unknown error occurred: {e}")
+            logger.exception(f"{self} an unknown error occurred: {e}")
-        await self.push_frame(LLMFullResponseEndFrame())
+        finally:
+            await self.push_frame(LLMFullResponseEndFrame())

src/pipecat/services/anthropic.py

@@ -122,7 +122,7 @@ class AnthropicLLMService(LLMService):
                     await self.push_frame(LLMResponseEndFrame())
 
         except Exception as e:
-            logger.error(f"{self} exception: {e}")
+            logger.exception(f"{self} exception: {e}")
         finally:
             await self.push_frame(LLMFullResponseEndFrame())
 

src/pipecat/services/cartesia.py

@@ -39,7 +39,7 @@ class CartesiaTTSService(TTSService):
             self._client = AsyncCartesia(api_key=self._api_key)
             self._voice = self._client.voices.get(id=voice_id)
         except Exception as e:
-            logger.error(f"{self} initialization error: {e}")
+            logger.exception(f"{self} initialization error: {e}")
 
     def can_generate_metrics(self) -> bool:
         return True
@@ -62,4 +62,4 @@ class CartesiaTTSService(TTSService):
                 await self.stop_ttfb_metrics()
                 yield AudioRawFrame(chunk["audio"], self._output_format["sample_rate"], 1)
         except Exception as e:
-            logger.error(f"{self} exception: {e}")
+            logger.exception(f"{self} exception: {e}")

src/pipecat/services/deepgram.py

@@ -91,7 +91,7 @@ class DeepgramTTSService(TTSService):
                     frame = AudioRawFrame(audio=data, sample_rate=16000, num_channels=1)
                     yield frame
         except Exception as e:
-            logger.error(f"{self} exception: {e}")
+            logger.exception(f"{self} exception: {e}")
 
 
 class DeepgramSTTService(AIService):

src/pipecat/services/google.py

@@ -104,10 +104,10 @@ class GoogleLLMService(LLMService):
                         logger.debug(
                             f"LLM refused to generate content for safety reasons - {messages}.")
                     else:
-                        logger.error(f"{self} error: {e}")
+                        logger.exception(f"{self} error: {e}")
 
         except Exception as e:
-            logger.error(f"{self} exception: {e}")
+            logger.exception(f"{self} exception: {e}")
         finally:
             await self.push_frame(LLMFullResponseEndFrame())
 

src/pipecat/services/openai.py

@@ -53,7 +53,7 @@ except ModuleNotFoundError as e:
     raise Exception(f"Missing module: {e}")
 
 
-class OpenAIUnhandledFunctionException(BaseException):
+class OpenAIUnhandledFunctionException(Exception):
     pass
 
 
@@ -109,10 +109,7 @@ class BaseOpenAILLMService(LLMService):
                 del message["data"]
                 del message["mime_type"]
 
-        try:
+        chunks = await self.get_chat_completions(context, messages)
-            chunks = await self.get_chat_completions(context, messages)
-        except Exception as e:
-            logger.error(f"{self} exception: {e}")
 
         return chunks
 
@@ -214,7 +211,7 @@ class BaseOpenAILLMService(LLMService):
         elif isinstance(result, type(None)):
             pass
         else:
-            raise BaseException(f"Unknown return type from function callback: {type(result)}")
+            raise TypeError(f"Unknown return type from function callback: {type(result)}")
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
@@ -336,4 +333,4 @@ class OpenAITTSService(TTSService):
                         frame = AudioRawFrame(chunk, 24_000, 1)
                         yield frame
         except BadRequestError as e:
-            logger.error(f"{self} error generating TTS: {e}")
+            logger.exception(f"{self} error generating TTS: {e}")

src/pipecat/services/playht.py

@@ -80,4 +80,4 @@ class PlayHTTTSService(TTSService):
                         frame = AudioRawFrame(chunk, 16000, 1)
                         yield frame
         except Exception as e:
-            logger.error(f"{self} error generating TTS: {e}")
+            logger.exception(f"{self} error generating TTS: {e}")

src/pipecat/transports/base_input.py

@@ -173,5 +173,5 @@ class BaseInputTransport(FrameProcessor):
                     await self._internal_push_frame(frame)
             except asyncio.CancelledError:
                 break
-            except BaseException as e:
+            except Exception as e:
-                logger.error(f"{self} error reading audio frames: {e}")
+                logger.exception(f"{self} error reading audio frames: {e}")

src/pipecat/transports/base_output.py

@@ -180,8 +180,8 @@ class BaseOutputTransport(FrameProcessor):
                 self._sink_queue.task_done()
             except asyncio.CancelledError:
                 break
-            except BaseException as e:
+            except Exception as e:
-                logger.error(f"{self} error processing sink queue: {e}")
+                logger.exception(f"{self} error processing sink queue: {e}")
 
     #
     # Push frames task
@@ -250,7 +250,7 @@ class BaseOutputTransport(FrameProcessor):
             except asyncio.CancelledError:
                 break
             except Exception as e:
-                logger.error(f"{self} error writing to camera: {e}")
+                logger.exception(f"{self} error writing to camera: {e}")
 
     #
     # Audio out

src/pipecat/transports/base_transport.py

@@ -82,5 +82,4 @@ class BaseTransport(ABC):
                 else:
                     handler(self, *args, **kwargs)
         except Exception as e:
-            logger.error(f"Exception in event handler {event_name}: {e}")
+            logger.exception(f"Exception in event handler {event_name}: {e}")
-            raise e

src/pipecat/transports/services/daily.py

@@ -835,8 +835,8 @@ class DailyTransport(BaseTransport):
                     logger.debug("Event dialin-ready was handled successfully")
             except asyncio.TimeoutError:
                 logger.error(f"Timeout handling dialin-ready event ({url})")
-            except BaseException as e:
+            except Exception as e:
-                logger.error(f"Error handling dialin-ready event ({url}): {e}")
+                logger.exception(f"Error handling dialin-ready event ({url}): {e}")
 
     async def _on_dialin_ready(self, sip_endpoint):
         if self._params.dialin_settings:

src/pipecat/utils/test_frame_processor.py

@@ -2,7 +2,7 @@ from typing import List
 from pipecat.processors.frame_processor import FrameProcessor
 
 
-class TestException(BaseException):
+class TestException(Exception):
     pass
 
 

src/pipecat/vad/silero.py

@@ -72,9 +72,9 @@ class SileroVADAnalyzer(VADAnalyzer):
                 self._last_reset_time = curr_time
 
             return new_confidence
-        except BaseException as e:
+        except Exception as e:
             # This comes from an empty audio array
-            logger.error(f"Error analyzing audio with Silero VAD: {e}")
+            logger.exception(f"Error analyzing audio with Silero VAD: {e}")
             return 0