frames: add new FatalErrorFrame

src/pipecat/frames/frames.py

@@ -238,11 +238,24 @@ class CancelFrame(SystemFrame):
 @dataclass
 class ErrorFrame(SystemFrame):
     """This is used notify upstream that an error has occurred downstream the
-    pipeline."""
+    pipeline. A fatal error indicates the error is unrecoverable and that the
+    bot should exit.
+
+    """
     error: str
+    fatal: bool = False
 
     def __str__(self):
-        return f"{self.name}(error: {self.error})"
+        return f"{self.name}(error: {self.error}, fatal: {self.fatal})"
+
+
+@dataclass
+class FatalErrorFrame(ErrorFrame):
+    """This is used notify upstream that an unrecoverable error has occurred and
+    that the bot should exit.
+
+    """
+    fatal: bool = field(default=True, init=False)
 
 
 @dataclass

src/pipecat/pipeline/task.py

@@ -50,11 +50,12 @@ class Source(FrameProcessor):
 
     async def _handle_upstream_frame(self, frame: Frame):
         if isinstance(frame, ErrorFrame):
-            logger.error(f"Error running app: {frame.error}")
+            logger.error(f"Error running app: {frame}")
-            # Cancel all tasks downstream.
+            if frame.fatal:
-            await self.push_frame(CancelFrame())
+                # Cancel all tasks downstream.
-            # Tell the task we should stop.
+                await self.push_frame(CancelFrame())
-            await self._up_queue.put(StopTaskFrame())
+                # Tell the task we should stop.
+                await self._up_queue.put(StopTaskFrame())
 
 
 class PipelineTask:

src/pipecat/processors/frame_processor.py

@@ -9,7 +9,13 @@ import time
 
 from enum import Enum
 
-from pipecat.frames.frames import ErrorFrame, Frame, MetricsFrame, StartFrame, StartInterruptionFrame, UserStoppedSpeakingFrame
+from pipecat.frames.frames import (
+    ErrorFrame,
+    Frame,
+    MetricsFrame,
+    StartFrame,
+    StartInterruptionFrame,
+    UserStoppedSpeakingFrame)
 from pipecat.utils.utils import obj_count, obj_id
 
 from loguru import logger