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

Merge pull request #4449 from pipecat-ai/aleix/base-object-task-manager

Browse Source 
Move create_task and cancel_task from FrameProcessor to BaseObject
This commit is contained in:
Aleix Conchillo Flaqué
2026-05-10 20:36:54 -07:00
committed by GitHub
parent 95bcebe774 77058b01c4
commit 71578e7151
10 changed files with 96 additions and 175 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
changelog/4449.changed.md Normal file
View File

@@ -0,0 +1 @@
- Moved `create_task`, `cancel_task`, the `task_manager` property, and `setup(task_manager)` up from `FrameProcessor` to `BaseObject`. Custom `BaseObject` subclasses (turn strategies, controllers, etc.) now inherit these methods directly instead of reimplementing the task manager wiring. Owners propagate the task manager to their child `BaseObject`s via `await child.setup(task_manager)`.

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

@@ -303,7 +303,7 @@ class PipelineTask(BasePipelineTask):
# This task maneger will handle all the asyncio tasks created by this
# PipelineTask and its frame processors.
self._task_manager = task_manager or TaskManager()
self._pipeline_task_manager = task_manager or TaskManager()
# This queue is the queue used to push frames to the pipeline.
self._push_queue = asyncio.Queue()
@@ -386,7 +386,7 @@ class PipelineTask(BasePipelineTask):
# The task observer acts as a proxy to the provided observers. This way,
# we only need to pass a single observer (using the StartFrame) which
# then just acts as a proxy.
self._observer = TaskObserver(observers=observers, task_manager=self._task_manager)
self._observer = TaskObserver(observers=observers)
# These events can be used to check which frames make it to the source
# or sink processors. Instead of calling the event handlers for every
@@ -654,32 +654,24 @@ class PipelineTask(BasePipelineTask):
async def _create_tasks(self):
"""Create and start all pipeline processing tasks."""
self._process_push_task = self._task_manager.create_task(
self._process_push_queue(), f"{self}::_process_push_queue"
)
self._process_push_task = self.create_task(self._process_push_queue())
return self._process_push_task
def _maybe_start_heartbeat_tasks(self):
"""Start heartbeat tasks if heartbeats are enabled and not already running."""
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"
)
self._heartbeat_monitor_task = self._task_manager.create_task(
self._heartbeat_monitor_handler(), f"{self}::_heartbeat_monitor_handler"
)
self._heartbeat_push_task = self.create_task(self._heartbeat_push_handler())
self._heartbeat_monitor_task = self.create_task(self._heartbeat_monitor_handler())
def _maybe_start_idle_task(self):
"""Start idle monitoring task if idle timeout is configured."""
if self._idle_timeout_secs:
self._idle_monitor_task = self._task_manager.create_task(
self._idle_monitor_handler(), f"{self}::_idle_monitor_handler"
)
self._idle_monitor_task = self.create_task(self._idle_monitor_handler())
async def _cancel_tasks(self):
"""Cancel all running pipeline tasks."""
if self._process_push_task:
await self._task_manager.cancel_task(self._process_push_task)
await self.cancel_task(self._process_push_task)
self._process_push_task = None
await self._maybe_cancel_heartbeat_tasks()
@@ -691,17 +683,17 @@ class PipelineTask(BasePipelineTask):
return
if self._heartbeat_push_task:
await self._task_manager.cancel_task(self._heartbeat_push_task)
await self.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)
await self.cancel_task(self._heartbeat_monitor_task)
self._heartbeat_monitor_task = None
async def _maybe_cancel_idle_task(self):
"""Cancel idle monitoring task if it is running."""
if self._idle_monitor_task:
await self._task_manager.cancel_task(self._idle_monitor_task)
await self.cancel_task(self._idle_monitor_task)
self._idle_monitor_task = None
def _initial_metrics_frame(self) -> MetricsFrame:
@@ -759,12 +751,14 @@ class PipelineTask(BasePipelineTask):
async def _setup(self, params: PipelineTaskParams):
"""Set up the pipeline task and all processors."""
await super().setup(self._pipeline_task_manager)
mgr_params = TaskManagerParams(loop=params.loop)
self._task_manager.setup(mgr_params)
self.task_manager.setup(mgr_params)
setup = FrameProcessorSetup(
clock=self._clock,
task_manager=self._task_manager,
task_manager=self.task_manager,
observer=self._observer,
pipeline_task=self,
# Populate the deprecated `tool_resources` field for backwards
@@ -780,6 +774,7 @@ class PipelineTask(BasePipelineTask):
await self._load_setup_files()
# Start task observer.
await self._observer.setup(self.task_manager)
await self._observer.start()
async def _cleanup(self, cleanup_pipeline: bool):
@@ -1019,7 +1014,7 @@ class PipelineTask(BasePipelineTask):
def _print_dangling_tasks(self):
"""Log any dangling tasks that haven't been properly cleaned up."""
tasks = [t.get_name() for t in self._task_manager.current_tasks()]
tasks = [t.get_name() for t in self.task_manager.current_tasks()]
if tasks:
logger.warning(f"{self} dangling tasks detected: {tasks}")

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

@@ -17,7 +17,6 @@ from typing import Any
from attr import dataclass
from pipecat.observers.base_observer import BaseObserver, FrameProcessed, FramePushed
from pipecat.utils.asyncio.task_manager import BaseTaskManager
@dataclass
@@ -62,19 +61,16 @@ class TaskObserver(BaseObserver):
self,
*,
observers: list[BaseObserver] | None = None,
task_manager: BaseTaskManager,
**kwargs,
):
"""Initialize the TaskObserver.
Args:
observers: List of observers to manage. Defaults to empty list.
task_manager: Task manager for creating and managing observer tasks.
**kwargs: Additional arguments passed to the base observer.
"""
super().__init__(**kwargs)
self._observers = observers or []
self._task_manager = task_manager
self._proxies: dict[BaseObserver, Proxy] | None = (
None # Becomes a dict after start() is called
)
@@ -106,7 +102,7 @@ class TaskObserver(BaseObserver):
# Remove the proxy so it doesn't get called anymore.
del self._proxies[observer]
# Cancel the proxy task right away.
await self._task_manager.cancel_task(proxy.task)
await self.cancel_task(proxy.task)
# Remove the observer from the list.
if observer in self._observers:
@@ -122,7 +118,7 @@ class TaskObserver(BaseObserver):
return
for proxy in self._proxies.values():
await self._task_manager.cancel_task(proxy.task)
await self.cancel_task(proxy.task)
async def cleanup(self):
"""Cleanup all proxy observers."""
@@ -157,9 +153,8 @@ class TaskObserver(BaseObserver):
def _create_proxy(self, observer: BaseObserver) -> Proxy:
"""Create a proxy for a single observer."""
queue = asyncio.Queue()
task = self._task_manager.create_task(
self._proxy_task_handler(queue, observer),
f"TaskObserver::{observer}::_proxy_task_handler",
task = self.create_task(
self._proxy_task_handler(queue, observer), f"{observer}::_proxy_task_handler"
)
proxy = Proxy(queue=queue, task=task, observer=observer)
return proxy

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

@@ -17,7 +17,7 @@ import asyncio
import dataclasses
import traceback
import warnings
from collections.abc import Awaitable, Callable, Coroutine
from collections.abc import Awaitable, Callable
from dataclasses import dataclass
from enum import Enum
from typing import (
@@ -217,9 +217,6 @@ class FrameProcessor(BaseObject):
# Clock
self._clock: BaseClock | None = None
# Task Manager
self._task_manager: BaseTaskManager | None = None
# Observer
self._observer: BaseObserver | None = None
@@ -368,20 +365,6 @@ class FrameProcessor(BaseObject):
"""
return self._report_only_initial_ttfb
@property
def task_manager(self) -> BaseTaskManager:
"""Get the task manager for this processor.
Returns:
The task manager instance.
Raises:
Exception: If the task manager is not initialized.
"""
if not self._task_manager:
raise Exception(f"{self} TaskManager is still not initialized.")
return self._task_manager
@property
def pipeline_task(self) -> PipelineTask | None:
"""Get the :class:`PipelineTask` this processor is running in.
@@ -511,43 +494,14 @@ class FrameProcessor(BaseObject):
await self.stop_processing_metrics()
await self.stop_text_aggregation_metrics()
def create_task(self, coroutine: Coroutine, name: str | None = None) -> asyncio.Task:
"""Create a new task managed by this processor.
Args:
coroutine: The coroutine to run in the task.
name: Optional name for the task.
Returns:
The created asyncio task.
"""
if name:
name = f"{self}::{name}"
else:
name = f"{self}::{coroutine.cr_code.co_name}"
return self.task_manager.create_task(coroutine, name)
async def cancel_task(self, task: asyncio.Task, timeout: float | None = 1.0):
"""Cancel a task managed by this processor.
A default timeout if 1 second is used in order to avoid potential
freezes caused by certain libraries that swallow
`asyncio.CancelledError`.
Args:
task: The task to cancel.
timeout: Optional timeout for task cancellation.
"""
await self.task_manager.cancel_task(task, timeout)
async def setup(self, setup: FrameProcessorSetup):
"""Set up the processor with required components.
Args:
setup: Configuration object containing setup parameters.
"""
await super().setup(setup.task_manager)
self._clock = setup.clock
self._task_manager = setup.task_manager
self._observer = setup.observer
self._pipeline_task = setup.pipeline_task
@@ -555,7 +509,7 @@ class FrameProcessor(BaseObject):
self.__create_input_task()
if self._metrics is not None:
await self._metrics.setup(self._task_manager)
await self._metrics.setup(self.task_manager)
async def cleanup(self):
"""Clean up processor resources."""

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

@@ -20,7 +20,6 @@ from pipecat.metrics.metrics import (
TTFBMetricsData,
TTSUsageMetricsData,
)
from pipecat.utils.asyncio.task_manager import BaseTaskManager
from pipecat.utils.base_object import BaseObject
@@ -40,36 +39,12 @@ class FrameProcessorMetrics(BaseObject):
processing times, and usage statistics.
"""
super().__init__()
self._task_manager = None
self._start_ttfb_time = 0
self._start_processing_time = 0
self._start_text_aggregation_time = 0
self._last_ttfb_time = 0
self._should_report_ttfb = True
async def setup(self, task_manager: BaseTaskManager):
"""Set up the metrics collector with a task manager.
Args:
task_manager: The task manager for handling async operations.
"""
self._task_manager = task_manager
async def cleanup(self):
"""Clean up metrics collection resources."""
await super().cleanup()
@property
def task_manager(self) -> BaseTaskManager:
"""Get the associated task manager.
Returns:
The task manager instance for async operations.
"""
if self._task_manager is None:
raise RuntimeError("task_manager not set; call setup() first")
return self._task_manager
@property
def ttfb(self) -> float | None:
"""Get the current TTFB value in seconds.

25
src/pipecat/turns/user_idle_controller.py
View File

@@ -19,7 +19,6 @@ from pipecat.frames.frames import (
UserStartedSpeakingFrame,
UserStoppedSpeakingFrame,
)
from pipecat.utils.asyncio.task_manager import BaseTaskManager
from pipecat.utils.base_object import BaseObject
@@ -63,29 +62,12 @@ class UserIdleController(BaseObject):
self._user_idle_timeout = user_idle_timeout
self._task_manager: BaseTaskManager | None = None
self._user_turn_in_progress: bool = False
self._function_calls_in_progress: int = 0
self._idle_timer_task: asyncio.Task | None = None
self._register_event_handler("on_user_turn_idle", sync=True)
@property
def task_manager(self) -> BaseTaskManager:
"""Returns the configured task manager."""
if not self._task_manager:
raise RuntimeError(f"{self} user idle controller was not properly setup")
return self._task_manager
async def setup(self, task_manager: BaseTaskManager):
"""Initialize the controller with the given task manager.
Args:
task_manager: The task manager to be associated with this instance.
"""
self._task_manager = task_manager
async def cleanup(self):
"""Cleanup the controller."""
await super().cleanup()
@@ -138,17 +120,14 @@ class UserIdleController(BaseObject):
if self._user_idle_timeout <= 0:
return
await self._cancel_idle_timer()
self._idle_timer_task = self.task_manager.create_task(
self._idle_timer_expired(),
f"{self}::idle_timer",
)
self._idle_timer_task = self.create_task(self._idle_timer_expired())
# Make sure the task is scheduled.
await asyncio.sleep(0)
async def _cancel_idle_timer(self):
"""Cancel the idle timer if running."""
if self._idle_timer_task:
await self.task_manager.cancel_task(self._idle_timer_task)
await self.cancel_task(self._idle_timer_task)
self._idle_timer_task = None
async def _idle_timer_expired(self):

17
src/pipecat/turns/user_start/base_user_turn_start_strategy.py
View File

@@ -11,7 +11,6 @@ from dataclasses import dataclass
from pipecat.frames.frames import Frame
from pipecat.processors.frame_processor import FrameDirection
from pipecat.turns.types import ProcessFrameResult
from pipecat.utils.asyncio.task_manager import BaseTaskManager
from pipecat.utils.base_object import BaseObject
@@ -72,27 +71,11 @@ class BaseUserTurnStartStrategy(BaseObject):
super().__init__(**kwargs)
self._enable_interruptions = enable_interruptions
self._enable_user_speaking_frames = enable_user_speaking_frames
self._task_manager: BaseTaskManager | None = None
self._register_event_handler("on_push_frame", sync=True)
self._register_event_handler("on_broadcast_frame", sync=True)
self._register_event_handler("on_user_turn_started", sync=True)
self._register_event_handler("on_reset_aggregation", sync=True)
@property
def task_manager(self) -> BaseTaskManager:
"""Returns the configured task manager."""
if not self._task_manager:
raise RuntimeError(f"{self} user turn start strategy was not properly setup")
return self._task_manager
async def setup(self, task_manager: BaseTaskManager):
"""Initialize the strategy with the given task manager.
Args:
task_manager: The task manager to be associated with this instance.
"""
self._task_manager = task_manager
async def cleanup(self):
"""Cleanup the strategy."""
pass

17
src/pipecat/turns/user_stop/base_user_turn_stop_strategy.py
View File

@@ -11,7 +11,6 @@ from dataclasses import dataclass
from pipecat.frames.frames import Frame
from pipecat.processors.frame_processor import FrameDirection
from pipecat.turns.types import ProcessFrameResult
from pipecat.utils.asyncio.task_manager import BaseTaskManager
from pipecat.utils.base_object import BaseObject
@@ -70,27 +69,11 @@ class BaseUserTurnStopStrategy(BaseObject):
"""
super().__init__(**kwargs)
self._enable_user_speaking_frames = enable_user_speaking_frames
self._task_manager: BaseTaskManager | None = None
self._register_event_handler("on_push_frame", sync=True)
self._register_event_handler("on_broadcast_frame", sync=True)
self._register_event_handler("on_user_turn_inference_triggered", sync=True)
self._register_event_handler("on_user_turn_stopped", sync=True)
@property
def task_manager(self) -> BaseTaskManager:
"""Returns the configured task manager."""
if not self._task_manager:
raise RuntimeError(f"{self} user turn stop strategy was not properly setup")
return self._task_manager
async def setup(self, task_manager: BaseTaskManager):
"""Initialize the strategy with the given task manager.
Args:
task_manager: The task manager to be associated with this instance.
"""
self._task_manager = task_manager
async def cleanup(self):
"""Cleanup the strategy."""
pass

18
src/pipecat/turns/user_turn_controller.py
View File

@@ -92,8 +92,6 @@ class UserTurnController(BaseObject):
self._user_turn_strategies = user_turn_strategies
self._user_turn_stop_timeout = user_turn_stop_timeout
self._task_manager: BaseTaskManager | None = None
self._user_speaking = False
self._user_turn = False
@@ -108,25 +106,17 @@ class UserTurnController(BaseObject):
self._register_event_handler("on_user_turn_stop_timeout", sync=True)
self._register_event_handler("on_reset_aggregation", sync=True)
@property
def task_manager(self) -> BaseTaskManager:
"""Returns the configured task manager."""
if not self._task_manager:
raise RuntimeError(f"{self} user turn controller was not properly setup")
return self._task_manager
async def setup(self, task_manager: BaseTaskManager):
"""Initialize the controller with the given task manager.
Args:
task_manager: The task manager to be associated with this instance.
"""
self._task_manager = task_manager
await super().setup(task_manager)
if not self._user_turn_stop_timeout_task:
self._user_turn_stop_timeout_task = self.task_manager.create_task(
self._user_turn_stop_timeout_task_handler(),
f"{self}::_user_turn_stop_timeout_task_handler",
self._user_turn_stop_timeout_task = self.create_task(
self._user_turn_stop_timeout_task_handler()
)
await self._setup_strategies()
@@ -136,7 +126,7 @@ class UserTurnController(BaseObject):
await super().cleanup()
if self._user_turn_stop_timeout_task:
await self.task_manager.cancel_task(self._user_turn_stop_timeout_task)
await self.cancel_task(self._user_turn_stop_timeout_task)
self._user_turn_stop_timeout_task = None
await self._cleanup_strategies()

66
src/pipecat/utils/base_object.py
View File

@@ -15,11 +15,13 @@ import asyncio
import inspect
import traceback
from abc import ABC
from collections.abc import Coroutine
from dataclasses import dataclass
from typing import Any
from loguru import logger
from pipecat.utils.asyncio.task_manager import BaseTaskManager
from pipecat.utils.utils import obj_count, obj_id
@@ -69,6 +71,10 @@ class BaseObject(ABC):
# event tasks still being executed.
self._event_tasks = set()
# Task manager. Populated by setup(); accessing the task_manager
# property before setup raises.
self._task_manager: BaseTaskManager | None = None
@property
def id(self) -> int:
"""Get the unique identifier for this object.
@@ -87,6 +93,66 @@ class BaseObject(ABC):
"""
return self._name
@property
def task_manager(self) -> BaseTaskManager:
"""Get the task manager for this object.
Returns:
The task manager instance.
Raises:
Exception: If the task manager is not initialized.
"""
if not self._task_manager:
raise Exception(f"{self}: TaskManager is not initialized.")
return self._task_manager
async def setup(self, task_manager: BaseTaskManager):
"""Wire the object up with a task manager.
Owners of a :class:`BaseObject` should call this on their child objects
to propagate the task manager down. Subclasses that own other
:class:`BaseObject` instances should override and forward::
async def setup(self, task_manager):
await super().setup(task_manager)
await self._child.setup(task_manager)
Args:
task_manager: The task manager to associate with this instance.
"""
self._task_manager = task_manager
def create_task(self, coroutine: Coroutine, name: str | None = None) -> asyncio.Task:
"""Create a new task managed by this object's task manager.
Args:
coroutine: The coroutine to run in the task.
name: Optional name for the task.
Returns:
The created asyncio task.
"""
if not name:
# Native coroutines expose ``cr_code``; fall back to a generic
# name for any other awaitable subtype.
cr_code = getattr(coroutine, "cr_code", None)
name = getattr(cr_code, "co_name", "task")
return self.task_manager.create_task(coroutine, f"{self}::{name}")
async def cancel_task(self, task: asyncio.Task, timeout: float | None = 1.0):
"""Cancel a task managed by this object's task manager.
A default timeout of 1 second is used in order to avoid potential
freezes caused by certain libraries that swallow
:class:`asyncio.CancelledError`.
Args:
task: The task to cancel.
timeout: Optional timeout for task cancellation.
"""
await self.task_manager.cancel_task(task, timeout)
async def cleanup(self):
"""Clean up resources and wait for running event handlers to complete.