Merge pull request #3936 from pipecat-ai/filipi/fix_push_aggregation

Fixed TTS context not being appended to the assistant message history

changelog/3936.fixed.md

@@ -0,0 +1 @@
+- Fixed TTS context not being appended to the assistant message history when using `TTSSpeakFrame` with `append_to_context=True` with some TTS providers.

src/pipecat/services/piper/tts.py

@@ -17,6 +17,7 @@ from loguru import logger
 from pipecat.frames.frames import (
     ErrorFrame,
     Frame,
+    TTSStoppedFrame,
 )
 from pipecat.services.settings import TTSSettings, _warn_deprecated_param
 from pipecat.services.tts_service import TTSService
@@ -289,6 +290,7 @@ class PiperHttpTTSService(TTSService):
                     yield ErrorFrame(
                         error=f"Error getting audio (status: {response.status}, error: {error})"
                     )
+                    yield TTSStoppedFrame(context_id=context_id)
                     return
 
                 await self.start_tts_usage_metrics(text)

src/pipecat/services/tts_service.py

@@ -768,6 +768,8 @@ class TTSService(AIService):
         # Clean up context when we see TTSStoppedFrame
         if isinstance(frame, TTSStoppedFrame) and frame.context_id:
             if frame.context_id in self._tts_contexts:
+                if self._tts_contexts[frame.context_id].push_assistant_aggregation:
+                    await self.push_frame(LLMAssistantPushAggregationFrame())
                 logger.debug(f"{self} cleaning up TTS context {frame.context_id}")
                 del self._tts_contexts[frame.context_id]
 
@@ -1009,14 +1011,8 @@ class TTSService(AIService):
             # For services using the audio context we are appending to the context, so it preserves the ordering.
             if self.audio_context_available(context_id):
                 await self.append_to_audio_context(context_id, frame)
-                if push_assistant_aggregation:
-                    await self.append_to_audio_context(
-                        context_id, LLMAssistantPushAggregationFrame()
-                    )
             else:
                 await self.push_frame(frame)
-                if push_assistant_aggregation:
-                    await self.push_frame(LLMAssistantPushAggregationFrame())
 
     async def tts_process_generator(
         self, context_id: str, generator: AsyncGenerator[Frame | None, None]
@@ -1042,23 +1038,27 @@ class TTSService(AIService):
         async for frame in generator:
             if frame:
                 await self.append_to_audio_context(context_id, frame)
-                is_yielding_frames = True
+                if isinstance(frame, TTSAudioRawFrame):
+                    is_yielding_frames = True
+
         self._is_yielding_frames_synchronously = is_yielding_frames
 
     async def _stop_frame_handler(self):
         has_started = False
+        context_id = None
         while True:
             try:
                 frame = await asyncio.wait_for(
                     self._stop_frame_queue.get(), timeout=self._stop_frame_timeout_s
                 )
                 if isinstance(frame, TTSStartedFrame):
+                    context_id = frame.context_id
                     has_started = True
                 elif isinstance(frame, (TTSStoppedFrame, InterruptionFrame)):
                     has_started = False
             except asyncio.TimeoutError:
                 if has_started:
-                    await self.push_frame(TTSStoppedFrame())
+                    await self.push_frame(TTSStoppedFrame(context_id=context_id))
                     has_started = False
 
     #
@@ -1142,9 +1142,6 @@ class TTSService(AIService):
                 frame.pts = self._word_last_pts
                 frame.context_id = context_id
                 await self.push_frame(frame)
-                if context_id in self._tts_contexts:
-                    if self._tts_contexts[context_id].push_assistant_aggregation:
-                        await self.push_frame(LLMAssistantPushAggregationFrame())
             else:
                 ts_ns = seconds_to_nanoseconds(timestamp)
                 if self._initial_word_timestamp == -1:

tests/test_piper_tts.py

@@ -136,7 +136,7 @@ async def test_run_piper_tts_error(aiohttp_client):
             TTSSpeakFrame(text="Error case.", append_to_context=False),
         ]
 
-        expected_down_frames = [AggregatedTextFrame, TTSStartedFrame, TTSTextFrame, TTSStoppedFrame]
+        expected_down_frames = [AggregatedTextFrame, TTSStartedFrame, TTSStoppedFrame, TTSTextFrame]
 
         expected_up_frames = [ErrorFrame]