diff --git a/CHANGELOG.md b/CHANGELOG.md index 72f00278c..e5d3ae6d1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added a new `StopFrame` which can be used to stop a pipeline task while + keeping the frame processors running. The frame processors could then be used + in a different pipeline. The difference between a `StopFrame` and a + `StopTaskFrame` is that, as with `EndFrame` and `EndTaskFrame`, the + `StopFrame` is pushed from the task and the `StopTaskFrame` is pushed upstream + inside the pipeline by any processor. + - Added a new `PipelineTask` parameter `observers` that replaces the previous `PipelineParams.observers`. diff --git a/examples/foundational/03a-local-still-frame.py b/examples/foundational/03a-local-still-frame.py index 8143c4765..59cc27b7b 100644 --- a/examples/foundational/03a-local-still-frame.py +++ b/examples/foundational/03a-local-still-frame.py @@ -18,8 +18,7 @@ from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineTask from pipecat.services.fal import FalImageGenService -from pipecat.transports.base_transport import TransportParams -from pipecat.transports.local.tk import TkLocalTransport +from pipecat.transports.local.tk import TkLocalTransport, TkTransportParams load_dotenv(override=True) @@ -34,7 +33,9 @@ async def main(): transport = TkLocalTransport( tk_root, - TransportParams(camera_out_enabled=True, camera_out_width=1024, camera_out_height=1024), + TkTransportParams( + camera_out_enabled=True, camera_out_width=1024, camera_out_height=1024 + ), ) imagegen = FalImageGenService( diff --git a/examples/foundational/05a-local-sync-speech-and-image.py b/examples/foundational/05a-local-sync-speech-and-image.py index 479ddfa94..7bd689fe6 100644 --- a/examples/foundational/05a-local-sync-speech-and-image.py +++ b/examples/foundational/05a-local-sync-speech-and-image.py @@ -30,8 +30,7 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.services.cartesia import CartesiaHttpTTSService from pipecat.services.fal import FalImageGenService from pipecat.services.openai import OpenAILLMService -from pipecat.transports.base_transport import TransportParams -from pipecat.transports.local.tk import TkLocalTransport, TkOutputTransport +from pipecat.transports.local.tk import TkLocalTransport, TkTransportParams load_dotenv(override=True) @@ -152,7 +151,7 @@ async def main(): transport = TkLocalTransport( tk_root, - TransportParams( + TkTransportParams( audio_out_enabled=True, camera_out_enabled=True, camera_out_width=1024, diff --git a/examples/foundational/09a-local-mirror.py b/examples/foundational/09a-local-mirror.py index 937ab4867..d9a010766 100644 --- a/examples/foundational/09a-local-mirror.py +++ b/examples/foundational/09a-local-mirror.py @@ -24,8 +24,7 @@ from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.transports.base_transport import TransportParams -from pipecat.transports.local.tk import TkLocalTransport +from pipecat.transports.local.tk import TkLocalTransport, TkTransportParams from pipecat.transports.services.daily import DailyParams, DailyTransport load_dotenv(override=True) @@ -67,7 +66,7 @@ async def main(): tk_transport = TkLocalTransport( tk_root, - TransportParams( + TkTransportParams( audio_out_enabled=True, camera_out_enabled=True, camera_out_is_live=True, 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..dde280fc1 100644 --- a/examples/foundational/22d-natural-conversation-gemini-audio.py +++ b/examples/foundational/22d-natural-conversation-gemini-audio.py @@ -25,10 +25,8 @@ from pipecat.frames.frames import ( InputAudioRawFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, - LLMMessagesFrame, StartFrame, StartInterruptionFrame, - StopInterruptionFrame, SystemFrame, TextFrame, TranscriptionFrame, @@ -555,6 +553,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 +601,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/frames/frames.py b/src/pipecat/frames/frames.py index c9b812a1c..b43aaaeae 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -23,7 +23,7 @@ from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.clocks.base_clock import BaseClock from pipecat.metrics.metrics import MetricsData from pipecat.transcriptions.language import Language -from pipecat.utils.asyncio import TaskManager +from pipecat.utils.asyncio import BaseTaskManager from pipecat.utils.time import nanoseconds_to_str from pipecat.utils.utils import obj_count, obj_id @@ -438,7 +438,7 @@ class StartFrame(SystemFrame): """This is the first frame that should be pushed down a pipeline.""" clock: BaseClock - task_manager: TaskManager + task_manager: BaseTaskManager audio_in_sample_rate: int = 16000 audio_out_sample_rate: int = 24000 allow_interruptions: bool = False @@ -513,9 +513,9 @@ class CancelTaskFrame(SystemFrame): @dataclass class StopTaskFrame(SystemFrame): - """Indicates that a pipeline task should be stopped but that the pipeline - processors should be kept in a running state. This is normally queued from - the pipeline task. + """This is used to notify the pipeline task that it should be stopped as + soon as possible (flushing all the queued frames) but that the pipeline + processors should be kept in a running state. """ @@ -722,6 +722,17 @@ class EndFrame(ControlFrame): pass +@dataclass +class StopFrame(ControlFrame): + """Indicates that a pipeline should be stopped but that the pipeline + processors should be kept in a running state. This is normally queued from + the pipeline task. + + """ + + pass + + @dataclass class LLMFullResponseStartFrame(ControlFrame): """Used to indicate the beginning of an LLM response. Following by one or 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/pipeline/task.py b/src/pipecat/pipeline/task.py index 873fc60d6..da18cca5d 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -5,7 +5,7 @@ # import asyncio -from typing import Any, AsyncIterable, Dict, Iterable, List +from typing import Any, AsyncIterable, Dict, Iterable, List, Optional from loguru import logger from pydantic import BaseModel, ConfigDict @@ -22,6 +22,7 @@ from pipecat.frames.frames import ( HeartbeatFrame, MetricsFrame, StartFrame, + StopFrame, StopTaskFrame, ) from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData @@ -30,7 +31,7 @@ from pipecat.pipeline.base_pipeline import BasePipeline from pipecat.pipeline.base_task import BaseTask from pipecat.pipeline.task_observer import TaskObserver from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.utils.asyncio import TaskManager +from pipecat.utils.asyncio import BaseTaskManager, TaskManager from pipecat.utils.utils import obj_count, obj_id HEARTBEAT_SECONDS = 1.0 @@ -133,6 +134,7 @@ class PipelineTask(BaseTask): params: PipelineParams = PipelineParams(), observers: List[BaseObserver] = [], clock: BaseClock = SystemClock(), + task_manager: Optional[BaseTaskManager] = None, check_dangling_tasks: bool = True, ): self._id: int = obj_id() @@ -163,9 +165,9 @@ class PipelineTask(BaseTask): # This is the heartbeat queue. When a heartbeat frame is received in the # down queue we add it to the heartbeat queue for processing. self._heartbeat_queue = asyncio.Queue() - # This event is used to indicate an EndFrame has been received in the - # down queue. - self._endframe_event = asyncio.Event() + # This event is used to indicate a finalize frame (e.g. EndFrame, + # StopFrame) has been received in the down queue. + self._pipeline_end_event = asyncio.Event() self._source = PipelineTaskSource(self._up_queue) self._source.link(pipeline) @@ -173,7 +175,7 @@ class PipelineTask(BaseTask): self._sink = PipelineTaskSink(self._down_queue) pipeline.link(self._sink) - self._task_manager = TaskManager() + self._task_manager = task_manager or TaskManager() self._observer = TaskObserver(observers=observers, task_manager=self._task_manager) @@ -224,9 +226,12 @@ class PipelineTask(BaseTask): """Starts and manages the pipeline execution until completion or cancellation.""" if self.has_finished(): return + cleanup_pipeline = True try: push_task = await self._create_tasks() await self._task_manager.wait_for_task(push_task) + # We have already cleaned up the pipeline inside the task. + cleanup_pipeline = False except asyncio.CancelledError: # We are awaiting on the push task and it might be cancelled # (e.g. Ctrl-C). This means we will get a CancelledError here as @@ -234,7 +239,7 @@ class PipelineTask(BaseTask): # awaiting a task. pass await self._cancel_tasks() - await self._cleanup() + await self._cleanup(cleanup_pipeline) if self._check_dangling_tasks: self._print_dangling_tasks() self._finished = True @@ -305,13 +310,14 @@ class PipelineTask(BaseTask): data.append(ProcessingMetricsData(processor=p.name, value=0.0)) return MetricsFrame(data=data) - async def _wait_for_endframe(self): - await self._endframe_event.wait() - self._endframe_event.clear() + async def _wait_for_pipeline_end(self): + await self._pipeline_end_event.wait() + self._pipeline_end_event.clear() - async def _cleanup(self): + async def _cleanup(self, cleanup_pipeline: bool): await self._source.cleanup() - await self._pipeline.cleanup() + if cleanup_pipeline: + await self._pipeline.cleanup() await self._sink.cleanup() async def _process_push_queue(self): @@ -342,18 +348,16 @@ class PipelineTask(BaseTask): await self._source.queue_frame(self._initial_metrics_frame(), FrameDirection.DOWNSTREAM) running = True - should_cleanup = True + cleanup_pipeline = True while running: frame = await self._push_queue.get() await self._source.queue_frame(frame, FrameDirection.DOWNSTREAM) - if isinstance(frame, EndFrame): - await self._wait_for_endframe() - running = not isinstance(frame, (CancelFrame, EndFrame, StopTaskFrame)) - should_cleanup = not isinstance(frame, StopTaskFrame) + if isinstance(frame, (EndFrame, StopFrame)): + await self._wait_for_pipeline_end() + running = not isinstance(frame, (CancelFrame, EndFrame, StopFrame)) + cleanup_pipeline = not isinstance(frame, StopFrame) self._push_queue.task_done() - # Cleanup only if we need to. - if should_cleanup: - await self._cleanup() + await self._cleanup(cleanup_pipeline) async def _process_up_queue(self): """This is the task that processes frames coming upstream from the @@ -371,7 +375,8 @@ class PipelineTask(BaseTask): # Tell the task we should end right away. await self.queue_frame(CancelFrame()) elif isinstance(frame, StopTaskFrame): - await self.queue_frame(StopTaskFrame()) + # Tell the task we should stop nicely. + await self.queue_frame(StopFrame()) elif isinstance(frame, ErrorFrame): logger.error(f"Error running app: {frame}") if frame.fatal: @@ -390,8 +395,8 @@ class PipelineTask(BaseTask): """ while True: frame = await self._down_queue.get() - if isinstance(frame, EndFrame): - self._endframe_event.set() + if isinstance(frame, (EndFrame, StopFrame)): + self._pipeline_end_event.set() elif isinstance(frame, HeartbeatFrame): await self._heartbeat_queue.put(frame) self._down_queue.task_done() diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py index abcbf513b..122038386 100644 --- a/src/pipecat/pipeline/task_observer.py +++ b/src/pipecat/pipeline/task_observer.py @@ -12,7 +12,7 @@ from attr import dataclass from pipecat.frames.frames import Frame from pipecat.observers.base_observer import BaseObserver from pipecat.processors.frame_processor import FrameDirection, FrameProcessor -from pipecat.utils.asyncio import TaskManager +from pipecat.utils.asyncio import BaseTaskManager from pipecat.utils.utils import obj_count, obj_id @@ -55,7 +55,7 @@ class TaskObserver(BaseObserver): """ - def __init__(self, *, observers: List[BaseObserver] = [], task_manager: TaskManager): + def __init__(self, *, observers: List[BaseObserver] = [], task_manager: BaseTaskManager): self._id: int = obj_id() self._name: str = f"{self.__class__.__name__}#{obj_count(self)}" self._observers = observers 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..da8ca63aa 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -246,11 +246,15 @@ class LLMUserContextAggregator(LLMContextResponseAggregator): await super().process_frame(frame, direction) if isinstance(frame, StartFrame): + # Push StartFrame before start(), because we want StartFrame to be + # processed by every processor before any other frame is processed. + await self.push_frame(frame, direction) await self._start(frame) - await self.push_frame(frame, direction) elif isinstance(frame, EndFrame): - await self._stop(frame) + # Push EndFrame before stop(), because stop() waits on the task to + # finish and the task finishes when EndFrame is processed. await self.push_frame(frame, direction) + await self._stop(frame) elif isinstance(frame, CancelFrame): await self._cancel(frame) await self.push_frame(frame, direction) @@ -309,7 +313,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/frame_processor.py b/src/pipecat/processors/frame_processor.py index f27c6d6c2..335585ec3 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -23,7 +23,7 @@ from pipecat.frames.frames import ( ) from pipecat.metrics.metrics import LLMTokenUsage, MetricsData from pipecat.processors.metrics.frame_processor_metrics import FrameProcessorMetrics -from pipecat.utils.asyncio import TaskManager +from pipecat.utils.asyncio import BaseTaskManager from pipecat.utils.utils import obj_count, obj_id @@ -52,7 +52,7 @@ class FrameProcessor: self._clock: Optional[BaseClock] = None # Task Manager - self._task_manager: Optional[TaskManager] = None + self._task_manager: Optional[BaseTaskManager] = None # Other properties self._allow_interruptions = False @@ -192,7 +192,7 @@ class FrameProcessor: raise Exception(f"{self} Clock is still not initialized.") return self._clock - def get_task_manager(self) -> TaskManager: + def get_task_manager(self) -> BaseTaskManager: if not self._task_manager: raise Exception(f"{self} TaskManager is still not initialized.") return self._task_manager @@ -240,7 +240,7 @@ class FrameProcessor: elif isinstance(frame, StopInterruptionFrame): self._should_report_ttfb = True elif isinstance(frame, CancelFrame): - self._cancelling = True + await self.__cancel(frame) async def push_error(self, error: ErrorFrame): await self.push_frame(error, FrameDirection.UPSTREAM) @@ -275,6 +275,11 @@ class FrameProcessor: self.__create_input_task() self.__create_push_task() + async def __cancel(self, frame: CancelFrame): + self._cancelling = True + await self.__cancel_input_task() + await self.__cancel_push_task() + # # Handle interruptions # 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 4fdcd0afa..a47320b7a 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): @@ -487,7 +487,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: @@ -663,9 +664,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: @@ -841,7 +843,8 @@ class SegmentedSTTService(STTService): async def start(self, frame: StartFrame): await super().start(frame) - (self._content, self._wave) = self._new_wave() + if not self._wave: + (self._content, self._wave) = self._new_wave() async def stop(self, frame: EndFrame): await super().stop(frame) diff --git a/src/pipecat/services/assemblyai.py b/src/pipecat/services/assemblyai.py index 84113f582..94f081c5b 100644 --- a/src/pipecat/services/assemblyai.py +++ b/src/pipecat/services/assemblyai.py @@ -91,6 +91,9 @@ class AssemblyAISTTService(STTService): AssemblyAI transcriber. """ + if self._transcriber: + return + def on_open(session_opened: aai.RealtimeSessionOpened): """Callback for when the connection to AssemblyAI is opened.""" logger.info(f"{self}: Connected to AssemblyAI") diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index fbdd08ab1..11c34214c 100644 --- a/src/pipecat/services/azure.py +++ b/src/pipecat/services/azure.py @@ -535,6 +535,10 @@ class AzureTTSService(AzureBaseTTSService): async def start(self, frame: StartFrame): await super().start(frame) + + if self._speech_config: + return + # Now self.sample_rate is properly initialized self._speech_config = SpeechConfig( subscription=self._api_key, @@ -624,6 +628,10 @@ class AzureHttpTTSService(AzureBaseTTSService): async def start(self, frame: StartFrame): await super().start(frame) + + if self._speech_config: + return + self._speech_config = SpeechConfig( subscription=self._api_key, region=self._region, @@ -678,15 +686,22 @@ class AzureSTTService(STTService): self._speech_config = SpeechConfig(subscription=api_key, region=region) self._speech_config.speech_recognition_language = language + self._audio_stream = None + self._speech_recognizer = None + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: await self.start_processing_metrics() - self._audio_stream.write(audio) + if self._audio_stream: + self._audio_stream.write(audio) await self.stop_processing_metrics() yield None async def start(self, frame: StartFrame): await super().start(frame) + if self._audio_stream: + return + stream_format = AudioStreamFormat(samples_per_second=self.sample_rate, channels=1) self._audio_stream = PushAudioInputStream(stream_format) @@ -700,13 +715,21 @@ class AzureSTTService(STTService): async def stop(self, frame: EndFrame): await super().stop(frame) - self._speech_recognizer.stop_continuous_recognition_async() - self._audio_stream.close() + + if self._speech_recognizer: + self._speech_recognizer.stop_continuous_recognition_async() + + if self._audio_stream: + self._audio_stream.close() async def cancel(self, frame: CancelFrame): await super().cancel(frame) - self._speech_recognizer.stop_continuous_recognition_async() - self._audio_stream.close() + + if self._speech_recognizer: + self._speech_recognizer.stop_continuous_recognition_async() + + if self._audio_stream: + self._audio_stream.close() def _on_handle_recognized(self, event): if event.result.reason == ResultReason.RecognizedSpeech and len(event.result.text) > 0: diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 0ec55aa0c..04a35087b 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: @@ -194,6 +194,8 @@ class CartesiaTTSService(AudioContextWordTTSService): async def _connect_websocket(self): try: + if self._websocket: + return logger.debug("Connecting to Cartesia") self._websocket = await websockets.connect( f"{self._url}?api_key={self._api_key}&cartesia_version={self._cartesia_version}" diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 47fa87b5a..3dc42975e 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: @@ -302,6 +308,9 @@ class ElevenLabsTTSService(InterruptibleWordTTSService): async def _connect_websocket(self): try: + if self._websocket: + return + logger.debug("Connecting to ElevenLabs") voice_id = self._voice_id diff --git a/src/pipecat/services/fish.py b/src/pipecat/services/fish.py index 1c5e6a08a..e61579bfb 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: @@ -117,6 +118,9 @@ class FishAudioTTSService(InterruptibleTTSService): async def _connect_websocket(self): try: + if self._websocket: + return + logger.debug("Connecting to Fish Audio") headers = {"Authorization": f"Bearer {self._api_key}"} self._websocket = await websockets.connect(self._base_url, extra_headers=headers) diff --git a/src/pipecat/services/gladia.py b/src/pipecat/services/gladia.py index c93dfef3b..f0a00b327 100644 --- a/src/pipecat/services/gladia.py +++ b/src/pipecat/services/gladia.py @@ -172,27 +172,38 @@ class GladiaSTTService(STTService): }, } self._confidence = confidence + self._websocket = None + self._receive_task = None def language_to_service_language(self, language: Language) -> Optional[str]: return language_to_gladia_language(language) async def start(self, frame: StartFrame): await super().start(frame) + if self._websocket: + return 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._websocket: + await self._websocket.close() + self._websocket = None + 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..8d4ca4803 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: @@ -122,6 +124,9 @@ class LmntTTSService(InterruptibleTTSService): async def _connect_websocket(self): """Connect to LMNT websocket.""" try: + if self._websocket: + return + logger.debug("Connecting to LMNT") # Build initial connection message diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index 828acbdb7..56179e34c 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: @@ -170,6 +171,9 @@ class PlayHTTTSService(InterruptibleTTSService): async def _connect_websocket(self): try: + if self._websocket: + return + logger.debug("Connecting to PlayHT") if not self._websocket_url: diff --git a/src/pipecat/services/rime.py b/src/pipecat/services/rime.py index 6e2daccd6..60d2d67ef 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.""" @@ -178,6 +180,9 @@ class RimeTTSService(AudioContextWordTTSService): async def _connect_websocket(self): """Connect to Rime websocket API with configured settings.""" try: + if self._websocket: + return + params = "&".join(f"{k}={v}" for k, v in self._settings.items()) url = f"{self._url}?{params}" headers = {"Authorization": f"Bearer {self._api_key}"} diff --git a/src/pipecat/services/riva.py b/src/pipecat/services/riva.py index 458cb4919..8d5ce79c4 100644 --- a/src/pipecat/services/riva.py +++ b/src/pipecat/services/riva.py @@ -166,6 +166,9 @@ class ParakeetSTTService(STTService): self._asr_service = riva.client.ASRService(auth) self._queue = asyncio.Queue() + self._config = None + self._thread_task = None + self._response_task = None def can_generate_metrics(self) -> bool: return False @@ -173,6 +176,9 @@ class ParakeetSTTService(STTService): async def start(self, frame: StartFrame): await super().start(frame) + if self._config: + return + config = riva.client.StreamingRecognitionConfig( config=riva.client.RecognitionConfig( encoding=riva.client.AudioEncoding.LINEAR_PCM, @@ -205,9 +211,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 +227,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..0fb490914 100644 --- a/src/pipecat/services/simli.py +++ b/src/pipecat/services/simli.py @@ -43,14 +43,21 @@ class SimliVideoService(FrameProcessor): self._pipecat_resampler: AudioResampler = None self._simli_resampler = AudioResampler("s16", "mono", 16000) + self._initialized = False self._audio_task: asyncio.Task = None self._video_task: asyncio.Task = None async def _start_connection(self): - await self._simli_client.Initialize() + if not self._initialized: + await self._simli_client.Initialize() + self._initialized = True + # 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 +124,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/services/xtts.py b/src/pipecat/services/xtts.py index 4d0f3fca6..9d2f3dd1e 100644 --- a/src/pipecat/services/xtts.py +++ b/src/pipecat/services/xtts.py @@ -99,6 +99,10 @@ class XTTSService(TTSService): async def start(self, frame: StartFrame): await super().start(frame) + + if self._studio_speakers: + return + async with self._aiohttp_session.get(self._settings["base_url"] + "/studio_speakers") as r: if r.status != 200: text = await r.text() 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/local/audio.py b/src/pipecat/transports/local/audio.py index e7b803f2b..bc8d2dd16 100644 --- a/src/pipecat/transports/local/audio.py +++ b/src/pipecat/transports/local/audio.py @@ -44,6 +44,9 @@ class LocalAudioInputTransport(BaseInputTransport): async def start(self, frame: StartFrame): await super().start(frame) + if self._in_stream: + return + self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate num_frames = int(self._sample_rate / 100) * 2 # 20ms of audio @@ -94,6 +97,9 @@ class LocalAudioOutputTransport(BaseOutputTransport): async def start(self, frame: StartFrame): await super().start(frame) + if self._out_stream: + return + self._sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate self._out_stream = self._py_audio.open( @@ -110,6 +116,7 @@ class LocalAudioOutputTransport(BaseOutputTransport): if self._out_stream: self._out_stream.stop_stream() self._out_stream.close() + self._out_stream = None async def write_raw_audio_frames(self, frames: bytes): if self._out_stream: diff --git a/src/pipecat/transports/local/tk.py b/src/pipecat/transports/local/tk.py index 73777c573..4338992e3 100644 --- a/src/pipecat/transports/local/tk.py +++ b/src/pipecat/transports/local/tk.py @@ -51,6 +51,9 @@ class TkInputTransport(BaseInputTransport): async def start(self, frame: StartFrame): await super().start(frame) + if self._in_stream: + return + self._sample_rate = self._params.audio_in_sample_rate or frame.audio_in_sample_rate num_frames = int(self._sample_rate / 100) * 2 # 20ms of audio @@ -70,6 +73,7 @@ class TkInputTransport(BaseInputTransport): if self._in_stream: self._in_stream.stop_stream() self._in_stream.close() + self._in_stream = None def _audio_in_callback(self, in_data, frame_count, time_info, status): frame = InputAudioRawFrame( @@ -86,7 +90,7 @@ class TkInputTransport(BaseInputTransport): class TkOutputTransport(BaseOutputTransport): _params: TkTransportParams - def __init__(self, tk_root: tk.Tk, py_audio: pyaudio.PyAudio, params: TransportParams): + def __init__(self, tk_root: tk.Tk, py_audio: pyaudio.PyAudio, params: TkTransportParams): super().__init__(params) self._py_audio = py_audio self._out_stream = None @@ -106,6 +110,9 @@ class TkOutputTransport(BaseOutputTransport): async def start(self, frame: StartFrame): await super().start(frame) + if self._out_stream: + return + self._sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate self._out_stream = self._py_audio.open( @@ -122,6 +129,7 @@ class TkOutputTransport(BaseOutputTransport): if self._out_stream: self._out_stream.stop_stream() self._out_stream.close() + self._out_stream = None async def write_raw_audio_frames(self, frames: bytes): if self._out_stream: @@ -145,7 +153,7 @@ class TkOutputTransport(BaseOutputTransport): class TkLocalTransport(BaseTransport): - def __init__(self, tk_root: tk.Tk, params: TransportParams): + def __init__(self, tk_root: tk.Tk, params: TkTransportParams): super().__init__() self._tk_root = tk_root self._params = params 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_client.py b/src/pipecat/transports/network/websocket_client.py index d7db9c4bd..eb2b5cfb8 100644 --- a/src/pipecat/transports/network/websocket_client.py +++ b/src/pipecat/transports/network/websocket_client.py @@ -30,7 +30,7 @@ from pipecat.serializers.protobuf import ProtobufFrameSerializer from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams -from pipecat.utils.asyncio import TaskManager +from pipecat.utils.asyncio import BaseTaskManager class WebsocketClientParams(TransportParams): @@ -57,12 +57,12 @@ class WebsocketClientSession: self._callbacks = callbacks self._transport_name = transport_name - self._task_manager: Optional[TaskManager] = None + self._task_manager: Optional[BaseTaskManager] = None self._websocket: Optional[websockets.WebSocketClientProtocol] = None @property - def task_manager(self) -> TaskManager: + def task_manager(self) -> BaseTaskManager: if not self._task_manager: raise Exception( f"{self._transport_name}::WebsocketClientSession: TaskManager not initialized (pipeline not started?)" diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index ca5d07d2e..19ebe4a45 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -22,7 +22,6 @@ from pipecat.frames.frames import ( OutputAudioRawFrame, StartFrame, StartInterruptionFrame, - TextFrame, TransportMessageFrame, TransportMessageUrgentFrame, ) @@ -81,22 +80,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 +120,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..7735ae91f 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -43,7 +43,7 @@ from pipecat.transcriptions.language import Language from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams -from pipecat.utils.asyncio import TaskManager +from pipecat.utils.asyncio import BaseTaskManager try: from daily import CallClient, Daily, EventHandler @@ -293,7 +293,7 @@ class DailyTransportClient(EventHandler): self._joined_event = asyncio.Event() self._leave_counter = 0 - self._task_manager: Optional[TaskManager] = None + self._task_manager: Optional[BaseTaskManager] = None # We use the executor to cleanup the client. We just do it from one # place, so only one thread is really needed. @@ -832,6 +832,9 @@ class DailyInputTransport(BaseInputTransport): self._video_renderers = {} + # Whether we have seen a StartFrame already. + self._initialized = False + # Task that gets audio data from a device or the network and queues it # internally to be processed. self._audio_in_task = None @@ -845,13 +848,19 @@ 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()) async def start(self, frame: StartFrame): # Parent start. await super().start(frame) + + if self._initialized: + return + + self._initialized = True + # Setup client. await self._client.setup(frame) # Join the room. @@ -980,9 +989,18 @@ class DailyOutputTransport(BaseOutputTransport): self._client = client + # Whether we have seen a StartFrame already. + self._initialized = False + async def start(self, frame: StartFrame): # Parent start. await super().start(frame) + + if self._initialized: + return + + self._initialized = True + # Setup client. await self._client.setup(frame) # Join the room. diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index f104e99c8..7018ea520 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -27,7 +27,7 @@ from pipecat.processors.frame_processor import FrameDirection from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport from pipecat.transports.base_transport import BaseTransport, TransportParams -from pipecat.utils.asyncio import TaskManager +from pipecat.utils.asyncio import BaseTaskManager try: from livekit import rtc @@ -88,7 +88,7 @@ class LiveKitTransportClient: self._audio_tracks = {} self._audio_queue = asyncio.Queue() self._other_participant_has_joined = False - self._task_manager: Optional[TaskManager] = None + self._task_manager: Optional[BaseTaskManager] = None @property def participant_id(self) -> str: @@ -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") diff --git a/src/pipecat/utils/asyncio.py b/src/pipecat/utils/asyncio.py index ab40a8195..acc4acec8 100644 --- a/src/pipecat/utils/asyncio.py +++ b/src/pipecat/utils/asyncio.py @@ -5,12 +5,76 @@ # import asyncio +from abc import ABC, abstractmethod from typing import Coroutine, Optional, Set from loguru import logger -class TaskManager: +class BaseTaskManager(ABC): + @abstractmethod + def set_event_loop(self, loop: asyncio.AbstractEventLoop): + pass + + @abstractmethod + def get_event_loop(self) -> asyncio.AbstractEventLoop: + pass + + @abstractmethod + def create_task(self, coroutine: Coroutine, name: str) -> asyncio.Task: + """ + Creates and schedules a new asyncio Task that runs the given coroutine. + + The task is added to a global set of created tasks. + + Args: + loop (asyncio.AbstractEventLoop): The event loop to use for creating the task. + coroutine (Coroutine): The coroutine to be executed within the task. + name (str): The name to assign to the task for identification. + + Returns: + asyncio.Task: The created task object. + """ + pass + + @abstractmethod + async def wait_for_task(self, task: asyncio.Task, timeout: Optional[float] = None): + """Wait for an asyncio.Task to complete with optional timeout handling. + + This function awaits the specified asyncio.Task and handles scenarios for + timeouts, cancellations, and other exceptions. It also ensures that the task + is removed from the set of registered tasks upon completion or failure. + + Args: + task (asyncio.Task): The asyncio Task to wait for. + timeout (Optional[float], optional): The maximum number of seconds + to wait for the task to complete. If None, waits indefinitely. + Defaults to None. + """ + pass + + @abstractmethod + async def cancel_task(self, task: asyncio.Task, timeout: Optional[float] = None): + """Cancels the given asyncio Task and awaits its completion with an + optional timeout. + + This function removes the task from the set of registered tasks upon + completion or failure. + + Args: + task (asyncio.Task): The task to be cancelled. + timeout (Optional[float]): The optional timeout in seconds to wait for the task to cancel. + + """ + pass + + @abstractmethod + def current_tasks(self) -> Set[asyncio.Task]: + """Returns the list of currently created/registered tasks.""" + pass + + +class TaskManager(BaseTaskManager): def __init__(self) -> None: self._tasks: Set[asyncio.Task] = set() self._loop: Optional[asyncio.AbstractEventLoop] = None @@ -80,6 +144,7 @@ class TaskManager: logger.warning(f"{name}: timed out waiting for task to finish") except asyncio.CancelledError: logger.trace(f"{name}: unexpected task cancellation (maybe Ctrl-C?)") + raise except Exception as e: logger.exception(f"{name}: unexpected exception while stopping task: {e}") finally: diff --git a/tests/integration/integration_openai_llm.py b/tests/integration/integration_openai_llm.py index c788936a1..2c84c8882 100644 --- a/tests/integration/integration_openai_llm.py +++ b/tests/integration/integration_openai_llm.py @@ -10,7 +10,7 @@ from openai.types.chat import ( ) from pipecat.frames.frames import LLMFullResponseEndFrame, LLMFullResponseStartFrame, TextFrame -from pipecat.processors.frame_processor import FrameDirection, FrameProcessor +from pipecat.processors.frame_processor import FrameDirection from pipecat.services.openai import OpenAILLMContext, OpenAILLMContextFrame, OpenAILLMService from pipecat.utils.test_frame_processor import TestFrameProcessor