diff --git a/CHANGELOG.md b/CHANGELOG.md index 984c033e1..83b6dc03e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- If an unexpected exception is caught, or if `FrameProcessor.push_error()` is + called with an exception, the file name and line number where the exception + occured are now logged. + - Improved interruption handling to prevent bots from repeating themselves. LLM services that return multiple sentences in a single response (e.g., `GoogleLLMService`) are now split into individual sentences before being sent diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 63390cf94..ec9ebab98 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -12,6 +12,7 @@ management, and frame flow control mechanisms. """ import asyncio +import traceback from dataclasses import dataclass from enum import Enum from typing import Any, Awaitable, Callable, Coroutine, List, Optional, Sequence, Tuple, Type @@ -677,7 +678,17 @@ class FrameProcessor(BaseObject): if not error.processor: error.processor = self await self._call_event_handler("on_error", error) - logger.error(f"{error.processor} error: {error.error}") + + if error.exception: + tb = traceback.extract_tb(error.exception.__traceback__) + last = tb[-1] + error_message = ( + f"{error.processor} exception ({last.filename}:{last.lineno}): {error.error}" + ) + else: + error_message = f"{error.processor} error: {error.error}" + + logger.error(error_message) await self.push_frame(error, FrameDirection.UPSTREAM) async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): diff --git a/src/pipecat/utils/asyncio/task_manager.py b/src/pipecat/utils/asyncio/task_manager.py index cf433edb1..b41a571c9 100644 --- a/src/pipecat/utils/asyncio/task_manager.py +++ b/src/pipecat/utils/asyncio/task_manager.py @@ -12,6 +12,7 @@ comprehensive monitoring and cleanup capabilities. """ import asyncio +import traceback from abc import ABC, abstractmethod from dataclasses import dataclass from typing import Coroutine, Dict, Optional, Sequence @@ -162,7 +163,9 @@ class TaskManager(BaseTaskManager): # Re-raise the exception to ensure the task is cancelled. raise except Exception as e: - logger.error(f"{name}: unexpected exception: {e}") + tb = traceback.extract_tb(e.__traceback__) + last = tb[-1] + logger.error(f"{name} unexpected exception ({last.filename}:{last.lineno}): {e}") if not self._params: raise Exception("TaskManager is not setup: unable to get event loop") @@ -197,9 +200,17 @@ class TaskManager(BaseTaskManager): # Here are sure the task is cancelled properly. pass except Exception as e: - logger.error(f"{name}: unexpected exception while cancelling task: {e}") + tb = traceback.extract_tb(e.__traceback__) + last = tb[-1] + logger.error( + f"{name} unexpected exception while cancelling task ({last.filename}:{last.lineno}): {e}" + ) except BaseException as e: - logger.critical(f"{name}: fatal base exception while cancelling task: {e}") + tb = traceback.extract_tb(e.__traceback__) + last = tb[-1] + logger.critical( + f"{name} fatal base exception while cancelling task ({last.filename}:{last.lineno}): {e}" + ) raise def current_tasks(self) -> Sequence[asyncio.Task]: