diff --git a/CHANGELOG.md b/CHANGELOG.md index ca4c9885b..d4a2da1e4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Allow specifying frame processors' name through a new `name` constructor + argument. + ### Changed - `daily_rest.DailyRoomProperties` now allows extra unknown parameters. diff --git a/src/pipecat/pipeline/pipeline.py b/src/pipecat/pipeline/pipeline.py index eb6bd78d1..c13cc61a1 100644 --- a/src/pipecat/pipeline/pipeline.py +++ b/src/pipecat/pipeline/pipeline.py @@ -4,11 +4,9 @@ # 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.frames.frames import Frame from pipecat.pipeline.base_pipeline import BasePipeline from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -81,9 +79,6 @@ class Pipeline(BasePipeline): async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) - if isinstance(frame, StartFrame) and self.metrics_enabled: - await self._send_initial_metrics() - if direction == FrameDirection.DOWNSTREAM: await self._source.process_frame(frame, FrameDirection.DOWNSTREAM) elif direction == FrameDirection.UPSTREAM: @@ -98,9 +93,3 @@ class Pipeline(BasePipeline): for curr in self._processors[1:]: prev.link(curr) prev = curr - - async def _send_initial_metrics(self): - 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/pipeline/task.py b/src/pipecat/pipeline/task.py index 76d98165b..32aeea61f 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -10,7 +10,8 @@ from typing import AsyncIterable, Iterable from pydantic import BaseModel -from pipecat.frames.frames import CancelFrame, EndFrame, ErrorFrame, Frame, StartFrame, StopTaskFrame +from pipecat.frames.frames import CancelFrame, EndFrame, ErrorFrame, Frame, MetricsFrame, StartFrame, StopTaskFrame +from pipecat.pipeline.base_pipeline import BasePipeline from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.utils.utils import obj_count, obj_id @@ -40,7 +41,7 @@ class Source(FrameProcessor): class PipelineTask: - def __init__(self, pipeline: FrameProcessor, params: PipelineParams = PipelineParams()): + def __init__(self, pipeline: BasePipeline, params: PipelineParams = PipelineParams()): self.id: int = obj_id() self.name: str = f"{self.__class__.__name__}#{obj_count(self)}" @@ -89,12 +90,18 @@ class PipelineTask: else: raise Exception("Frames must be an iterable or async iterable") + def _initial_metrics_frame(self) -> MetricsFrame: + processors = self._pipeline.processors_with_metrics() + ttfb = dict(zip([p.name for p in processors], [0] * len(processors))) + return MetricsFrame(ttfb=ttfb) + async def _process_down_queue(self): start_frame = StartFrame( allow_interruptions=self._params.allow_interruptions, enable_metrics=self._params.enable_metrics, ) await self._source.process_frame(start_frame, FrameDirection.DOWNSTREAM) + await self._source.process_frame(self._initial_metrics_frame(), FrameDirection.DOWNSTREAM) running = True should_cleanup = True diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 49651c7a2..1197f4490 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -22,9 +22,9 @@ class FrameDirection(Enum): class FrameProcessor: - def __init__(self, loop: asyncio.AbstractEventLoop | None = None): + def __init__(self, name: str | None = None, loop: asyncio.AbstractEventLoop | None = None): self.id: int = obj_id() - self.name = f"{self.__class__.__name__}#{obj_count(self)}" + self.name = name or f"{self.__class__.__name__}#{obj_count(self)}" self._prev: "FrameProcessor" | None = None self._next: "FrameProcessor" | None = None self._loop: asyncio.AbstractEventLoop = loop or asyncio.get_running_loop() diff --git a/src/pipecat/services/ai_services.py b/src/pipecat/services/ai_services.py index 82175b797..6837c2b71 100644 --- a/src/pipecat/services/ai_services.py +++ b/src/pipecat/services/ai_services.py @@ -28,8 +28,8 @@ from pipecat.utils.utils import exp_smoothing class AIService(FrameProcessor): - def __init__(self): - super().__init__() + def __init__(self, **kwargs): + super().__init__(**kwargs) async def start(self, frame: StartFrame): pass @@ -61,8 +61,8 @@ class AIService(FrameProcessor): class LLMService(AIService): """This class is a no-op but serves as a base class for LLM services.""" - def __init__(self): - super().__init__() + def __init__(self, **kwargs): + super().__init__(**kwargs) self._callbacks = {} self._start_callbacks = {} @@ -91,8 +91,8 @@ class LLMService(AIService): class TTSService(AIService): - def __init__(self, aggregate_sentences: bool = True): - super().__init__() + def __init__(self, aggregate_sentences: bool = True, **kwargs): + super().__init__(**kwargs) self._aggregate_sentences: bool = aggregate_sentences self._current_sentence: str = "" @@ -146,8 +146,9 @@ class STTService(AIService): max_silence_secs: float = 0.3, max_buffer_secs: float = 1.5, sample_rate: int = 16000, - num_channels: int = 1): - super().__init__() + num_channels: int = 1, + **kwargs): + super().__init__(**kwargs) self._min_volume = min_volume self._max_silence_secs = max_silence_secs self._max_buffer_secs = max_buffer_secs @@ -216,8 +217,8 @@ class STTService(AIService): class ImageGenService(AIService): - def __init__(self): - super().__init__() + def __init__(self, **kwargs): + super().__init__(**kwargs) # Renders the image. Returns an Image object. @abstractmethod @@ -237,8 +238,8 @@ class ImageGenService(AIService): class VisionService(AIService): """VisionService is a base class for vision services.""" - def __init__(self): - super().__init__() + def __init__(self, **kwargs): + super().__init__(**kwargs) self._describe_text = None @abstractmethod diff --git a/src/pipecat/services/openai.py b/src/pipecat/services/openai.py index daa6125b3..c2f4d8915 100644 --- a/src/pipecat/services/openai.py +++ b/src/pipecat/services/openai.py @@ -41,7 +41,6 @@ from pipecat.services.ai_services import ( try: from openai import AsyncOpenAI, AsyncStream, BadRequestError from openai.types.chat import ( - ChatCompletion, ChatCompletionChunk, ChatCompletionFunctionMessageParam, ChatCompletionMessageParam, @@ -68,8 +67,8 @@ class BaseOpenAILLMService(LLMService): calls from the LLM. """ - def __init__(self, model: str, api_key=None, base_url=None): - super().__init__() + def __init__(self, model: str, api_key=None, base_url=None, **kwargs): + super().__init__(**kwargs) self._model: str = model self._client = self.create_client(api_key=api_key, base_url=base_url) diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 303979816..0d4817e62 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -28,8 +28,8 @@ from loguru import logger class BaseInputTransport(FrameProcessor): - def __init__(self, params: TransportParams): - super().__init__() + def __init__(self, params: TransportParams, **kwargs): + super().__init__(**kwargs) self._params = params diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index cfbfce6b3..86802d552 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -37,8 +37,8 @@ from loguru import logger class BaseOutputTransport(FrameProcessor): - def __init__(self, params: TransportParams): - super().__init__() + def __init__(self, params: TransportParams, **kwargs): + super().__init__(**kwargs) self._params = params @@ -135,6 +135,9 @@ class BaseOutputTransport(FrameProcessor): elif isinstance(frame, StartInterruptionFrame) or isinstance(frame, StopInterruptionFrame): await self._handle_interruptions(frame) await self.push_frame(frame, direction) + elif isinstance(frame, MetricsFrame): + self.send_metrics(frame) + await self.push_frame(frame, direction) elif isinstance(frame, SystemFrame): await self.push_frame(frame, direction) elif isinstance(frame, AudioRawFrame): @@ -182,8 +185,6 @@ class BaseOutputTransport(FrameProcessor): self._set_camera_images(frame.images) elif isinstance(frame, TransportMessageFrame): self.send_message(frame) - elif isinstance(frame, MetricsFrame): - self.send_metrics(frame) else: future = asyncio.run_coroutine_threadsafe( self._internal_push_frame(frame), self.get_event_loop()) diff --git a/src/pipecat/transports/base_transport.py b/src/pipecat/transports/base_transport.py index 7034b81eb..d04ea2f61 100644 --- a/src/pipecat/transports/base_transport.py +++ b/src/pipecat/transports/base_transport.py @@ -41,7 +41,12 @@ class TransportParams(BaseModel): class BaseTransport(ABC): - def __init__(self, loop: asyncio.AbstractEventLoop | None): + def __init__(self, + input_name: str | None = None, + output_name: str | None = None, + loop: asyncio.AbstractEventLoop | None = None): + self._input_name = input_name + self._output_name = output_name self._loop = loop or asyncio.get_running_loop() self._event_handlers: dict = {} diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index bf51732e1..b3f2b2e33 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/src/pipecat/transports/network/websocket_server.py @@ -42,8 +42,9 @@ class WebsocketServerInputTransport(BaseInputTransport): host: str, port: int, params: WebsocketServerParams, - callbacks: WebsocketServerCallbacks): - super().__init__(params) + callbacks: WebsocketServerCallbacks, + **kwargs): + super().__init__(params, **kwargs) self._host = host self._port = port @@ -98,8 +99,8 @@ class WebsocketServerInputTransport(BaseInputTransport): class WebsocketServerOutputTransport(BaseOutputTransport): - def __init__(self, params: WebsocketServerParams): - super().__init__(params) + def __init__(self, params: WebsocketServerParams, **kwargs): + super().__init__(params, **kwargs) self._params = params @@ -153,8 +154,10 @@ class WebsocketServerTransport(BaseTransport): host: str = "localhost", port: int = 8765, params: WebsocketServerParams = WebsocketServerParams(), + input_name: str | None = None, + output_name: str | None = None, loop: asyncio.AbstractEventLoop | None = None): - super().__init__(loop) + super().__init__(input_name=input_name, output_name=output_name, loop=loop) self._host = host self._port = port self._params = params @@ -175,12 +178,12 @@ class WebsocketServerTransport(BaseTransport): def input(self) -> FrameProcessor: if not self._input: self._input = WebsocketServerInputTransport( - self._host, self._port, self._params, self._callbacks) + self._host, self._port, self._params, self._callbacks, name=self._input_name) return self._input def output(self) -> FrameProcessor: if not self._output: - self._output = WebsocketServerOutputTransport(self._params) + self._output = WebsocketServerOutputTransport(self._params, name=self._output_name) return self._output async def _on_client_connected(self, websocket): diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 30e857c2d..b0a9d8f8e 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -473,8 +473,8 @@ class DailyTransportClient(EventHandler): class DailyInputTransport(BaseInputTransport): - def __init__(self, client: DailyTransportClient, params: DailyParams): - super().__init__(params) + def __init__(self, client: DailyTransportClient, params: DailyParams, **kwargs): + super().__init__(params, **kwargs) self._client = client @@ -609,8 +609,8 @@ class DailyInputTransport(BaseInputTransport): class DailyOutputTransport(BaseOutputTransport): - def __init__(self, client: DailyTransportClient, params: DailyParams): - super().__init__(params) + def __init__(self, client: DailyTransportClient, params: DailyParams, **kwargs): + super().__init__(params, **kwargs) self._client = client @@ -662,8 +662,10 @@ class DailyTransport(BaseTransport): token: str | None, bot_name: str, params: DailyParams, + input_name: str | None = None, + output_name: str | None = None, loop: asyncio.AbstractEventLoop | None = None): - super().__init__(loop) + super().__init__(input_name=input_name, output_name=output_name, loop=loop) callbacks = DailyCallbacks( on_joined=self._on_joined, @@ -708,12 +710,12 @@ class DailyTransport(BaseTransport): def input(self) -> FrameProcessor: if not self._input: - self._input = DailyInputTransport(self._client, self._params) + self._input = DailyInputTransport(self._client, self._params, name=self._input_name) return self._input def output(self) -> FrameProcessor: if not self._output: - self._output = DailyOutputTransport(self._client, self._params) + self._output = DailyOutputTransport(self._client, self._params, name=self._output_name) return self._output #