Save

Add missing `Params` export to AWS Nova Sonic module

examples/foundational/25-google-audio-in.py

@@ -34,6 +34,7 @@ from pipecat.processors.aggregators.openai_llm_context import (
 from pipecat.processors.frame_processor import FrameProcessor
 from pipecat.services.cartesia.tts import CartesiaTTSService
 from pipecat.services.google.llm import GoogleLLMContext, GoogleLLMService
+from pipecat.services.openai.llm import OpenAILLMService
 from pipecat.transports.base_transport import TransportParams
 from pipecat.transports.network.small_webrtc import SmallWebRTCTransport
 from pipecat.transports.network.webrtc_connection import SmallWebRTCConnection
@@ -286,16 +287,18 @@ async def run_bot(webrtc_connection: SmallWebRTCConnection, _: argparse.Namespac
         voice_id="71a7ad14-091c-4e8e-a314-022ece01c121",  # British Reading Lady
     )
 
-    conversation_llm = GoogleLLMService(
-        name="Conversation",
-        model="gemini-2.0-flash-001",
-        # model="gemini-exp-1121",
-        api_key=os.getenv("GOOGLE_API_KEY"),
-        # we can give the GoogleLLMService a system instruction to use directly
-        # in the GenerativeModel constructor. Let's do that rather than put
-        # our system message in the messages list.
-        system_instruction=conversation_system_message,
-    )
+    # conversation_llm = GoogleLLMService(
+    #     name="Conversation",
+    #     model="gemini-2.0-flash-001",
+    #     # model="gemini-exp-1121",
+    #     api_key=os.getenv("GOOGLE_API_KEY"),
+    #     # we can give the GoogleLLMService a system instruction to use directly
+    #     # in the GenerativeModel constructor. Let's do that rather than put
+    #     # our system message in the messages list.
+    #     system_instruction=conversation_system_message,
+    # )
+
+    conversation_llm = OpenAILLMService(name="Conversation", api_key=os.getenv("OPENAI_API_KEY"))
 
     input_transcription_llm = GoogleLLMService(
         name="Transcription",

src/pipecat/frames/frames.py

@@ -77,8 +77,8 @@ class Frame:
 
 @dataclass
 class SystemFrame(Frame):
-    """A frame that takes higher priority than other frames. System frames are
-    handled in order and are not affected by user interruptions.
+    """System frames are frames that are not internally queued by any of the
+    frame processors and should be processed immediately.
 
     """
 
@@ -87,9 +87,8 @@ class SystemFrame(Frame):
 
 @dataclass
 class DataFrame(Frame):
-    """A frame that is processed in order and usually contains data such as LLM
-    context, text, audio or images. Data frames are cancelled by user
-    interruptions.
+    """Data frames are frames that will be processed in order and usually
+    contain data such as LLM context, text, audio or images.
 
     """
 
@@ -98,9 +97,9 @@ class DataFrame(Frame):
 
 @dataclass
 class ControlFrame(Frame):
-    """A frame that, as data frames, is processed in order and usually contains
-    control information such as update settings or to end the pipeline after
-    everything is flushed. Control frames are cancelled by user interruptions.
+    """Control frames are frames that, similar to data frames, will be processed
+    in order and usually contain control information such as frames to update
+    settings or to end the pipeline.
 
     """
 
@@ -691,7 +690,7 @@ class FunctionCallResultFrame(SystemFrame):
 
 @dataclass
 class STTMuteFrame(SystemFrame):
-    """A frame to mute/unmute the STT service."""
+    """System frame to mute/unmute the STT service."""
 
     mute: bool
 
@@ -797,7 +796,7 @@ class EndFrame(ControlFrame):
     should be shut down. If the transport receives this frame, it will stop
     sending frames to its output channel(s) and close all its threads. Note,
     that this is a control frame, which means it will received in the order it
-    was sent.
+    was sent (unline system frames).
 
     """
 

src/pipecat/processors/frame_processor.py

@@ -5,7 +5,6 @@
 #
 
 import asyncio
-from dataclasses import dataclass
 from enum import Enum
 from typing import Awaitable, Callable, Coroutine, Optional
 
@@ -33,51 +32,6 @@ class FrameDirection(Enum):
     UPSTREAM = 2
 
 
-@dataclass
-class FrameProcessorQueueItem:
-    frame: Frame
-    direction: FrameDirection
-    callback: Optional[Callable[["FrameProcessor", Frame, FrameDirection], Awaitable[None]]]
-
-
-class FrameProcessorQueue:
-    def __init__(self):
-        self._queue = asyncio.Queue()
-        self._urgent_queue = asyncio.Queue()
-        self._event = asyncio.Event()
-
-    async def put(self, item: FrameProcessorQueueItem):
-        if isinstance(item.frame, SystemFrame):
-            await self._urgent_queue.put(item)
-        else:
-            await self._queue.put(item)
-        self._event.set()
-
-    async def get(self) -> FrameProcessorQueueItem:
-        # Wait for an item in any of the queues.
-        await self._event.wait()
-
-        if self._urgent_queue.empty():
-            item = await self._queue.get()
-            self._queue.task_done()
-        else:
-            item = await self._urgent_queue.get()
-            self._urgent_queue.task_done()
-
-        # Clear the event only if all queues are empty.
-        if self._queue.empty() and self._urgent_queue.empty():
-            self._event.clear()
-
-        return item
-
-    def clear(self):
-        self._queue = asyncio.Queue()
-
-        # Clear the event only if all queues are empty.
-        if self._queue.empty() and self._urgent_queue.empty():
-            self._event.clear()
-
-
 class FrameProcessor(BaseObject):
     def __init__(
         self,
@@ -115,21 +69,18 @@ class FrameProcessor(BaseObject):
         self._metrics = metrics or FrameProcessorMetrics()
         self._metrics.set_processor_name(self.name)
 
-        # Processors receive frames on a streaming queue which are then
-        # processed by a streaming task. This guarantees that all frames are
-        # processed in the same task. By default, the streaming queue is
-        # processed immediately but it may block if `pause_processing_frames()`
+        # Processors have an input queue. The input queue will be processed
+        # immediately (default) or it will block if `pause_processing_frames()`
         # is called. To resume processing frames we need to call
         # `resume_processing_frames()` which will wake up the event.
         self.__should_block_frames = False
-        self.__streaming_event = asyncio.Event()
-        self.__streaming_queue = FrameProcessorQueue()
-        self.__streaming_frame_task: Optional[asyncio.Task] = None
+        self.__input_event = asyncio.Event()
+        self.__input_frame_task: Optional[asyncio.Task] = None
 
-        self.__process_queue = asyncio.Queue()
-        self.__process_task: Optional[asyncio.Task] = None
-        self.__process_urgent_queue = asyncio.Queue()
-        self.__process_urgent_task: Optional[asyncio.Task] = None
+        # Every processor in Pipecat should only output frames from a single
+        # task. This avoid problems like audio overlapping. System frames are the
+        # exception to this rule. This create this task.
+        self.__push_frame_task: Optional[asyncio.Task] = None
 
     @property
     def id(self) -> int:
@@ -219,8 +170,7 @@ class FrameProcessor(BaseObject):
     async def cleanup(self):
         await super().cleanup()
         await self.__cancel_input_task()
-        await self.__cancel_process_task()
-        await self.__cancel_process_urgent_task()
+        await self.__cancel_push_task()
 
     def link(self, processor: "FrameProcessor"):
         self._next = processor
@@ -265,7 +215,7 @@ class FrameProcessor(BaseObject):
             await self.process_frame(frame, direction)
         else:
             # We queue everything else.
-            await self.__streaming_queue.put(FrameProcessorQueueItem(frame, direction, callback))
+            await self.__input_queue.put((frame, direction, callback))
 
     async def pause_processing_frames(self):
         logger.trace(f"{self}: pausing frame processing")
@@ -273,7 +223,7 @@ class FrameProcessor(BaseObject):
 
     async def resume_processing_frames(self):
         logger.trace(f"{self}: resuming frame processing")
-        self.__streaming_event.set()
+        self.__input_event.set()
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         if isinstance(frame, StartFrame):
@@ -300,6 +250,47 @@ class FrameProcessor(BaseObject):
         if not self._check_ready(frame):
             return
 
+        if isinstance(frame, SystemFrame):
+            await self.__internal_push_frame(frame, direction)
+        else:
+            await self.__push_queue.put((frame, direction))
+
+    async def __start(self, frame: StartFrame):
+        self.__create_input_task()
+        self.__create_push_task()
+
+    async def __cancel(self, frame: CancelFrame):
+        self._cancelling = True
+        await self.__cancel_input_task()
+        await self.__cancel_push_task()
+
+    #
+    # Handle interruptions
+    #
+
+    async def _start_interruption(self):
+        try:
+            # Cancel the push frame task. This will stop pushing frames downstream.
+            await self.__cancel_push_task()
+
+            # Cancel the input task. This will stop processing queued frames.
+            await self.__cancel_input_task()
+        except Exception as e:
+            logger.exception(f"Uncaught exception in {self}: {e}")
+            await self.push_error(ErrorFrame(str(e)))
+            raise
+
+        # Create a new input queue and task.
+        self.__create_input_task()
+
+        # Create a new output queue and task.
+        self.__create_push_task()
+
+    async def _stop_interruption(self):
+        # Nothing to do right now.
+        pass
+
+    async def __internal_push_frame(self, frame: Frame, direction: FrameDirection):
         try:
             timestamp = self._clock.get_time() if self._clock else 0
             if direction == FrameDirection.DOWNSTREAM and self._next:
@@ -332,49 +323,6 @@ class FrameProcessor(BaseObject):
             await self.push_error(ErrorFrame(str(e)))
             raise
 
-    async def __start(self, frame: StartFrame):
-        self.__create_process_task()
-        self.__create_process_urgent_task()
-        self.__create_input_task()
-
-    async def __cancel(self, frame: CancelFrame):
-        self._cancelling = True
-        await self.__cancel_input_task()
-        await self.__cancel_process_task()
-        await self.__cancel_process_urgent_task()
-
-    #
-    # Handle interruptions
-    #
-
-    async def _start_interruption(self):
-        try:
-            # Cancel the streaming task.
-            await self.__cancel_input_task()
-
-            # Cancel the task processing frames. We do not cancel the task that
-            # is processing urgent frames.
-            await self.__cancel_process_task()
-
-            # If there's an interruption we should not block frames anymore.
-            self.__should_block_frames = False
-
-            # Clear the streaming queue, since we don't want to process its
-            # frame anymore (except system and urgent frames).
-            self.__streaming_queue.clear()
-        except Exception as e:
-            logger.exception(f"Uncaught exception in {self}: {e}")
-            await self.push_error(ErrorFrame(str(e)))
-            raise
-
-        # Create a new tasks.
-        self.__create_process_task()
-        self.__create_input_task()
-
-    async def _stop_interruption(self):
-        # Nothing to do right now.
-        pass
-
     def _check_ready(self, frame: Frame):
         # If we are trying to push a frame but we still have no clock, it means
         # we didn't process a StartFrame.
@@ -386,60 +334,49 @@ class FrameProcessor(BaseObject):
         return True
 
     def __create_input_task(self):
-        if not self.__streaming_frame_task:
-            self.__streaming_frame_task = self.create_task(self.__streaming_frame_task_handler())
+        if not self.__input_frame_task:
+            self.__should_block_frames = False
+            self.__input_event.clear()
+            self.__input_queue = asyncio.Queue()
+            self.__input_frame_task = self.create_task(self.__input_frame_task_handler())
 
     async def __cancel_input_task(self):
-        if self.__streaming_frame_task:
-            await self.cancel_task(self.__streaming_frame_task)
-            self.__streaming_frame_task = None
+        if self.__input_frame_task:
+            await self.cancel_task(self.__input_frame_task)
+            self.__input_frame_task = None
 
-    def __create_process_task(self):
-        if not self.__process_task:
-            self.__process_queue = asyncio.Queue()
-            self.__process_task = self.create_task(
-                self.__process_task_handler(self.__process_queue)
-            )
-
-    async def __cancel_process_task(self):
-        if self.__process_task:
-            await self.cancel_task(self.__process_task)
-            self.__process_task = None
-
-    def __create_process_urgent_task(self):
-        if not self.__process_urgent_task:
-            self.__process_urgent_task = self.create_task(
-                self.__process_task_handler(self.__process_urgent_queue)
-            )
-
-    async def __cancel_process_urgent_task(self):
-        if self.__process_urgent_task:
-            await self.cancel_task(self.__process_urgent_task)
-            self.__process_urgent_task = None
-
-    async def __streaming_frame_task_handler(self):
+    async def __input_frame_task_handler(self):
         while True:
             if self.__should_block_frames:
                 logger.trace(f"{self}: frame processing paused")
-                await self.__streaming_event.wait()
-                self.__streaming_event.clear()
+                await self.__input_event.wait()
+                self.__input_event.clear()
                 self.__should_block_frames = False
                 logger.trace(f"{self}: frame processing resumed")
 
-            item = await self.__streaming_queue.get()
-
-            if isinstance(item.frame, SystemFrame):
-                await self.__process_urgent_queue.put(item)
-            else:
-                await self.__process_queue.put(item)
-
-    async def __process_task_handler(self, queue: asyncio.Queue):
-        while True:
-            item = await queue.get()
+            (frame, direction, callback) = await self.__input_queue.get()
 
             # Process the frame.
-            await self.process_frame(item.frame, item.direction)
+            await self.process_frame(frame, direction)
 
             # If this frame has an associated callback, call it now.
-            if item.callback:
-                await item.callback(self, item.frame, item.direction)
+            if callback:
+                await callback(self, frame, direction)
+
+            self.__input_queue.task_done()
+
+    def __create_push_task(self):
+        if not self.__push_frame_task:
+            self.__push_queue = asyncio.Queue()
+            self.__push_frame_task = self.create_task(self.__push_frame_task_handler())
+
+    async def __cancel_push_task(self):
+        if self.__push_frame_task:
+            await self.cancel_task(self.__push_frame_task)
+            self.__push_frame_task = None
+
+    async def __push_frame_task_handler(self):
+        while True:
+            (frame, direction) = await self.__push_queue.get()
+            await self.__internal_push_frame(frame, direction)
+            self.__push_queue.task_done()

src/pipecat/services/aws_nova_sonic/__init__.py

@@ -1 +1 @@
-from .aws import AWSNovaSonicLLMService
+from .aws import AWSNovaSonicLLMService, Params

tests/test_filters.py

@@ -9,7 +9,6 @@ import unittest
 from pipecat.frames.frames import (
     EndFrame,
     Frame,
-    StartInterruptionFrame,
     TextFrame,
     TranscriptionFrame,
     UserStartedSpeakingFrame,
@@ -58,8 +57,8 @@ class TestFrameFilter(unittest.IsolatedAsyncioTestCase):
 
     async def test_system_frame(self):
         filter = FrameFilter(types=())
-        frames_to_send = [StartInterruptionFrame()]
-        expected_down_frames = [StartInterruptionFrame]
+        frames_to_send = [UserStartedSpeakingFrame()]
+        expected_down_frames = [UserStartedSpeakingFrame]
         await run_test(
             filter,
             frames_to_send=frames_to_send,