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

Wire TaskObserver via setup() instead of constructor

Browse Source 
TaskObserver previously took a TaskManager in __init__ and reached into
it directly. Since BaseObject now provides task_manager / create_task /
cancel_task, drop the constructor argument and call
`observer.setup(task_manager)` from PipelineTask._setup() before
starting it.
This commit is contained in:
Aleix Conchillo Flaqué
2026-05-08 14:00:46 -07:00
parent 784667bad2
commit 15531c8112
2 changed files with 8 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
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 # This task maneger will handle all the asyncio tasks created by this
# PipelineTask and its frame processors. # 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. # This queue is the queue used to push frames to the pipeline.
self._push_queue = asyncio.Queue() 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, # 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 # we only need to pass a single observer (using the StartFrame) which
# then just acts as a proxy. # 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 # 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 # or sink processors. Instead of calling the event handlers for every
@@ -751,7 +751,7 @@ class PipelineTask(BasePipelineTask):
async def _setup(self, params: PipelineTaskParams): async def _setup(self, params: PipelineTaskParams):
"""Set up the pipeline task and all processors.""" """Set up the pipeline task and all processors."""
await super().setup(self._task_manager) await super().setup(self._pipeline_task_manager)
mgr_params = TaskManagerParams(loop=params.loop) mgr_params = TaskManagerParams(loop=params.loop)
self.task_manager.setup(mgr_params) self.task_manager.setup(mgr_params)
@@ -774,6 +774,7 @@ class PipelineTask(BasePipelineTask):
await self._load_setup_files() await self._load_setup_files()
# Start task observer. # Start task observer.
await self._observer.setup(self.task_manager)
await self._observer.start() await self._observer.start()
async def _cleanup(self, cleanup_pipeline: bool): async def _cleanup(self, cleanup_pipeline: bool):

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

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