diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ace173b8..a74e0fc82 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,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. diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 986b59bcd..9bd08c489 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/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))