From ead4e97ab5b135911a0b3ad32a859def76f07939 Mon Sep 17 00:00:00 2001 From: richtermb Date: Fri, 1 Aug 2025 11:14:50 -0700 Subject: [PATCH 1/5] Add source parameter to ErrorFrame and set it in FrameProcessor. Updated error handling in AnthropicLLMService and DeepgramSTTService to include ErrorFrame with source information. --- src/pipecat/frames/frames.py | 2 ++ src/pipecat/processors/frame_processor.py | 2 ++ src/pipecat/services/anthropic/llm.py | 2 ++ src/pipecat/services/deepgram/stt.py | 2 ++ 4 files changed, 8 insertions(+) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index bc9e8a06d..7f0a46efc 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -652,10 +652,12 @@ class ErrorFrame(SystemFrame): Parameters: error: Description of the error that occurred. fatal: Whether the error is fatal and requires bot shutdown. + source: The frame processor that generated the error. """ error: str fatal: bool = False + source: Optional[FrameProcessor] = None def __str__(self): return f"{self.name}(error: {self.error}, fatal: {self.fatal})" diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index a049fabdb..ac053007f 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -494,6 +494,8 @@ class FrameProcessor(BaseObject): Args: error: The error frame to push. """ + if not error.source: + error.source = self await self.push_frame(error, FrameDirection.UPSTREAM) async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index d7c703ebb..15c4cebee 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -26,6 +26,7 @@ from pydantic import BaseModel, Field from pipecat.adapters.services.anthropic_adapter import AnthropicLLMAdapter from pipecat.frames.frames import ( + ErrorFrame, Frame, FunctionCallCancelFrame, FunctionCallInProgressFrame, @@ -346,6 +347,7 @@ class AnthropicLLMService(LLMService): 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}")) finally: await self.stop_processing_metrics() await self.push_frame(LLMFullResponseEndFrame()) diff --git a/src/pipecat/services/deepgram/stt.py b/src/pipecat/services/deepgram/stt.py index 56658cad5..8381135d9 100644 --- a/src/pipecat/services/deepgram/stt.py +++ b/src/pipecat/services/deepgram/stt.py @@ -13,6 +13,7 @@ from loguru import logger from pipecat.frames.frames import ( CancelFrame, EndFrame, + ErrorFrame, Frame, InterimTranscriptionFrame, StartFrame, @@ -246,6 +247,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(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, From 165d6b4c1d5848409372f6cfff9eab4dc378a2f0 Mon Sep 17 00:00:00 2001 From: richtermb Date: Fri, 1 Aug 2025 12:25:29 -0700 Subject: [PATCH 2/5] Update CHANGELOG to include new `source` field in `ErrorFrame` for error tracking. --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 685ae8481..6690b54fa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -65,6 +65,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `on_transcription_stopped` and `on_transcription_error` to Daily callbacks. +- Added `source` field to `ErrorFrame` to indicate `FrameProcessor` that generated the error. + ### Changed - Changed the default `url` for `NeuphonicTTSService` to From 91568eeddca874fbd45537df7c798aac2e8b9fb3 Mon Sep 17 00:00:00 2001 From: richtermb Date: Fri, 1 Aug 2025 12:52:56 -0700 Subject: [PATCH 3/5] Update type hint for `source` in `ErrorFrame` to use forward declaration for improved clarity. --- src/pipecat/frames/frames.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 7f0a46efc..6ee7a858c 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -657,7 +657,7 @@ class ErrorFrame(SystemFrame): error: str fatal: bool = False - source: Optional[FrameProcessor] = None + source: Optional["FrameProcessor"] = None def __str__(self): return f"{self.name}(error: {self.error}, fatal: {self.fatal})" From 6814c390ba36c21440a2e8f4ef301677a88f69c4 Mon Sep 17 00:00:00 2001 From: richtermb Date: Fri, 1 Aug 2025 14:47:57 -0700 Subject: [PATCH 4/5] Update CHANGELOG to reflect the addition of the `source` field in `ErrorFrame` for improved error tracking. --- CHANGELOG.md | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6eb35808b..e3ad46575 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue in `LiveKitTransport` where empty `AudioRawFrame`s were pushed down the pipeline. This resulted in warnings by the STT processor. +- Added `source` field to `ErrorFrame` to indicate `FrameProcessor` that generated the error. + ## [0.0.77] - 2025-07-31 ### Added @@ -72,8 +74,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `on_transcription_stopped` and `on_transcription_error` to Daily callbacks. -- Added `source` field to `ErrorFrame` to indicate `FrameProcessor` that generated the error. - ### Changed - Changed the default `url` for `NeuphonicTTSService` to From efddc4732c68e345c2a1f599f4ffe9ef9abbb20c Mon Sep 17 00:00:00 2001 From: richtermb Date: Tue, 5 Aug 2025 13:25:08 -0700 Subject: [PATCH 5/5] Refactor ErrorFrame: rename `source` field to `processor` for clarity and update related references in FrameProcessor. --- CHANGELOG.md | 2 +- src/pipecat/frames/frames.py | 2 +- src/pipecat/processors/frame_processor.py | 4 ++-- 3 files changed, 4 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ee43fad6c..f99864d10 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -72,7 +72,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 removed in a future version. Use the `/start` endpoint in its place. In the meantime, both endpoints work and deliver equivalent functionality. -- Added `source` field to `ErrorFrame` to indicate `FrameProcessor` that generated the error. +- Added `processor` field to `ErrorFrame` to indicate `FrameProcessor` that generated the error. ## [0.0.77] - 2025-07-31 diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 6ee7a858c..30f8eb7b4 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -657,7 +657,7 @@ class ErrorFrame(SystemFrame): error: str fatal: bool = False - source: Optional["FrameProcessor"] = None + processor: Optional["FrameProcessor"] = None def __str__(self): return f"{self.name}(error: {self.error}, fatal: {self.fatal})" diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index ac053007f..e6611a161 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -494,8 +494,8 @@ class FrameProcessor(BaseObject): Args: error: The error frame to push. """ - if not error.source: - error.source = self + if not error.processor: + error.processor = self await self.push_frame(error, FrameDirection.UPSTREAM) async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM):