introduce PipelineParams

pipecat-ai · May 28, 2024 · 7a4cccf · 7a4cccf
1 parent 650a2b4
commit 7a4cccf
Show file tree

Hide file tree

Showing 5 changed files with 17 additions and 11 deletions.
diff --git a/examples/foundational/07-interruptible.py b/examples/foundational/07-interruptible.py
@@ -12,7 +12,7 @@
 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
@@ -12,7 +12,7 @@
 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/14-wake-phrase.py b/examples/foundational/14-wake-phrase.py
@@ -12,7 +12,7 @@
 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
@@ -7,7 +7,7 @@
 
 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
@@ -8,13 +8,19 @@
 
 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
 
 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 @@ async def process_frame(self, frame: Frame, direction: FrameDirection):
 
 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 @@ async def queue_frames(self, frames: Iterable[Frame] | AsyncIterable[Frame]):
 
     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: