From cad271068ef5b8e0719a42077226146fd8f9f8e0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 29 May 2025 19:03:36 -0700 Subject: [PATCH 1/8] BaseInputTransport: handle StopFrame and pause pushing frames --- CHANGELOG.md | 5 ++ src/pipecat/transports/base_input.py | 47 +++++++++++++++---- .../transports/network/small_webrtc.py | 8 ++-- src/pipecat/transports/services/daily.py | 14 +++--- 4 files changed, 53 insertions(+), 21 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c971a1c23..025731588 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- `BaseInputTransport` now handles `StopFrame`. When a `StopFrame` is received + the transport will pause sending frames downstream until a new `StartFrame` is + received. This allows the transport to be reused (keeping the same connection) + in a different pipeline. + - Updated AssemblyAI STT service to support their latest streaming speech-to-text model with improved transcription latency and endpointing. diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 0dea8efdc..2a2343883 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -24,9 +24,11 @@ from pipecat.frames.frames import ( FilterUpdateSettingsFrame, Frame, InputAudioRawFrame, + InputImageRawFrame, MetricsFrame, StartFrame, StartInterruptionFrame, + StopFrame, StopInterruptionFrame, SystemFrame, UserStartedSpeakingFrame, @@ -57,6 +59,11 @@ class BaseInputTransport(FrameProcessor): # if passthrough is enabled. self._audio_task = None + # If the transport is stopped with `StopFrame` we might still be + # receiving frames from the transport but we really don't want to push + # them downstream until we get another `StartFrame`. + self._paused = False + if self._params.vad_enabled: import warnings @@ -117,6 +124,8 @@ class BaseInputTransport(FrameProcessor): return self._params.turn_analyzer async def start(self, frame: StartFrame): + self._paused = False + self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate # Configure VAD analyzer. @@ -133,28 +142,33 @@ class BaseInputTransport(FrameProcessor): async def stop(self, frame: EndFrame): # Cancel and wait for the audio input task to finish. - if self._audio_task and self._params.audio_in_enabled: - await self.cancel_task(self._audio_task) - self._audio_task = None + await self._cancel_audio_task() # Stop audio filter. if self._params.audio_in_filter: await self._params.audio_in_filter.stop() + async def pause(self, frame: StopFrame): + self._paused = True + # Cancel task so we clear the queue + await self._cancel_audio_task() + # Retart the task + self._create_audio_task() + async def cancel(self, frame: CancelFrame): # Cancel and wait for the audio input task to finish. - if self._audio_task and self._params.audio_in_enabled: - await self.cancel_task(self._audio_task) - self._audio_task = None + await self._cancel_audio_task() async def set_transport_ready(self, frame: StartFrame): """To be called when the transport is ready to stream.""" # Create audio input queue and task if needed. - if not self._audio_task and self._params.audio_in_enabled: - self._audio_in_queue = asyncio.Queue() - self._audio_task = self.create_task(self._audio_task_handler()) + self._create_audio_task() + + async def push_video_frame(self, frame: InputImageRawFrame): + if self._params.video_in_enabled and not self._paused: + await self.push_frame(frame) async def push_audio_frame(self, frame: InputAudioRawFrame): - if self._params.audio_in_enabled: + if self._params.audio_in_enabled and not self._paused: await self._audio_in_queue.put(frame) # @@ -190,6 +204,9 @@ class BaseInputTransport(FrameProcessor): # finish and the task finishes when EndFrame is processed. await self.push_frame(frame, direction) await self.stop(frame) + elif isinstance(frame, StopFrame): + await self.push_frame(frame, direction) + await self.pause(frame) elif isinstance(frame, VADParamsUpdateFrame): if self.vad_analyzer: self.vad_analyzer.set_params(frame.params) @@ -231,6 +248,16 @@ class BaseInputTransport(FrameProcessor): # Audio input # + def _create_audio_task(self): + if not self._audio_task and self._params.audio_in_enabled: + self._audio_in_queue = asyncio.Queue() + self._audio_task = self.create_task(self._audio_task_handler()) + + async def _cancel_audio_task(self): + if self._audio_task: + await self.cancel_task(self._audio_task) + self._audio_task = None + async def _vad_analyze(self, audio_frame: InputAudioRawFrame) -> VADState: state = VADState.QUIET if self.vad_analyzer: diff --git a/src/pipecat/transports/network/small_webrtc.py b/src/pipecat/transports/network/small_webrtc.py index c773860db..f9dd38791 100644 --- a/src/pipecat/transports/network/small_webrtc.py +++ b/src/pipecat/transports/network/small_webrtc.py @@ -19,7 +19,6 @@ from pipecat.frames.frames import ( EndFrame, Frame, InputAudioRawFrame, - InputImageRawFrame, OutputAudioRawFrame, OutputImageRawFrame, SpriteFrame, @@ -232,7 +231,8 @@ class SmallWebRTCClient: frame_array = frame.to_ndarray(format=format_name) frame_rgb = self._convert_frame(frame_array, format_name) - image_frame = InputImageRawFrame( + image_frame = UserImageRawFrame( + user_id=self._webrtc_connection.pc_id, image=frame_rgb.tobytes(), size=(frame.width, frame.height), format="RGB", @@ -424,7 +424,7 @@ class SmallWebRTCInputTransport(BaseInputTransport): try: async for video_frame in self._client.read_video_frame(): if video_frame: - await self.push_frame(video_frame) + await self.push_video_frame(video_frame) # Check if there are any pending image requests and create UserImageRawFrame if self._image_requests: @@ -438,7 +438,7 @@ class SmallWebRTCInputTransport(BaseInputTransport): format=video_frame.format, ) # Push the frame to the pipeline - await self.push_frame(image_frame) + await self.push_video_frame(image_frame) # Remove from pending requests del self._image_requests[req_id] diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 13ea557ee..78bf62062 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -1006,14 +1006,14 @@ class DailyInputTransport(BaseInputTransport): await self._transport.cleanup() async def start(self, frame: StartFrame): + # Parent start. + await super().start(frame) + if self._initialized: return self._initialized = True - # Parent start. - await super().start(frame) - # Setup client. await self._client.start(frame) @@ -1148,7 +1148,7 @@ class DailyInputTransport(BaseInputTransport): format=video_frame.color_format, ) frame.transport_source = video_source - await self.push_frame(frame) + await self.push_video_frame(frame) self._video_renderers[participant_id][video_source]["timestamp"] = curr_time @@ -1183,14 +1183,14 @@ class DailyOutputTransport(BaseOutputTransport): await self._transport.cleanup() async def start(self, frame: StartFrame): + # Parent start. + await super().start(frame) + if self._initialized: return self._initialized = True - # Parent start. - await super().start(frame) - # Setup client. await self._client.start(frame) From bc6d7b7bbd67fa5601002068e4bf019007e6c6d4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Thu, 29 May 2025 19:04:11 -0700 Subject: [PATCH 2/8] examples(phone-chatbot): don't show dangling tasks in first pipeline --- examples/phone-chatbot/voicemail_detection.py | 1 + 1 file changed, 1 insertion(+) diff --git a/examples/phone-chatbot/voicemail_detection.py b/examples/phone-chatbot/voicemail_detection.py index 8603f19de..e5b51eb55 100644 --- a/examples/phone-chatbot/voicemail_detection.py +++ b/examples/phone-chatbot/voicemail_detection.py @@ -279,6 +279,7 @@ async def main( voicemail_detection_pipeline_task = PipelineTask( voicemail_detection_pipeline, params=PipelineParams(allow_interruptions=True), + check_dangling_tasks=False, ) # ------------ EVENT HANDLERS ------------ From 24e19db29eebba23b5e131fc083bac813e76e44a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 30 May 2025 07:42:15 -0700 Subject: [PATCH 3/8] TavusTransport: don't initialize if a second StartFrame is received --- src/pipecat/transports/services/tavus.py | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/src/pipecat/transports/services/tavus.py b/src/pipecat/transports/services/tavus.py index 301a0b7df..2c76b9b33 100644 --- a/src/pipecat/transports/services/tavus.py +++ b/src/pipecat/transports/services/tavus.py @@ -18,7 +18,6 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, OutputAudioRawFrame, - OutputImageRawFrame, StartFrame, StartInterruptionFrame, TransportMessageFrame, @@ -317,6 +316,9 @@ class TavusInputTransport(BaseInputTransport): self._params = params self._resampler = create_default_resampler() + # Whether we have seen a StartFrame already. + self._initialized = False + async def setup(self, setup: FrameProcessorSetup): await super().setup(setup) await self._client.setup(setup) @@ -327,6 +329,12 @@ class TavusInputTransport(BaseInputTransport): async def start(self, frame: StartFrame): await super().start(frame) + + if self._initialized: + return + + self._initialized = True + await self._client.start(frame) await self.set_transport_ready(frame) @@ -375,6 +383,9 @@ class TavusOutputTransport(BaseOutputTransport): self._start_time = None self._current_idx_str: Optional[str] = None + # Whether we have seen a StartFrame already. + self._initialized = False + async def setup(self, setup: FrameProcessorSetup): await super().setup(setup) await self._client.setup(setup) @@ -385,6 +396,12 @@ class TavusOutputTransport(BaseOutputTransport): async def start(self, frame: StartFrame): await super().start(frame) + + if self._initialized: + return + + self._initialized = True + await self._client.start(frame) await self.set_transport_ready(frame) From 0c7e3cfcb2c0f0ac324b6d378c6c63991be0bc8e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 30 May 2025 07:42:58 -0700 Subject: [PATCH 4/8] LiveKitTransport: don't initialize if a second StartFrame is received --- src/pipecat/transports/services/livekit.py | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index c998cfc06..736054ab9 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -363,6 +363,11 @@ class LiveKitInputTransport(BaseInputTransport): self._audio_in_task = None self._vad_analyzer: Optional[VADAnalyzer] = params.vad_analyzer self._resampler = create_default_resampler() + if self._initialized: + return + + # Whether we have seen a StartFrame already. + self._initialized = False @property def vad_analyzer(self) -> Optional[VADAnalyzer]: @@ -370,6 +375,12 @@ class LiveKitInputTransport(BaseInputTransport): async def start(self, frame: StartFrame): await super().start(frame) + + if self._initialized: + return + + self._initialized = True + await self._client.start(frame) await self._client.connect() if not self._audio_in_task and self._params.audio_in_enabled: @@ -447,8 +458,17 @@ class LiveKitOutputTransport(BaseOutputTransport): self._transport = transport self._client = client + # Whether we have seen a StartFrame already. + self._initialized = False + async def start(self, frame: StartFrame): await super().start(frame) + + if self._initialized: + return + + self._initialized = True + await self._client.start(frame) await self._client.connect() await self.set_transport_ready(frame) From f2eb869b028bbf8a54ce01f210aa035077e0c6fe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 30 May 2025 07:49:58 -0700 Subject: [PATCH 5/8] FastAPIWebsocketTransport: don't initialize if a second StartFrame is received --- .../transports/network/fastapi_websocket.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index f19bc0a9c..205fb8c9e 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -122,8 +122,17 @@ class FastAPIWebsocketInputTransport(BaseInputTransport): self._receive_task = None self._monitor_websocket_task = None + # Whether we have seen a StartFrame already. + self._initialized = False + async def start(self, frame: StartFrame): await super().start(frame) + + if self._initialized: + return + + self._initialized = True + await self._client.setup(frame) if self._params.serializer: await self._params.serializer.setup(frame) @@ -204,8 +213,17 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport): self._send_interval = 0 self._next_send_time = 0 + # Whether we have seen a StartFrame already. + self._initialized = False + async def start(self, frame: StartFrame): await super().start(frame) + + if self._initialized: + return + + self._initialized = True + await self._client.setup(frame) if self._params.serializer: await self._params.serializer.setup(frame) From 9a5a1edb6ba652fa1af86a7e28680f52b2e62c77 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 30 May 2025 07:50:26 -0700 Subject: [PATCH 6/8] WebsocketServerTransport: don't initialize if a second StartFrame is received --- .../transports/network/websocket_server.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index 15db99f6a..1fe33f4a6 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -78,8 +78,17 @@ class WebsocketServerInputTransport(BaseInputTransport): self._stop_server_event = asyncio.Event() + # Whether we have seen a StartFrame already. + self._initialized = False + async def start(self, frame: StartFrame): await super().start(frame) + + if self._initialized: + return + + self._initialized = True + if self._params.serializer: await self._params.serializer.setup(frame) if not self._server_task: @@ -190,6 +199,9 @@ class WebsocketServerOutputTransport(BaseOutputTransport): self._send_interval = 0 self._next_send_time = 0 + # Whether we have seen a StartFrame already. + self._initialized = False + async def set_client_connection(self, websocket: Optional[websockets.WebSocketServerProtocol]): if self._websocket: await self._websocket.close() @@ -198,6 +210,12 @@ class WebsocketServerOutputTransport(BaseOutputTransport): async def start(self, frame: StartFrame): await super().start(frame) + + if self._initialized: + return + + self._initialized = True + if self._params.serializer: await self._params.serializer.setup(frame) self._send_interval = (self.audio_chunk_size / self.sample_rate) / 2 From 1c3d3f2f4b016fb7d802f941394e018d4ef6fd01 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 30 May 2025 07:50:44 -0700 Subject: [PATCH 7/8] WebsocketClientTransport: don't initialize if a second StartFrame is received --- .../transports/network/websocket_client.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/pipecat/transports/network/websocket_client.py b/src/pipecat/transports/network/websocket_client.py index d5a276249..738904546 100644 --- a/src/pipecat/transports/network/websocket_client.py +++ b/src/pipecat/transports/network/websocket_client.py @@ -132,12 +132,21 @@ class WebsocketClientInputTransport(BaseInputTransport): self._session = session self._params = params + # Whether we have seen a StartFrame already. + self._initialized = False + async def setup(self, setup: FrameProcessorSetup): await super().setup(setup) await self._session.setup(setup.task_manager) async def start(self, frame: StartFrame): await super().start(frame) + + if self._initialized: + return + + self._initialized = True + if self._params.serializer: await self._params.serializer.setup(frame) await self._session.connect() @@ -188,12 +197,21 @@ class WebsocketClientOutputTransport(BaseOutputTransport): self._send_interval = 0 self._next_send_time = 0 + # Whether we have seen a StartFrame already. + self._initialized = False + async def setup(self, setup: FrameProcessorSetup): await super().setup(setup) await self._session.setup(setup.task_manager) async def start(self, frame: StartFrame): await super().start(frame) + + if self._initialized: + return + + self._initialized = True + self._send_interval = (self.audio_chunk_size / self.sample_rate) / 2 if self._params.serializer: await self._params.serializer.setup(frame) From b33897ffb969abc986d1f1aa0f47c50dc129c532 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 30 May 2025 07:50:54 -0700 Subject: [PATCH 8/8] SmallWebRTCTransport: don't initialize if a second StartFrame is received --- src/pipecat/transports/network/small_webrtc.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/src/pipecat/transports/network/small_webrtc.py b/src/pipecat/transports/network/small_webrtc.py index f9dd38791..ddc32d06a 100644 --- a/src/pipecat/transports/network/small_webrtc.py +++ b/src/pipecat/transports/network/small_webrtc.py @@ -377,6 +377,9 @@ class SmallWebRTCInputTransport(BaseInputTransport): self._receive_video_task = None self._image_requests = {} + # Whether we have seen a StartFrame already. + self._initialized = False + async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -385,6 +388,12 @@ class SmallWebRTCInputTransport(BaseInputTransport): async def start(self, frame: StartFrame): await super().start(frame) + + if self._initialized: + return + + self._initialized = True + await self._client.setup(self._params, frame) await self._client.connect() if not self._receive_audio_task and self._params.audio_in_enabled: @@ -480,8 +489,17 @@ class SmallWebRTCOutputTransport(BaseOutputTransport): self._client = client self._params = params + # Whether we have seen a StartFrame already. + self._initialized = False + async def start(self, frame: StartFrame): await super().start(frame) + + if self._initialized: + return + + self._initialized = True + await self._client.setup(self._params, frame) await self._client.connect() await self.set_transport_ready(frame)