Merge pull request #2708 from pipecat-ai/aleix/base-output-transport-only-push-if-send-successful

BaseOutputTransport: only push downstream if transport write successful

CHANGELOG.md

@@ -68,6 +68,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- `BaseOutputTransport` methods `write_audio_frame` and `write_video_frame` now
+  return a boolean to indicate if the transport implementation was able to write
+  the given frame or not.
+
 - Updated Silero VAD model to v6.
 
 - Updated `livekit` to 1.0.13.
@@ -98,6 +102,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Fixed a `BaseOutputTransport` issue that could produce large saved
+  `AudioBufferProcessor` files when using an audio mixer.
+
 - Fixed a `PipelineRunner` issue on Windows where setting up SIGINT and SIGTERM
   was raising an exception.
 

examples/foundational/04b-transports-livekit.py

@@ -18,8 +18,8 @@ from pipecat.audio.vad.silero import SileroVADAnalyzer
 from pipecat.audio.vad.vad_analyzer import VADParams
 from pipecat.frames.frames import (
     InterruptionFrame,
-    TextFrame,
     TranscriptionFrame,
+    TTSSpeakFrame,
     UserStartedSpeakingFrame,
     UserStoppedSpeakingFrame,
 )
@@ -103,7 +103,7 @@ async def main():
     async def on_first_participant_joined(transport, participant_id):
         await asyncio.sleep(1)
         await task.queue_frame(
-            TextFrame(
+            TTSSpeakFrame(
                 "Hello there! How are you doing today? Would you like to talk about the weather?"
             )
         )

src/pipecat/processors/frameworks/strands_agents.py

@@ -84,11 +84,11 @@ class StrandsAgentsProcessor(FrameProcessor):
             text: The user input text to process through the agent or graph.
         """
         logger.debug(f"Invoking Strands agent with: {text}")
+        ttfb_tracking = True
         try:
             await self.push_frame(LLMFullResponseStartFrame())
             await self.start_processing_metrics()
             await self.start_ttfb_metrics()
-            ttfb_tracking = True
 
             if self.graph:
                 # Graph does not stream; await full result then emit assistant text
@@ -108,9 +108,9 @@ class StrandsAgentsProcessor(FrameProcessor):
                                     await self.push_frame(LLMTextFrame(str(block["text"])))
                         # Update usage metrics
                         await self._report_usage_metrics(
-                            agent_result.metrics.accumulated_usage.get('inputTokens', 0), 
+                            agent_result.metrics.accumulated_usage.get("inputTokens", 0),
-                            agent_result.metrics.accumulated_usage.get('outputTokens', 0), 
+                            agent_result.metrics.accumulated_usage.get("outputTokens", 0),
-                            agent_result.metrics.accumulated_usage.get('totalTokens', 0)
+                            agent_result.metrics.accumulated_usage.get("totalTokens", 0),
                         )
                 except Exception as parse_err:
                     logger.warning(f"Failed to extract messages from GraphResult: {parse_err}")
@@ -125,10 +125,18 @@ class StrandsAgentsProcessor(FrameProcessor):
                             ttfb_tracking = False
 
                     # Update usage metrics
-                    if isinstance(event, dict) and "event" in event and "metadata" in event['event']:
+                    if (
-                        if 'usage' in event['event']['metadata']:
+                        isinstance(event, dict)
-                            usage = event['event']['metadata']['usage']
+                        and "event" in event
-                            await self._report_usage_metrics(usage.get('inputTokens', 0), usage.get('outputTokens', 0), usage.get('totalTokens', 0))
+                        and "metadata" in event["event"]
+                    ):
+                        if "usage" in event["event"]["metadata"]:
+                            usage = event["event"]["metadata"]["usage"]
+                            await self._report_usage_metrics(
+                                usage.get("inputTokens", 0),
+                                usage.get("outputTokens", 0),
+                                usage.get("totalTokens", 0),
+                            )
         except GeneratorExit:
             logger.warning(f"{self} generator was closed prematurely")
         except Exception as e:
@@ -149,14 +157,11 @@ class StrandsAgentsProcessor(FrameProcessor):
         return True
 
     async def _report_usage_metrics(
-        self,
+        self, prompt_tokens: int, completion_tokens: int, total_tokens: int
-        prompt_tokens: int,
-        completion_tokens: int,
-        total_tokens: int
     ):
         tokens = LLMTokenUsage(
             prompt_tokens=prompt_tokens,
             completion_tokens=completion_tokens,
-            total_tokens=total_tokens
+            total_tokens=total_tokens,
         )
         await self.start_llm_usage_metrics(tokens)

src/pipecat/transports/base_output.py

@@ -202,21 +202,27 @@ class BaseOutputTransport(FrameProcessor):
         """
         pass
 
-    async def write_video_frame(self, frame: OutputImageRawFrame):
+    async def write_video_frame(self, frame: OutputImageRawFrame) -> bool:
         """Write a video frame to the transport.
 
         Args:
             frame: The output video frame to write.
-        """
-        pass
 
-    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+        Returns:
+            True if the video frame was written successfully, False otherwise.
+        """
+        return False
+
+    async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the transport.
 
         Args:
             frame: The output audio frame to write.
+
+        Returns:
+            True if the audio frame was written successfully, False otherwise.
         """
-        pass
+        return False
 
     async def write_dtmf(self, frame: OutputDTMFFrame | OutputDTMFUrgentFrame):
         """Write a DTMF tone using the transport's preferred method.
@@ -740,12 +746,22 @@ class BaseOutputTransport(FrameProcessor):
                 # Handle frame.
                 await self._handle_frame(frame)
 
-                # Also, push frame downstream in case anyone else needs it.
+                # If we are not able to write to the transport we shouldn't
-                await self._transport.push_frame(frame)
+                # pushb downstream.
+                push_downstream = True
 
-                # Send audio.
+                # Try to send audio to the transport.
-                if isinstance(frame, OutputAudioRawFrame):
+                try:
-                    await self._transport.write_audio_frame(frame)
+                    if isinstance(frame, OutputAudioRawFrame):
+                        push_downstream = await self._transport.write_audio_frame(frame)
+                except Exception as e:
+                    logger.error(f"{self} Error writing {frame} to transport: {e}")
+                    push_downstream = False
+
+                # If we were able to send to the transport, push the frame
+                # downstream in case anyone else needs it.
+                if push_downstream:
+                    await self._transport.push_frame(frame)
 
         #
         # Video handling

src/pipecat/transports/daily/transport.py

@@ -506,11 +506,14 @@ class DailyTransportClient(EventHandler):
         self._custom_audio_tracks[destination] = await self.add_custom_audio_track(destination)
         self._client.update_publishing({"customAudio": {destination: True}})
 
-    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the appropriate audio track.
 
         Args:
             frame: The audio frame to write.
+
+        Returns:
+            True if the audio frame was written successfully, False otherwise.
         """
         future = self._get_event_loop().create_future()
 
@@ -526,18 +529,24 @@ class DailyTransportClient(EventHandler):
             audio_source.write_frames(frame.audio, completion=completion_callback(future))
         else:
             logger.warning(f"{self} unable to write audio frames to destination [{destination}]")
-            future.set_result(None)
+            future.set_result(0)
 
-        await future
+        num_frames = await future
+        return num_frames > 0
 
-    async def write_video_frame(self, frame: OutputImageRawFrame):
+    async def write_video_frame(self, frame: OutputImageRawFrame) -> bool:
         """Write a video frame to the camera device.
 
         Args:
             frame: The image frame to write.
+
+        Returns:
+            True if the video frame was written successfully, False otherwise.
         """
         if not frame.transport_destination and self._camera:
             self._camera.write_frame(frame.image)
+            return True
+        return False
 
     async def setup(self, setup: FrameProcessorSetup):
         """Setup the client with task manager and event queues.
@@ -1835,24 +1844,33 @@ class DailyOutputTransport(BaseOutputTransport):
 
         Args:
             destination: The destination identifier to register.
+
+        Returns:
+            True if the audio frame was written successfully, False otherwise.
         """
         await self._client.register_audio_destination(destination)
 
-    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the Daily call.
 
         Args:
             frame: The audio frame to write.
-        """
-        await self._client.write_audio_frame(frame)
 
-    async def write_video_frame(self, frame: OutputImageRawFrame):
+        Returns:
+            True if the audio frame was written successfully, False otherwise.
+        """
+        return await self._client.write_audio_frame(frame)
+
+    async def write_video_frame(self, frame: OutputImageRawFrame) -> bool:
         """Write a video frame to the Daily call.
 
         Args:
             frame: The video frame to write.
+
+        Returns:
+            True if the video frame was written successfully, False otherwise.
         """
-        await self._client.write_video_frame(frame)
+        return await self._client.write_video_frame(frame)
 
     def _supports_native_dtmf(self) -> bool:
         """Daily supports native DTMF via telephone events.

src/pipecat/transports/livekit/transport.py

@@ -329,19 +329,21 @@ class LiveKitTransportClient:
         except Exception as e:
             logger.error(f"Error sending DTMF tone {digit}: {e}")
 
-    async def publish_audio(self, audio_frame: rtc.AudioFrame):
+    async def publish_audio(self, audio_frame: rtc.AudioFrame) -> bool:
         """Publish an audio frame to the room.
 
         Args:
             audio_frame: The LiveKit audio frame to publish.
         """
         if not self._connected or not self._audio_source:
-            return
+            return False
 
         try:
             await self._audio_source.capture_frame(audio_frame)
+            return True
         except Exception as e:
             logger.error(f"Error publishing audio: {e}")
+            return False
 
     def get_participants(self) -> List[str]:
         """Get list of participant IDs in the room.
@@ -849,14 +851,17 @@ class LiveKitOutputTransport(BaseOutputTransport):
         else:
             await self._client.send_data(message.encode())
 
-    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the LiveKit room.
 
         Args:
             frame: The audio frame to write.
+
+        Returns:
+            True if the audio frame was written successfully, False otherwise.
         """
         livekit_audio = self._convert_pipecat_audio_to_livekit(frame.audio)
-        await self._client.publish_audio(livekit_audio)
+        return await self._client.publish_audio(livekit_audio)
 
     def _supports_native_dtmf(self) -> bool:
         """LiveKit supports native DTMF via telephone events.

src/pipecat/transports/local/audio.py

@@ -172,16 +172,21 @@ class LocalAudioOutputTransport(BaseOutputTransport):
             self._out_stream.close()
             self._out_stream = None
 
-    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the output stream.
 
         Args:
             frame: The audio frame to write to the output device.
+
+        Returns:
+            True if the audio frame was written successfully, False otherwise.
         """
         if self._out_stream:
             await self.get_event_loop().run_in_executor(
                 self._executor, self._out_stream.write, frame.audio
             )
+            return True
+        return False
 
 
 class LocalAudioTransport(BaseTransport):

src/pipecat/transports/local/tk.py

@@ -191,24 +191,33 @@ class TkOutputTransport(BaseOutputTransport):
             self._out_stream.close()
             self._out_stream = None
 
-    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the output stream.
 
         Args:
             frame: The audio frame to write to the output device.
+
+        Returns:
+            True if the audio frame was written successfully, False otherwise.
         """
         if self._out_stream:
             await self.get_event_loop().run_in_executor(
                 self._executor, self._out_stream.write, frame.audio
             )
+            return True
+        return False
 
-    async def write_video_frame(self, frame: OutputImageRawFrame):
+    async def write_video_frame(self, frame: OutputImageRawFrame) -> bool:
         """Write a video frame to the Tkinter display.
 
         Args:
             frame: The video frame to display in the Tkinter window.
+
+        Returns:
+            True if the video frame was written successfully, False otherwise.
         """
         self.get_event_loop().call_soon(self._write_frame_to_tk, frame)
+        return True
 
     def _write_frame_to_tk(self, frame: OutputImageRawFrame):
         """Write frame data to the Tkinter image label."""

src/pipecat/transports/smallwebrtc/transport.py

@@ -399,23 +399,33 @@ class SmallWebRTCClient:
 
             del frame  # free original AudioFrame
 
-    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the WebRTC connection.
 
         Args:
             frame: The audio frame to transmit.
+
+        Returns:
+            True if the audio frame was written successfully, False otherwise.
         """
         if self._can_send() and self._audio_output_track:
             await self._audio_output_track.add_audio_bytes(frame.audio)
+            return True
+        return False
 
-    async def write_video_frame(self, frame: OutputImageRawFrame):
+    async def write_video_frame(self, frame: OutputImageRawFrame) -> bool:
         """Write a video frame to the WebRTC connection.
 
         Args:
             frame: The video frame to transmit.
+
+        Returns:
+            True if the video frame was written successfully, False otherwise.
         """
         if self._can_send() and self._video_output_track:
             self._video_output_track.add_video_frame(frame)
+            return True
+        return False
 
     async def setup(self, _params: TransportParams, frame):
         """Set up the client with transport parameters.
@@ -818,21 +828,27 @@ class SmallWebRTCOutputTransport(BaseOutputTransport):
         """
         await self._client.send_message(frame)
 
-    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the WebRTC connection.
 
         Args:
             frame: The output audio frame to transmit.
-        """
-        await self._client.write_audio_frame(frame)
 
-    async def write_video_frame(self, frame: OutputImageRawFrame):
+        Returns:
+            True if the audio frame was written successfully, False otherwise.
+        """
+        return await self._client.write_audio_frame(frame)
+
+    async def write_video_frame(self, frame: OutputImageRawFrame) -> bool:
         """Write a video frame to the WebRTC connection.
 
         Args:
             frame: The output video frame to transmit.
+
+        Returns:
+            True if the video frame was written successfully, False otherwise.
         """
-        await self._client.write_video_frame(frame)
+        return await self._client.write_video_frame(frame)
 
 
 class SmallWebRTCTransport(BaseTransport):

src/pipecat/transports/tavus/transport.py

@@ -395,15 +395,18 @@ class TavusTransportClient:
             participant_settings=participant_settings, profile_settings=profile_settings
         )
 
-    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the transport.
 
         Args:
             frame: The audio frame to write.
+
+        Returns:
+            True if the audio frame was written successfully, False otherwise.
         """
         if not self._client:
-            return
+            return False
-        await self._client.write_audio_frame(frame)
+        return await self._client.write_audio_frame(frame)
 
     async def register_audio_destination(self, destination: str):
         """Register an audio destination for output.
@@ -625,15 +628,18 @@ class TavusOutputTransport(BaseOutputTransport):
         """Handle interruption events by sending interrupt message."""
         await self._client.send_interrupt_message()
 
-    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the Tavus transport.
 
         Args:
             frame: The audio frame to write.
+
+        Returns:
+            True if the audio frame was written successfully, False otherwise.
         """
         # This is the custom track destination expected by Tavus
         frame.transport_destination = self._transport_destination
-        await self._client.write_audio_frame(frame)
+        return await self._client.write_audio_frame(frame)
 
     async def register_audio_destination(self, destination: str):
         """Register an audio destination.

src/pipecat/transports/websocket/client.py

@@ -150,17 +150,39 @@ class WebsocketClientSession:
         await self._websocket.close()
         self._websocket = None
 
-    async def send(self, message: websockets.Data):
+    async def send(self, message: websockets.Data) -> bool:
         """Send a message through the WebSocket connection.
 
         Args:
             message: The message data to send.
         """
+        result = False
         try:
             if self._websocket:
                 await self._websocket.send(message)
+                result = True
         except Exception as e:
             logger.error(f"{self} exception sending data: {e.__class__.__name__} ({e})")
+        finally:
+            return result
+
+    @property
+    def is_connected(self) -> bool:
+        """Check if the WebSocket is currently connected.
+
+        Returns:
+            True if the WebSocket is in connected state.
+        """
+        return self._websocket.state == websockets.State.OPEN if self._websocket else False
+
+    @property
+    def is_closing(self) -> bool:
+        """Check if the WebSocket is currently closing.
+
+        Returns:
+            True if the WebSocket is in the process of closing.
+        """
+        return self._websocket.state == websockets.State.CLOSING if self._websocket else False
 
     async def _client_task_handler(self):
         """Handle incoming messages from the WebSocket connection."""
@@ -371,12 +393,18 @@ class WebsocketClientOutputTransport(BaseOutputTransport):
         """
         await self._write_frame(frame)
 
-    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the WebSocket with optional WAV header.
 
         Args:
             frame: The output audio frame to write.
+
+        Returns:
+            True if the audio frame was written successfully, False otherwise.
         """
+        if self._session.is_closing or not self._session.is_connected:
+            return False
+
         frame = OutputAudioRawFrame(
             audio=frame.audio,
             sample_rate=self.sample_rate,
@@ -402,10 +430,16 @@ class WebsocketClientOutputTransport(BaseOutputTransport):
         # Simulate audio playback with a sleep.
         await self._write_audio_sleep()
 
+        return True
+
     async def _write_frame(self, frame: Frame):
         """Write a frame to the WebSocket after serialization."""
+        if self._session.is_closing or not self._session.is_connected:
+            return
+
         if not self._params.serializer:
             return
+
         payload = await self._params.serializer.serialize(frame)
         if payload:
             await self._session.send(payload)

src/pipecat/transports/websocket/fastapi.py

@@ -410,14 +410,17 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
         """
         await self._write_frame(frame)
 
-    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the WebSocket with timing simulation.
 
         Args:
             frame: The output audio frame to write.
+
+        Returns:
+            True if the audio frame was written successfully, False otherwise.
         """
         if self._client.is_closing or not self._client.is_connected:
-            return
+            return False
 
         frame = OutputAudioRawFrame(
             audio=frame.audio,
@@ -444,6 +447,8 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
         # Simulate audio playback with a sleep.
         await self._write_audio_sleep()
 
+        return True
+
     async def _write_frame(self, frame: Frame):
         """Serialize and send a frame through the WebSocket."""
         if self._client.is_closing or not self._client.is_connected:

src/pipecat/transports/websocket/server.py

@@ -346,14 +346,17 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
         """
         await self._write_frame(frame)
 
-    async def write_audio_frame(self, frame: OutputAudioRawFrame):
+    async def write_audio_frame(self, frame: OutputAudioRawFrame) -> bool:
         """Write an audio frame to the WebSocket client with timing control.
 
         Args:
             frame: The output audio frame to write.
+
+        Returns:
+            True if the audio frame was written successfully, False otherwise.
         """
         if not self._websocket:
-            return
+            return False
 
         frame = OutputAudioRawFrame(
             audio=frame.audio,
@@ -380,6 +383,8 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
         # Simulate audio playback with a sleep.
         await self._write_audio_sleep()
 
+        return True
+
     async def _write_frame(self, frame: Frame):
         """Serialize and send a frame to the WebSocket client."""
         if not self._params.serializer: