wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

initial interruptions support

Browse Source
This commit is contained in:
Aleix Conchillo Flaqué
2024-05-15 23:33:15 -07:00
parent ba6ecf541f
commit 7384b63b1d
11 changed files with 124 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
CHANGELOG.md
View File

@@ -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

2
src/pipecat/frames/frames.py
View File

@@ -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

6
src/pipecat/pipeline/task.py
View File

@@ -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()

5
src/pipecat/processors/aggregators/llm_response.py
View File

@@ -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

5
src/pipecat/processors/aggregators/user_response.py
View File

@@ -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

5
src/pipecat/processors/frame_processor.py
View File

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

49
src/pipecat/transports/base_input.py
View File

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

60
src/pipecat/transports/base_output.py
View File

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

9
src/pipecat/transports/local/audio.py
View File

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

10
src/pipecat/transports/local/tk.py
View File

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

12
src/pipecat/transports/services/daily.py
View File

@@ -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()