diff --git a/CHANGELOG.md b/CHANGELOG.md index 4ada4e408..5c590528d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 `RTVIBotLLMProcessor`) are now deprecated, instantiate an `RTVIObserver` instead. +### Fixed + +- Fixed an `RTVI` issue that was causing `bot-tts-text` messages to be sent + before being processed by the output transport. + ## [0.0.56] - 2025-02-06 ### Changed diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index b5965c64b..186a26cf0 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -58,6 +58,7 @@ from pipecat.processors.aggregators.openai_llm_context import ( OpenAILLMContextFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.transports.base_output import BaseOutputTransport from pipecat.utils.string import match_endofsentence RTVI_PROTOCOL_VERSION = "0.3.0" @@ -632,10 +633,18 @@ class RTVIMetricsProcessor(RTVIFrameProcessor): class RTVIObserver(BaseObserver): - """This is a pipeline frame observer that is used to send RTVI server - messages to clients. The observer does not handle incoming RTVI client - messages, which is done by the RTVIProcessor. + """Pipeline frame observer for RTVI server message handling. + This observer monitors pipeline frames and converts them into appropriate RTVI messages + for client communication. It handles various frame types including speech events, + transcriptions, LLM responses, and TTS events. + + Note: + This observer only handles outgoing messages. Incoming RTVI client messages + are handled by the RTVIProcessor. + + Args: + rtvi (FrameProcessor): The RTVI processor to push frames to. """ def __init__(self, rtvi: FrameProcessor): @@ -652,10 +661,22 @@ class RTVIObserver(BaseObserver): direction: FrameDirection, timestamp: int, ): + """Process a frame being pushed through the pipeline. + + Args: + src: Source processor pushing the frame + dst: Destination processor receiving the frame + frame: The frame being pushed + direction: Direction of frame flow in pipeline + timestamp: Time when frame was pushed + """ # If we have already seen this frame, let's skip it. if frame.id in self._frames_seen: return - self._frames_seen.add(frame.id) + + # This tells whether the frame is already processed. If false, we will try + # again the next time we see the frame. + mark_as_seen = True if isinstance(frame, (UserStartedSpeakingFrame, UserStoppedSpeakingFrame)): await self._handle_interruptions(frame) @@ -678,12 +699,24 @@ class RTVIObserver(BaseObserver): elif isinstance(frame, TTSStoppedFrame): await self.push_transport_message_urgent(RTVIBotTTSStoppedMessage()) elif isinstance(frame, TTSTextFrame): - message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text)) - await self.push_transport_message_urgent(message) + if isinstance(src, BaseOutputTransport): + message = RTVIBotTTSTextMessage(data=RTVITextMessageData(text=frame.text)) + await self.push_transport_message_urgent(message) + else: + mark_as_seen = False elif isinstance(frame, MetricsFrame): await self._handle_metrics(frame) + if mark_as_seen: + self._frames_seen.add(frame.id) + async def push_transport_message_urgent(self, model: BaseModel, exclude_none: bool = True): + """Push an urgent transport message to the RTVI processor. + + Args: + model: The message model to send + exclude_none: Whether to exclude None values from the model dump + """ frame = TransportMessageUrgentFrame(message=model.model_dump(exclude_none=exclude_none)) await self._rtvi.push_frame(frame)