diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 56df719d5..4b471c08e 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -57,7 +57,7 @@ from pipecat.utils.tracing.tracing_context import TracingContext from pipecat.utils.tracing.turn_trace_observer import TurnTraceObserver HEARTBEAT_SECS = 1.0 -HEARTBEAT_MONITOR_SECS = HEARTBEAT_SECS * 10 +HEARTBEAT_MONITOR_SECS = 10.0 IDLE_TIMEOUT_SECS = 300 @@ -123,6 +123,8 @@ class PipelineParams(BaseModel): enable_metrics: Whether to enable metrics collection. enable_usage_metrics: Whether to enable usage metrics. heartbeats_period_secs: Period between heartbeats in seconds. + heartbeats_monitor_secs: Timeout (in seconds) before warning about + missed heartbeats. Defaults to 10 seconds. interruption_strategies: [deprecated] Strategies for bot interruption behavior. .. deprecated:: 0.0.99 @@ -147,6 +149,7 @@ class PipelineParams(BaseModel): enable_metrics: bool = False enable_usage_metrics: bool = False heartbeats_period_secs: float = HEARTBEAT_SECS + heartbeats_monitor_secs: float = HEARTBEAT_MONITOR_SECS interruption_strategies: List[BaseInterruptionStrategy] = Field(default_factory=list) observers: List[BaseObserver] = Field(default_factory=list) report_only_initial_ttfb: bool = False @@ -964,7 +967,7 @@ class PipelineTask(BasePipelineTask): the time that a heartbeat frame takes to processes, that is how long it takes for the heartbeat frame to traverse all the pipeline. """ - wait_time = HEARTBEAT_MONITOR_SECS + wait_time = self._params.heartbeats_monitor_secs while True: try: frame = await asyncio.wait_for(self._heartbeat_queue.get(), timeout=wait_time) diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 04601bf14..de62d9739 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -5,9 +5,12 @@ # import asyncio +import io import time import unittest +from loguru import logger + from pipecat.frames.frames import ( CancelFrame, EndFrame, @@ -383,6 +386,45 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): pass assert heartbeats_counter == expected_heartbeats + async def test_heartbeat_monitor_respects_custom_timeout(self): + """Verify the heartbeat monitor uses heartbeats_monitor_secs from params.""" + + class HeartbeatBlocker(FrameProcessor): + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if not isinstance(frame, HeartbeatFrame): + await self.push_frame(frame, direction) + + log_output = io.StringIO() + handler_id = logger.add(log_output, level="WARNING", format="{message}") + + custom_monitor_secs = 0.3 + + try: + pipeline = Pipeline([HeartbeatBlocker()]) + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_heartbeats=True, + heartbeats_period_secs=0.1, + heartbeats_monitor_secs=custom_monitor_secs, + ), + cancel_on_idle_timeout=False, + ) + + try: + await asyncio.wait_for( + task.run(PipelineTaskParams(loop=asyncio.get_event_loop())), + timeout=0.6, + ) + except asyncio.TimeoutError: + pass + + log_text = log_output.getvalue() + assert f"more than {custom_monitor_secs} seconds" in log_text + finally: + logger.remove(handler_id) + async def test_idle_task(self): identity = IdentityFilter() pipeline = Pipeline([identity])