Merge pull request #184 from pipecat-ai/aleix/introduce-pipelineparams

introduce PipelineParams

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

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):

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):

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):

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):

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")

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: