Merge pull request #2345 from pipecat-ai/aleix/task-observer-signature-performance

TaskObserver: don't inspect on_push_frame signature for every frame

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

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
                 )

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,