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