diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index a3c373eac..f828c2878 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/examples/foundational/22b-natural-conversation-proposal.py @@ -138,6 +138,7 @@ class OutputGate(FrameProcessor): self._gate_open = start_open self._frames_buffer = [] self._notifier = notifier + self._gate_task = None def close_gate(self): self._gate_open = False @@ -178,10 +179,13 @@ class OutputGate(FrameProcessor): async def _start(self): self._frames_buffer = [] - self._gate_task = self.create_task(self._gate_task_handler()) + if not self._gate_task: + self._gate_task = self.create_task(self._gate_task_handler()) async def _stop(self): - await self.cancel_task(self._gate_task) + if self._gate_task: + await self.cancel_task(self._gate_task) + self._gate_task = None async def _gate_task_handler(self): while True: diff --git a/examples/foundational/22c-natural-conversation-mixed-llms.py b/examples/foundational/22c-natural-conversation-mixed-llms.py index 92f47ae92..283a7e518 100644 --- a/examples/foundational/22c-natural-conversation-mixed-llms.py +++ b/examples/foundational/22c-natural-conversation-mixed-llms.py @@ -342,6 +342,7 @@ class OutputGate(FrameProcessor): self._gate_open = start_open self._frames_buffer = [] self._notifier = notifier + self._gate_task = None def close_gate(self): self._gate_open = False @@ -382,10 +383,13 @@ class OutputGate(FrameProcessor): async def _start(self): self._frames_buffer = [] - self._gate_task = self.create_task(self._gate_task_handler()) + if not self._gate_task: + self._gate_task = self.create_task(self._gate_task_handler()) async def _stop(self): - await self.cancel_task(self._gate_task) + if self._gate_task: + await self.cancel_task(self._gate_task) + self._gate_task = None async def _gate_task_handler(self): while True: diff --git a/examples/foundational/22d-natural-conversation-gemini-audio.py b/examples/foundational/22d-natural-conversation-gemini-audio.py index 15ee835e2..0e1feee57 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -555,6 +555,7 @@ class OutputGate(FrameProcessor): self._notifier = notifier self._context = context self._transcription_buffer = user_transcription_buffer + self._gate_task = None def close_gate(self): self._gate_open = False @@ -602,10 +603,13 @@ class OutputGate(FrameProcessor): async def _start(self): self._frames_buffer = [] - self._gate_task = self.create_task(self._gate_task_handler()) + if not self._gate_task: + self._gate_task = self.create_task(self._gate_task_handler()) async def _stop(self): - await self.cancel_task(self._gate_task) + if self._gate_task: + await self.cancel_task(self._gate_task) + self._gate_task = None async def _gate_task_handler(self): while True: diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py index f77d09cc6..c6f4de2df 100644 --- a/src/pipecat/pipeline/parallel_pipeline.py +++ b/src/pipecat/pipeline/parallel_pipeline.py @@ -81,6 +81,8 @@ class ParallelPipeline(BasePipeline): self._seen_ids = set() self._endframe_counter: Dict[int, int] = {} + self._up_task = None + self._down_task = None self._up_queue = asyncio.Queue() self._down_queue = asyncio.Queue() @@ -150,19 +152,30 @@ class ParallelPipeline(BasePipeline): await self._create_tasks() async def _stop(self): - # The up task doesn't receive an EndFrame, so we just cancel it. - await self.cancel_task(self._up_task) - # The down tasks waits for the last EndFrame sent by the internal - # pipelines. - await self._down_task + if self._up_task: + # The up task doesn't receive an EndFrame, so we just cancel it. + await self.cancel_task(self._up_task) + self._up_task = None + + if self._down_task: + # The down tasks waits for the last EndFrame sent by the internal + # pipelines. + await self._down_task + self._down_task = None async def _cancel(self): - await self.cancel_task(self._up_task) - await self.cancel_task(self._down_task) + if self._up_task: + await self.cancel_task(self._up_task) + self._up_task = None + if self._down_task: + await self.cancel_task(self._down_task) + self._down_task = None async def _create_tasks(self): - self._up_task = self.create_task(self._process_up_queue()) - self._down_task = self.create_task(self._process_down_queue()) + if not self._up_task: + self._up_task = self.create_task(self._process_up_queue()) + if not self._down_task: + self._down_task = self.create_task(self._process_down_queue()) async def _drain_queues(self): while not self._up_queue.empty: diff --git a/src/pipecat/processors/aggregators/gated_openai_llm_context.py b/src/pipecat/processors/aggregators/gated_openai_llm_context.py index f3ae78121..9973e3d02 100644 --- a/src/pipecat/processors/aggregators/gated_openai_llm_context.py +++ b/src/pipecat/processors/aggregators/gated_openai_llm_context.py @@ -21,6 +21,7 @@ class GatedOpenAILLMContextAggregator(FrameProcessor): self._notifier = notifier self._start_open = start_open self._last_context_frame = None + self._gate_task = None async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -41,10 +42,13 @@ class GatedOpenAILLMContextAggregator(FrameProcessor): await self.push_frame(frame, direction) async def _start(self): - self._gate_task = self.create_task(self._gate_task_handler()) + if not self._gate_task: + self._gate_task = self.create_task(self._gate_task_handler()) async def _stop(self): - await self.cancel_task(self._gate_task) + if self._gate_task: + await self.cancel_task(self._gate_task) + self._gate_task = None async def _gate_task_handler(self): while True: diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index ae0b23507..f25bf7644 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -309,7 +309,8 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): self._seen_interim_results = True def _create_aggregation_task(self): - self._aggregation_task = self.create_task(self._aggregation_task_handler()) + if not self._aggregation_task: + self._aggregation_task = self.create_task(self._aggregation_task_handler()) async def _cancel_aggregation_task(self): if self._aggregation_task: diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 5b28846a2..af55fbd42 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -956,8 +956,10 @@ class RTVIProcessor(FrameProcessor): await self._pipeline.cleanup() async def _start(self, frame: StartFrame): - self._action_task = self.create_task(self._action_task_handler()) - self._message_task = self.create_task(self._message_task_handler()) + if not self._action_task: + self._action_task = self.create_task(self._action_task_handler()) + if not self._message_task: + self._message_task = self.create_task(self._message_task_handler()) await self._call_event_handler("on_bot_started") async def _stop(self, frame: EndFrame): diff --git a/src/pipecat/processors/idle_frame_processor.py b/src/pipecat/processors/idle_frame_processor.py index 85ea215cf..ab98df771 100644 --- a/src/pipecat/processors/idle_frame_processor.py +++ b/src/pipecat/processors/idle_frame_processor.py @@ -30,6 +30,7 @@ class IdleFrameProcessor(FrameProcessor): self._callback = callback self._timeout = timeout self._types = types + self._idle_task = None async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -49,11 +50,13 @@ class IdleFrameProcessor(FrameProcessor): self._idle_event.set() async def cleanup(self): - await self.cancel_task(self._idle_task) + if self._idle_task: + await self.cancel_task(self._idle_task) def _create_idle_task(self): - self._idle_event = asyncio.Event() - self._idle_task = self.create_task(self._idle_task_handler()) + if not self._idle_task: + self._idle_event = asyncio.Event() + self._idle_task = self.create_task(self._idle_task_handler()) async def _idle_task_handler(self): while True: diff --git a/src/pipecat/processors/user_idle_processor.py b/src/pipecat/processors/user_idle_processor.py index e65bd437f..e88df3540 100644 --- a/src/pipecat/processors/user_idle_processor.py +++ b/src/pipecat/processors/user_idle_processor.py @@ -102,7 +102,7 @@ class UserIdleProcessor(FrameProcessor): def _create_idle_task(self) -> None: """Creates the idle task if it hasn't been created yet.""" - if self._idle_task is None: + if not self._idle_task: self._idle_task = self.create_task(self._idle_task_handler()) @property @@ -112,7 +112,7 @@ class UserIdleProcessor(FrameProcessor): async def _stop(self) -> None: """Stops and cleans up the idle monitoring task.""" - if self._idle_task is not None: + if self._idle_task: await self.cancel_task(self._idle_task) self._idle_task = None diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 8f0df1b37..594f7704d 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -273,7 +273,7 @@ class TTSService(AIService): async def start(self, frame: StartFrame): await super().start(frame) self._sample_rate = self._init_sample_rate or frame.audio_out_sample_rate - if self._push_stop_frames: + if self._push_stop_frames and not self._stop_frame_task: self._stop_frame_task = self.create_task(self._stop_frame_handler()) async def stop(self, frame: EndFrame): @@ -484,7 +484,8 @@ class WordTTSService(TTSService): self.reset_word_timestamps() def _create_words_task(self): - self._words_task = self.create_task(self._words_task_handler()) + if not self._words_task: + self._words_task = self.create_task(self._words_task_handler()) async def _stop_words_task(self): if self._words_task: @@ -660,9 +661,10 @@ class AudioContextWordTTSService(WebsocketWordTTSService): self._create_audio_context_task() def _create_audio_context_task(self): - self._contexts_queue = asyncio.Queue() - self._contexts: Dict[str, asyncio.Queue] = {} - self._audio_context_task = self.create_task(self._audio_context_task_handler()) + if not self._audio_context_task: + self._contexts_queue = asyncio.Queue() + self._contexts: Dict[str, asyncio.Queue] = {} + self._audio_context_task = self.create_task(self._audio_context_task_handler()) async def _stop_audio_context_task(self): if self._audio_context_task: diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 0ec55aa0c..3aafb4023 100644 --- a/src/pipecat/services/cartesia.py +++ b/src/pipecat/services/cartesia.py @@ -182,8 +182,8 @@ class CartesiaTTSService(AudioContextWordTTSService): async def _connect(self): await self._connect_websocket() - - self._receive_task = self.create_task(self._receive_task_handler(self.push_error)) + if not self._receive_task: + self._receive_task = self.create_task(self._receive_task_handler(self.push_error)) async def _disconnect(self): if self._receive_task: diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 47fa87b5a..6c808f66b 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/src/pipecat/services/elevenlabs.py @@ -214,6 +214,9 @@ class ElevenLabsTTSService(InterruptibleWordTTSService): self._started = False self._cumulative_time = 0 + self._receive_task = None + self._keepalive_task = None + def can_generate_metrics(self) -> bool: return True @@ -286,8 +289,11 @@ class ElevenLabsTTSService(InterruptibleWordTTSService): async def _connect(self): await self._connect_websocket() - self._receive_task = self.create_task(self._receive_task_handler(self.push_error)) - self._keepalive_task = self.create_task(self._keepalive_task_handler()) + if not self._receive_task: + self._receive_task = self.create_task(self._receive_task_handler(self.push_error)) + + if not self._keepalive_task: + self._keepalive_task = self.create_task(self._keepalive_task_handler()) async def _disconnect(self): if self._receive_task: diff --git a/src/pipecat/services/fish.py b/src/pipecat/services/fish.py index 1c5e6a08a..645ee1496 100644 --- a/src/pipecat/services/fish.py +++ b/src/pipecat/services/fish.py @@ -106,7 +106,8 @@ class FishAudioTTSService(InterruptibleTTSService): async def _connect(self): await self._connect_websocket() - self._receive_task = self.create_task(self._receive_task_handler(self.push_error)) + if not self._receive_task: + self._receive_task = self.create_task(self._receive_task_handler(self.push_error)) async def _disconnect(self): if self._receive_task: diff --git a/src/pipecat/services/gladia.py b/src/pipecat/services/gladia.py index c93dfef3b..2fbf31882 100644 --- a/src/pipecat/services/gladia.py +++ b/src/pipecat/services/gladia.py @@ -172,6 +172,7 @@ class GladiaSTTService(STTService): }, } self._confidence = confidence + self._receive_task = None def language_to_service_language(self, language: Language) -> Optional[str]: return language_to_gladia_language(language) @@ -181,18 +182,23 @@ class GladiaSTTService(STTService): self._settings["sample_rate"] = self.sample_rate response = await self._setup_gladia() self._websocket = await websockets.connect(response["url"]) - self._receive_task = self.create_task(self._receive_task_handler()) + if not self._receive_task: + self._receive_task = self.create_task(self._receive_task_handler()) async def stop(self, frame: EndFrame): await super().stop(frame) await self._send_stop_recording() await self._websocket.close() - await self.wait_for_task(self._receive_task) + if self._receive_task: + await self.wait_for_task(self._receive_task) + self._receive_task = None async def cancel(self, frame: CancelFrame): await super().cancel(frame) await self._websocket.close() - await self.cancel_task(self._receive_task) + if self._receive_task: + await self.cancel_task(self._receive_task) + self._receive_task = None async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: await self.start_processing_metrics() diff --git a/src/pipecat/services/lmnt.py b/src/pipecat/services/lmnt.py index dd6b9e3b8..4b3a3ce34 100644 --- a/src/pipecat/services/lmnt.py +++ b/src/pipecat/services/lmnt.py @@ -83,6 +83,7 @@ class LmntTTSService(InterruptibleTTSService): "format": "raw", # Use raw format for direct PCM data } self._started = False + self._receive_task = None def can_generate_metrics(self) -> bool: return True @@ -110,7 +111,8 @@ class LmntTTSService(InterruptibleTTSService): async def _connect(self): await self._connect_websocket() - self._receive_task = self.create_task(self._receive_task_handler(self.push_error)) + if not self._receive_task: + self._receive_task = self.create_task(self._receive_task_handler(self.push_error)) async def _disconnect(self): if self._receive_task: diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index 828acbdb7..4b4bd6b10 100644 --- a/src/pipecat/services/playht.py +++ b/src/pipecat/services/playht.py @@ -159,7 +159,8 @@ class PlayHTTTSService(InterruptibleTTSService): async def _connect(self): await self._connect_websocket() - self._receive_task = self.create_task(self._receive_task_handler(self.push_error)) + if not self._receive_task: + self._receive_task = self.create_task(self._receive_task_handler(self.push_error)) async def _disconnect(self): if self._receive_task: diff --git a/src/pipecat/services/rime.py b/src/pipecat/services/rime.py index 6e2daccd6..b9a2d20da 100644 --- a/src/pipecat/services/rime.py +++ b/src/pipecat/services/rime.py @@ -165,7 +165,9 @@ class RimeTTSService(AudioContextWordTTSService): async def _connect(self): """Establish websocket connection and start receive task.""" await self._connect_websocket() - self._receive_task = self.create_task(self._receive_task_handler(self.push_error)) + + if not self._receive_task: + self._receive_task = self.create_task(self._receive_task_handler(self.push_error)) async def _disconnect(self): """Close websocket connection and clean up tasks.""" diff --git a/src/pipecat/services/riva.py b/src/pipecat/services/riva.py index 458cb4919..89c965c12 100644 --- a/src/pipecat/services/riva.py +++ b/src/pipecat/services/riva.py @@ -166,6 +166,8 @@ class ParakeetSTTService(STTService): self._asr_service = riva.client.ASRService(auth) self._queue = asyncio.Queue() + self._thread_task = None + self._response_task = None def can_generate_metrics(self) -> bool: return False @@ -205,9 +207,12 @@ class ParakeetSTTService(STTService): self._config = config - self._thread_task = self.create_task(self._thread_task_handler()) - self._response_task = self.create_task(self._response_task_handler()) - self._response_queue = asyncio.Queue() + if not self._thread_task: + self._thread_task = self.create_task(self._thread_task_handler()) + + if not self._response_task: + self._response_queue = asyncio.Queue() + self._response_task = self.create_task(self._response_task_handler()) async def stop(self, frame: EndFrame): await super().stop(frame) @@ -218,8 +223,13 @@ class ParakeetSTTService(STTService): await self._stop_tasks() async def _stop_tasks(self): - await self.cancel_task(self._thread_task) - await self.cancel_task(self._response_task) + if self._thread_task: + await self.cancel_task(self._thread_task) + self._thread_task = None + + if self._response_task: + await self.cancel_task(self._response_task) + self._response_task = None def _response_handler(self): responses = self._asr_service.streaming_response_generator( diff --git a/src/pipecat/services/simli.py b/src/pipecat/services/simli.py index 01aaeedfb..10f7ae68c 100644 --- a/src/pipecat/services/simli.py +++ b/src/pipecat/services/simli.py @@ -49,8 +49,11 @@ class SimliVideoService(FrameProcessor): async def _start_connection(self): await self._simli_client.Initialize() # Create task to consume and process audio and video - self._audio_task = self.create_task(self._consume_and_process_audio()) - self._video_task = self.create_task(self._consume_and_process_video()) + if not self._audio_task: + self._audio_task = self.create_task(self._consume_and_process_audio()) + + if not self._video_task: + self._video_task = self.create_task(self._consume_and_process_video()) async def _consume_and_process_audio(self): await self._pipecat_resampler_event.wait() @@ -117,5 +120,7 @@ class SimliVideoService(FrameProcessor): await self._simli_client.stop() if self._audio_task: await self.cancel_task(self._audio_task) + self._audio_task = None if self._video_task: await self.cancel_task(self._video_task) + self._video_task = None diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index ee7ae7f7c..782ad1333 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -74,7 +74,7 @@ class BaseInputTransport(FrameProcessor): if self._params.audio_in_filter: await self._params.audio_in_filter.start(self._sample_rate) # Create audio input queue and task if needed. - if self._params.audio_in_enabled or self._params.vad_enabled: + if not self._audio_task and (self._params.audio_in_enabled or self._params.vad_enabled): self._audio_in_queue = asyncio.Queue() self._audio_task = self.create_task(self._audio_task_handler()) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 1b6d1e833..2322ade9d 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -238,10 +238,12 @@ class BaseOutputTransport(FrameProcessor): # def _create_sink_tasks(self): - self._sink_queue = asyncio.Queue() - self._sink_clock_queue = asyncio.PriorityQueue() - self._sink_task = self.create_task(self._sink_task_handler()) - self._sink_clock_task = self.create_task(self._sink_clock_task_handler()) + if not self._sink_task: + self._sink_queue = asyncio.Queue() + self._sink_task = self.create_task(self._sink_task_handler()) + if not self._sink_clock_task: + self._sink_clock_queue = asyncio.PriorityQueue() + self._sink_clock_task = self.create_task(self._sink_clock_task_handler()) async def _cancel_sink_tasks(self): # Stop sink tasks. @@ -358,7 +360,7 @@ class BaseOutputTransport(FrameProcessor): def _create_camera_task(self): # Create camera output queue and task if needed. - if self._params.camera_out_enabled: + if not self._camera_out_task and self._params.camera_out_enabled: self._camera_out_queue = asyncio.Queue() self._camera_out_task = self.create_task(self._camera_out_task_handler()) diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index 565d1906c..bfb1e8146 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -115,15 +115,19 @@ class FastAPIWebsocketInputTransport(BaseInputTransport): async def start(self, frame: StartFrame): await super().start(frame) await self._params.serializer.setup(frame) - if self._params.session_timeout: + if not self._monitor_websocket_task and self._params.session_timeout: self._monitor_websocket_task = self.create_task(self._monitor_websocket()) await self._client.trigger_client_connected() - self._receive_task = self.create_task(self._receive_messages()) + if not self._receive_task: + self._receive_task = self.create_task(self._receive_messages()) async def _stop_tasks(self): if self._monitor_websocket_task: await self.cancel_task(self._monitor_websocket_task) - await self.cancel_task(self._receive_task) + self._monitor_websocket_task = None + if self._receive_task: + await self.cancel_task(self._receive_task) + self._receive_task = None async def stop(self, frame: EndFrame): await super().stop(frame) diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index ca5d07d2e..4d1e07230 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -81,22 +81,27 @@ class WebsocketServerInputTransport(BaseInputTransport): async def start(self, frame: StartFrame): await super().start(frame) await self._params.serializer.setup(frame) - self._server_task = self.create_task(self._server_task_handler()) + if not self._server_task: + self._server_task = self.create_task(self._server_task_handler()) async def stop(self, frame: EndFrame): await super().stop(frame) self._stop_server_event.set() if self._monitor_task: await self.cancel_task(self._monitor_task) + self._monitor_task = None if self._server_task: await self.wait_for_task(self._server_task) + self._server_task = None async def cancel(self, frame: CancelFrame): await super().cancel(frame) if self._monitor_task: await self.cancel_task(self._monitor_task) + self._monitor_task = None if self._server_task: await self.cancel_task(self._server_task) + self._server_task = None async def _server_task_handler(self): logger.info(f"Starting websocket server on {self._host}:{self._port}") @@ -116,7 +121,7 @@ class WebsocketServerInputTransport(BaseInputTransport): await self._callbacks.on_client_connected(websocket) # Create a task to monitor the websocket connection - if self._params.session_timeout: + if not self._monitor_task and self._params.session_timeout: self._monitor_task = self.create_task( self._monitor_websocket(websocket, self._params.session_timeout) ) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 7ce329659..0bb0dfd6d 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -845,7 +845,7 @@ class DailyInputTransport(BaseInputTransport): def start_audio_in_streaming(self): # Create audio task. It reads audio frames from Daily and push them # internally for VAD processing. - if self._params.audio_in_enabled or self._params.vad_enabled: + if not self._audio_in_task and (self._params.audio_in_enabled or self._params.vad_enabled): logger.debug(f"Start receiving audio") self._audio_in_task = self.create_task(self._audio_in_task_handler()) diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index f104e99c8..9d26e1bfc 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -360,7 +360,7 @@ class LiveKitInputTransport(BaseInputTransport): await super().start(frame) await self._client.setup(frame) await self._client.connect() - if self._params.audio_in_enabled or self._params.vad_enabled: + if not self._audio_in_task and (self._params.audio_in_enabled or self._params.vad_enabled): self._audio_in_task = self.create_task(self._audio_in_task_handler()) logger.info("LiveKitInputTransport started")