Merge pull request #4172 from pipecat-ai/mb/rime-tts-fixes

Fix Rime TTS stop-frame handling and handle done message

changelog/4172.fixed.md

@@ -0,0 +1 @@
+- Fixed duplicate `TTSStoppedFrame` being pushed in TTS services using `push_stop_frames=True`. When the stop-frame timeout fired, a second `TTSStoppedFrame` could be pushed after the normal one at context completion.

changelog/4172.performance.md

@@ -0,0 +1 @@
+- `RimeTTSService` now handles Rime's `done` WebSocket message to complete audio contexts immediately, eliminating the 3-second idle timeout that previously added latency at the end of each utterance.

src/pipecat/services/rime/tts.py

@@ -24,13 +24,11 @@ from pipecat.frames.frames import (
     EndFrame,
     ErrorFrame,
     Frame,
-    InterruptionFrame,
     StartFrame,
     TTSAudioRawFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
 )
-from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.settings import NOT_GIVEN, TTSSettings, _NotGiven
 from pipecat.services.tts_service import (
     InterruptibleTTSService,
@@ -273,7 +271,6 @@ class RimeTTSService(WebsocketTTSService):
             text_aggregation_mode=text_aggregation_mode,
             aggregate_sentences=aggregate_sentences,
             push_text_frames=False,
-            push_stop_frames=True,
             pause_frame_processing=True,
             append_trailing_space=True,
             sample_rate=sample_rate,
@@ -521,9 +518,8 @@ class RimeTTSService(WebsocketTTSService):
     async def on_audio_context_completed(self, context_id: str):
         """Clear server-side state and stop metrics after the Rime context finishes playing.
 
-        Rime does not send a server-side completion signal (e.g. ``done`` / ``end_of_stream`` /
+        Sends a ``clear`` message to clean up any residual server-side state
-        ``audio_end``), so we explicitly send a ``clear`` message to clean up
+        once all audio has been delivered.
-        any residual server-side state once all audio has been delivered.
         """
         await self._close_context(context_id)
         await super().on_audio_context_completed(context_id)
@@ -600,6 +596,13 @@ class RimeTTSService(WebsocketTTSService):
                         self._cumulative_time = ends[-1] + self._cumulative_time
                         logger.debug(f"Updated cumulative time to: {self._cumulative_time}")
 
+            elif msg["type"] == "done":
+                await self.stop_ttfb_metrics()
+                await self.append_to_audio_context(
+                    context_id, TTSStoppedFrame(context_id=context_id)
+                )
+                await self.remove_audio_context(context_id)
+
             elif msg["type"] == "error":
                 await self.push_frame(TTSStoppedFrame())
                 await self.stop_all_metrics()

src/pipecat/services/tts_service.py

@@ -1403,6 +1403,7 @@ class TTSService(AIService):
                 logger.trace(f"{self} time out on audio context {context_id}")
                 if should_push_stop_frame and self._push_stop_frames:
                     await self.push_frame(TTSStoppedFrame(context_id=context_id))
+                    should_push_stop_frame = False
                 break
 
         if should_push_stop_frame and self._push_stop_frames: