From 61f6669926c100456f043e40a9a710bca8a775f3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 5 Feb 2025 10:01:18 -0800 Subject: [PATCH] task: allow passing StartFrame metadata via start_metadata param --- CHANGELOG.md | 3 +++ src/pipecat/frames/frames.py | 19 +++++++++++++++---- src/pipecat/pipeline/task.py | 20 +++++++++++--------- 3 files changed, 29 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 26187a4bb..a00c6f45b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added a new `start_metadata` field to `PipelineParams`. The provided metadata + will be set to the initial `StartFrame` being pushed from the `PipelineTask`. + - Added new fields to `PipelineParams` to control audio input and output sample rates for the whole pipeline. This allows controlling sample rates from a single place instead of having to specify sample rates in each diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index d34f56208..30549f739 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -6,7 +6,18 @@ from dataclasses import dataclass, field from enum import Enum -from typing import TYPE_CHECKING, Any, Awaitable, Callable, List, Literal, Mapping, Optional, Tuple +from typing import ( + TYPE_CHECKING, + Any, + Awaitable, + Callable, + Dict, + List, + Literal, + Mapping, + Optional, + Tuple, +) from pipecat.audio.vad.vad_analyzer import VADParams from pipecat.clocks.base_clock import BaseClock @@ -48,13 +59,13 @@ class Frame: id: int = field(init=False) name: str = field(init=False) pts: Optional[int] = field(init=False) - metadata: dict = field(init=False) + metadata: Dict[str, Any] = field(init=False) def __post_init__(self): self.id: int = obj_id() self.name: str = f"{self.__class__.__name__}#{obj_count(self)}" self.pts: Optional[int] = None - self.metadata: dict = {} + self.metadata: Dict[str, Any] = {} def __str__(self): return self.name @@ -433,8 +444,8 @@ class StartFrame(SystemFrame): allow_interruptions: bool = False enable_metrics: bool = False enable_usage_metrics: bool = False - report_only_initial_ttfb: bool = False observer: Optional["BaseObserver"] = None + report_only_initial_ttfb: bool = False @dataclass diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index da300100d..21b6af78b 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -5,7 +5,7 @@ # import asyncio -from typing import AsyncIterable, Iterable, List +from typing import Any, AsyncIterable, Dict, Iterable, List from loguru import logger from pydantic import BaseModel, ConfigDict @@ -40,16 +40,17 @@ HEARTBEAT_MONITOR_SECONDS = HEARTBEAT_SECONDS * 5 class PipelineParams(BaseModel): model_config = ConfigDict(arbitrary_types_allowed=True) + allow_interruptions: bool = False audio_in_sample_rate: int = 16000 audio_out_sample_rate: int = 24000 - allow_interruptions: bool = False enable_heartbeats: bool = False enable_metrics: bool = False enable_usage_metrics: bool = False - send_initial_empty_metrics: bool = True - report_only_initial_ttfb: bool = False - observers: List[BaseObserver] = [] heartbeats_period_secs: float = HEARTBEAT_SECONDS + observers: List[BaseObserver] = [] + report_only_initial_ttfb: bool = False + send_initial_empty_metrics: bool = True + start_metadata: Dict[str, Any] = {} class PipelineTaskSource(FrameProcessor): @@ -278,13 +279,14 @@ class PipelineTask(BaseTask): clock=self._clock, task_manager=self._task_manager, allow_interruptions=self._params.allow_interruptions, - enable_metrics=self._params.enable_metrics, - enable_usage_metrics=self._params.enable_usage_metrics, - report_only_initial_ttfb=self._params.report_only_initial_ttfb, - observer=self._observer, audio_in_sample_rate=self._params.audio_in_sample_rate, audio_out_sample_rate=self._params.audio_out_sample_rate, + enable_metrics=self._params.enable_metrics, + enable_usage_metrics=self._params.enable_usage_metrics, + observer=self._observer, + report_only_initial_ttfb=self._params.report_only_initial_ttfb, ) + start_frame.metadata = self._params.start_metadata await self._source.queue_frame(start_frame, FrameDirection.DOWNSTREAM) if self._params.enable_metrics and self._params.send_initial_empty_metrics: