From 1afe6901d9a77a7a98ee5734e1a5107596531c9d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 7 Jun 2024 13:38:21 -0700 Subject: [PATCH] processors: add processors_with_metrics() and can_generate_metrics() --- CHANGELOG.md | 8 ++++++-- src/pipecat/pipeline/base_pipeline.py | 2 +- src/pipecat/pipeline/parallel_pipeline.py | 4 ++-- src/pipecat/pipeline/parallel_task.py | 4 ++-- src/pipecat/pipeline/pipeline.py | 15 ++++++++------- src/pipecat/processors/frame_processor.py | 3 +++ src/pipecat/services/anthropic.py | 3 +++ src/pipecat/services/azure.py | 3 +++ src/pipecat/services/cartesia.py | 3 +++ src/pipecat/services/deepgram.py | 3 +++ src/pipecat/services/elevenlabs.py | 3 +++ src/pipecat/services/google.py | 3 +++ src/pipecat/services/openai.py | 6 ++++++ src/pipecat/services/playht.py | 4 ++-- src/pipecat/services/whisper.py | 3 +++ 15 files changed, 51 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d38ba2e4a..b42244ea9 100644 --- a/CHANGELOG.md +++ b/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`. diff --git a/src/pipecat/pipeline/base_pipeline.py b/src/pipecat/pipeline/base_pipeline.py index 83dbc2730..54f6499a9 100644 --- a/src/pipecat/pipeline/base_pipeline.py +++ b/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 diff --git a/src/pipecat/pipeline/parallel_pipeline.py b/src/pipecat/pipeline/parallel_pipeline.py index f89e61ca7..d045c3493 100644 --- a/src/pipecat/pipeline/parallel_pipeline.py +++ b/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 diff --git a/src/pipecat/pipeline/parallel_task.py b/src/pipecat/pipeline/parallel_task.py index b1f3c520c..306a772b7 100644 --- a/src/pipecat/pipeline/parallel_task.py +++ b/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 diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py index 1d7e8a024..eb6bd78d1 100644 --- a/src/pipecat/pipeline/pipeline.py +++ b/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) diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index e0a072477..49651c7a2 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/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() diff --git a/src/pipecat/services/anthropic.py b/src/pipecat/services/anthropic.py index 51e1c60f1..0f3d7d241 100644 --- a/src/pipecat/services/anthropic.py +++ b/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() diff --git a/src/pipecat/services/azure.py b/src/pipecat/services/azure.py index d35584303..f4184f647 100644 --- a/src/pipecat/services/azure.py +++ b/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}") diff --git a/src/pipecat/services/cartesia.py b/src/pipecat/services/cartesia.py index 474600ffa..161f0589b 100644 --- a/src/pipecat/services/cartesia.py +++ b/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}]") diff --git a/src/pipecat/services/deepgram.py b/src/pipecat/services/deepgram.py index aaf0c01ee..ce418916b 100644 --- a/src/pipecat/services/deepgram.py +++ b/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}]") diff --git a/src/pipecat/services/elevenlabs.py b/src/pipecat/services/elevenlabs.py index 717460fd3..256f077c3 100644 --- a/src/pipecat/services/elevenlabs.py +++ b/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}]") diff --git a/src/pipecat/services/google.py b/src/pipecat/services/google.py index ffe9a9fee..b1da9ddf6 100644 --- a/src/pipecat/services/google.py +++ b/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() diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index a42fddbf9..daa6125b3 100644 --- a/src/pipecat/services/openai.py +++ b/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}]") diff --git a/src/pipecat/services/playht.py b/src/pipecat/services/playht.py index cf38e125d..c44719047 100644 --- a/src/pipecat/services/playht.py +++ b/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}]") diff --git a/src/pipecat/services/whisper.py b/src/pipecat/services/whisper.py index f1b37712e..6dfc5080e 100644 --- a/src/pipecat/services/whisper.py +++ b/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."""