diff --git a/CHANGELOG.md b/CHANGELOG.md index bdbb634f1..ad49ae5d7 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/examples/foundational/04b-transports-livekit.py b/examples/foundational/04b-transports-livekit.py index 4e5a7e820..402fd79ca 100644 --- a/examples/foundational/04b-transports-livekit.py +++ b/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?" ) ) diff --git a/src/pipecat/processors/frameworks/strands_agents.py b/src/pipecat/processors/frameworks/strands_agents.py index 67005d74b..d129c2063 100644 --- a/src/pipecat/processors/frameworks/strands_agents.py +++ b/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('outputTokens', 0), - agent_result.metrics.accumulated_usage.get('totalTokens', 0) + agent_result.metrics.accumulated_usage.get("inputTokens", 0), + agent_result.metrics.accumulated_usage.get("outputTokens", 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}") @@ -123,12 +123,20 @@ class StrandsAgentsProcessor(FrameProcessor): if ttfb_tracking: await self.stop_ttfb_metrics() ttfb_tracking = False - + # Update usage metrics - if isinstance(event, dict) and "event" in event 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)) + if ( + isinstance(event, dict) + and "event" in event + 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: @@ -139,7 +147,7 @@ class StrandsAgentsProcessor(FrameProcessor): ttfb_tracking = False await self.stop_processing_metrics() await self.push_frame(LLMFullResponseEndFrame()) - + def can_generate_metrics(self) -> bool: """Check if this service can generate performance metrics. @@ -149,14 +157,11 @@ class StrandsAgentsProcessor(FrameProcessor): return True async def _report_usage_metrics( - self, - prompt_tokens: int, - completion_tokens: int, - total_tokens: int + self, 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) \ No newline at end of file + await self.start_llm_usage_metrics(tokens) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index ed60a4584..9d1b2b3bb 100644 --- a/src/pipecat/transports/base_output.py +++ b/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. - await self._transport.push_frame(frame) + # If we are not able to write to the transport we shouldn't + # pushb downstream. + push_downstream = True - # Send audio. - if isinstance(frame, OutputAudioRawFrame): - await self._transport.write_audio_frame(frame) + # Try to send audio to the transport. + try: + 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 diff --git a/src/pipecat/transports/daily/transport.py b/src/pipecat/transports/daily/transport.py index 030f1ad6a..edb25531f 100644 --- a/src/pipecat/transports/daily/transport.py +++ b/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. diff --git a/src/pipecat/transports/livekit/transport.py b/src/pipecat/transports/livekit/transport.py index 227a9f4f9..334d48ecb 100644 --- a/src/pipecat/transports/livekit/transport.py +++ b/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. diff --git a/src/pipecat/transports/local/audio.py b/src/pipecat/transports/local/audio.py index b38d1dd8e..7f809b5d4 100644 --- a/src/pipecat/transports/local/audio.py +++ b/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): diff --git a/src/pipecat/transports/local/tk.py b/src/pipecat/transports/local/tk.py index c687370ce..6cceb2cf1 100644 --- a/src/pipecat/transports/local/tk.py +++ b/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.""" diff --git a/src/pipecat/transports/smallwebrtc/transport.py b/src/pipecat/transports/smallwebrtc/transport.py index 0ec8bf998..4a9ba9341 100644 --- a/src/pipecat/transports/smallwebrtc/transport.py +++ b/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): diff --git a/src/pipecat/transports/tavus/transport.py b/src/pipecat/transports/tavus/transport.py index 23e14f97b..b199de09c 100644 --- a/src/pipecat/transports/tavus/transport.py +++ b/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 - await self._client.write_audio_frame(frame) + return False + 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. diff --git a/src/pipecat/transports/websocket/client.py b/src/pipecat/transports/websocket/client.py index d141b52f3..c18c0d22a 100644 --- a/src/pipecat/transports/websocket/client.py +++ b/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) diff --git a/src/pipecat/transports/websocket/fastapi.py b/src/pipecat/transports/websocket/fastapi.py index cfa68f5cb..72654204a 100644 --- a/src/pipecat/transports/websocket/fastapi.py +++ b/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: diff --git a/src/pipecat/transports/websocket/server.py b/src/pipecat/transports/websocket/server.py index 67631ab04..095407422 100644 --- a/src/pipecat/transports/websocket/server.py +++ b/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: