From d743586bfb304e33545815ce050daf86f20e5f14 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 15 Aug 2025 10:36:05 -0700 Subject: [PATCH] BasePipeline: move processors_with_metrics() to FrameProcessor --- src/pipecat/pipeline/base_pipeline.py | 21 +------------------ src/pipecat/pipeline/parallel_pipeline.py | 6 +----- src/pipecat/pipeline/pipeline.py | 13 ++++-------- .../pipeline/sync_parallel_pipeline.py | 6 +----- src/pipecat/pipeline/task.py | 3 +-- src/pipecat/processors/frame_processor.py | 11 ++++++++++ 6 files changed, 19 insertions(+), 41 deletions(-) diff --git a/src/pipecat/pipeline/base_pipeline.py b/src/pipecat/pipeline/base_pipeline.py index b27467b01..8bc9da524 100644 --- a/src/pipecat/pipeline/base_pipeline.py +++ b/src/pipecat/pipeline/base_pipeline.py @@ -6,31 +6,12 @@ """Base pipeline implementation for frame processing.""" -from abc import abstractmethod -from typing import List - from pipecat.processors.frame_processor import FrameProcessor class BasePipeline(FrameProcessor): - """Base class for all pipeline implementations. - - Provides the foundation for pipeline processors that need to support - metrics collection from their contained processors. - """ + """Base class for all pipeline implementations.""" def __init__(self, **kwargs): """Initialize the base pipeline.""" super().__init__(**kwargs) - - @abstractmethod - def processors_with_metrics(self) -> List[FrameProcessor]: - """Return processors that can generate metrics. - - Implementing classes should collect and return all processors within - their pipeline that support metrics generation. - - Returns: - List of frame processors that support metrics collection. - """ - pass diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py index b2c20e74e..0e9ad0e4e 100644 --- a/src/pipecat/pipeline/parallel_pipeline.py +++ b/src/pipecat/pipeline/parallel_pipeline.py @@ -149,7 +149,7 @@ class ParallelPipeline(BasePipeline): self._down_task = None # - # BasePipeline + # Frame processor # def processors_with_metrics(self) -> List[FrameProcessor]: @@ -160,10 +160,6 @@ class ParallelPipeline(BasePipeline): """ return list(chain.from_iterable(p.processors_with_metrics() for p in self._pipelines)) - # - # Frame processor - # - async def setup(self, setup: FrameProcessorSetup): """Set up the parallel pipeline and all its branches. diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py index 227fc7139..d9068767b 100644 --- a/src/pipecat/pipeline/pipeline.py +++ b/src/pipecat/pipeline/pipeline.py @@ -111,7 +111,7 @@ class Pipeline(BasePipeline): self._link_processors() # - # BasePipeline + # Frame processor # def processors_with_metrics(self): @@ -124,17 +124,12 @@ class Pipeline(BasePipeline): List of frame processors that can generate metrics. """ services = [] - for p in self._processors: - if isinstance(p, BasePipeline): - services.extend(p.processors_with_metrics()) - elif p.can_generate_metrics(): + for p in self.processors: + if p.can_generate_metrics(): services.append(p) + services.extend(p.processors_with_metrics()) return services - # - # Frame processor - # - async def setup(self, setup: FrameProcessorSetup): """Set up the pipeline and all contained processors. diff --git a/src/pipecat/pipeline/sync_parallel_pipeline.py b/src/pipecat/pipeline/sync_parallel_pipeline.py index c4cffe112..332e5d35c 100644 --- a/src/pipecat/pipeline/sync_parallel_pipeline.py +++ b/src/pipecat/pipeline/sync_parallel_pipeline.py @@ -134,7 +134,7 @@ class SyncParallelPipeline(BasePipeline): self._pipelines = [] # - # BasePipeline + # Frame processor # def processors_with_metrics(self) -> List[FrameProcessor]: @@ -145,10 +145,6 @@ class SyncParallelPipeline(BasePipeline): """ return list(chain.from_iterable(p.processors_with_metrics() for p in self._pipelines)) - # - # Frame processor - # - async def setup(self, setup: FrameProcessorSetup): """Set up the parallel pipeline and all contained processors. diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 5ad22077e..fc03cc9c9 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -41,7 +41,6 @@ from pipecat.frames.frames import ( from pipecat.metrics.metrics import ProcessingMetricsData, TTFBMetricsData from pipecat.observers.base_observer import BaseObserver from pipecat.observers.turn_tracking_observer import TurnTrackingObserver -from pipecat.pipeline.base_pipeline import BasePipeline from pipecat.pipeline.base_task import BasePipelineTask, PipelineTaskParams from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.task_observer import TaskObserver @@ -197,7 +196,7 @@ class PipelineTask(BasePipelineTask): def __init__( self, - pipeline: BasePipeline, + pipeline: FrameProcessor, *, params: Optional[PipelineParams] = None, additional_span_attributes: Optional[dict] = None, diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index b03930c67..368152b46 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -339,6 +339,17 @@ class FrameProcessor(BaseObject): raise Exception(f"{self} TaskManager is still not initialized.") return self._task_manager + def processors_with_metrics(self): + """Return processors that can generate metrics. + + Recursively collects all processors that support metrics generation, + including those from nested processors. + + Returns: + List of frame processors that can generate metrics. + """ + return [] + def can_generate_metrics(self) -> bool: """Check if this processor can generate metrics.