From 89c801f82c81d3cd36c449d43ca6bd31073ca145 Mon Sep 17 00:00:00 2001 From: Abhishek Date: Thu, 26 Jun 2025 23:28:37 +0530 Subject: [PATCH] Start HeartBeat when all processors have processed StartFrame Some of the processors like STTService and TTSService don't push StartFrame ahead in the pipeline, unless they have connected with their service providers. This delays StartFrame in downstream processors. If we receive HeartBeat frame before StartFrame, we will get AttributeError `'Processor' object has no attribute '_FrameProcessor__input_queue'`. Idea is to start HeartBeats after StartFrame has been processed by all the Processors in the pipeline. --- src/pipecat/pipeline/task.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 8b30d1a4e..62d8ccf71 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -472,7 +472,7 @@ class PipelineTask(WatchdogReseter, BasePipelineTask): return self._process_push_task def _maybe_start_heartbeat_tasks(self): - if self._params.enable_heartbeats: + if self._params.enable_heartbeats and self._heartbeat_push_task is None: self._heartbeat_push_task = self._task_manager.create_task( self._heartbeat_push_handler(), f"{self}::_heartbeat_push_handler" ) @@ -570,7 +570,6 @@ class PipelineTask(WatchdogReseter, BasePipelineTask): """ self._clock.start() - self._maybe_start_heartbeat_tasks() self._maybe_start_idle_task() start_frame = StartFrame( @@ -652,6 +651,10 @@ class PipelineTask(WatchdogReseter, BasePipelineTask): if isinstance(frame, StartFrame): await self._call_event_handler("on_pipeline_started", frame) + + # Start heartbeat tasks now that StartFrame has been processed + # by all processors in the pipeline + self._maybe_start_heartbeat_tasks() elif isinstance(frame, EndFrame): await self._call_event_handler("on_pipeline_ended", frame) self._pipeline_end_event.set()