wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add source parameter to ErrorFrame and set it in FrameProcessor. Updated error handling in AnthropicLLMService and DeepgramSTTService to include ErrorFrame with source information.

Browse Source
This commit is contained in:
richtermb
2025-08-01 11:14:50 -07:00
parent b8eef4f93b
commit ead4e97ab5
4 changed files with 8 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/pipecat/frames/frames.py
View File

@@ -652,10 +652,12 @@ class ErrorFrame(SystemFrame):
Parameters: Parameters:
error: Description of the error that occurred. error: Description of the error that occurred.
fatal: Whether the error is fatal and requires bot shutdown. fatal: Whether the error is fatal and requires bot shutdown.
source: The frame processor that generated the error.
""" """
error: str error: str
fatal: bool = False fatal: bool = False
source: Optional[FrameProcessor] = None
def __str__(self): def __str__(self):
return f"{self.name}(error: {self.error}, fatal: {self.fatal})" return f"{self.name}(error: {self.error}, fatal: {self.fatal})"

2
src/pipecat/processors/frame_processor.py
View File

@@ -494,6 +494,8 @@ class FrameProcessor(BaseObject):
Args: Args:
error: The error frame to push. error: The error frame to push.
""" """
if not error.source:
error.source = self
await self.push_frame(error, FrameDirection.UPSTREAM) await self.push_frame(error, FrameDirection.UPSTREAM)
async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):

2
src/pipecat/services/anthropic/llm.py
View File

@@ -26,6 +26,7 @@ from pydantic import BaseModel, Field
from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter
from pipecat.frames.frames import ( from pipecat.frames.frames import (
ErrorFrame,
Frame, Frame,
FunctionCallCancelFrame, FunctionCallCancelFrame,
FunctionCallInProgressFrame, FunctionCallInProgressFrame,
@@ -346,6 +347,7 @@ class AnthropicLLMService(LLMService):
await self._call_event_handler("on_completion_timeout") await self._call_event_handler("on_completion_timeout")
except Exception as e: except Exception as e:
logger.exception(f"{self} exception: {e}") logger.exception(f"{self} exception: {e}")
await self.push_error(ErrorFrame(f"{e}"))
finally: finally:
await self.stop_processing_metrics() await self.stop_processing_metrics()
await self.push_frame(LLMFullResponseEndFrame()) await self.push_frame(LLMFullResponseEndFrame())

2
src/pipecat/services/deepgram/stt.py
View File

@@ -13,6 +13,7 @@ from loguru import logger
from pipecat.frames.frames import ( from pipecat.frames.frames import (
CancelFrame, CancelFrame,
EndFrame, EndFrame,
ErrorFrame,
Frame, Frame,
InterimTranscriptionFrame, InterimTranscriptionFrame,
StartFrame, StartFrame,
@@ -246,6 +247,7 @@ class DeepgramSTTService(STTService):
async def _on_error(self, *args, **kwargs): async def _on_error(self, *args, **kwargs):
error: ErrorResponse = kwargs["error"] error: ErrorResponse = kwargs["error"]
logger.warning(f"{self} connection error, will retry: {error}") logger.warning(f"{self} connection error, will retry: {error}")
await self.push_error(ErrorFrame(f"{error}"))
await self.stop_all_metrics() await self.stop_all_metrics()
# NOTE(aleix): we don't disconnect (i.e. call finish on the connection) # NOTE(aleix): we don't disconnect (i.e. call finish on the connection)
# because this triggers more errors internally in the Deepgram SDK. So, # because this triggers more errors internally in the Deepgram SDK. So,