From d6d39fc8737c68e1b9232350d22bc58f1b87639a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sat, 2 Aug 2025 20:39:59 -0700 Subject: [PATCH 1/2] TaskObserver: don't inspect on_push_frame signature for every frame --- CHANGELOG.md | 5 +++++ src/pipecat/pipeline/task_observer.py | 29 ++++++++++++++------------- 2 files changed, 20 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 84f7c7047..671e0c771 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Fixed an issue with the `TavusVideoService` where an error was thrown due to missing transcription callbacks. +### Performance + +- Fixed an issue in `TaskObserver` (a proxy to all observers) that was degrading + global performance. + ## [0.0.77] - 2025-07-31 ### Added diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py index cd46f85ef..532ef7977 100644 --- a/src/pipecat/pipeline/task_observer.py +++ b/src/pipecat/pipeline/task_observer.py @@ -153,22 +153,23 @@ class TaskObserver(BaseObserver): async def _proxy_task_handler(self, queue: asyncio.Queue, observer: BaseObserver): """Handle frame processing for a single observer.""" - warning_reported = False + on_push_frame_deprecated = False + signature = inspect.signature(observer.on_push_frame) + if len(signature.parameters) > 1: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Observer `on_push_frame(source, destination, frame, direction, timestamp)` is deprecated, us `on_push_frame(data: FramePushed)` instead.", + DeprecationWarning, + ) + + on_push_frame_deprecated = True + while True: data = await queue.get() - - signature = inspect.signature(observer.on_push_frame) - if len(signature.parameters) > 1: - if not warning_reported: - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Observer `on_push_frame(source, destination, frame, direction, timestamp)` is deprecated, us `on_push_frame(data: FramePushed)` instead.", - DeprecationWarning, - ) - warning_reported = True + if on_push_frame_deprecated: await observer.on_push_frame( data.src, data.dst, data.frame, data.direction, data.timestamp ) From 3a72e94d0ca146ab1d6b31672e5eb41ac767365e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Sun, 3 Aug 2025 09:59:14 -0700 Subject: [PATCH 2/2] LLMService: only do handle function inspection once --- src/pipecat/services/llm_service.py | 28 ++++++++++++++++------------ 1 file changed, 16 insertions(+), 12 deletions(-) diff --git a/src/pipecat/services/llm_service.py b/src/pipecat/services/llm_service.py index f3e2b843a..c3ad69e0e 100644 --- a/src/pipecat/services/llm_service.py +++ b/src/pipecat/services/llm_service.py @@ -25,7 +25,6 @@ from loguru import logger from pipecat.adapters.base_llm_adapter import BaseLLMAdapter from pipecat.adapters.schemas.direct_function import DirectFunction, DirectFunctionWrapper -from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.services.open_ai_adapter import OpenAILLMAdapter from pipecat.frames.frames import ( CancelFrame, @@ -108,6 +107,7 @@ class FunctionCallRegistryItem: function_name: Optional[str] handler: FunctionCallHandler | "DirectFunctionWrapper" cancel_on_interruption: bool + handler_deprecated: bool @dataclass @@ -282,12 +282,25 @@ class LLMService(AIService): cancel_on_interruption: Whether to cancel this function call when an interruption occurs. Defaults to True. """ + signature = inspect.signature(handler) + handler_deprecated = len(signature.parameters) > 1 + if handler_deprecated: + import warnings + + with warnings.catch_warnings(): + warnings.simplefilter("always") + warnings.warn( + "Function calls with parameters `(function_name, tool_call_id, arguments, llm, context, result_callback)` are deprecated, use a single `FunctionCallParams` parameter instead.", + DeprecationWarning, + ) + # Registering a function with the function_name set to None will run # that handler for all functions self._functions[function_name] = FunctionCallRegistryItem( function_name=function_name, handler=handler, cancel_on_interruption=cancel_on_interruption, + handler_deprecated=handler_deprecated, ) # Start callbacks are now deprecated. @@ -325,6 +338,7 @@ class LLMService(AIService): function_name=wrapper.name, handler=wrapper, cancel_on_interruption=cancel_on_interruption, + handler_deprecated=False, ) def unregister_function(self, function_name: Optional[str]): @@ -552,17 +566,7 @@ class LLMService(AIService): ) else: # Handler is a FunctionCallHandler - signature = inspect.signature(item.handler) - if len(signature.parameters) > 1: - import warnings - - with warnings.catch_warnings(): - warnings.simplefilter("always") - warnings.warn( - "Function calls with parameters `(function_name, tool_call_id, arguments, llm, context, result_callback)` are deprecated, use a single `FunctionCallParams` parameter instead.", - DeprecationWarning, - ) - + if item.handler_deprecated: await item.handler( runner_item.function_name, runner_item.tool_call_id,