FrameProcessor: add new entry_processors() method

CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Added new `FrameProcessor.entry_processor()` method. This allows you to access
+  the first non-compound processor in a pipeline.
+
 - Added `FrameProcessor` properties `processors`, `next` and `previous`.
 
 - `ElevenLabsTTSService` now supports additional runtime changes to the `model`,

src/pipecat/pipeline/parallel_pipeline.py

@@ -164,6 +164,20 @@ class ParallelPipeline(BasePipeline):
         """
         return self._pipelines
 
+    @property
+    def entry_processors(self) -> List["FrameProcessor"]:
+        """Return the list of entry processors for this processor.
+
+        Entry processors are the first processors in a compound processor
+        (e.g. pipelines, parallel pipelines). Note that pipelines can also be an
+        entry processor as pipelines are processors themselves. Non-compound
+        processors will simply return an empty list.
+
+        Returns:
+            The list of entry processors.
+        """
+        return self._pipelines
+
     def processors_with_metrics(self) -> List[FrameProcessor]:
         """Collect processors that can generate metrics from all parallel branches.
 

src/pipecat/pipeline/pipeline.py

@@ -116,8 +116,30 @@ class Pipeline(BasePipeline):
 
     @property
     def processors(self):
+        """Return the list of sub-processors contained within this processor.
+
+        Only compound processors (e.g. pipelines and parallel pipelines) have
+        sub-processors. Non-compound processors will return an empty list.
+
+        Returns:
+            The list of sub-processors if this is a compound processor.
+        """
         return self._processors
 
+    @property
+    def entry_processors(self) -> List["FrameProcessor"]:
+        """Return the list of entry processors for this processor.
+
+        Entry processors are the first processors in a compound processor
+        (e.g. pipelines, parallel pipelines). Note that pipelines can also be an
+        entry processor as pipelines are processors themselves. Non-compound
+        processors will simply return an empty list.
+
+        Returns:
+            The list of entry processors.
+        """
+        return [self._source]
+
     def processors_with_metrics(self):
         """Return processors that can generate metrics.
 

src/pipecat/pipeline/sync_parallel_pipeline.py

@@ -149,6 +149,20 @@ class SyncParallelPipeline(BasePipeline):
         """
         return self._pipelines
 
+    @property
+    def entry_processors(self) -> List["FrameProcessor"]:
+        """Return the list of entry processors for this processor.
+
+        Entry processors are the first processors in a compound processor
+        (e.g. pipelines, parallel pipelines). Note that pipelines can also be an
+        entry processor as pipelines are processors themselves. Non-compound
+        processors will simply return an empty list.
+
+        Returns:
+            The list of entry processors.
+        """
+        return self._sources
+
     def processors_with_metrics(self) -> List[FrameProcessor]:
         """Collect processors that can generate metrics from all parallel pipelines.
 

src/pipecat/processors/frame_processor.py

@@ -262,6 +262,20 @@ class FrameProcessor(BaseObject):
         """
         return []
 
+    @property
+    def entry_processors(self) -> List["FrameProcessor"]:
+        """Return the list of entry processors for this processor.
+
+        Entry processors are the first processors in a compound processor
+        (e.g. pipelines, parallel pipelines). Note that pipelines can also be an
+        entry processor as pipelines are processors themselves. Non-compound
+        processors will simply return an empty list.
+
+        Returns:
+            The list of entry processors.
+        """
+        return []
+
     @property
     def next(self) -> Optional["FrameProcessor"]:
         """Get the next processor.