diff --git a/changelog/3623.fixed.md b/changelog/3623.fixed.md new file mode 100644 index 000000000..3eab6d90c --- /dev/null +++ b/changelog/3623.fixed.md @@ -0,0 +1 @@ +- Fixed `PipelineTask` to also call `set_bot_ready()` when an external `RTVIProcessor` is provided. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 6ae3c55a9..4ccc879b8 100644 --- a/src/pipecat/pipeline/task.py +++ b/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( - self, processor: FrameProcessor, processor_type: Type[FrameProcessor] - ) -> Optional[FrameProcessor]: + def _find_processor(self, processor: FrameProcessor, processor_type: Type[T]) -> Optional[T]: """Recursively find a processor of the given type in the pipeline.""" if isinstance(processor, processor_type): return processor