From 7384b63b1d5cfdd6b29cc4440c5e5147278cf2f2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 15 May 2024 23:33:15 -0700 Subject: [PATCH] initial interruptions support --- CHANGELOG.md | 6 ++ src/pipecat/frames/frames.py | 2 +- src/pipecat/pipeline/task.py | 6 +- .../processors/aggregators/llm_response.py | 5 ++ .../processors/aggregators/user_response.py | 5 ++ src/pipecat/processors/frame_processor.py | 5 +- src/pipecat/transports/base_input.py | 49 +++++++++++---- src/pipecat/transports/base_output.py | 60 ++++++++++++++----- src/pipecat/transports/local/audio.py | 9 +-- src/pipecat/transports/local/tk.py | 10 ++-- src/pipecat/transports/services/daily.py | 12 ++-- 11 files changed, 124 insertions(+), 45 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ee98f3b02..5f5049a57 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,12 @@ All notable changes to **pipecat** will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added + +- Added initial interruptions support. + ## [0.0.16] - 2024-05-16 ### Fixed diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index 1278a7de1..7fcb0b6c2 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -187,7 +187,7 @@ class SystemFrame(Frame): @dataclass class StartFrame(SystemFrame): """This is the first frame that should be pushed down a pipeline.""" - pass + allow_interruptions: bool = False @dataclass diff --git a/src/pipecat/pipeline/task.py b/src/pipecat/pipeline/task.py index 2ecbe84ca..b693c0d58 100644 --- a/src/pipecat/pipeline/task.py +++ b/src/pipecat/pipeline/task.py @@ -31,11 +31,12 @@ class Source(FrameProcessor): class PipelineTask: - def __init__(self, pipeline: FrameProcessor): + def __init__(self, pipeline: FrameProcessor, allow_interruptions=False): self.id: int = obj_id() self.name: str = f"{self.__class__.__name__}#{obj_count(self)}" self._pipeline = pipeline + self._allow_interruptions = allow_interruptions self._task_queue = asyncio.Queue() self._up_queue = asyncio.Queue() @@ -70,7 +71,8 @@ class PipelineTask: raise Exception("Frames must be an iterable or async iterable") async def _process_task_queue(self): - await self._source.process_frame(StartFrame(), FrameDirection.DOWNSTREAM) + await self._source.process_frame( + StartFrame(allow_interruptions=self._allow_interruptions), FrameDirection.DOWNSTREAM) running = True while running: frame = await self._task_queue.get() diff --git a/src/pipecat/processors/aggregators/llm_response.py b/src/pipecat/processors/aggregators/llm_response.py index 37f8dab2e..82288a4e9 100644 --- a/src/pipecat/processors/aggregators/llm_response.py +++ b/src/pipecat/processors/aggregators/llm_response.py @@ -72,6 +72,8 @@ class LLMResponseAggregator(FrameProcessor): if isinstance(frame, self._start_frame): self._seen_start_frame = True self._aggregating = True + + await self.push_frame(frame, direction) elif isinstance(frame, self._end_frame): self._seen_end_frame = True @@ -83,6 +85,8 @@ class LLMResponseAggregator(FrameProcessor): # Send the aggregation if we are not aggregating anymore (i.e. no # more interim results received). send_aggregation = not self._aggregating + + await self.push_frame(frame, direction) elif isinstance(frame, self._accumulator_frame): if self._aggregating: self._aggregation += f" {frame.text}" @@ -109,6 +113,7 @@ class LLMResponseAggregator(FrameProcessor): # Reset self._aggregation = "" + self._aggregating = False self._seen_start_frame = False self._seen_end_frame = False self._seen_interim_results = False diff --git a/src/pipecat/processors/aggregators/user_response.py b/src/pipecat/processors/aggregators/user_response.py index e112fd3e4..ae063b3cd 100644 --- a/src/pipecat/processors/aggregators/user_response.py +++ b/src/pipecat/processors/aggregators/user_response.py @@ -89,6 +89,8 @@ class ResponseAggregator(FrameProcessor): if isinstance(frame, self._start_frame): self._seen_start_frame = True self._aggregating = True + + await self.push_frame(frame, direction) elif isinstance(frame, self._end_frame): self._seen_end_frame = True @@ -100,6 +102,8 @@ class ResponseAggregator(FrameProcessor): # Send the aggregation if we are not aggregating anymore (i.e. no # more interim results received). send_aggregation = not self._aggregating + + await self.push_frame(frame, direction) elif isinstance(frame, self._accumulator_frame): if self._aggregating: self._aggregation += f" {frame.text}" @@ -124,6 +128,7 @@ class ResponseAggregator(FrameProcessor): # Reset self._aggregation = "" + self._aggregating = False self._seen_start_frame = False self._seen_end_frame = False self._seen_interim_results = False diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 3bb750218..a7d330680 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -8,7 +8,7 @@ import asyncio from asyncio import AbstractEventLoop from enum import Enum -from pipecat.frames.frames import ErrorFrame, Frame +from pipecat.frames.frames import AudioRawFrame, ErrorFrame, Frame from pipecat.utils.utils import obj_count, obj_id from loguru import logger @@ -47,7 +47,8 @@ class FrameProcessor: async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): if direction == FrameDirection.DOWNSTREAM and self._next: - logger.trace(f"Pushing {frame} from {self} to {self._next}") + if not isinstance(frame, AudioRawFrame): + logger.trace(f"Pushing {frame} from {self} to {self._next}") await self._next.process_frame(frame, direction) elif direction == FrameDirection.UPSTREAM and self._prev: logger.trace(f"Pushing {frame} upstream from {self} to {self._prev}") diff --git a/src/pipecat/transports/base_input.py b/src/pipecat/transports/base_input.py index 265c8e6c4..a4f2a4cc6 100644 --- a/src/pipecat/transports/base_input.py +++ b/src/pipecat/transports/base_input.py @@ -30,23 +30,23 @@ class BaseInputTransport(FrameProcessor): self._params = params self._running = False + self._allow_interruptions = False # Start media threads. if self._params.audio_in_enabled or self._params.vad_enabled: self._audio_in_queue = queue.Queue() - # Start push frame task. This is the task that will push frames in + # Create push frame task. This is the task that will push frames in # order. So, a transport guarantees that all frames are pushed in the # same task. - loop = self.get_event_loop() - self._push_frame_task = loop.create_task(self._push_frame_task_handler()) - self._push_queue = asyncio.Queue() + self._create_push_task() - async def start(self): + async def start(self, frame: StartFrame): if self._running: return self._running = True + self._allow_interruptions = frame.allow_interruptions if self._params.audio_in_enabled or self._params.vad_enabled: loop = self.get_event_loop() @@ -65,6 +65,9 @@ class BaseInputTransport(FrameProcessor): await self._audio_in_thread await self._audio_out_thread + await self._internal_push_frame(None, None) + await self._push_frame_task + def vad_analyze(self, audio_frames: bytes) -> VADState: pass @@ -79,10 +82,15 @@ class BaseInputTransport(FrameProcessor): pass async def process_frame(self, frame: Frame, direction: FrameDirection): - if isinstance(frame, StartFrame): - await self.start() + if isinstance(frame, CancelFrame): + await self.stop() + # We don't queue a CancelFrame since we want to stop ASAP. + await self.push_frame(frame, direction) + elif isinstance(frame, StartFrame): + self._allow_interruption = frame.allow_interruptions + await self.start(frame) await self._internal_push_frame(frame, direction) - elif isinstance(frame, CancelFrame) or isinstance(frame, EndFrame): + elif isinstance(frame, EndFrame): await self.stop() await self._internal_push_frame(frame, direction) else: @@ -92,10 +100,15 @@ class BaseInputTransport(FrameProcessor): # Push frames task # + def _create_push_task(self): + loop = self.get_event_loop() + self._push_frame_task = loop.create_task(self._push_frame_task_handler()) + self._push_queue = asyncio.Queue() + async def _internal_push_frame( self, - frame: Frame, - direction: FrameDirection = FrameDirection.DOWNSTREAM): + frame: Frame | None, + direction: FrameDirection | None = FrameDirection.DOWNSTREAM): await self._push_queue.put((frame, direction)) async def _push_frame_task_handler(self): @@ -106,6 +119,16 @@ class BaseInputTransport(FrameProcessor): await self.push_frame(frame, direction) running = frame is not None + # + # Handle interruptions + # + + async def _handle_interruptions(self, frame: Frame): + if self._allow_interruptions and isinstance(frame, UserStartedSpeakingFrame): + self._push_frame_task.cancel() + self._create_push_task() + await self._internal_push_frame(frame) + # # Audio input # @@ -118,11 +141,13 @@ class BaseInputTransport(FrameProcessor): frame = UserStartedSpeakingFrame() elif new_vad_state == VADState.QUIET: frame = UserStoppedSpeakingFrame() + if frame: future = asyncio.run_coroutine_threadsafe( - self._internal_push_frame(frame), self.get_event_loop()) + self._handle_interruptions(frame), self.get_event_loop()) future.result() - vad_state = new_vad_state + + vad_state = new_vad_state return vad_state def _audio_in_thread_handler(self): diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index f46d629f8..520372803 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -7,9 +7,9 @@ import asyncio import itertools -from multiprocessing.context import _force_start_method import queue import time +import threading from PIL import Image from typing import List @@ -23,7 +23,9 @@ from pipecat.frames.frames import ( EndFrame, Frame, ImageRawFrame, - TransportMessageFrame) + TransportMessageFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame) from pipecat.transports.base_transport import TransportParams from loguru import logger @@ -37,6 +39,7 @@ class BaseOutputTransport(FrameProcessor): self._params = params self._running = False + self._allow_interruptions = False # These are the images that we should send to the camera at our desired # framerate. @@ -48,12 +51,14 @@ class BaseOutputTransport(FrameProcessor): self._sink_queue = queue.Queue() self._stopped_event = asyncio.Event() + self._is_interrupted = threading.Event() - async def start(self): + async def start(self, frame: StartFrame): if self._running: return self._running = True + self._allow_interruptions = frame.allow_interruptions loop = self.get_event_loop() @@ -93,12 +98,15 @@ class BaseOutputTransport(FrameProcessor): async def process_frame(self, frame: Frame, direction: FrameDirection): if isinstance(frame, StartFrame): - await self.start() + await self.start(frame) await self.push_frame(frame, direction) # EndFrame is managed in the queue handler. elif isinstance(frame, CancelFrame): await self.stop() await self.push_frame(frame, direction) + elif isinstance(frame, UserStartedSpeakingFrame) or isinstance(frame, UserStoppedSpeakingFrame): + await self._handle_interruptions(frame) + await self.push_frame(frame, direction) elif self._frame_managed_by_sink(frame): self._sink_queue.put(frame) else: @@ -117,28 +125,50 @@ class BaseOutputTransport(FrameProcessor): or isinstance(frame, TransportMessageFrame) or isinstance(frame, EndFrame)) + async def _handle_interruptions(self, frame: Frame): + if not self._allow_interruptions: + return + + if isinstance(frame, UserStartedSpeakingFrame): + self._is_interrupted.set() + elif isinstance(frame, UserStoppedSpeakingFrame): + self._is_interrupted.clear() + def _sink_thread_handler(self): - buffer = bytearray() + # 10ms bytes bytes_size_10ms = int(self._params.audio_out_sample_rate / 100) * \ self._params.audio_out_channels * 2 + + # We will send at least 100ms bytes. + smallest_write_size = bytes_size_10ms * 10 + + # Audio accumlation buffer + buffer = bytearray() while self._running: try: frame = self._sink_queue.get(timeout=1) + if isinstance(frame, EndFrame): # Send all remaining audio before stopping (multiple of 10ms of audio). self._send_audio_truncated(buffer, bytes_size_10ms) future = asyncio.run_coroutine_threadsafe(self.stop(), self.get_event_loop()) future.result() - elif isinstance(frame, AudioRawFrame): - if self._params.audio_out_enabled: - buffer.extend(frame.audio) - buffer = self._send_audio_truncated(buffer, bytes_size_10ms) - elif isinstance(frame, ImageRawFrame) and self._params.camera_out_enabled: - self._set_camera_image(frame) - elif isinstance(frame, SpriteFrame) and self._params.camera_out_enabled: - self._set_camera_images(frame.images) - elif isinstance(frame, TransportMessageFrame): - self.send_message(frame) + + if not self._is_interrupted.is_set(): + if isinstance(frame, AudioRawFrame): + if self._params.audio_out_enabled: + buffer.extend(frame.audio) + buffer = self._send_audio_truncated(buffer, smallest_write_size) + elif isinstance(frame, ImageRawFrame) and self._params.camera_out_enabled: + self._set_camera_image(frame) + elif isinstance(frame, SpriteFrame) and self._params.camera_out_enabled: + self._set_camera_images(frame.images) + elif isinstance(frame, TransportMessageFrame): + self.send_message(frame) + else: + # Send any remaining audio + self._send_audio_truncated(buffer, bytes_size_10ms) + buffer = bytearray() except queue.Empty: pass except BaseException as e: diff --git a/src/pipecat/transports/local/audio.py b/src/pipecat/transports/local/audio.py index ee85bcf61..771715111 100644 --- a/src/pipecat/transports/local/audio.py +++ b/src/pipecat/transports/local/audio.py @@ -6,6 +6,7 @@ import asyncio +from pipecat.frames.frames import StartFrame from pipecat.processors.frame_processor import FrameProcessor from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport @@ -37,8 +38,8 @@ class LocalAudioInputTransport(BaseInputTransport): def read_raw_audio_frames(self, frame_count: int) -> bytes: return self._in_stream.read(frame_count, exception_on_overflow=False) - async def start(self): - await super().start() + async def start(self, frame: StartFrame): + await super().start(frame) self._in_stream.start_stream() async def stop(self): @@ -68,8 +69,8 @@ class LocalAudioOutputTransport(BaseOutputTransport): def write_raw_audio_frames(self, frames: bytes): self._out_stream.write(frames) - async def start(self): - await super().start() + async def start(self, frame: StartFrame): + await super().start(frame) self._out_stream.start_stream() async def stop(self): diff --git a/src/pipecat/transports/local/tk.py b/src/pipecat/transports/local/tk.py index 4165f941c..782c01dae 100644 --- a/src/pipecat/transports/local/tk.py +++ b/src/pipecat/transports/local/tk.py @@ -9,7 +9,7 @@ import asyncio import numpy as np import tkinter as tk -from pipecat.frames.frames import ImageRawFrame +from pipecat.frames.frames import ImageRawFrame, StartFrame from pipecat.processors.frame_processor import FrameProcessor from pipecat.transports.base_input import BaseInputTransport from pipecat.transports.base_output import BaseOutputTransport @@ -48,8 +48,8 @@ class TkInputTransport(BaseInputTransport): def read_raw_audio_frames(self, frame_count: int) -> bytes: return self._in_stream.read(frame_count, exception_on_overflow=False) - async def start(self): - await super().start() + async def start(self, frame: StartFrame): + await super().start(frame) self._in_stream.start_stream() async def stop(self): @@ -89,8 +89,8 @@ class TkOutputTransport(BaseOutputTransport): def write_frame_to_camera(self, frame: ImageRawFrame): self.get_event_loop().call_soon(self._write_frame_to_tk, frame) - async def start(self): - await super().start() + async def start(self, frame: StartFrame): + await super().start(frame) self._out_stream.start_stream() async def stop(self): diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index cca69a284..506c962ea 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -29,6 +29,7 @@ from pipecat.frames.frames import ( ImageRawFrame, InterimTranscriptionFrame, SpriteFrame, + StartFrame, TranscriptionFrame, TransportMessageFrame, UserImageRawFrame, @@ -283,6 +284,7 @@ class DailyTransportClient(EventHandler): error_msg = f"Error joining {self._room_url}: {error}" logger.error(error_msg) self._callbacks.on_error(error_msg) + self._sync_response["join"].task_done() except queue.Empty: error_msg = f"Time out joining {self._room_url}" logger.error(error_msg) @@ -320,6 +322,7 @@ class DailyTransportClient(EventHandler): error_msg = f"Error leaving {self._room_url}: {error}" logger.error(error_msg) self._callbacks.on_error(error_msg) + self._sync_response["leave"].task_done() except queue.Empty: error_msg = f"Time out leaving {self._room_url}" logger.error(error_msg) @@ -432,13 +435,13 @@ class DailyInputTransport(BaseInputTransport): self._video_renderers = {} self._camera_in_queue = queue.Queue() - async def start(self): + async def start(self, frame: StartFrame): if self._running: return # Join the room. await self._client.join() # This will set _running=True - await super().start() + await super().start(frame) # Create camera in thread (runs if _running is true). loop = asyncio.get_running_loop() self._camera_in_thread = loop.run_in_executor(None, self._camera_in_thread_handler) @@ -547,6 +550,7 @@ class DailyInputTransport(BaseInputTransport): future = asyncio.run_coroutine_threadsafe( self._internal_push_frame(frame), self.get_event_loop()) future.result() + self._camera_in_queue.task_done() except queue.Empty: pass except BaseException as e: @@ -560,11 +564,11 @@ class DailyOutputTransport(BaseOutputTransport): self._client = client - async def start(self): + async def start(self, frame: StartFrame): if self._running: return # This will set _running=True - await super().start() + await super().start(frame) # Join the room. await self._client.join()