Merge pull request #3623 from pipecat-ai/aleix/pipeline-task-rtvi-always-set-bot-ready

PipelineTask: also call set_bot_ready() for external RTVI processors

changelog/3623.fixed.md

@@ -0,0 +1 @@
+- Fixed `PipelineTask` to also call `set_bot_ready()` when an external `RTVIProcessor` is provided.

src/pipecat/pipeline/task.py

@@ -15,7 +15,7 @@ import asyncio
 import importlib.util
 import os
 from pathlib import Path
-from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Set, Tuple, Type
+from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Set, Tuple, Type, TypeVar
 
 from loguru import logger
 from pydantic import BaseModel, ConfigDict, Field
@@ -62,6 +62,9 @@ IDLE_TIMEOUT_SECS = 300
 CANCEL_TIMEOUT_SECS = 20.0
 
 
+T = TypeVar("T")
+
+
 class IdleFrameObserver(BaseObserver):
     """Idle timeout observer.
 
@@ -333,15 +336,17 @@ class PipelineTask(BasePipelineTask):
                 f"{self}: RTVIProcessor and RTVIObserver found, skipping default ones. "
                 "They are both added by default, no need to add them yourself."
             )
+            self._rtvi = external_rtvi
         elif enable_rtvi:
             self._rtvi = rtvi_processor or RTVIProcessor()
+            observers.append(self._rtvi.create_rtvi_observer(params=rtvi_observer_params))
 
+        if self._rtvi:
+            # Automatically call RTVIProcessor.set_bot_ready()
             @self.rtvi.event_handler("on_client_ready")
             async def on_client_ready(rtvi: RTVIProcessor):
                 await rtvi.set_bot_ready()
 
-            observers.append(self._rtvi.create_rtvi_observer(params=rtvi_observer_params))
-
         # This is the idle event. When selected frames are pushed from any
         # processor we consider the pipeline is not idle. We use an observer
         # which will be listening any part of the pipeline.
@@ -1039,9 +1044,7 @@ class PipelineTask(BasePipelineTask):
 
         return start_metadata
 
-    def _find_processor(
+    def _find_processor(self, processor: FrameProcessor, processor_type: Type[T]) -> Optional[T]:
-        self, processor: FrameProcessor, processor_type: Type[FrameProcessor]
-    ) -> Optional[FrameProcessor]:
         """Recursively find a processor of the given type in the pipeline."""
         if isinstance(processor, processor_type):
             return processor