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

watchdog timers are disabled by default use enable_watchdog_timers

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2025-06-25 13:36:52 -07:00
parent 327973657f
commit 357934a644
35 changed files with 256 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/pipecat/frames/frames.py
View File

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

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

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

35
src/pipecat/pipeline/sync_parallel_pipeline.py
View File

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

48
src/pipecat/pipeline/task.py
View File

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

9
src/pipecat/pipeline/task_observer.py
View File

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

23
src/pipecat/processors/frame_processor.py
View File

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

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

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

2
src/pipecat/processors/metrics/frame_processor_metrics.py
View File

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

6
src/pipecat/processors/metrics/sentry.py
View File

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

2
src/pipecat/processors/producer_processor.py
View File

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

4
src/pipecat/services/anthropic/llm.py
View File

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

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

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

4
src/pipecat/services/elevenlabs/tts.py
View File

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

4
src/pipecat/services/gemini_multimodal_live/gemini.py
View File

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

4
src/pipecat/services/gladia/stt.py
View File

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

4
src/pipecat/services/google/llm.py
View File

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

4
src/pipecat/services/google/llm_openai.py
View File

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

4
src/pipecat/services/google/stt.py
View File

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

4
src/pipecat/services/openai/base_llm.py
View File

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

4
src/pipecat/services/openai_realtime_beta/openai.py
View File

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

4
src/pipecat/services/riva/stt.py
View File

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

4
src/pipecat/services/sambanova/llm.py
View File

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

8
src/pipecat/services/simli/video.py
View File

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

3
src/pipecat/services/tavus/video.py
View File

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

6
src/pipecat/services/tts_service.py
View File

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

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

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

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

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

8
src/pipecat/transports/network/small_webrtc.py
View File

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

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

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

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

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

11
src/pipecat/utils/asyncio.py
View File

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

16
src/pipecat/utils/watchdog_async_iterator.py
View File

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

15
src/pipecat/utils/watchdog_event.py
View File

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

25
src/pipecat/utils/watchdog_priority_queue.py
View File

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

25
src/pipecat/utils/watchdog_queue.py
View File

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