processors(rtvi): make sure to send bot-ready when transport is joined

src/pipecat/processors/frameworks/rtvi.py

@@ -23,9 +23,11 @@ from pipecat.frames.frames import (
     FunctionCallResultFrame,
     UserStoppedSpeakingFrame)
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.transports.base_transport import BaseTransport
 
 from loguru import logger
 
+
 RTVI_PROTOCOL_VERSION = "0.1"
 
 ActionResult = Union[bool, int, float, str, list, dict]
@@ -241,13 +243,24 @@ class RTVIUserStoppedSpeakingMessage(BaseModel):
     type: Literal["user-stopped-speaking"] = "user-stopped-speaking"
 
 
+class RTVIProcessorParams(BaseModel):
+    send_bot_ready: bool = True
+
+
 class RTVIProcessor(FrameProcessor):
 
-    def __init__(self, config: RTVIConfig):
+    def __init__(self,
+                 *,
+                 transport: BaseTransport,
+                 config: RTVIConfig = RTVIConfig(config=[]),
+                 params: RTVIProcessorParams = RTVIProcessorParams()):
         super().__init__()
         self._config = config
+        self._params = params
 
         self._pipeline: FrameProcessor | None = None
+        self._pipeline_started = False
+        self._transport_joined = False
 
         self._registered_actions: Dict[str, RTVIAction] = {}
         self._registered_services: Dict[str, RTVIService] = {}
@@ -255,6 +268,10 @@ class RTVIProcessor(FrameProcessor):
         self._frame_handler_task = self.get_event_loop().create_task(self._frame_handler())
         self._frame_queue = asyncio.Queue()
 
+        # TODO(aleix): This is very Daily specific. There should be a generic
+        # way to do this.
+        transport.add_event_handler("on_joined", self._transport_on_joined)
+
     def register_action(self, action: RTVIAction):
         id = self._action_id(action.service, action.action)
         self._registered_actions[id] = action
@@ -334,8 +351,9 @@ class RTVIProcessor(FrameProcessor):
             await self._pipeline.cleanup()
 
     async def _start(self, frame: StartFrame):
+        self._pipeline_started = True
         await self._update_config(self._config)
-        await self._send_bot_ready()
+        await self._maybe_send_bot_ready()
 
     async def _stop(self, frame: EndFrame):
         await self._frame_handler_task
@@ -503,7 +521,17 @@ class RTVIProcessor(FrameProcessor):
         frame = TransportMessageFrame(message=message.model_dump(exclude_none=True))
         await self.push_frame(frame)
 
+    async def _transport_on_joined(self, transport, participant):
+        self._transport_joined = True
+
+    async def _maybe_send_bot_ready(self):
+        if self._pipeline_started and self._transport_joined:
+            await self._send_bot_ready()
+
     async def _send_bot_ready(self):
+        if not self._params.send_bot_ready:
+            return
+
         message = RTVIBotReady(
             data=RTVIBotReadyData(
                 version=RTVI_PROTOCOL_VERSION,

src/pipecat/services/openai.py

@@ -8,7 +8,6 @@ import aiohttp
 import base64
 import io
 import json
-from anthropic.types import tool_use_block
 import httpx
 from dataclasses import dataclass
 
@@ -25,7 +24,6 @@ from pipecat.frames.frames import (
     LLMFullResponseStartFrame,
     LLMMessagesFrame,
     LLMModelUpdateFrame,
-    MetricsFrame,
     TextFrame,
     URLImageRawFrame,
     VisionImageRawFrame,
@@ -48,12 +46,7 @@ from pipecat.services.ai_services import (
 
 try:
     from openai import AsyncOpenAI, AsyncStream, DefaultAsyncHttpxClient, BadRequestError
-    from openai.types.chat import (
+    from openai.types.chat import ChatCompletionChunk, ChatCompletionMessageParam
-        ChatCompletionChunk,
-        ChatCompletionFunctionMessageParam,
-        ChatCompletionMessageParam,
-        ChatCompletionToolParam
-    )
 except ModuleNotFoundError as e:
     logger.error(f"Exception: {e}")
     logger.error(
@@ -65,7 +58,6 @@ class OpenAIUnhandledFunctionException(Exception):
     pass
 
 
-
 class BaseOpenAILLMService(LLMService):
     """This is the base for all services that use the AsyncOpenAI client.
 
@@ -94,8 +86,6 @@ class BaseOpenAILLMService(LLMService):
     def can_generate_metrics(self) -> bool:
         return True
 
-
-
     async def get_chat_completions(
             self,
             context: OpenAILLMContext,
@@ -208,7 +198,6 @@ class BaseOpenAILLMService(LLMService):
             arguments=arguments
         )
 
-
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
@@ -232,6 +221,7 @@ class BaseOpenAILLMService(LLMService):
             await self.stop_processing_metrics()
             await self.push_frame(LLMFullResponseEndFrame())
 
+
 @dataclass
 class OpenAIContextAggregatorPair:
     _user: 'OpenAIUserContextAggregator'
@@ -243,12 +233,13 @@ class OpenAIContextAggregatorPair:
     def assistant(self) -> str:
         return self._assistant
 
+
 class OpenAILLMService(BaseOpenAILLMService):
 
     def __init__(self, *, model: str = "gpt-4o", **kwargs):
         super().__init__(model=model, **kwargs)
 
-    @ staticmethod
+    @staticmethod
     def create_context_aggregator(context: OpenAILLMContext) -> OpenAIContextAggregatorPair:
         user = OpenAIUserContextAggregator(context)
         assistant = OpenAIAssistantContextAggregator(user)
@@ -256,6 +247,8 @@ class OpenAILLMService(BaseOpenAILLMService):
             _user=user,
             _assistant=assistant
         )
+
+
 class OpenAIImageGenService(ImageGenService):
 
     def __init__(
@@ -357,6 +350,7 @@ class OpenAITTSService(TTSService):
         except BadRequestError as e:
             logger.exception(f"{self} error generating TTS: {e}")
 
+
 class OpenAIUserContextAggregator(LLMUserContextAggregator):
     def __init__(self, context: OpenAILLMContext):
         super().__init__(context=context)
@@ -393,7 +387,6 @@ class OpenAIAssistantContextAggregator(LLMAssistantContextAggregator):
                 self._function_call_in_progress = None
                 self._function_call_result = None
 
-    
     def add_message(self, message):
         self._user_context_aggregator.add_message(message)