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

Start HeartBeat when all processors have processed StartFrame

Browse Source 
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.
This commit is contained in:
Abhishek
2025-06-26 23:28:37 +05:30
committed by GitHub
parent 4d34aa7cd6
commit 89c801f82c
1 changed files with 5 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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