diff --git a/CHANGELOG.md b/CHANGELOG.md index 6efebf31a..bd2cda2dc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -231,6 +231,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed a `PipelineTask` issue that was causing an idle timeout for frames that + were being generated but not reaching the end of the pipeline. Since the exact + point when frames are discarded is unknown, we now monitor pipeline frames + using an observer. If the observer detects frames are being generated, it will + prevent the pipeline from being considered idle. + - Fixed an issue in `HumeTTSService` that was only using Octave 2, which does not support the `description` field. Now, if a description is provided, it switches to Octave 1. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index a511db12a..1f9cbf63a 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -12,7 +12,6 @@ including heartbeats, idle detection, and observer integration. """ import asyncio -import time from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Tuple, Type from loguru import logger @@ -39,7 +38,7 @@ from pipecat.frames.frames import ( UserSpeakingFrame, ) from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData -from pipecat.observers.base_observer import BaseObserver +from pipecat.observers.base_observer import BaseObserver, FramePushed from pipecat.observers.turn_tracking_observer import TurnTrackingObserver from pipecat.pipeline.base_task import BasePipelineTask, PipelineTaskParams from pipecat.pipeline.pipeline import Pipeline, PipelineSink, PipelineSource @@ -57,6 +56,43 @@ IDLE_TIMEOUT_SECS = 300 CANCEL_TIMEOUT_SECS = 20.0 +class IdleFrameObserver(BaseObserver): + """Idle timeout observer. + + This observer waits for specific frames being generated in the pipeline. If + the frames are generated the given asyncio event is set. If the event is not + set it means the pipeline is probably idle. + + """ + + def __init__(self, *, idle_event: asyncio.Event, idle_timeout_frames: Tuple[Type[Frame], ...]): + """Initialize the observer. + + Args: + idle_event: The event to set if the idle timeout frames are being pushed. + idle_timeout_frames: A tuple with the frames that should set the event when received + """ + super().__init__() + self._idle_event = idle_event + self._idle_timeout_frames = idle_timeout_frames + self._processed_frames = set() + + async def on_push_frame(self, data: FramePushed): + """Callback executed when a frame is pushed in the pipeline. + + Args: + data: The frame push event data. + """ + # Skip already processed frames + if data.frame.id in self._processed_frames: + return + + self._processed_frames.add(data.frame.id) + + if isinstance(data.frame, StartFrame) or isinstance(data.frame, self._idle_timeout_frames): + self._idle_event.set() + + class PipelineParams(BaseModel): """Configuration parameters for pipeline execution. @@ -215,7 +251,6 @@ class PipelineTask(BasePipelineTask): self._conversation_id = conversation_id self._enable_tracing = enable_tracing and is_tracing_available() self._enable_turn_tracking = enable_turn_tracking - self._idle_timeout_frames = idle_timeout_frames self._idle_timeout_secs = idle_timeout_secs if self._params.observers: import warnings @@ -250,16 +285,24 @@ class PipelineTask(BasePipelineTask): # This queue is the queue used to push frames to the pipeline. self._push_queue = asyncio.Queue() 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 = asyncio.Queue() 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 = asyncio.Queue() + + # This is the idle event. When selected frames are pushed from any + # processor we consider the pipeline is not idle. We use an observer + # which will be listening any part of the pipeline. + self._idle_event = asyncio.Event() self._idle_monitor_task: Optional[asyncio.Task] = None + if self._idle_timeout_secs: + idle_frame_observer = IdleFrameObserver( + idle_event=self._idle_event, + idle_timeout_frames=idle_timeout_frames, + ) + observers.append(idle_frame_observer) # This event is used to indicate the StartFrame has been received at the # end of the pipeline. @@ -530,7 +573,7 @@ class PipelineTask(BasePipelineTask): async def _maybe_cancel_idle_task(self): """Cancel idle monitoring task if it is running.""" - if self._idle_timeout_secs and self._idle_monitor_task: + if self._idle_monitor_task: await self._task_manager.cancel_task(self._idle_monitor_task) self._idle_monitor_task = None @@ -706,10 +749,6 @@ class PipelineTask(BasePipelineTask): processors have handled the EndFrame and therefore we can exit the task cleanly. """ - # Queue received frame to the idle queue so we can monitor idle - # pipelines. - await self._idle_queue.put(frame) - if isinstance(frame, self._reached_downstream_types): await self._call_event_handler("on_frame_reached_downstream", frame) @@ -772,33 +811,10 @@ class PipelineTask(BasePipelineTask): Note: Heartbeats are excluded from idle detection. """ running = True - last_frame_time = 0 - while running: try: - frame = await asyncio.wait_for( - self._idle_queue.get(), timeout=self._idle_timeout_secs - ) - - if isinstance(frame, StartFrame) or isinstance(frame, self._idle_timeout_frames): - # If we find a StartFrame or one of the frames that prevents a - # time out we update the time. - last_frame_time = time.time() - else: - # If we find any other frame we check if the pipeline is - # idle by checking the last time we received one of the - # valid frames. - diff_time = time.time() - last_frame_time - if diff_time >= self._idle_timeout_secs: - running = await self._idle_timeout_detected() - # Reset `last_frame_time` so we don't trigger another - # immediate idle timeout if we are not cancelling. For - # example, we might want to force the bot to say goodbye - # and then clean nicely with an `EndFrame`. - last_frame_time = time.time() - - self._idle_queue.task_done() - + await asyncio.wait_for(self._idle_event.wait(), timeout=self._idle_timeout_secs) + self._idle_event.clear() except asyncio.TimeoutError: running = await self._idle_timeout_detected() @@ -810,7 +826,7 @@ class PipelineTask(BasePipelineTask): """ # If we are cancelling, just exit the task. if self._cancelled: - return True + return False logger.warning("Idle timeout detected.") await self._call_event_handler("on_idle_timeout") diff --git a/src/pipecat/pipeline/task_observer.py b/src/pipecat/pipeline/task_observer.py index 98ff7c91e..11aef58cb 100644 --- a/src/pipecat/pipeline/task_observer.py +++ b/src/pipecat/pipeline/task_observer.py @@ -129,7 +129,7 @@ class TaskObserver(BaseObserver): for proxy in self._proxies: await proxy.cleanup() - async def on_process_frame(self, data: FramePushed): + async def on_process_frame(self, data: FrameProcessed): """Queue frame data for all managed observers. Args: diff --git a/src/pipecat/tests/utils.py b/src/pipecat/tests/utils.py index c92eb6309..6ccce4b31 100644 --- a/src/pipecat/tests/utils.py +++ b/src/pipecat/tests/utils.py @@ -8,7 +8,7 @@ import asyncio from dataclasses import dataclass -from typing import Any, Awaitable, Callable, Dict, List, Optional, Sequence, Tuple +from typing import Awaitable, Callable, List, Optional, Sequence, Tuple from pipecat.frames.frames import ( EndFrame, diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 7d9ebec6f..3c7f50453 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -24,6 +24,7 @@ from pipecat.pipeline.base_task import PipelineTaskParams from pipecat.pipeline.parallel_pipeline import ParallelPipeline from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.filters.frame_filter import FrameFilter from pipecat.processors.filters.identity_filter import IdentityFilter from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.tests.utils import HeartbeatsObserver, run_test @@ -383,6 +384,7 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): idle_timeout_secs = 0.2 sleep_time_secs = idle_timeout_secs / 2 + # Use the identify filter so the frames just reach the end of the pipeline. identity = IdentityFilter() pipeline = Pipeline([identity]) task = PipelineTask( @@ -392,6 +394,12 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): ) async def delayed_frames(): + """Sending multiple text frames. + + The total amount of elapsed time in this function should be greater + than the task idle timeout. If an idle timeout event is triggered it + means we haven't detected that the TextFrames have been pushed. + """ await asyncio.sleep(sleep_time_secs) await task.queue_frame(TextFrame("Hello Pipecat!")) await asyncio.sleep(sleep_time_secs) @@ -415,6 +423,51 @@ class TestPipelineTask(unittest.IsolatedAsyncioTestCase): # Wait for the pending tasks to complete. await asyncio.gather(*pending) + async def test_idle_task_swallowed_frames(self): + idle_timeout_secs = 0.2 + sleep_time_secs = idle_timeout_secs / 2 + + # Block all frames (except system frames). Here, we are testing that + # generated frames don't trigger an idle timeout (they don't need to + # reach the end of the pipeline). + filter = FrameFilter(types=()) + pipeline = Pipeline([filter]) + task = PipelineTask( + pipeline, + idle_timeout_secs=idle_timeout_secs, + idle_timeout_frames=(TextFrame,), + ) + + start_time = time.time() + + async def delayed_frames(): + """Sending multiple text frames. + + The total amount of elapsed time in this function should be greater + than the task idle timeout. If an idle timeout event is triggered it + means we haven't detected that the TextFrames have been pushed. + """ + await asyncio.sleep(sleep_time_secs) + await task.queue_frame(TextFrame("Hello Pipecat!")) + await asyncio.sleep(sleep_time_secs) + await task.queue_frame(TextFrame("Hello Pipecat!")) + await asyncio.sleep(sleep_time_secs) + await task.queue_frame(TextFrame("Hello Pipecat!")) + + tasks = [ + asyncio.create_task(task.run(PipelineTaskParams(loop=asyncio.get_event_loop()))), + asyncio.create_task(delayed_frames()), + ] + + _, pending = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED) + + diff_time = time.time() - start_time + + self.assertGreater(diff_time, sleep_time_secs * 3) + + # Wait for the pending tasks to complete. + await asyncio.gather(*pending) + async def test_task_cancel_timeout(self): class CancelFilter(FrameProcessor): def __init__(self, **kwargs):