processors: add processors_with_metrics() and can_generate_metrics()

CHANGELOG.md

@@ -12,9 +12,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - Added a new `FunctionFilter`. This filter will let you filter frames based on
   a given function, except system messages which should never be filtered.
 
+- Added `FrameProcessor.can_generate_metrics()` method to indicate if a
+  processor can generate metrics. In the future this might get an extra argument
+  to ask for a specific type of metric.
+
 - Added `BasePipeline`. All pipeline classes should be based on this class. All
-  subclasses should implement a `services()` method that returns a list of
-  all `AIServices` in the pipeline.
+  subclasses should implement a `processors_with_metrics()` method that returns
+  a list of all `FrameProcessor`s in the pipeline that can generate metrics.
 
 - Added `enable_metrics` to `PipelineParams`.
 

src/pipecat/pipeline/base_pipeline.py

@@ -17,5 +17,5 @@ class BasePipeline(FrameProcessor):
         super().__init__()
 
     @abstractmethod
-    def services(self) -> List[FrameProcessor]:
+    def processors_with_metrics(self) -> List[FrameProcessor]:
         pass

src/pipecat/pipeline/parallel_pipeline.py

@@ -89,8 +89,8 @@ class ParallelPipeline(BasePipeline):
     # BasePipeline
     #
 
-    def services(self) -> List[FrameProcessor]:
-        return list(chain.from_iterable(p.services() for p in self._pipelines))
+    def processors_with_metrics(self) -> List[FrameProcessor]:
+        return list(chain.from_iterable(p.processors_with_metrics() for p in self._pipelines))
 
     #
     # Frame processor

src/pipecat/pipeline/parallel_task.py

@@ -85,8 +85,8 @@ class ParallelTask(BasePipeline):
     # BasePipeline
     #
 
-    def services(self) -> List[FrameProcessor]:
-        return list(chain.from_iterable(p.services() for p in self._pipelines))
+    def processors_with_metrics(self) -> List[FrameProcessor]:
+        return list(chain.from_iterable(p.processors_with_metrics() for p in self._pipelines))
 
     #
     # Frame processor

src/pipecat/pipeline/pipeline.py

@@ -4,12 +4,13 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
+from itertools import chain
+
 from typing import Callable, Coroutine, List
 
 from pipecat.frames.frames import Frame, MetricsFrame, StartFrame
 from pipecat.pipeline.base_pipeline import BasePipeline
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
-from pipecat.services.ai_services import AIService
 
 
 class PipelineSource(FrameProcessor):
@@ -61,13 +62,13 @@ class Pipeline(BasePipeline):
     # BasePipeline
     #
 
-    def services(self):
+    def processors_with_metrics(self):
         services = []
         for p in self._processors:
-            if isinstance(p, AIService):
+            if isinstance(p, BasePipeline):
+                services += p.processors_with_metrics()
+            elif p.can_generate_metrics():
                 services.append(p)
-            elif isinstance(p, Pipeline):
-                services += p.services()
         return services
 
     #
@@ -99,7 +100,7 @@ class Pipeline(BasePipeline):
             prev = curr
 
     async def _send_initial_metrics(self):
-        services = self.services()
-        ttfb = dict(zip([s.name for s in services], [0] * len(services)))
+        processors = self.processors_with_metrics()
+        ttfb = dict(zip([p.name for p in processors], [0] * len(processors)))
         frame = MetricsFrame(ttfb=ttfb)
         await self._source.process_frame(frame, FrameDirection.DOWNSTREAM)

src/pipecat/processors/frame_processor.py

@@ -44,6 +44,9 @@ class FrameProcessor:
     def metrics_enabled(self):
         return self._enable_metrics
 
+    def can_generate_metrics(self) -> bool:
+        return False
+
     async def start_ttfb_metrics(self):
         if self.metrics_enabled:
             self._start_ttfb_time = time.time()

src/pipecat/services/anthropic.py

@@ -49,6 +49,9 @@ class AnthropicLLMService(LLMService):
         self._model = model
         self._max_tokens = max_tokens
 
+    def can_generate_metrics(self) -> bool:
+        return True
+
     def _get_messages_from_openai_context(
             self, context: OpenAILLMContext):
         openai_messages = context.get_messages()

src/pipecat/services/azure.py

@@ -44,6 +44,9 @@ class AzureTTSService(TTSService):
         )
         self._voice = voice
 
+    def can_generate_metrics(self) -> bool:
+        return True
+
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating TTS: {text}")
 

src/pipecat/services/cartesia.py

@@ -39,6 +39,9 @@ class CartesiaTTSService(TTSService):
         except Exception as e:
             logger.error(f"Cartesia initialization error: {e}")
 
+    def can_generate_metrics(self) -> bool:
+        return True
+
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating TTS: [{text}]")
 

src/pipecat/services/deepgram.py

@@ -29,6 +29,9 @@ class DeepgramTTSService(TTSService):
         self._api_key = api_key
         self._aiohttp_session = aiohttp_session
 
+    def can_generate_metrics(self) -> bool:
+        return True
+
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating TTS: [{text}]")
 

src/pipecat/services/elevenlabs.py

@@ -31,6 +31,9 @@ class ElevenLabsTTSService(TTSService):
         self._aiohttp_session = aiohttp_session
         self._model = model
 
+    def can_generate_metrics(self) -> bool:
+        return True
+
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating TTS: [{text}]")
 

src/pipecat/services/google.py

@@ -47,6 +47,9 @@ class GoogleLLMService(LLMService):
         gai.configure(api_key=api_key)
         self._client = gai.GenerativeModel(model)
 
+    def can_generate_metrics(self) -> bool:
+        return True
+
     def _get_messages_from_openai_context(
             self, context: OpenAILLMContext) -> List[glm.Content]:
         openai_messages = context.get_messages()

src/pipecat/services/openai.py

@@ -76,6 +76,9 @@ class BaseOpenAILLMService(LLMService):
     def create_client(self, api_key=None, base_url=None):
         return AsyncOpenAI(api_key=api_key, base_url=base_url)
 
+    def can_generate_metrics(self) -> bool:
+        return True
+
     async def _stream_chat_completions(
         self, context: OpenAILLMContext
     ) -> AsyncStream[ChatCompletionChunk]:
@@ -298,6 +301,9 @@ class OpenAITTSService(TTSService):
 
         self._client = AsyncOpenAI(api_key=api_key)
 
+    def can_generate_metrics(self) -> bool:
+        return True
+
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating TTS: [{text}]")
 

src/pipecat/services/playht.py

@@ -43,8 +43,8 @@ class PlayHTTTSService(TTSService):
             quality="higher",
             format=Format.FORMAT_WAV)
 
-    def __del__(self):
-        self._client.close()
+    def can_generate_metrics(self) -> bool:
+        return True
 
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating TTS: [{text}]")

src/pipecat/services/whisper.py

@@ -56,6 +56,9 @@ class WhisperSTTService(STTService):
         self._model: WhisperModel | None = None
         self._load()
 
+    def can_generate_metrics(self) -> bool:
+        return True
+
     def _load(self):
         """Loads the Whisper model. Note that if this is the first time
         this model is being run, it will take time to download."""