From 6e44a2ab493371e6c28ca823eb118abf3696bcac Mon Sep 17 00:00:00 2001 From: Om Chauhan Date: Wed, 21 Jan 2026 08:14:48 +0530 Subject: [PATCH 1/3] feat(task): add additive filter methods for frame monitoring --- src/pipecat/pipeline/task.py | 42 ++++++++++++++++++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 9a2f1d33d..20b4adf36 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -427,6 +427,24 @@ class PipelineTask(BasePipelineTask): raise Exception(f"{self} RTVI is not enabled.") return self._rtvi + @property + def reached_upstream_types(self) -> Tuple[Type[Frame], ...]: + """Get the currently configured upstream frame type filters. + + Returns: + Tuple of frame types that trigger the on_frame_reached_upstream event. + """ + return self._reached_upstream_types + + @property + def reached_downstream_types(self) -> Tuple[Type[Frame], ...]: + """Get the currently configured downstream frame type filters. + + Returns: + Tuple of frame types that trigger the on_frame_reached_downstream event. + """ + return self._reached_downstream_types + def event_handler(self, event_name: str): """Decorator for registering event handlers. @@ -480,6 +498,30 @@ class PipelineTask(BasePipelineTask): """ self._reached_downstream_types = types + def add_reached_upstream_filter(self, types: Tuple[Type[Frame], ...]): + """Add frame types to trigger the on_frame_reached_upstream event. + + Args: + types: Tuple of frame types to add to upstream monitoring. + """ + if not types: + return + current = set(self._reached_upstream_types) + current.update(types) + self._reached_upstream_types = tuple(current) + + def add_reached_downstream_filter(self, types: Tuple[Type[Frame], ...]): + """Add frame types to trigger the on_frame_reached_downstream event. + + Args: + types: Tuple of frame types to add to downstream monitoring. + """ + if not types: + return + current = set(self._reached_downstream_types) + current.update(types) + self._reached_downstream_types = tuple(current) + def has_finished(self) -> bool: """Check if the pipeline task has finished execution. From 9c0bf892472cf0d8b92f87d7547342156856fca2 Mon Sep 17 00:00:00 2001 From: Om Chauhan Date: Wed, 21 Jan 2026 08:36:20 +0530 Subject: [PATCH 2/3] added changelog --- changelog/3510.added.2.md | 1 + changelog/3510.added.md | 1 + 2 files changed, 2 insertions(+) create mode 100644 changelog/3510.added.2.md create mode 100644 changelog/3510.added.md diff --git a/changelog/3510.added.2.md b/changelog/3510.added.2.md new file mode 100644 index 000000000..e03b81ee3 --- /dev/null +++ b/changelog/3510.added.2.md @@ -0,0 +1 @@ +- Added `add_reached_upstream_filter()` and `add_reached_downstream_filter()` methods to `PipelineTask` for appending frame types. \ No newline at end of file diff --git a/changelog/3510.added.md b/changelog/3510.added.md new file mode 100644 index 000000000..93d9ef179 --- /dev/null +++ b/changelog/3510.added.md @@ -0,0 +1 @@ +- Added `reached_upstream_types` and `reached_downstream_types` read-only properties to `PipelineTask` for inspecting current frame filters. \ No newline at end of file From 87c12f3098c4176788a8090c1890bdf7dbd00bf0 Mon Sep 17 00:00:00 2001 From: Om Chauhan Date: Thu, 22 Jan 2026 08:39:31 +0530 Subject: [PATCH 3/3] changed frame filter storage type from tuples to sets --- changelog/3510.changed.3.md | 1 + src/pipecat/pipeline/task.py | 30 +++++++++++------------------- 2 files changed, 12 insertions(+), 19 deletions(-) create mode 100644 changelog/3510.changed.3.md diff --git a/changelog/3510.changed.3.md b/changelog/3510.changed.3.md new file mode 100644 index 000000000..ac299f143 --- /dev/null +++ b/changelog/3510.changed.3.md @@ -0,0 +1 @@ +- Changed frame filter storage from tuples to sets in `PipelineTask`. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 20b4adf36..60df77ff9 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -15,7 +15,7 @@ import asyncio import importlib.util import os from pathlib import Path -from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Tuple, Type +from typing import Any, AsyncIterable, Dict, Iterable, List, Optional, Set, Tuple, Type from loguru import logger from pydantic import BaseModel, ConfigDict, Field @@ -366,8 +366,8 @@ class PipelineTask(BasePipelineTask): # in. This is mainly for efficiency reason because each event handler # creates a task and most likely you only care about one or two frame # types. - self._reached_upstream_types: Tuple[Type[Frame], ...] = () - self._reached_downstream_types: Tuple[Type[Frame], ...] = () + self._reached_upstream_types: Set[Type[Frame]] = set() + self._reached_downstream_types: Set[Type[Frame]] = set() self._register_event_handler("on_frame_reached_upstream") self._register_event_handler("on_frame_reached_downstream") self._register_event_handler("on_idle_timeout") @@ -434,7 +434,7 @@ class PipelineTask(BasePipelineTask): Returns: Tuple of frame types that trigger the on_frame_reached_upstream event. """ - return self._reached_upstream_types + return tuple(self._reached_upstream_types) @property def reached_downstream_types(self) -> Tuple[Type[Frame], ...]: @@ -443,7 +443,7 @@ class PipelineTask(BasePipelineTask): Returns: Tuple of frame types that trigger the on_frame_reached_downstream event. """ - return self._reached_downstream_types + return tuple(self._reached_downstream_types) def event_handler(self, event_name: str): """Decorator for registering event handlers. @@ -488,7 +488,7 @@ class PipelineTask(BasePipelineTask): Args: types: Tuple of frame types to monitor for upstream events. """ - self._reached_upstream_types = types + self._reached_upstream_types = set(types) def set_reached_downstream_filter(self, types: Tuple[Type[Frame], ...]): """Set which frame types trigger the on_frame_reached_downstream event. @@ -496,7 +496,7 @@ class PipelineTask(BasePipelineTask): Args: types: Tuple of frame types to monitor for downstream events. """ - self._reached_downstream_types = types + self._reached_downstream_types = set(types) def add_reached_upstream_filter(self, types: Tuple[Type[Frame], ...]): """Add frame types to trigger the on_frame_reached_upstream event. @@ -504,11 +504,7 @@ class PipelineTask(BasePipelineTask): Args: types: Tuple of frame types to add to upstream monitoring. """ - if not types: - return - current = set(self._reached_upstream_types) - current.update(types) - self._reached_upstream_types = tuple(current) + self._reached_upstream_types.update(types) def add_reached_downstream_filter(self, types: Tuple[Type[Frame], ...]): """Add frame types to trigger the on_frame_reached_downstream event. @@ -516,11 +512,7 @@ class PipelineTask(BasePipelineTask): Args: types: Tuple of frame types to add to downstream monitoring. """ - if not types: - return - current = set(self._reached_downstream_types) - current.update(types) - self._reached_downstream_types = tuple(current) + self._reached_downstream_types.update(types) def has_finished(self) -> bool: """Check if the pipeline task has finished execution. @@ -820,7 +812,7 @@ class PipelineTask(BasePipelineTask): pipeline to be stopped (e.g. EndTaskFrame) in which case we would send an EndFrame down the pipeline. """ - if isinstance(frame, self._reached_upstream_types): + if isinstance(frame, tuple(self._reached_upstream_types)): await self._call_event_handler("on_frame_reached_upstream", frame) if isinstance(frame, EndTaskFrame): @@ -859,7 +851,7 @@ class PipelineTask(BasePipelineTask): processors have handled the EndFrame and therefore we can exit the task cleanly. """ - if isinstance(frame, self._reached_downstream_types): + if isinstance(frame, tuple(self._reached_downstream_types)): await self._call_event_handler("on_frame_reached_downstream", frame) if isinstance(frame, StartFrame):