From 357934a644adaf75d53230e767dbb4001234a8c7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 25 Jun 2025 13:36:52 -0700 Subject: [PATCH] watchdog timers are disabled by default use enable_watchdog_timers --- src/pipecat/frames/frames.py | 2 +- src/pipecat/pipeline/parallel_pipeline.py | 41 ++++++++-------- .../pipeline/sync_parallel_pipeline.py | 35 +++++++------- src/pipecat/pipeline/task.py | 48 ++++++++++++++----- src/pipecat/pipeline/task_observer.py | 9 +++- src/pipecat/processors/frame_processor.py | 23 ++++++--- src/pipecat/processors/frameworks/rtvi.py | 4 +- .../metrics/frame_processor_metrics.py | 2 +- src/pipecat/processors/metrics/sentry.py | 6 +-- src/pipecat/processors/producer_processor.py | 2 +- src/pipecat/services/anthropic/llm.py | 4 +- src/pipecat/services/cartesia/tts.py | 4 +- src/pipecat/services/elevenlabs/tts.py | 4 +- .../services/gemini_multimodal_live/gemini.py | 4 +- src/pipecat/services/gladia/stt.py | 4 +- src/pipecat/services/google/llm.py | 4 +- src/pipecat/services/google/llm_openai.py | 4 +- src/pipecat/services/google/stt.py | 4 +- src/pipecat/services/openai/base_llm.py | 4 +- .../services/openai_realtime_beta/openai.py | 4 +- src/pipecat/services/riva/stt.py | 4 +- src/pipecat/services/sambanova/llm.py | 4 +- src/pipecat/services/simli/video.py | 8 +++- src/pipecat/services/tavus/video.py | 3 +- src/pipecat/services/tts_service.py | 6 ++- src/pipecat/transports/base_output.py | 4 +- .../transports/network/fastapi_websocket.py | 4 +- .../transports/network/small_webrtc.py | 8 +++- src/pipecat/transports/services/daily.py | 9 ++-- src/pipecat/transports/services/livekit.py | 4 +- src/pipecat/utils/asyncio.py | 11 ++++- src/pipecat/utils/watchdog_async_iterator.py | 16 ++++++- src/pipecat/utils/watchdog_event.py | 15 +++++- src/pipecat/utils/watchdog_priority_queue.py | 25 ++++++++-- src/pipecat/utils/watchdog_queue.py | 25 ++++++++-- 35 files changed, 256 insertions(+), 102 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 4b2d934ab..3a602a3e2 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -453,8 +453,8 @@ class StartFrame(SystemFrame): allow_interruptions: bool = False enable_metrics: bool = False enable_usage_metrics: bool = False - report_only_initial_ttfb: bool = False interruption_strategies: List[BaseInterruptionStrategy] = field(default_factory=list) + report_only_initial_ttfb: bool = False @dataclass diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py index 4794ea708..f6ac78827 100644 --- a/src/pipecat/pipeline/parallel_pipeline.py +++ b/src/pipecat/pipeline/parallel_pipeline.py @@ -77,20 +77,36 @@ class ParallelPipeline(BasePipeline): if len(args) == 0: raise Exception(f"ParallelPipeline needs at least one argument") + self._args = args self._sources = [] self._sinks = [] + self._pipelines = [] + self._seen_ids = set() self._endframe_counter: Dict[int, int] = {} self._up_task = None self._down_task = None - self._up_queue = WatchdogQueue(self) - self._down_queue = WatchdogQueue(self) - self._pipelines = [] + # + # BasePipeline + # + + def processors_with_metrics(self) -> List[FrameProcessor]: + return list(chain.from_iterable(p.processors_with_metrics() for p in self._pipelines)) + + # + # Frame processor + # + + async def setup(self, setup: FrameProcessorSetup): + await super().setup(setup) + + self._up_queue = WatchdogQueue(self, watchdog_enabled=setup.watchdog_timers_enabled) + self._down_queue = WatchdogQueue(self, watchdog_enabled=setup.watchdog_timers_enabled) logger.debug(f"Creating {self} pipelines") - for processors in args: + for processors in self._args: if not isinstance(processors, list): raise TypeError(f"ParallelPipeline argument {processors} is not a list") @@ -108,19 +124,6 @@ class ParallelPipeline(BasePipeline): logger.debug(f"Finished creating {self} pipelines") - # - # BasePipeline - # - - def processors_with_metrics(self) -> List[FrameProcessor]: - return list(chain.from_iterable(p.processors_with_metrics() for p in self._pipelines)) - - # - # Frame processor - # - - async def setup(self, setup: FrameProcessorSetup): - await super().setup(setup) await asyncio.gather(*[s.setup(setup) for s in self._sources]) await asyncio.gather(*[p.setup(setup) for p in self._pipelines]) await asyncio.gather(*[s.setup(setup) for s in self._sinks]) @@ -135,7 +138,7 @@ class ParallelPipeline(BasePipeline): await super().process_frame(frame, direction) if isinstance(frame, StartFrame): - await self._start() + await self._start(frame) elif isinstance(frame, EndFrame): self._endframe_counter[frame.id] = len(self._pipelines) elif isinstance(frame, CancelFrame): @@ -155,7 +158,7 @@ class ParallelPipeline(BasePipeline): elif isinstance(frame, EndFrame): await self._stop() - async def _start(self): + async def _start(self, frame: StartFrame): await self._create_tasks() async def _stop(self): diff --git a/src/pipecat/pipeline/sync_parallel_pipeline.py b/src/pipecat/pipeline/sync_parallel_pipeline.py index 4cf9f5033..f78ca0de3 100644 --- a/src/pipecat/pipeline/sync_parallel_pipeline.py +++ b/src/pipecat/pipeline/sync_parallel_pipeline.py @@ -15,6 +15,7 @@ from pipecat.frames.frames import ControlFrame, EndFrame, Frame, SystemFrame from pipecat.pipeline.base_pipeline import BasePipeline from pipecat.pipeline.pipeline import Pipeline from pipecat.processors.frame_processor import FrameDirection, FrameProcessor, FrameProcessorSetup +from pipecat.utils.watchdog_queue import WatchdogQueue @dataclass @@ -61,15 +62,30 @@ class SyncParallelPipeline(BasePipeline): if len(args) == 0: raise Exception(f"SyncParallelPipeline needs at least one argument") + self._args = args self._sinks = [] self._sources = [] self._pipelines = [] - self._up_queue = asyncio.Queue() - self._down_queue = asyncio.Queue() + # + # BasePipeline + # + + def processors_with_metrics(self) -> List[FrameProcessor]: + return list(chain.from_iterable(p.processors_with_metrics() for p in self._pipelines)) + + # + # Frame processor + # + + async def setup(self, setup: FrameProcessorSetup): + await super().setup(setup) + + self._up_queue = WatchdogQueue(self, watchdog_enabled=setup.watchdog_timers_enabled) + self._down_queue = WatchdogQueue(self, watchdog_enabled=setup.watchdog_timers_enabled) logger.debug(f"Creating {self} pipelines") - for processors in args: + for processors in self._args: if not isinstance(processors, list): raise TypeError(f"SyncParallelPipeline argument {processors} is not a list") @@ -92,19 +108,6 @@ class SyncParallelPipeline(BasePipeline): logger.debug(f"Finished creating {self} pipelines") - # - # BasePipeline - # - - def processors_with_metrics(self) -> List[FrameProcessor]: - return list(chain.from_iterable(p.processors_with_metrics() for p in self._pipelines)) - - # - # Frame processor - # - - async def setup(self, setup: FrameProcessorSetup): - await super().setup(setup) await asyncio.gather(*[s["processor"].setup(setup) for s in self._sources]) await asyncio.gather(*[p.setup(setup) for p in self._pipelines]) await asyncio.gather(*[s["processor"].setup(setup) for s in self._sinks]) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index ad7a5cda6..68b76401e 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -190,6 +190,7 @@ class PipelineTask(WatchdogReseter, BasePipelineTask): enable_tracing: Whether to enable tracing. enable_turn_tracking: Whether to enable turn tracking. enable_watchdog_logging: Whether to print task processing times. + enable_watchdog_timers: Whether to enable task watchdog timers. idle_timeout_frames: A tuple with the frames that should trigger an idle timeout if not received withing `idle_timeout_seconds`. idle_timeout_secs: Timeout (in seconds) to consider pipeline idle or @@ -213,6 +214,7 @@ class PipelineTask(WatchdogReseter, BasePipelineTask): enable_tracing: bool = False, enable_turn_tracking: bool = True, enable_watchdog_logging: bool = False, + enable_watchdog_timers: bool = False, idle_timeout_frames: Tuple[Type[Frame], ...] = ( BotSpeakingFrame, LLMFullResponseEndFrame, @@ -233,6 +235,7 @@ class PipelineTask(WatchdogReseter, BasePipelineTask): self._enable_tracing = enable_tracing and is_tracing_available() self._enable_turn_tracking = enable_turn_tracking self._enable_watchdog_logging = enable_watchdog_logging + self._enable_watchdog_timers = enable_watchdog_timers self._idle_timeout_frames = idle_timeout_frames self._idle_timeout_secs = idle_timeout_secs self._watchdog_timeout_secs = watchdog_timeout_secs @@ -263,18 +266,24 @@ class PipelineTask(WatchdogReseter, BasePipelineTask): self._cancelled = False # This queue receives frames coming from the pipeline upstream. - self._up_queue = WatchdogQueue(self) + self._up_queue = WatchdogQueue(self, watchdog_enabled=enable_watchdog_timers) + self._process_up_task: Optional[asyncio.Task] = None # This queue receives frames coming from the pipeline downstream. - self._down_queue = WatchdogQueue(self) + self._down_queue = WatchdogQueue(self, watchdog_enabled=enable_watchdog_timers) + self._process_down_task: Optional[asyncio.Task] = None # This queue is the queue used to push frames to the pipeline. - self._push_queue = WatchdogQueue(self) + self._push_queue = WatchdogQueue(self, watchdog_enabled=enable_watchdog_timers) + self._process_push_task: Optional[asyncio.Task] = None # 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 = WatchdogQueue(self) + self._heartbeat_queue = WatchdogQueue(self, watchdog_enabled=enable_watchdog_timers) + self._heartbeat_push_task: Optional[asyncio.Task] = None + self._heartbeat_monitor_task: Optional[asyncio.Task] = None # This is the idle queue. When frames are received downstream they are # put in the queue. If no frame is received the pipeline is considered # idle. - self._idle_queue = WatchdogQueue(self) + self._idle_queue = WatchdogQueue(self, watchdog_enabled=enable_watchdog_timers) + self._idle_monitor_task: Optional[asyncio.Task] = None # 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() @@ -438,7 +447,9 @@ class PipelineTask(WatchdogReseter, BasePipelineTask): # we want to cancel right away. await self._source.push_frame(CancelFrame()) # Only cancel the push task. Everything else will be cancelled in run(). - await self._task_manager.cancel_task(self._process_push_task) + if self._process_push_task: + await self._task_manager.cancel_task(self._process_push_task) + self._process_push_task = None async def _create_tasks(self): self._process_up_task = self._task_manager.create_task( @@ -451,7 +462,7 @@ class PipelineTask(WatchdogReseter, BasePipelineTask): self._process_push_queue(), f"{self}::_process_push_queue" ) - await self._observer.start() + await self._observer.start(self._enable_watchdog_timers) return self._process_push_task @@ -473,20 +484,33 @@ class PipelineTask(WatchdogReseter, BasePipelineTask): async def _cancel_tasks(self): await self._observer.stop() - await self._task_manager.cancel_task(self._process_up_task) - await self._task_manager.cancel_task(self._process_down_task) + if self._process_up_task: + await self._task_manager.cancel_task(self._process_up_task) + self._process_up_task = None + + if self._process_down_task: + await self._task_manager.cancel_task(self._process_down_task) + self._process_down_task = None await self._maybe_cancel_heartbeat_tasks() await self._maybe_cancel_idle_task() async def _maybe_cancel_heartbeat_tasks(self): - if self._params.enable_heartbeats: + if not self._params.enable_heartbeats: + return + + if self._heartbeat_push_task: await self._task_manager.cancel_task(self._heartbeat_push_task) + self._heartbeat_push_task = None + + if self._heartbeat_monitor_task: await self._task_manager.cancel_task(self._heartbeat_monitor_task) + self._heartbeat_monitor_task = None async def _maybe_cancel_idle_task(self): - if self._idle_timeout_secs: + if self._idle_timeout_secs and self._idle_monitor_task: await self._task_manager.cancel_task(self._idle_monitor_task) + self._idle_monitor_task = None def _initial_metrics_frame(self) -> MetricsFrame: processors = self._pipeline.processors_with_metrics() @@ -504,6 +528,7 @@ class PipelineTask(WatchdogReseter, BasePipelineTask): mgr_params = TaskManagerParams( loop=params.loop, enable_watchdog_logging=self._enable_watchdog_logging, + enable_watchdog_timers=self._enable_watchdog_timers, watchdog_timeout=self._watchdog_timeout_secs, ) self._task_manager.setup(mgr_params) @@ -512,6 +537,7 @@ class PipelineTask(WatchdogReseter, BasePipelineTask): clock=self._clock, task_manager=self._task_manager, observer=self._observer, + watchdog_timers_enabled=self._enable_watchdog_timers, ) await self._source.setup(setup) await self._pipeline.setup(setup) diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py index 2f7450a75..40d4ef3ed 100644 --- a/src/pipecat/pipeline/task_observer.py +++ b/src/pipecat/pipeline/task_observer.py @@ -54,6 +54,7 @@ class TaskObserver(WatchdogReseter, BaseObserver): self._proxies: Optional[Dict[BaseObserver, Proxy]] = ( None # Becomes a dict after start() is called ) + self._watchdog_timers_enabled = False def add_observer(self, observer: BaseObserver): # Add the observer to the list. @@ -78,12 +79,16 @@ class TaskObserver(WatchdogReseter, BaseObserver): if observer in self._observers: self._observers.remove(observer) - async def start(self): + async def start(self, watchdog_timers_enabled: bool = False): """Starts all proxy observer tasks.""" + self._watchdog_timers_enabled = watchdog_timers_enabled self._proxies = self._create_proxies(self._observers) async def stop(self): """Stops all proxy observer tasks.""" + if not self._proxies: + return + for proxy in self._proxies.values(): await self._task_manager.cancel_task(proxy.task) @@ -98,7 +103,7 @@ class TaskObserver(WatchdogReseter, BaseObserver): return self._proxies is not None def _create_proxy(self, observer: BaseObserver) -> Proxy: - queue = WatchdogQueue(self) + queue = WatchdogQueue(self, watchdog_enabled=self._watchdog_timers_enabled) task = self._task_manager.create_task( self._proxy_task_handler(queue, observer), f"TaskObserver::{observer}::_proxy_task_handler", diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index d7723d868..9c61bd93a 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -46,6 +46,7 @@ class FrameProcessorSetup: clock: BaseClock task_manager: BaseTaskManager observer: Optional[BaseObserver] = None + watchdog_timers_enabled: bool = False class FrameProcessor(WatchdogReseter, BaseObject): @@ -84,6 +85,7 @@ class FrameProcessor(WatchdogReseter, BaseObject): self._enable_usage_metrics = False self._report_only_initial_ttfb = False self._interruption_strategies: List[BaseInterruptionStrategy] = [] + self._watchdog_timers_enabled = False # Indicates whether we have received the StartFrame. self.__started = False @@ -104,7 +106,7 @@ class FrameProcessor(WatchdogReseter, BaseObject): # is called. To resume processing frames we need to call # `resume_processing_frames()` which will wake up the event. self.__should_block_frames = False - self.__input_event = WatchdogEvent(self) + self.__input_event = None self.__input_frame_task: Optional[asyncio.Task] = None # Every processor in Pipecat should only output frames from a single @@ -140,6 +142,10 @@ class FrameProcessor(WatchdogReseter, BaseObject): def interruption_strategies(self) -> Sequence[BaseInterruptionStrategy]: return self._interruption_strategies + @property + def watchdog_timers_enabled(self): + return self._watchdog_timers_enabled + def can_generate_metrics(self) -> bool: return False @@ -220,8 +226,9 @@ class FrameProcessor(WatchdogReseter, BaseObject): self._clock = setup.clock self._task_manager = setup.task_manager self._observer = setup.observer + self._watchdog_timers_enabled = setup.watchdog_timers_enabled if self._metrics is not None: - await self._metrics.setup(self._task_manager) + await self._metrics.setup(self._task_manager, self.watchdog_timers_enabled) async def cleanup(self): await super().cleanup() @@ -313,8 +320,8 @@ class FrameProcessor(WatchdogReseter, BaseObject): self._allow_interruptions = frame.allow_interruptions self._enable_metrics = frame.enable_metrics self._enable_usage_metrics = frame.enable_usage_metrics - self._report_only_initial_ttfb = frame.report_only_initial_ttfb self._interruption_strategies = frame.interruption_strategies + self._report_only_initial_ttfb = frame.report_only_initial_ttfb self.__create_input_task() self.__create_push_task() @@ -396,8 +403,12 @@ class FrameProcessor(WatchdogReseter, BaseObject): def __create_input_task(self): if not self.__input_frame_task: self.__should_block_frames = False + if not self.__input_event: + self.__input_event = WatchdogEvent( + self, watchdog_enabled=self.watchdog_timers_enabled + ) self.__input_event.clear() - self.__input_queue = WatchdogQueue(self) + self.__input_queue = WatchdogQueue(self, watchdog_enabled=self.watchdog_timers_enabled) self.__input_frame_task = self.create_task(self.__input_frame_task_handler()) async def __cancel_input_task(self): @@ -407,7 +418,7 @@ class FrameProcessor(WatchdogReseter, BaseObject): async def __input_frame_task_handler(self): while True: - if self.__should_block_frames: + if self.__should_block_frames and self.__input_event: logger.trace(f"{self}: frame processing paused") await self.__input_event.wait() self.__input_event.clear() @@ -429,7 +440,7 @@ class FrameProcessor(WatchdogReseter, BaseObject): def __create_push_task(self): if not self.__push_frame_task: - self.__push_queue = WatchdogQueue(self) + self.__push_queue = WatchdogQueue(self, watchdog_enabled=self.watchdog_timers_enabled) self.__push_frame_task = self.create_task(self.__push_frame_task_handler()) async def __cancel_push_task(self): diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index e35f72e0b..1646a9fa6 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -651,11 +651,9 @@ class RTVIProcessor(FrameProcessor): self._registered_services: Dict[str, RTVIService] = {} # A task to process incoming action frames. - self._action_queue = WatchdogQueue(self) self._action_task: Optional[asyncio.Task] = None # A task to process incoming transport messages. - self._message_queue = WatchdogQueue(self) self._message_task: Optional[asyncio.Task] = None self._register_event_handler("on_bot_started") @@ -757,8 +755,10 @@ class RTVIProcessor(FrameProcessor): async def _start(self, frame: StartFrame): if not self._action_task: + self._action_queue = WatchdogQueue(self, watchdog_enabled=self.watchdog_timers_enabled) self._action_task = self.create_task(self._action_task_handler()) if not self._message_task: + self._message_queue = WatchdogQueue(self, watchdog_enabled=self.watchdog_timers_enabled) self._message_task = self.create_task(self._message_task_handler()) await self._call_event_handler("on_bot_started") diff --git a/src/pipecat/processors/metrics/frame_processor_metrics.py b/src/pipecat/processors/metrics/frame_processor_metrics.py index 40a83fa38..9ee1ccdd3 100644 --- a/src/pipecat/processors/metrics/frame_processor_metrics.py +++ b/src/pipecat/processors/metrics/frame_processor_metrics.py @@ -31,7 +31,7 @@ class FrameProcessorMetrics(BaseObject): self._last_ttfb_time = 0 self._should_report_ttfb = True - async def setup(self, task_manager: TaskManager): + async def setup(self, task_manager: TaskManager, watchdog_timers_enabled: bool = False): self._task_manager = task_manager async def cleanup(self): diff --git a/src/pipecat/processors/metrics/sentry.py b/src/pipecat/processors/metrics/sentry.py index ac8ca2095..083ff621b 100644 --- a/src/pipecat/processors/metrics/sentry.py +++ b/src/pipecat/processors/metrics/sentry.py @@ -32,10 +32,10 @@ class SentryMetrics(WatchdogReseter, FrameProcessorMetrics): logger.warning("Sentry SDK not initialized. Sentry features will be disabled.") self._sentry_task = None - async def setup(self, task_manager: TaskManager): - await super().setup(task_manager) + async def setup(self, task_manager: TaskManager, watchdog_timers_enabled: bool = False): + await super().setup(task_manager, watchdog_timers_enabled) if self._sentry_available: - self._sentry_queue = WatchdogQueue(self) + self._sentry_queue = WatchdogQueue(self, watchdog_enabled=watchdog_timers_enabled) self._sentry_task = self.task_manager.create_task( self._sentry_task_handler(), name=f"{self}::_sentry_task_handler" ) diff --git a/src/pipecat/processors/producer_processor.py b/src/pipecat/processors/producer_processor.py index 6dd381a51..ad08802e2 100644 --- a/src/pipecat/processors/producer_processor.py +++ b/src/pipecat/processors/producer_processor.py @@ -44,7 +44,7 @@ class ProducerProcessor(FrameProcessor): Returns: asyncio.Queue: The queue for the newly added consumer. """ - queue = WatchdogQueue(consumer) + queue = WatchdogQueue(consumer, watchdog_enabled=self.watchdog_timers_enabled) self._consumers.append(queue) return queue diff --git a/src/pipecat/services/anthropic/llm.py b/src/pipecat/services/anthropic/llm.py index 0d92e8b30..b5334c383 100644 --- a/src/pipecat/services/anthropic/llm.py +++ b/src/pipecat/services/anthropic/llm.py @@ -204,7 +204,9 @@ class AnthropicLLMService(LLMService): json_accumulator = "" function_calls = [] - async for event in WatchdogAsyncIterator(response, reseter=self): + async for event in WatchdogAsyncIterator( + response, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): # Aggregate streaming content, create frames, trigger events if event.type == "content_block_delta": diff --git a/src/pipecat/services/cartesia/tts.py b/src/pipecat/services/cartesia/tts.py index 30f5c0754..8ac997f27 100644 --- a/src/pipecat/services/cartesia/tts.py +++ b/src/pipecat/services/cartesia/tts.py @@ -256,7 +256,9 @@ class CartesiaTTSService(AudioContextWordTTSService): self._context_id = None async def _receive_messages(self): - async for message in WatchdogAsyncIterator(self._get_websocket(), reseter=self): + async for message in WatchdogAsyncIterator( + self._get_websocket(), reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): msg = json.loads(message) if not msg or not self.audio_context_available(msg["context_id"]): continue diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index d0bd29f5b..fdb7bf1a8 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -395,7 +395,9 @@ class ElevenLabsTTSService(AudioContextWordTTSService): self._started = False async def _receive_messages(self): - async for message in WatchdogAsyncIterator(self._get_websocket(), reseter=self): + async for message in WatchdogAsyncIterator( + self._get_websocket(), reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): msg = json.loads(message) received_ctx_id = msg.get("contextId") diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 44829500f..c713c3cab 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -687,7 +687,9 @@ class GeminiMultimodalLiveLLMService(LLMService): # async def _receive_task_handler(self): - async for message in WatchdogAsyncIterator(self._websocket, reseter=self): + async for message in WatchdogAsyncIterator( + self._websocket, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): evt = events.parse_server_event(message) # logger.debug(f"Received event: {message[:500]}") # logger.debug(f"Received event: {evt}") diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py index ef73c9c97..a21c26ad5 100644 --- a/src/pipecat/services/gladia/stt.py +++ b/src/pipecat/services/gladia/stt.py @@ -502,7 +502,9 @@ class GladiaSTTService(STTService): async def _receive_task_handler(self): try: - async for message in WatchdogAsyncIterator(self._websocket, reseter=self): + async for message in WatchdogAsyncIterator( + self._websocket, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): content = json.loads(message) # Handle audio chunk acknowledgments diff --git a/src/pipecat/services/google/llm.py b/src/pipecat/services/google/llm.py index 38fba45b9..5fe005fbd 100644 --- a/src/pipecat/services/google/llm.py +++ b/src/pipecat/services/google/llm.py @@ -558,7 +558,9 @@ class GoogleLLMService(LLMService): ) function_calls = [] - async for chunk in WatchdogAsyncIterator(response, reseter=self): + async for chunk in WatchdogAsyncIterator( + response, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): # Stop TTFB metrics after the first chunk await self.stop_ttfb_metrics() if chunk.usage_metadata: diff --git a/src/pipecat/services/google/llm_openai.py b/src/pipecat/services/google/llm_openai.py index e7fcfdb0f..e76ac1886 100644 --- a/src/pipecat/services/google/llm_openai.py +++ b/src/pipecat/services/google/llm_openai.py @@ -54,7 +54,9 @@ class GoogleLLMOpenAIBetaService(OpenAILLMService): context ) - async for chunk in WatchdogAsyncIterator(chunk_stream, reseter=self): + async for chunk in WatchdogAsyncIterator( + chunk_stream, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): if chunk.usage: tokens = LLMTokenUsage( prompt_tokens=chunk.usage.prompt_tokens, diff --git a/src/pipecat/services/google/stt.py b/src/pipecat/services/google/stt.py index c87c857e5..80f061b44 100644 --- a/src/pipecat/services/google/stt.py +++ b/src/pipecat/services/google/stt.py @@ -784,7 +784,9 @@ class GoogleSTTService(STTService): async def _process_responses(self, streaming_recognize): """Process streaming recognition responses.""" try: - async for response in WatchdogAsyncIterator(streaming_recognize, reseter=self): + async for response in WatchdogAsyncIterator( + streaming_recognize, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): # Check streaming limit if (int(time.time() * 1000) - self._stream_start_time) > self.STREAMING_LIMIT: logger.debug("Stream timeout reached in response processing") diff --git a/src/pipecat/services/openai/base_llm.py b/src/pipecat/services/openai/base_llm.py index 9651e0f99..f0029ad22 100644 --- a/src/pipecat/services/openai/base_llm.py +++ b/src/pipecat/services/openai/base_llm.py @@ -193,7 +193,9 @@ class BaseOpenAILLMService(LLMService): context ) - async for chunk in WatchdogAsyncIterator(chunk_stream, reseter=self): + async for chunk in WatchdogAsyncIterator( + chunk_stream, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): if chunk.usage: tokens = LLMTokenUsage( prompt_tokens=chunk.usage.prompt_tokens, diff --git a/src/pipecat/services/openai_realtime_beta/openai.py b/src/pipecat/services/openai_realtime_beta/openai.py index 49e3383e7..8d5168c70 100644 --- a/src/pipecat/services/openai_realtime_beta/openai.py +++ b/src/pipecat/services/openai_realtime_beta/openai.py @@ -370,7 +370,9 @@ class OpenAIRealtimeBetaLLMService(LLMService): # async def _receive_task_handler(self): - async for message in WatchdogAsyncIterator(self._websocket, reseter=self): + async for message in WatchdogAsyncIterator( + self._websocket, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): evt = events.parse_server_event(message) if evt.type == "session.created": await self._handle_evt_session_created(evt) diff --git a/src/pipecat/services/riva/stt.py b/src/pipecat/services/riva/stt.py index 16d9528c5..284252adb 100644 --- a/src/pipecat/services/riva/stt.py +++ b/src/pipecat/services/riva/stt.py @@ -199,7 +199,9 @@ class RivaSTTService(STTService): self._thread_task = self.create_task(self._thread_task_handler()) if not self._response_task: - self._response_queue = WatchdogQueue(self) + self._response_queue = WatchdogQueue( + self, watchdog_enabled=self.watchdog_timers_enabled + ) self._response_task = self.create_task(self._response_task_handler()) async def stop(self, frame: EndFrame): diff --git a/src/pipecat/services/sambanova/llm.py b/src/pipecat/services/sambanova/llm.py index 6c44215a0..3ca2ee5be 100644 --- a/src/pipecat/services/sambanova/llm.py +++ b/src/pipecat/services/sambanova/llm.py @@ -95,7 +95,9 @@ class SambaNovaLLMService(OpenAILLMService): # type: ignore context ) - async for chunk in WatchdogAsyncIterator(chunk_stream, reseter=self): + async for chunk in WatchdogAsyncIterator( + chunk_stream, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): if chunk.usage: tokens = LLMTokenUsage( prompt_tokens=chunk.usage.prompt_tokens, diff --git a/src/pipecat/services/simli/video.py b/src/pipecat/services/simli/video.py index 19774ab4f..2bca697cb 100644 --- a/src/pipecat/services/simli/video.py +++ b/src/pipecat/services/simli/video.py @@ -63,7 +63,9 @@ class SimliVideoService(FrameProcessor): async def _consume_and_process_audio(self): await self._pipecat_resampler_event.wait() audio_iterator = self._simli_client.getAudioStreamIterator() - async for audio_frame in WatchdogAsyncIterator(audio_iterator, reseter=self): + async for audio_frame in WatchdogAsyncIterator( + audio_iterator, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): resampled_frames = self._pipecat_resampler.resample(audio_frame) for resampled_frame in resampled_frames: audio_array = resampled_frame.to_ndarray() @@ -80,7 +82,9 @@ class SimliVideoService(FrameProcessor): async def _consume_and_process_video(self): await self._pipecat_resampler_event.wait() video_iterator = self._simli_client.getVideoStreamIterator(targetFormat="rgb24") - async for video_frame in WatchdogAsyncIterator(video_iterator, reseter=self): + async for video_frame in WatchdogAsyncIterator( + video_iterator, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): # Process the video frame convertedFrame: OutputImageRawFrame = OutputImageRawFrame( image=video_frame.to_rgb().to_image().tobytes(), diff --git a/src/pipecat/services/tavus/video.py b/src/pipecat/services/tavus/video.py index 41202d0e5..9688aaebc 100644 --- a/src/pipecat/services/tavus/video.py +++ b/src/pipecat/services/tavus/video.py @@ -72,7 +72,6 @@ class TavusVideoService(AIService): self._resampler = create_default_resampler() self._audio_buffer = bytearray() - self._queue = WatchdogQueue(self) self._send_task: Optional[asyncio.Task] = None # This is the custom track destination expected by Tavus self._transport_destination: Optional[str] = "stream" @@ -189,7 +188,7 @@ class TavusVideoService(AIService): async def _create_send_task(self): if not self._send_task: - self._queue = WatchdogQueue(self) + self._queue = WatchdogQueue(self, watchdog_enabled=self.watchdog_timers_enabled) self._send_task = self.create_task(self._send_task_handler()) async def _cancel_send_task(self): diff --git a/src/pipecat/services/tts_service.py b/src/pipecat/services/tts_service.py index a623c9531..2c3bb7b61 100644 --- a/src/pipecat/services/tts_service.py +++ b/src/pipecat/services/tts_service.py @@ -330,7 +330,6 @@ class WordTTSService(TTSService): def __init__(self, **kwargs): super().__init__(**kwargs) self._initial_word_timestamp = -1 - self._words_queue = WatchdogQueue(self) self._words_task = None self._llm_response_started: bool = False @@ -372,6 +371,7 @@ class WordTTSService(TTSService): def _create_words_task(self): if not self._words_task: + self._words_queue = WatchdogQueue(self, watchdog_enabled=self.watchdog_timers_enabled) self._words_task = self.create_task(self._words_task_handler()) async def _stop_words_task(self): @@ -581,7 +581,9 @@ class AudioContextWordTTSService(WebsocketWordTTSService): def _create_audio_context_task(self): if not self._audio_context_task: - self._contexts_queue = WatchdogQueue(self) + self._contexts_queue = WatchdogQueue( + self, watchdog_enabled=self.watchdog_timers_enabled + ) self._contexts: Dict[str, asyncio.Queue] = {} self._audio_context_task = self.create_task(self._audio_context_task_handler()) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index 2b37e7ddf..f477be447 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -602,7 +602,9 @@ class BaseOutputTransport(FrameProcessor): def _create_clock_task(self): if not self._clock_task: - self._clock_queue = WatchdogPriorityQueue(self._transport) + self._clock_queue = WatchdogPriorityQueue( + self._transport, watchdog_enabled=self._transport.watchdog_timers_enabled + ) self._clock_task = self._transport.create_task(self._clock_task_handler()) async def _cancel_clock_task(self): diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index 710c715a9..3d19cb05f 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/src/pipecat/transports/network/fastapi_websocket.py @@ -179,7 +179,9 @@ class FastAPIWebsocketInputTransport(BaseInputTransport): async def _receive_messages(self): try: - async for message in WatchdogAsyncIterator(self._client.receive(), reseter=self): + async for message in WatchdogAsyncIterator( + self._client.receive(), reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): if not self._params.serializer: continue diff --git a/src/pipecat/transports/network/small_webrtc.py b/src/pipecat/transports/network/small_webrtc.py index 48ca9d53f..086aa383c 100644 --- a/src/pipecat/transports/network/small_webrtc.py +++ b/src/pipecat/transports/network/small_webrtc.py @@ -424,7 +424,9 @@ class SmallWebRTCInputTransport(BaseInputTransport): async def _receive_audio(self): try: audio_iterator = self._client.read_audio_frame() - async for audio_frame in WatchdogAsyncIterator(audio_iterator, reseter=self): + async for audio_frame in WatchdogAsyncIterator( + audio_iterator, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): if audio_frame: await self.push_audio_frame(audio_frame) @@ -434,7 +436,9 @@ class SmallWebRTCInputTransport(BaseInputTransport): async def _receive_video(self): try: video_iterator = self._client.read_video_frame() - async for video_frame in WatchdogAsyncIterator(video_iterator, reseter=self): + async for video_frame in WatchdogAsyncIterator( + video_iterator, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): if video_frame: await self.push_video_frame(video_frame) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 61afcb6c9..f92d632ca 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -306,6 +306,7 @@ class DailyTransportClient(WatchdogReseter, EventHandler): self._leave_counter = 0 self._task_manager: Optional[BaseTaskManager] = None + self._watchdog_timers_enabled = False # We use the executor to cleanup the client. We just do it from one # place, so only one thread is really needed. @@ -322,9 +323,6 @@ class DailyTransportClient(WatchdogReseter, EventHandler): # waits for it to finish using completions (and a future) we will # deadlock because completions use event handlers (which are holding the # GIL). - self._event_queue = WatchdogQueue(self) - self._audio_queue = WatchdogQueue(self) - self._video_queue = WatchdogQueue(self) self._event_task = None self._audio_task = None self._video_task = None @@ -406,6 +404,9 @@ class DailyTransportClient(WatchdogReseter, EventHandler): return self._task_manager = setup.task_manager + self._watchdog_timers_enabled = setup.watchdog_timers_enabled + + self._event_queue = WatchdogQueue(self, watchdog_enabled=self._watchdog_timers_enabled) self._event_task = self._task_manager.create_task( self._callback_task_handler(self._event_queue), f"{self}::event_callback_task", @@ -430,12 +431,14 @@ class DailyTransportClient(WatchdogReseter, EventHandler): self._out_sample_rate = self._params.audio_out_sample_rate or frame.audio_out_sample_rate if self._params.audio_in_enabled and not self._audio_task and self._task_manager: + self._audio_queue = WatchdogQueue(self, watchdog_enabled=self._watchdog_timers_enabled) self._audio_task = self._task_manager.create_task( self._callback_task_handler(self._audio_queue), f"{self}::audio_callback_task", ) if self._params.video_in_enabled and not self._video_task and self._task_manager: + self._video_queue = WatchdogQueue(self, watchdog_enabled=self._watchdog_timers_enabled) self._video_task = self._task_manager.create_task( self._callback_task_handler(self._video_queue), f"{self}::video_callback_task", diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index 5a4c38069..67ea6b32a 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -416,7 +416,9 @@ class LiveKitInputTransport(BaseInputTransport): async def _audio_in_task_handler(self): logger.info("Audio input task started") audio_iterator = self._client.get_next_audio_frame() - async for audio_data in WatchdogAsyncIterator(audio_iterator, reseter=self): + async for audio_data in WatchdogAsyncIterator( + audio_iterator, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): if audio_data: audio_frame_event, participant_id = audio_data pipecat_audio_frame = await self._convert_livekit_audio_to_pipecat( diff --git a/src/pipecat/utils/asyncio.py b/src/pipecat/utils/asyncio.py index a4b286273..1d1ed36e6 100644 --- a/src/pipecat/utils/asyncio.py +++ b/src/pipecat/utils/asyncio.py @@ -18,6 +18,7 @@ WATCHDOG_TIMEOUT = 5.0 @dataclass class TaskManagerParams: loop: asyncio.AbstractEventLoop + enable_watchdog_timers: bool = False enable_watchdog_logging: bool = False watchdog_timeout: float = WATCHDOG_TIMEOUT @@ -255,6 +256,9 @@ class TaskManager(BaseTaskManager): will be logged indicating the task is stalling. """ + if self._params and not self._params.enable_watchdog_timers: + return + name = task.get_name() if name in self._tasks: self._tasks[name].watchdog_timer.set() @@ -264,8 +268,11 @@ class TaskManager(BaseTaskManager): def _add_task(self, task_data: TaskData): name = task_data.task.get_name() self._tasks[name] = task_data - watchdog_task = self.get_event_loop().create_task(self._watchdog_task_handler(task_data)) - task_data.watchdog_task = watchdog_task + if self._params and self._params.enable_watchdog_timers: + watchdog_task = self.get_event_loop().create_task( + self._watchdog_task_handler(task_data) + ) + task_data.watchdog_task = watchdog_task async def _remove_task(self, task: asyncio.Task): name = task.get_name() diff --git a/src/pipecat/utils/watchdog_async_iterator.py b/src/pipecat/utils/watchdog_async_iterator.py index db126a363..62b6d1a23 100644 --- a/src/pipecat/utils/watchdog_async_iterator.py +++ b/src/pipecat/utils/watchdog_async_iterator.py @@ -17,12 +17,20 @@ class WatchdogAsyncIterator: """ - def __init__(self, async_iterable, *, reseter: WatchdogReseter, timeout: float = 2.0): + def __init__( + self, + async_iterable, + *, + reseter: WatchdogReseter, + timeout: float = 2.0, + watchdog_enabled: bool = False, + ): self._async_iterable = async_iterable self._reseter = reseter self._timeout = timeout self._iter: Optional[AsyncIterator] = None self._current_anext_task: Optional[asyncio.Task] = None + self._watchdog_enabled = watchdog_enabled def __aiter__(self): return self @@ -31,6 +39,12 @@ class WatchdogAsyncIterator: if not self._iter: self._iter = await self._ensure_async_iterator(self._async_iterable) + if self._watchdog_enabled: + return await self._watchdog_anext() + else: + return await self._iter.__anext__() + + async def _watchdog_anext(self): while True: try: if not self._current_anext_task: diff --git a/src/pipecat/utils/watchdog_event.py b/src/pipecat/utils/watchdog_event.py index 3165c31df..001a0cf26 100644 --- a/src/pipecat/utils/watchdog_event.py +++ b/src/pipecat/utils/watchdog_event.py @@ -16,12 +16,25 @@ class WatchdogEvent(asyncio.Event): """ - def __init__(self, reseter: WatchdogReseter, timeout: float = 2.0) -> None: + def __init__( + self, + reseter: WatchdogReseter, + *, + timeout: float = 2.0, + watchdog_enabled: bool = False, + ) -> None: super().__init__() self._reseter = reseter self._timeout = timeout + self._watchdog_enabled = watchdog_enabled async def wait(self): + if self._watchdog_enabled: + return await self._watchdog_wait() + else: + return await super().wait() + + async def _watchdog_wait(self): while True: try: await asyncio.wait_for(super().wait(), timeout=self._timeout) diff --git a/src/pipecat/utils/watchdog_priority_queue.py b/src/pipecat/utils/watchdog_priority_queue.py index 34782fbf8..a3635667c 100644 --- a/src/pipecat/utils/watchdog_priority_queue.py +++ b/src/pipecat/utils/watchdog_priority_queue.py @@ -16,12 +16,31 @@ class WatchdogPriorityQueue(asyncio.PriorityQueue): """ - def __init__(self, reseter: WatchdogReseter, maxsize: int = 0, timeout: float = 2.0) -> None: + def __init__( + self, + reseter: WatchdogReseter, + *, + maxsize: int = 0, + timeout: float = 2.0, + watchdog_enabled: bool = False, + ) -> None: super().__init__(maxsize) self._reseter = reseter self._timeout = timeout + self._watchdog_enabled = watchdog_enabled async def get(self): + if self._watchdog_enabled: + return await self._watchdog_get() + else: + return await super().get() + + def task_done(self): + if self._watchdog_enabled: + self._reseter.reset_watchdog() + super().task_done() + + async def _watchdog_get(self): while True: try: item = await asyncio.wait_for(super().get(), timeout=self._timeout) @@ -29,7 +48,3 @@ class WatchdogPriorityQueue(asyncio.PriorityQueue): return item except asyncio.TimeoutError: self._reseter.reset_watchdog() - - def task_done(self): - self._reseter.reset_watchdog() - super().task_done() diff --git a/src/pipecat/utils/watchdog_queue.py b/src/pipecat/utils/watchdog_queue.py index 0d9ffd7a8..6eb9dab10 100644 --- a/src/pipecat/utils/watchdog_queue.py +++ b/src/pipecat/utils/watchdog_queue.py @@ -16,12 +16,31 @@ class WatchdogQueue(asyncio.Queue): """ - def __init__(self, reseter: WatchdogReseter, maxsize: int = 0, timeout: float = 2.0) -> None: + def __init__( + self, + reseter: WatchdogReseter, + *, + maxsize: int = 0, + timeout: float = 2.0, + watchdog_enabled: bool = False, + ) -> None: super().__init__(maxsize) self._reseter = reseter self._timeout = timeout + self._watchdog_enabled = watchdog_enabled async def get(self): + if self._watchdog_enabled: + return await self._watchdog_get() + else: + return await super().get() + + def task_done(self): + if self._watchdog_enabled: + self._reseter.reset_watchdog() + super().task_done() + + async def _watchdog_get(self): while True: try: item = await asyncio.wait_for(super().get(), timeout=self._timeout) @@ -29,7 +48,3 @@ class WatchdogQueue(asyncio.Queue): return item except asyncio.TimeoutError: self._reseter.reset_watchdog() - - def task_done(self): - self._reseter.reset_watchdog() - super().task_done()