BasePipeline: move processors_with_metrics() to FrameProcessor

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.
+    """Base class for all pipeline implementations."""
-
-    Provides the foundation for pipeline processors that need to support
-    metrics collection from their contained processors.
-    """
 
     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

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.
 

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:
+        for p in self.processors:
-            if isinstance(p, BasePipeline):
+            if p.can_generate_metrics():
-                services.extend(p.processors_with_metrics())
-            elif 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.
 

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.
 

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,

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.