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

fix: add logging and error handling for issue #721 (#755)

Browse Source
This commit is contained in:
James Hush
2024-11-29 13:06:45 +08:00
committed by GitHub
parent c5324df807
commit 5e22ef251d
1 changed files with 19 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -6,10 +6,11 @@
import asyncio import asyncio
import inspect import inspect
from enum import Enum from enum import Enum
from typing import Awaitable, Callable, Optional from typing import Awaitable, Callable, Optional
from loguru import logger
from pipecat.clocks.base_clock import BaseClock from pipecat.clocks.base_clock import BaseClock
from pipecat.frames.frames import ( from pipecat.frames.frames import (
EndFrame, EndFrame,
@@ -24,8 +25,6 @@ from pipecat.metrics.metrics import LLMTokenUsage, MetricsData
from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics
from pipecat.utils.utils import obj_count, obj_id from pipecat.utils.utils import obj_count, obj_id
from loguru import logger
class FrameDirection(Enum): class FrameDirection(Enum):
DOWNSTREAM = 1 DOWNSTREAM = 1
@@ -220,11 +219,16 @@ class FrameProcessor:
# #
async def _start_interruption(self): async def _start_interruption(self):
# Cancel the push frame task. This will stop pushing frames downstream. try:
await self.__cancel_push_task() # Cancel the push frame task. This will stop pushing frames downstream.
await self.__cancel_push_task()
# Cancel the input task. This will stop processing queued frames. # Cancel the input task. This will stop processing queued frames.
await self.__cancel_input_task() await self.__cancel_input_task()
except Exception as e:
logger.exception(f"Uncaught exception in {self}: {e}")
await self.push_error(ErrorFrame(str(e)))
raise
# Create a new input queue and task. # Create a new input queue and task.
self.__create_input_task() self.__create_input_task()
@@ -281,7 +285,11 @@ class FrameProcessor:
self.__input_queue.task_done() self.__input_queue.task_done()
except asyncio.CancelledError: except asyncio.CancelledError:
logger.trace(f"Cancelled input task in {self}")
break break
except Exception as e:
logger.exception(f"Uncaught exception in {self}: {e}")
await self.push_error(ErrorFrame(str(e)))
def __create_push_task(self): def __create_push_task(self):
self.__push_queue = asyncio.Queue() self.__push_queue = asyncio.Queue()
@@ -300,7 +308,11 @@ class FrameProcessor:
running = not isinstance(frame, EndFrame) running = not isinstance(frame, EndFrame)
self.__push_queue.task_done() self.__push_queue.task_done()
except asyncio.CancelledError: except asyncio.CancelledError:
logger.trace(f"Cancelled push task in {self}")
break break
except Exception as e:
logger.exception(f"Uncaught exception in {self}: {e}")
await self.push_error(ErrorFrame(str(e)))
async def _call_event_handler(self, event_name: str, *args, **kwargs): async def _call_event_handler(self, event_name: str, *args, **kwargs):
try: try: