wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

make sure we don't create tasks if already created

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-02-24 22:32:52 -08:00
parent 4536d03e82
commit 1ec68bd071
25 changed files with 142 additions and 61 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
examples/foundational/22b-natural-conversation-proposal.py
View File

@@ -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:

8
examples/foundational/22c-natural-conversation-mixed-llms.py
View File

@@ -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:

8
examples/foundational/22d-natural-conversation-gemini-audio.py
View File

@@ -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:

31
src/pipecat/pipeline/parallel_pipeline.py
View File

@@ -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:

8
src/pipecat/processors/aggregators/gated_openai_llm_context.py
View File

@@ -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:

3
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -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:

6
src/pipecat/processors/frameworks/rtvi.py
View File

@@ -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):

9
src/pipecat/processors/idle_frame_processor.py
View File

@@ -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:

4
src/pipecat/processors/user_idle_processor.py
View File

@@ -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

12
src/pipecat/services/ai_services.py
View File

@@ -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:

4
src/pipecat/services/cartesia.py
View File

@@ -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:

10
src/pipecat/services/elevenlabs.py
View File

@@ -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:

3
src/pipecat/services/fish.py
View File

@@ -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:

12
src/pipecat/services/gladia.py
View File

@@ -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()

4
src/pipecat/services/lmnt.py
View File

@@ -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:

3
src/pipecat/services/playht.py
View File

@@ -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:

4
src/pipecat/services/rime.py
View File

@@ -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."""

20
src/pipecat/services/riva.py
View File

@@ -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(

9
src/pipecat/services/simli.py
View File

@@ -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

2
src/pipecat/transports/base_input.py
View File

@@ -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())

12
src/pipecat/transports/base_output.py
View File

@@ -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())

10
src/pipecat/transports/network/fastapi_websocket.py
View File

@@ -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)

9
src/pipecat/transports/network/websocket_server.py
View File

@@ -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)
)

2
src/pipecat/transports/services/daily.py
View File

@@ -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())

2
src/pipecat/transports/services/livekit.py
View File

@@ -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")