From a5c7b02a736491b7c8a28a68de2a5a9affd3138e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 20 Nov 2024 10:23:51 +0100 Subject: [PATCH 1/9] frames: input frames are now system frames Input frames from a transport should be processed fast and there's no need for them to be queued internally in each element. --- src/pipecat/frames/frames.py | 289 +++++++++++++++++++---------------- 1 file changed, 157 insertions(+), 132 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index e057e97da..d10a1cd21 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -21,6 +21,8 @@ def format_pts(pts: int | None): @dataclass class Frame: + """Base frame class.""" + id: int = field(init=False) name: str = field(init=False) pts: Optional[int] = field(init=False) @@ -35,13 +37,47 @@ class Frame: @dataclass -class DataFrame(Frame): +class SystemFrame(Frame): + """System frames are frames that are not internally queued by any of the + frame processors and should be processed immediately. + + """ + pass @dataclass -class AudioRawFrame(DataFrame): - """A chunk of audio.""" +class DataFrame(Frame): + """Data frames are frames that will be processed in order and usually + contain data such as LLM context, text, audio or images. + + """ + + pass + + +@dataclass +class ControlFrame(Frame): + """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. + + """ + + pass + + +# +# Data frames. +# + + +@dataclass +class OutputAudioRawFrame(DataFrame): + """A chunk of audio. Will be played by the output transport if the + transport's microphone has been enabled. + + """ audio: bytes sample_rate: int @@ -57,32 +93,8 @@ class AudioRawFrame(DataFrame): @dataclass -class InputAudioRawFrame(AudioRawFrame): - """A chunk of audio usually coming from an input transport.""" - - pass - - -@dataclass -class OutputAudioRawFrame(AudioRawFrame): - """A chunk of audio. Will be played by the output transport if the - transport's microphone has been enabled. - - """ - - pass - - -@dataclass -class TTSAudioRawFrame(OutputAudioRawFrame): - """A chunk of output audio generated by a TTS service.""" - - pass - - -@dataclass -class ImageRawFrame(DataFrame): - """An image. Will be shown by the transport if the transport's camera is +class OutputImageRawFrame(DataFrame): + """An image that will be shown by the transport if the transport's camera is enabled. """ @@ -97,47 +109,16 @@ class ImageRawFrame(DataFrame): @dataclass -class InputImageRawFrame(ImageRawFrame): +class TTSAudioRawFrame(OutputAudioRawFrame): + """A chunk of output audio generated by a TTS service.""" + pass -@dataclass -class OutputImageRawFrame(ImageRawFrame): - pass - - -@dataclass -class UserImageRawFrame(InputImageRawFrame): - """An image associated to a user. Will be shown by the transport if the - transport's camera is enabled. - - """ - - user_id: str - - def __str__(self): - pts = format_pts(self.pts) - return f"{self.name}(pts: {pts}, user: {self.user_id}, size: {self.size}, format: {self.format})" - - -@dataclass -class VisionImageRawFrame(InputImageRawFrame): - """An image with an associated text to ask for a description of it. Will be - shown by the transport if the transport's camera is enabled. - - """ - - text: str | None - - def __str__(self): - pts = format_pts(self.pts) - return f"{self.name}(pts: {pts}, text: [{self.text}], size: {self.size}, format: {self.format})" - - @dataclass class URLImageRawFrame(OutputImageRawFrame): - """An image with an associated URL. Will be shown by the transport if the - transport's camera is enabled. + """An output image with an associated URL. These images are usually + generated by third-party services that provide a URL to download the image. """ @@ -149,14 +130,14 @@ class URLImageRawFrame(OutputImageRawFrame): @dataclass -class SpriteFrame(Frame): +class SpriteFrame(DataFrame): """An animated sprite. Will be shown by the transport if the transport's camera is enabled. Will play at the framerate specified in the transport's `camera_out_framerate` constructor parameter. """ - images: List[ImageRawFrame] + images: List[OutputImageRawFrame] def __str__(self): pts = format_pts(self.pts) @@ -166,7 +147,7 @@ class SpriteFrame(Frame): @dataclass class TextFrame(DataFrame): """A chunk of text. Emitted by LLM services, consumed by TTS services, can - be used to send text through pipelines. + be used to send text through processors. """ @@ -177,41 +158,13 @@ class TextFrame(DataFrame): return f"{self.name}(pts: {pts}, text: [{self.text}])" -@dataclass -class TranscriptionFrame(TextFrame): - """A text frame with transcription-specific data. Will be placed in the - transport's receive queue when a participant speaks. - - """ - - user_id: str - timestamp: str - language: Language | None = None - - def __str__(self): - return f"{self.name}(user: {self.user_id}, text: [{self.text}], language: {self.language}, timestamp: {self.timestamp})" - - -@dataclass -class InterimTranscriptionFrame(TextFrame): - """A text frame with interim transcription-specific data. Will be placed in - the transport's receive queue when a participant speaks.""" - - user_id: str - timestamp: str - language: Language | None = None - - def __str__(self): - return f"{self.name}(user: {self.user_id}, text: [{self.text}], language: {self.language}, timestamp: {self.timestamp})" - - @dataclass class LLMMessagesFrame(DataFrame): """A frame containing a list of LLM messages. Used to signal that an LLM - service should run a chat completion and emit an LLMStartFrames, TextFrames - and an LLMEndFrame. Note that the messages property on this class is - mutable, and will be be updated by various ResponseAggregator frame - processors. + service should run a chat completion and emit an LLMFullResponseStartFrame, + TextFrames and an LLMFullResponseStartFrame. Note that the `messages` + property in this class is mutable, and will be be updated by various + aggregators. """ @@ -220,7 +173,7 @@ class LLMMessagesFrame(DataFrame): @dataclass class LLMMessagesAppendFrame(DataFrame): - """A frame containing a list of LLM messages that neeed to be added to the + """A frame containing a list of LLM messages that need to be added to the current context. """ @@ -274,17 +227,6 @@ class TransportMessageFrame(DataFrame): return f"{self.name}(message: {self.message})" -@dataclass -class FunctionCallResultFrame(DataFrame): - """A frame containing the result of an LLM function (tool) call.""" - - function_name: str - tool_call_id: str - arguments: str - result: Any - run_llm: bool = True - - # # App frames. Application user-defined frames. # @@ -300,11 +242,6 @@ class AppFrame(Frame): # -@dataclass -class SystemFrame(Frame): - pass - - @dataclass class StartFrame(SystemFrame): """This is the first frame that should be pushed down a pipeline.""" @@ -461,14 +398,10 @@ class BotSpeakingFrame(SystemFrame): @dataclass -class UserImageRequestFrame(SystemFrame): - """A frame user to request an image from the given user.""" +class MetricsFrame(SystemFrame): + """Emitted by processor that can compute metrics like latencies.""" - user_id: str - context: Optional[Any] = None - - def __str__(self): - return f"{self.name}, user: {self.user_id}" + data: List[MetricsData] @dataclass @@ -480,6 +413,17 @@ class FunctionCallInProgressFrame(SystemFrame): arguments: str +@dataclass +class FunctionCallResultFrame(SystemFrame): + """A frame containing the result of an LLM function (tool) call.""" + + function_name: str + tool_call_id: str + arguments: str + result: Any + run_llm: bool = True + + @dataclass class TransportMessageUrgentFrame(SystemFrame): message: Any @@ -489,10 +433,96 @@ class TransportMessageUrgentFrame(SystemFrame): @dataclass -class MetricsFrame(SystemFrame): - """Emitted by processor that can compute metrics like latencies.""" +class TranscriptionFrame(SystemFrame): + """A text frame with transcription-specific data. Will be placed in the + transport's receive queue when a participant speaks. - data: List[MetricsData] + """ + + text: str + user_id: str + timestamp: str + language: Language | None = None + + def __str__(self): + return f"{self.name}(user: {self.user_id}, text: [{self.text}], language: {self.language}, timestamp: {self.timestamp})" + + +@dataclass +class InterimTranscriptionFrame(SystemFrame): + """A text frame with interim transcription-specific data. Will be placed in + the transport's receive queue when a participant speaks.""" + + text: str + user_id: str + timestamp: str + language: Language | None = None + + def __str__(self): + return f"{self.name}(user: {self.user_id}, text: [{self.text}], language: {self.language}, timestamp: {self.timestamp})" + + +@dataclass +class UserImageRequestFrame(SystemFrame): + """A frame user to request an image from the given user.""" + + user_id: str + context: Optional[Any] = None + + def __str__(self): + return f"{self.name}, user: {self.user_id}" + + +@dataclass +class InputAudioRawFrame(SystemFrame): + """A chunk of audio usually coming from an input transport.""" + + audio: bytes + sample_rate: int + num_channels: int + + def __post_init__(self): + super().__post_init__() + self.num_frames = int(len(self.audio) / (self.num_channels * 2)) + + def __str__(self): + pts = format_pts(self.pts) + return f"{self.name}(pts: {pts}, size: {len(self.audio)}, frames: {self.num_frames}, sample_rate: {self.sample_rate}, channels: {self.num_channels})" + + +@dataclass +class InputImageRawFrame(SystemFrame): + """An image usually coming from an input transport.""" + + image: bytes + size: Tuple[int, int] + format: str | None + + def __str__(self): + pts = format_pts(self.pts) + return f"{self.name}(pts: {pts}, size: {self.size}, format: {self.format})" + + +@dataclass +class UserImageRawFrame(InputImageRawFrame): + """An image associated to a user.""" + + user_id: str + + def __str__(self): + pts = format_pts(self.pts) + return f"{self.name}(pts: {pts}, user: {self.user_id}, size: {self.size}, format: {self.format})" + + +@dataclass +class VisionImageRawFrame(InputImageRawFrame): + """An image with an associated text to ask for a description of it.""" + + text: str | None + + def __str__(self): + pts = format_pts(self.pts) + return f"{self.name}(pts: {pts}, text: [{self.text}], size: {self.size}, format: {self.format})" # @@ -500,11 +530,6 @@ class MetricsFrame(SystemFrame): # -@dataclass -class ControlFrame(Frame): - pass - - @dataclass class EndFrame(ControlFrame): """Indicates that a pipeline has ended and frame processors and pipelines From d6c944cdc18346f5634dca637d35e759f39380a5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 2 Dec 2024 13:06:30 -0800 Subject: [PATCH 2/9] processors(audio): fix AudioBufferProcessor interruptions --- src/pipecat/audio/utils.py | 31 +++++ .../audio/audio_buffer_processor.py | 124 +++++++++--------- 2 files changed, 90 insertions(+), 65 deletions(-) diff --git a/src/pipecat/audio/utils.py b/src/pipecat/audio/utils.py index f2260c57b..057942e04 100644 --- a/src/pipecat/audio/utils.py +++ b/src/pipecat/audio/utils.py @@ -18,6 +18,37 @@ def resample_audio(audio: bytes, original_rate: int, target_rate: int) -> bytes: return resampled_audio.astype(np.int16).tobytes() +def mix_audio(audio1: bytes, audio2: bytes) -> bytes: + data1 = np.frombuffer(audio1, dtype=np.int16) + data2 = np.frombuffer(audio2, dtype=np.int16) + + # Max length + max_length = max(len(data1), len(data2)) + + # Zero-pad the arrays to the same length + padded1 = np.pad(data1, (0, max_length - len(data1)), mode="constant") + padded2 = np.pad(data2, (0, max_length - len(data2)), mode="constant") + + # Mix the arrays + mixed_audio = padded1.astype(np.int32) + padded2.astype(np.int32) + mixed_audio = np.clip(mixed_audio, -32768, 32767).astype(np.int16) + + return mixed_audio.astype(np.int16).tobytes() + + +def interleave_stereo_audio(left_audio: bytes, right_audio: bytes) -> bytes: + left = np.frombuffer(left_audio, dtype=np.int16) + right = np.frombuffer(right_audio, dtype=np.int16) + + min_length = min(len(left), len(right)) + left = left[:min_length] + right = right[:min_length] + + stereo = np.column_stack((left, right)) + + return stereo.astype(np.int16).tobytes() + + def normalize_value(value, min_value, max_value): normalized = (value - min_value) / (max_value - min_value) normalized_clamped = max(0, min(1, normalized)) diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index 25a1f0237..d7c07c611 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -7,8 +7,8 @@ import wave from io import BytesIO +from pipecat.audio.utils import interleave_stereo_audio, mix_audio, resample_audio from pipecat.frames.frames import ( - AudioRawFrame, Frame, InputAudioRawFrame, OutputAudioRawFrame, @@ -17,84 +17,78 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor class AudioBufferProcessor(FrameProcessor): - def __init__(self, **kwargs): - """ - Initialize the AudioBufferProcessor. + """This processor buffers audio raw frames (input and output) that can later + be obtained as an in-memory WAV. You can provide the desired output + `sample_rate` and incoming audio frames will resampled to match it. Also, + you can provide the number of channels, 1 for mono and 2 for stereo. With + mono audio user and bot audio will be mixed, in the case of stereo the left + channel will be used for the user's audio and the right channel for the bot. - This constructor sets up the initial state for audio processing: - - audio_buffer: A bytearray to store incoming audio data. - - num_channels: The number of audio channels (initialized as None). - - sample_rate: The sample rate of the audio (initialized as None). + """ - The num_channels and sample_rate are set to None initially and will be - populated when the first audio frame is processed. - """ + def __init__(self, *, sample_rate: int = 24000, num_channels: int = 1, **kwargs): super().__init__(**kwargs) - self._user_audio_buffer = bytearray() - self._assistant_audio_buffer = bytearray() - self._num_channels = None - self._sample_rate = None + self._sample_rate = sample_rate + self._num_channels = num_channels - def _buffer_has_audio(self, buffer: bytearray): + self._user_audio_buffer = bytearray() + self._bot_audio_buffer = bytearray() + + def _buffer_has_audio(self, buffer: bytearray) -> bool: return buffer is not None and len(buffer) > 0 - def has_audio(self): - return ( - self._buffer_has_audio(self._user_audio_buffer) - and self._buffer_has_audio(self._assistant_audio_buffer) - and self._sample_rate is not None + def has_audio(self) -> bool: + return self._buffer_has_audio(self._user_audio_buffer) and self._buffer_has_audio( + self._bot_audio_buffer ) def reset_audio_buffer(self): self._user_audio_buffer = bytearray() - self._assistant_audio_buffer = bytearray() + self._bot_audio_buffer = bytearray() - def merge_audio_buffers(self): + def merge_audio_buffers(self) -> bytes: + if self._num_channels == 1: + return self._merge_mono() + elif self._num_channels == 2: + return self._merge_stereo() + else: + return b"" + + async def process_frame(self, frame: Frame, direction: FrameDirection): + await super().process_frame(frame, direction) + + # Include all audio from the user. + if isinstance(frame, InputAudioRawFrame): + resampled = resample_audio(frame.audio, frame.sample_rate, self._sample_rate) + self._user_audio_buffer.extend(resampled) + # Sync the bot's buffer to the user's buffer by adding silence if needed + if len(self._user_audio_buffer) > len(self._bot_audio_buffer): + silence = b"\x00" * len(resampled) + self._bot_audio_buffer.extend(silence) + + # If the bot is speaking, include all audio from the bot. + if isinstance(frame, OutputAudioRawFrame): + resampled = resample_audio(frame.audio, frame.sample_rate, self._sample_rate) + self._bot_audio_buffer.extend(resampled) + + def _merge_mono(self) -> bytes: + with BytesIO() as buffer: + with wave.open(buffer, "wb") as wf: + wf.setnchannels(1) + wf.setsampwidth(2) + wf.setframerate(self._sample_rate) + mixed = mix_audio(bytes(self._user_audio_buffer), bytes(self._bot_audio_buffer)) + wf.writeframes(mixed) + return buffer.getvalue() + + def _merge_stereo(self) -> bytes: with BytesIO() as buffer: with wave.open(buffer, "wb") as wf: wf.setnchannels(2) wf.setsampwidth(2) wf.setframerate(self._sample_rate) - # Interleave the two audio streams - max_length = max(len(self._user_audio_buffer), len(self._assistant_audio_buffer)) - interleaved = bytearray(max_length * 2) - - for i in range(0, max_length, 2): - if i < len(self._user_audio_buffer): - interleaved[i * 2] = self._user_audio_buffer[i] - interleaved[i * 2 + 1] = self._user_audio_buffer[i + 1] - else: - interleaved[i * 2] = 0 - interleaved[i * 2 + 1] = 0 - - if i < len(self._assistant_audio_buffer): - interleaved[i * 2 + 2] = self._assistant_audio_buffer[i] - interleaved[i * 2 + 3] = self._assistant_audio_buffer[i + 1] - else: - interleaved[i * 2 + 2] = 0 - interleaved[i * 2 + 3] = 0 - - wf.writeframes(interleaved) + stereo = interleave_stereo_audio( + bytes(self._user_audio_buffer), bytes(self._bot_audio_buffer) + ) + wf.writeframes(stereo) return buffer.getvalue() - - async def process_frame(self, frame: Frame, direction: FrameDirection): - await super().process_frame(frame, direction) - if isinstance(frame, AudioRawFrame) and self._sample_rate is None: - self._sample_rate = frame.sample_rate - - # include all audio from the user - if isinstance(frame, InputAudioRawFrame): - self._user_audio_buffer.extend(frame.audio) - # Sync the assistant's buffer to the user's buffer by adding silence if needed - if len(self._user_audio_buffer) > len(self._assistant_audio_buffer): - silence_length = len(self._user_audio_buffer) - len(self._assistant_audio_buffer) - silence = b"\x00" * silence_length - self._assistant_audio_buffer.extend(silence) - - # if the assistant is speaking, include all audio from the assistant, - if isinstance(frame, OutputAudioRawFrame): - self._assistant_audio_buffer.extend(frame.audio) - - # do not push the user's audio frame, doing so will result in echo - if not isinstance(frame, InputAudioRawFrame): - await self.push_frame(frame, direction) From a6606a4040523a28a897699014568c1f59bb5b8b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 2 Dec 2024 13:07:27 -0800 Subject: [PATCH 3/9] transports(base_output): remove unused code --- src/pipecat/transports/base_output.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/src/pipecat/transports/base_output.py b/src/pipecat/transports/base_output.py index afb93ebf8..c8aa5ef70 100644 --- a/src/pipecat/transports/base_output.py +++ b/src/pipecat/transports/base_output.py @@ -443,11 +443,6 @@ class BaseOutputTransport(FrameProcessor): return without_mixer(vad_stop_secs) async def _audio_out_task_handler(self): - wait_time = ( - self._params.vad_analyzer.params.stop_secs - if self._params.vad_analyzer - else VAD_STOP_SECS - ) try: async for frame in self._next_audio_frame(): # Notify the bot started speaking upstream if necessary and that From 42658ecd92753ffad33eb5bebe2f4ab085f283f2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 2 Dec 2024 14:49:28 -0800 Subject: [PATCH 4/9] frames: use mixins for audio and image data --- src/pipecat/frames/frames.py | 48 +++++++++++++++++++++--------------- 1 file changed, 28 insertions(+), 20 deletions(-) diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index d10a1cd21..b99a7307c 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -67,22 +67,42 @@ class ControlFrame(Frame): pass +# +# Mixins +# + + +@dataclass +class AudioRawFrame: + """A chunk of audio.""" + + audio: bytes + sample_rate: int + num_channels: int + num_frames: int = field(init=False) + + +@dataclass +class ImageRawFrame: + """A raw image.""" + + image: bytes + size: Tuple[int, int] + format: str | None + + # # Data frames. # @dataclass -class OutputAudioRawFrame(DataFrame): +class OutputAudioRawFrame(DataFrame, AudioRawFrame): """A chunk of audio. Will be played by the output transport if the transport's microphone has been enabled. """ - audio: bytes - sample_rate: int - num_channels: int - def __post_init__(self): super().__post_init__() self.num_frames = int(len(self.audio) / (self.num_channels * 2)) @@ -93,16 +113,12 @@ class OutputAudioRawFrame(DataFrame): @dataclass -class OutputImageRawFrame(DataFrame): +class OutputImageRawFrame(DataFrame, ImageRawFrame): """An image that will be shown by the transport if the transport's camera is enabled. """ - image: bytes - size: Tuple[int, int] - format: str | None - def __str__(self): pts = format_pts(self.pts) return f"{self.name}(pts: {pts}, size: {self.size}, format: {self.format})" @@ -474,13 +490,9 @@ class UserImageRequestFrame(SystemFrame): @dataclass -class InputAudioRawFrame(SystemFrame): +class InputAudioRawFrame(SystemFrame, AudioRawFrame): """A chunk of audio usually coming from an input transport.""" - audio: bytes - sample_rate: int - num_channels: int - def __post_init__(self): super().__post_init__() self.num_frames = int(len(self.audio) / (self.num_channels * 2)) @@ -491,13 +503,9 @@ class InputAudioRawFrame(SystemFrame): @dataclass -class InputImageRawFrame(SystemFrame): +class InputImageRawFrame(SystemFrame, ImageRawFrame): """An image usually coming from an input transport.""" - image: bytes - size: Tuple[int, int] - format: str | None - def __str__(self): pts = format_pts(self.pts) return f"{self.name}(pts: {pts}, size: {self.size}, format: {self.format})" From 52cefaa9d680023605a1e33fc7a0d032d55ab9c7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 2 Dec 2024 14:49:52 -0800 Subject: [PATCH 5/9] frames: remove AppFrame --- docs/frame.md | 3 --- examples/foundational/05-sync-speech-and-image.py | 4 ++-- src/pipecat/frames/frames.py | 10 ---------- src/pipecat/processors/filters/frame_filter.py | 8 ++------ 4 files changed, 4 insertions(+), 21 deletions(-) diff --git a/docs/frame.md b/docs/frame.md index 9026f30c1..7f50ffe62 100644 --- a/docs/frame.md +++ b/docs/frame.md @@ -96,9 +96,6 @@ Notable control frames: ## 7. Special Purpose Frames -### AppFrame -Base class for application-specific custom frames. - ### MetricsFrame Contains performance metrics data. diff --git a/examples/foundational/05-sync-speech-and-image.py b/examples/foundational/05-sync-speech-and-image.py index 5477d0691..64f85930b 100644 --- a/examples/foundational/05-sync-speech-and-image.py +++ b/examples/foundational/05-sync-speech-and-image.py @@ -12,7 +12,7 @@ import sys from dataclasses import dataclass from pipecat.frames.frames import ( - AppFrame, + DataFrame, Frame, LLMFullResponseStartFrame, LLMMessagesFrame, @@ -42,7 +42,7 @@ logger.add(sys.stderr, level="DEBUG") @dataclass -class MonthFrame(AppFrame): +class MonthFrame(DataFrame): month: str def __str__(self): diff --git a/src/pipecat/frames/frames.py b/src/pipecat/frames/frames.py index b99a7307c..17059700a 100644 --- a/src/pipecat/frames/frames.py +++ b/src/pipecat/frames/frames.py @@ -243,16 +243,6 @@ class TransportMessageFrame(DataFrame): return f"{self.name}(message: {self.message})" -# -# App frames. Application user-defined frames. -# - - -@dataclass -class AppFrame(Frame): - pass - - # # System frames # diff --git a/src/pipecat/processors/filters/frame_filter.py b/src/pipecat/processors/filters/frame_filter.py index f4c4b0f61..11f2e601a 100644 --- a/src/pipecat/processors/filters/frame_filter.py +++ b/src/pipecat/processors/filters/frame_filter.py @@ -6,7 +6,7 @@ from typing import Tuple, Type -from pipecat.frames.frames import AppFrame, ControlFrame, Frame, SystemFrame +from pipecat.frames.frames import ControlFrame, Frame, SystemFrame from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -23,11 +23,7 @@ class FrameFilter(FrameProcessor): if isinstance(frame, self._types): return True - return ( - isinstance(frame, AppFrame) - or isinstance(frame, ControlFrame) - or isinstance(frame, SystemFrame) - ) + return isinstance(frame, ControlFrame) or isinstance(frame, SystemFrame) async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) From a6a49109319e6ae53bc48ddbb75fd4cb82434213 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 3 Dec 2024 10:48:50 -0800 Subject: [PATCH 6/9] transports(services): incoming transport messages should be urgent --- src/pipecat/transports/services/daily.py | 2 +- src/pipecat/transports/services/livekit.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/src/pipecat/transports/services/daily.py b/src/pipecat/transports/services/daily.py index 7b0cd6c00..a5936e80a 100644 --- a/src/pipecat/transports/services/daily.py +++ b/src/pipecat/transports/services/daily.py @@ -723,7 +723,7 @@ class DailyInputTransport(BaseInputTransport): await self.push_frame(frame) async def push_app_message(self, message: Any, sender: str): - frame = DailyTransportMessageFrame(message=message, participant_id=sender) + frame = DailyTransportMessageUrgentFrame(message=message, participant_id=sender) await self.push_frame(frame) # diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index 5d4fbdd6c..6c99f6a66 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -342,7 +342,7 @@ class LiveKitInputTransport(BaseInputTransport): return self._vad_analyzer async def push_app_message(self, message: Any, sender: str): - frame = LiveKitTransportMessageFrame(message=message, participant_id=sender) + frame = LiveKitTransportMessageUrgentFrame(message=message, participant_id=sender) await self.push_frame(frame) async def _audio_in_task_handler(self): From 322dd0cea11474371dd1656fe432ab4b087f70b9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 3 Dec 2024 14:10:50 -0800 Subject: [PATCH 7/9] AudioBufferProcessor: use on_audio_data event handler to retrieve audio --- examples/canonical-metrics/bot.py | 1 - examples/chatbot-audio-recording/bot.py | 29 +++--- .../chatbot-audio-recording/requirements.txt | 1 + .../audio/audio_buffer_processor.py | 88 ++++++++++--------- src/pipecat/processors/frameworks/rtvi.py | 1 + src/pipecat/services/canonical.py | 42 ++++++--- 6 files changed, 97 insertions(+), 65 deletions(-) diff --git a/examples/canonical-metrics/bot.py b/examples/canonical-metrics/bot.py index efe4823fb..71aca70f3 100644 --- a/examples/canonical-metrics/bot.py +++ b/examples/canonical-metrics/bot.py @@ -102,7 +102,6 @@ async def main(): audio_buffer_processor=audio_buffer_processor, aiohttp_session=session, api_key=os.getenv("CANONICAL_API_KEY"), - api_url=os.getenv("CANONICAL_API_URL"), call_id=str(uuid.uuid4()), assistant="pipecat-chatbot", assistant_speaks_first=True, diff --git a/examples/chatbot-audio-recording/bot.py b/examples/chatbot-audio-recording/bot.py index 708de522d..f020a9626 100644 --- a/examples/chatbot-audio-recording/bot.py +++ b/examples/chatbot-audio-recording/bot.py @@ -4,7 +4,9 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import aiofiles import asyncio +import io import os import sys @@ -32,15 +34,17 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def save_audio(audiobuffer): - if audiobuffer.has_audio(): - merged_audio = audiobuffer.merge_audio_buffers() +async def save_audio(audio: bytes, sample_rate: int, num_channels: int): + if len(audio) > 0: filename = f"conversation_recording{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.wav" - with wave.open(filename, "wb") as wf: - wf.setnchannels(2) - wf.setsampwidth(2) - wf.setframerate(audiobuffer._sample_rate) - wf.writeframes(merged_audio) + with io.BytesIO() as buffer: + with wave.open(buffer, "wb") as wf: + wf.setsampwidth(2) + wf.setnchannels(num_channels) + wf.setframerate(sample_rate) + wf.writeframes(audio) + async with aiofiles.open(filename, "wb") as file: + await file.write(buffer.getvalue()) print(f"Merged audio saved to {filename}") else: print("No audio data to save") @@ -106,7 +110,9 @@ async def main(): context = OpenAILLMContext(messages) context_aggregator = llm.create_context_aggregator(context) - audiobuffer = AudioBufferProcessor() + # Save audio every 10 seconds. + audiobuffer = AudioBufferProcessor(buffer_size=480000) + pipeline = Pipeline( [ transport.input(), # microphone @@ -121,6 +127,10 @@ async def main(): task = PipelineTask(pipeline, PipelineParams(allow_interruptions=True)) + @audiobuffer.event_handler("on_audio_data") + async def on_audio_data(buffer, audio, sample_rate, num_channels): + await save_audio(audio, sample_rate, num_channels) + @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): await transport.capture_participant_transcription(participant["id"]) @@ -130,7 +140,6 @@ async def main(): async def on_participant_left(transport, participant, reason): print(f"Participant left: {participant}") await task.queue_frame(EndFrame()) - await save_audio(audiobuffer) runner = PipelineRunner() diff --git a/examples/chatbot-audio-recording/requirements.txt b/examples/chatbot-audio-recording/requirements.txt index 9786b52de..4cafe3680 100644 --- a/examples/chatbot-audio-recording/requirements.txt +++ b/examples/chatbot-audio-recording/requirements.txt @@ -1,3 +1,4 @@ +aiofiles python-dotenv fastapi[all] uvicorn diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index d7c07c611..488a251f0 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -4,9 +4,6 @@ # SPDX-License-Identifier: BSD 2-Clause License # -import wave -from io import BytesIO - from pipecat.audio.utils import interleave_stereo_audio, mix_audio, resample_audio from pipecat.frames.frames import ( Frame, @@ -17,43 +14,59 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor class AudioBufferProcessor(FrameProcessor): - """This processor buffers audio raw frames (input and output) that can later - be obtained as an in-memory WAV. You can provide the desired output - `sample_rate` and incoming audio frames will resampled to match it. Also, - you can provide the number of channels, 1 for mono and 2 for stereo. With - mono audio user and bot audio will be mixed, in the case of stereo the left - channel will be used for the user's audio and the right channel for the bot. + """This processor buffers audio raw frames (input and output). The mixed + audio can be obtained by calling `get_audio()` (if `buffer_size` is 0) or by + registering an "on_audio_data" event handler. The event handler will be + called every time `buffer_size` is reached. + + You can provide the desired output `sample_rate` and incoming audio frames + will resampled to match it. Also, you can provide the number of channels, 1 + for mono and 2 for stereo. With mono audio user and bot audio will be mixed, + in the case of stereo the left channel will be used for the user's audio and + the right channel for the bot. """ - def __init__(self, *, sample_rate: int = 24000, num_channels: int = 1, **kwargs): + def __init__( + self, *, sample_rate: int = 24000, num_channels: int = 1, buffer_size: int = 0, **kwargs + ): super().__init__(**kwargs) self._sample_rate = sample_rate self._num_channels = num_channels + self._buffer_size = buffer_size self._user_audio_buffer = bytearray() self._bot_audio_buffer = bytearray() - def _buffer_has_audio(self, buffer: bytearray) -> bool: - return buffer is not None and len(buffer) > 0 + self._register_event_handler("on_audio_data") + + @property + def sample_rate(self) -> int: + return self._sample_rate + + @property + def num_channels(self) -> int: + return self._num_channels def has_audio(self) -> bool: return self._buffer_has_audio(self._user_audio_buffer) and self._buffer_has_audio( self._bot_audio_buffer ) - def reset_audio_buffer(self): - self._user_audio_buffer = bytearray() - self._bot_audio_buffer = bytearray() - def merge_audio_buffers(self) -> bytes: if self._num_channels == 1: - return self._merge_mono() + return mix_audio(bytes(self._user_audio_buffer), bytes(self._bot_audio_buffer)) elif self._num_channels == 2: - return self._merge_stereo() + return interleave_stereo_audio( + bytes(self._user_audio_buffer), bytes(self._bot_audio_buffer) + ) else: return b"" + def reset_audio_buffers(self): + self._user_audio_buffer = bytearray() + self._bot_audio_buffer = bytearray() + async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) @@ -65,30 +78,25 @@ class AudioBufferProcessor(FrameProcessor): if len(self._user_audio_buffer) > len(self._bot_audio_buffer): silence = b"\x00" * len(resampled) self._bot_audio_buffer.extend(silence) - # If the bot is speaking, include all audio from the bot. - if isinstance(frame, OutputAudioRawFrame): + elif isinstance(frame, OutputAudioRawFrame): resampled = resample_audio(frame.audio, frame.sample_rate, self._sample_rate) self._bot_audio_buffer.extend(resampled) - def _merge_mono(self) -> bytes: - with BytesIO() as buffer: - with wave.open(buffer, "wb") as wf: - wf.setnchannels(1) - wf.setsampwidth(2) - wf.setframerate(self._sample_rate) - mixed = mix_audio(bytes(self._user_audio_buffer), bytes(self._bot_audio_buffer)) - wf.writeframes(mixed) - return buffer.getvalue() + if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size: + await self._call_on_audio_data_handler() - def _merge_stereo(self) -> bytes: - with BytesIO() as buffer: - with wave.open(buffer, "wb") as wf: - wf.setnchannels(2) - wf.setsampwidth(2) - wf.setframerate(self._sample_rate) - stereo = interleave_stereo_audio( - bytes(self._user_audio_buffer), bytes(self._bot_audio_buffer) - ) - wf.writeframes(stereo) - return buffer.getvalue() + await self.push_frame(frame, direction) + + async def _call_on_audio_data_handler(self): + if not self.has_audio(): + return + + merged_audio = self.merge_audio_buffers() + await self._call_event_handler( + "on_audio_data", merged_audio, self._sample_rate, self._num_channels + ) + self.reset_audio_buffers() + + def _buffer_has_audio(self, buffer: bytearray) -> bool: + return buffer is not None and len(buffer) > 0 diff --git a/src/pipecat/processors/frameworks/rtvi.py b/src/pipecat/processors/frameworks/rtvi.py index 4644e748f..0e1f3cf5c 100644 --- a/src/pipecat/processors/frameworks/rtvi.py +++ b/src/pipecat/processors/frameworks/rtvi.py @@ -676,6 +676,7 @@ class RTVIProcessor(FrameProcessor): await self.push_frame(frame, direction) async def cleanup(self): + await super().cleanup() if self._pipeline: await self._pipeline.cleanup() diff --git a/src/pipecat/services/canonical.py b/src/pipecat/services/canonical.py index 048a6a4ee..265cc1b1b 100644 --- a/src/pipecat/services/canonical.py +++ b/src/pipecat/services/canonical.py @@ -5,13 +5,16 @@ # import aiohttp +import io import os import uuid +import wave from datetime import datetime from typing import Dict, List, Tuple from pipecat.frames.frames import CancelFrame, EndFrame, Frame +from pipecat.processors.audio import audio_buffer_processor from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor from pipecat.processors.frame_processor import FrameDirection from pipecat.services.ai_services import AIService @@ -81,9 +84,11 @@ class CanonicalMetricsService(AIService): self._output_dir = output_dir async def stop(self, frame: EndFrame): + await super().stop(frame) await self._process_audio() async def cancel(self, frame: CancelFrame): + await super().cancel(frame) await self._process_audio() async def process_frame(self, frame: Frame, direction: FrameDirection): @@ -91,23 +96,32 @@ class CanonicalMetricsService(AIService): await self.push_frame(frame, direction) async def _process_audio(self): - pipeline = self._audio_buffer_processor - if pipeline.has_audio(): - os.makedirs(self._output_dir, exist_ok=True) - filename = self._get_output_filename() - wave_data = pipeline.merge_audio_buffers() + audio_buffer_processor = self._audio_buffer_processor + if not audio_buffer_processor.has_audio(): + return + + os.makedirs(self._output_dir, exist_ok=True) + filename = self._get_output_filename() + audio = audio_buffer_processor.merge_audio_buffers() + + with io.BytesIO() as buffer: + with wave.open(buffer, "wb") as wf: + wf.setsampwidth(2) + wf.setnchannels(audio_buffer_processor.num_channels) + wf.setframerate(audio_buffer_processor.sample_rate) + wf.writeframes(audio) async with aiofiles.open(filename, "wb") as file: - await file.write(wave_data) + await file.write(buffer.getvalue()) - try: - await self._multipart_upload(filename) - pipeline.reset_audio_buffer() - await aiofiles.os.remove(filename) - except FileNotFoundError: - pass - except Exception as e: - logger.error(f"Failed to upload recording: {e}") + try: + await self._multipart_upload(filename) + await aiofiles.os.remove(filename) + audio_buffer_processor.reset_audio_buffers() + except FileNotFoundError: + pass + except Exception as e: + logger.error(f"Failed to upload recording: {e}") def _get_output_filename(self): timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") From d278996d5b6eb27d39c3f574ea032ad0d72a0135 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 3 Dec 2024 14:29:49 -0800 Subject: [PATCH 8/9] updated CHANGELOG --- CHANGELOG.md | 23 +++++++++++++++++++++++ 1 file changed, 23 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 203276f1a..ff84cc17d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,19 +9,42 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- In order to obtain the audio stored by the `AudioBufferProcessor` you can now + also register an `on_audio_data` event handler. The `on_audio_data` handler + will be called every time `buffer_size` (a new constructor argument) is + reached. If `buffer_size` is 0 (default) you need to manually get the audio as + before using `AudioBufferProcessor.merge_audio_buffers()`. + +``` +@audiobuffer.event_handler("on_audio_data") +async def on_audio_data(processor, audio, sample_rate, num_channels): + await save_audio(audio, sample_rate, num_channels) +``` + - Added a new RTVI message called `disconnect-bot`, which when handled pushes an `EndFrame` to trigger the pipeline to stop. ### Changed +- All input frames (text, audio, image, etc.) are now system frames. This means + they are processed immediately by all processors instead of being queued + internally. + - Expanded the transcriptions.language module to support a superset of languages. - Updated STT and TTS services with language options that match the supported languages for each service. +### Removed + +- Removed `AppFrame`. This was used as a special user custom frame, but there's + actually no use case for that. + ### Fixed +- `AudioBufferProcessor` now handles interruptions properly. + - Fixed a `WebsocketServerTransport` issue that would prevent interruptions with `TwilioSerializer` from working. From 1cf93f1dcb5bc2451c8ed2e2e0653976393bc094 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Tue, 3 Dec 2024 16:26:29 -0800 Subject: [PATCH 9/9] FrameProcessor: ignore other frames during CancelFrame --- CHANGELOG.md | 3 +++ src/pipecat/processors/frame_processor.py | 14 ++++++++++++++ 2 files changed, 17 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index ff84cc17d..19dd9aab5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -43,6 +43,9 @@ async def on_audio_data(processor, audio, sample_rate, num_channels): ### Fixed +- Fixed an issue where other frames were being processed while a `CancelFrame` + was being pushed down the pipeline. + - `AudioBufferProcessor` now handles interruptions properly. - Fixed a `WebsocketServerTransport` issue that would prevent interruptions with diff --git a/src/pipecat/processors/frame_processor.py b/src/pipecat/processors/frame_processor.py index 93829c8fc..52066b4f4 100644 --- a/src/pipecat/processors/frame_processor.py +++ b/src/pipecat/processors/frame_processor.py @@ -13,6 +13,7 @@ from loguru import logger from pipecat.clocks.base_clock import BaseClock from pipecat.frames.frames import ( + CancelFrame, EndFrame, ErrorFrame, Frame, @@ -58,6 +59,13 @@ class FrameProcessor: self._enable_usage_metrics = False self._report_only_initial_ttfb = False + # Cancellation is done through CancelFrame (a system frame). This could + # cause other events being triggered (e.g. closing a transport) which + # could also cause other frames to be pushed from other tasks + # (e.g. EndFrame). So, when we are cancelling we don't want anything + # else to be pushed. + self._cancelling = False + # Metrics self._metrics = metrics or FrameProcessorMetrics() self._metrics.set_processor_name(self.name) @@ -161,6 +169,10 @@ class FrameProcessor: Callable[["FrameProcessor", Frame, FrameDirection], Awaitable[None]] ] = None, ): + # If we are cancelling we don't want to process any other frame. + if self._cancelling: + return + if isinstance(frame, SystemFrame): # We don't want to queue system frames. await self.process_frame(frame, direction) @@ -187,6 +199,8 @@ class FrameProcessor: await self.stop_all_metrics() elif isinstance(frame, StopInterruptionFrame): self._should_report_ttfb = True + elif isinstance(frame, CancelFrame): + self._cancelling = True async def push_error(self, error: ErrorFrame): await self.push_frame(error, FrameDirection.UPSTREAM)