Merge pull request #326 from pipecat-ai/aleix/rtvi-bot-ready-fixes

rtvi: send bot-ready when pipeline is ready and first participant joins

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Added
+
+- Transports now allow you to register event handlers without decorators.
+
 ### Changed
 
 - `StartFrame` is now a control frame similar to `EndFrame`.
@@ -16,6 +20,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- RTVI's `bot-ready` message is now sent when the RTVI pipeline is ready and
+  a first participant joins.
+
 - Fixed a `BaseInputTransport` issue that was causing incoming system frames to
   be queued instead of being pushed immediately.
 

src/pipecat/processors/frameworks/rtvi.py

@@ -289,6 +289,13 @@ class RTVIProcessor(FrameProcessor):
         self._llm: FrameProcessor | None = None
         self._tts: FrameProcessor | None = None
         self._pipeline: FrameProcessor | None = None
+        self._first_participant_joined: bool = False
+
+        # Register transport event so we can send a `bot-ready` event (and maybe
+        # others) when the participant joins.
+        transport.add_event_handler(
+            "on_first_participant_joined",
+            self._on_first_participant_joined)
 
         self._frame_handler_task = self.get_event_loop().create_task(self._frame_handler())
         self._frame_queue = asyncio.Queue()
@@ -446,7 +453,6 @@ class RTVIProcessor(FrameProcessor):
             self._tma_out,
             self._transport.output(),
         ])
-        self._pipeline = pipeline
 
         parent = self.get_parent()
         if parent and self._start_frame:
@@ -459,14 +465,14 @@ class RTVIProcessor(FrameProcessor):
 
             # Send new initial metrics with the new processors
             processors = parent.processors_with_metrics()
-            processors.extend(self._pipeline.processors_with_metrics())
+            processors.extend(pipeline.processors_with_metrics())
             ttfb = [{"processor": p.name, "value": 0.0} for p in processors]
             processing = [{"processor": p.name, "value": 0.0} for p in processors]
             await self.push_frame(MetricsFrame(ttfb=ttfb, processing=processing))
 
-        message = RTVIBotReady()
-        frame = TransportMessageFrame(message=message.model_dump(exclude_none=True))
-        await self.push_frame(frame)
+        self._pipeline = pipeline
+
+        await self._maybe_send_bot_ready()
 
     async def _handle_config_update(self, config: RTVIConfig):
         # Change voice before LLM updates, so we can hear the new vocie.
@@ -506,6 +512,16 @@ class RTVIProcessor(FrameProcessor):
     async def _handle_tts_interrupt(self):
         await self.push_frame(BotInterruptionFrame(), FrameDirection.UPSTREAM)
 
+    async def _on_first_participant_joined(self, transport, participant):
+        self._first_participant_joined = True
+        await self._maybe_send_bot_ready()
+
+    async def _maybe_send_bot_ready(self):
+        if self._pipeline and self._first_participant_joined:
+            message = RTVIBotReady()
+            frame = TransportMessageFrame(message=message.model_dump(exclude_none=True))
+            await self.push_frame(frame)
+
     async def _send_error(self, error: str):
         message = RTVIError(data=RTVIErrorData(message=error))
         frame = TransportMessageFrame(message=message.model_dump(exclude_none=True))

src/pipecat/transports/base_transport.py

@@ -60,20 +60,20 @@ class BaseTransport(ABC):
 
     def event_handler(self, event_name: str):
         def decorator(handler):
-            self._add_event_handler(event_name, handler)
+            self.add_event_handler(event_name, handler)
             return handler
         return decorator
 
+    def add_event_handler(self, event_name: str, handler):
+        if event_name not in self._event_handlers:
+            raise Exception(f"Event handler {event_name} not registered")
+        self._event_handlers[event_name].append(handler)
+
     def _register_event_handler(self, event_name: str):
         if event_name in self._event_handlers:
             raise Exception(f"Event handler {event_name} already registered")
         self._event_handlers[event_name] = []
 
-    def _add_event_handler(self, event_name: str, handler):
-        if event_name not in self._event_handlers:
-            raise Exception(f"Event handler {event_name} not registered")
-        self._event_handlers[event_name].append(handler)
-
     async def _call_event_handler(self, event_name: str, *args, **kwargs):
         try:
             for handler in self._event_handlers[event_name]: