From 71e1d0a334dfbeea093c1a830ce502b101071092 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 12 Jun 2024 11:33:59 -0700 Subject: [PATCH] pipeline: send initial TTFB initial metrics from PipelineTask --- src/pipecat/pipeline/pipeline.py | 13 +------------ src/pipecat/pipeline/task.py | 11 +++++++++-- 2 files changed, 10 insertions(+), 14 deletions(-) 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