From 6d6d9bea5ae529d21dc929e2578fc4f6b512e4ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 28 May 2024 16:03:28 -0700 Subject: [PATCH] introduce PipelineParams --- CHANGELOG.md | 4 ++++ examples/foundational/07-interruptible.py | 4 ++-- examples/foundational/07a-interruptible-anthropic.py | 4 ++-- examples/foundational/07c-interruptible-deepgram.py | 4 ++-- examples/foundational/14-wake-phrase.py | 4 ++-- examples/simple-chatbot/bot.py | 4 ++-- src/pipecat/pipeline/task.py | 12 +++++++++--- 7 files changed, 23 insertions(+), 13 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 84a6ae873..9e083a678 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Changed +- Added `PipelineParams`. This replaces the `allow_interruptions` argument in + `PipelineTask` and will allow future parameters in the future. + - Fixed Deepgram Aura TTS base_url and added ErrorFrame reporting. + - GoogleLLMService `api_key` argument is now mandatory. ### Fixed diff --git a/examples/foundational/07-interruptible.py b/examples/foundational/07-interruptible.py index 44c785d2c..ce37344f0 100644 --- a/examples/foundational/07-interruptible.py +++ b/examples/foundational/07-interruptible.py @@ -12,7 +12,7 @@ import sys from pipecat.frames.frames import LLMMessagesFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineTask +from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_response import ( LLMAssistantResponseAggregator, LLMUserResponseAggregator) from pipecat.services.elevenlabs import ElevenLabsTTSService @@ -74,7 +74,7 @@ async def main(room_url: str, token): tma_out # Assistant spoken responses ]) - task = PipelineTask(pipeline, allow_interruptions=True) + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): diff --git a/examples/foundational/07a-interruptible-anthropic.py b/examples/foundational/07a-interruptible-anthropic.py index 260cd74e8..151fe4f18 100644 --- a/examples/foundational/07a-interruptible-anthropic.py +++ b/examples/foundational/07a-interruptible-anthropic.py @@ -12,7 +12,7 @@ import sys from pipecat.frames.frames import LLMMessagesFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineTask +from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_response import ( LLMAssistantResponseAggregator, LLMUserResponseAggregator) from pipecat.services.elevenlabs import ElevenLabsTTSService @@ -77,7 +77,7 @@ async def main(room_url: str, token): tma_out # Assistant spoken responses ]) - task = PipelineTask(pipeline, allow_interruptions=True) + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): diff --git a/examples/foundational/07c-interruptible-deepgram.py b/examples/foundational/07c-interruptible-deepgram.py index 30703c07e..27245b02b 100644 --- a/examples/foundational/07c-interruptible-deepgram.py +++ b/examples/foundational/07c-interruptible-deepgram.py @@ -12,7 +12,7 @@ import sys from pipecat.frames.frames import LLMMessagesFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineTask +from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_response import ( LLMAssistantResponseAggregator, LLMUserResponseAggregator) from pipecat.services.deepgram import DeepgramTTSService @@ -74,7 +74,7 @@ async def main(room_url: str, token): tma_out # Assistant spoken responses ]) - task = PipelineTask(pipeline, allow_interruptions=True) + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): diff --git a/examples/foundational/14-wake-phrase.py b/examples/foundational/14-wake-phrase.py index 96875409f..853bac4e3 100644 --- a/examples/foundational/14-wake-phrase.py +++ b/examples/foundational/14-wake-phrase.py @@ -12,7 +12,7 @@ import sys from pipecat.processors.filters.wake_check_filter import WakeCheckFilter from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineTask +from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_response import ( LLMAssistantResponseAggregator, LLMUserResponseAggregator) from pipecat.services.elevenlabs import ElevenLabsTTSService @@ -77,7 +77,7 @@ async def main(room_url: str, token): tma_out # Assistant spoken responses ]) - task = PipelineTask(pipeline, allow_interruptions=True) + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): diff --git a/examples/simple-chatbot/bot.py b/examples/simple-chatbot/bot.py index 2c03a70f4..a63b215ab 100644 --- a/examples/simple-chatbot/bot.py +++ b/examples/simple-chatbot/bot.py @@ -7,7 +7,7 @@ from PIL import Image from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineTask +from pipecat.pipeline.task import PipelineParams, PipelineTask from pipecat.processors.aggregators.llm_response import LLMAssistantResponseAggregator, LLMUserResponseAggregator from pipecat.frames.frames import ( AudioRawFrame, @@ -149,7 +149,7 @@ async def main(room_url: str, token): assistant_response, ]) - task = PipelineTask(pipeline, allow_interruptions=True) + task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) await task.queue_frame(quiet_frame) @transport.event_handler("on_first_participant_joined") diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 400adcbfd..f81281820 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -8,6 +8,8 @@ import asyncio from typing import AsyncIterable, Iterable +from pydantic import BaseModel + from pipecat.frames.frames import CancelFrame, EndFrame, ErrorFrame, Frame, StartFrame, StopTaskFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor from pipecat.utils.utils import obj_count, obj_id @@ -15,6 +17,10 @@ from pipecat.utils.utils import obj_count, obj_id from loguru import logger +class PipelineParams(BaseModel): + allow_interruptions: bool = False + + class Source(FrameProcessor): def __init__(self, up_queue: asyncio.Queue): @@ -31,12 +37,12 @@ class Source(FrameProcessor): class PipelineTask: - def __init__(self, pipeline: FrameProcessor, allow_interruptions=False): + def __init__(self, pipeline: FrameProcessor, params: PipelineParams = PipelineParams()): self.id: int = obj_id() self.name: str = f"{self.__class__.__name__}#{obj_count(self)}" self._pipeline = pipeline - self._allow_interruptions = allow_interruptions + self._params = params self._down_queue = asyncio.Queue() self._up_queue = asyncio.Queue() @@ -77,7 +83,7 @@ class PipelineTask: async def _process_down_queue(self): await self._source.process_frame( - StartFrame(allow_interruptions=self._allow_interruptions), FrameDirection.DOWNSTREAM) + StartFrame(allow_interruptions=self._params.allow_interruptions), FrameDirection.DOWNSTREAM) running = True should_cleanup = True while running: