From 06c742a2ad147536db88c6d25dd31c9da2e3358b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 5 Mar 2025 17:20:02 -0800 Subject: [PATCH 1/3] AudioBufferProcessor: add on_user_turn_audio_data and on_bot_turn_audio_data --- CHANGELOG.md | 4 + .../audio/audio_buffer_processor.py | 73 +++++++++++++++++-- 2 files changed, 70 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4ff473ea2..c631bb0e2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `on_user_turn_audio_data` and `on_bot_turn_audio_data` to + `AudioBufferProcessor`. This gives the ability to grab the audio of only that + turn for both the user and the bot. + - Added new base class `BaseObject` which is now the base class of `FrameProcessor`, `PipelineRunner`, `PipelineTask` and `BaseTransport`. The new `BaseObject` adds supports for event handlers. diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index 1863a0ee6..c1b2eb810 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -10,12 +10,16 @@ from typing import Optional from pipecat.audio.utils import create_default_resampler, interleave_stereo_audio, mix_audio from pipecat.frames.frames import ( AudioRawFrame, + BotStartedSpeakingFrame, + BotStoppedSpeakingFrame, CancelFrame, EndFrame, Frame, InputAudioRawFrame, OutputAudioRawFrame, StartFrame, + UserStartedSpeakingFrame, + UserStoppedSpeakingFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -30,12 +34,15 @@ class AudioBufferProcessor(FrameProcessor): Events: on_audio_data: Triggered when buffer_size is reached, providing merged audio on_track_audio_data: Triggered when buffer_size is reached, providing separate tracks + on_user_turn_audio_data: Triggered when user turn has ended, providing that user turn's audio + on_bot_turn_audio_data: Triggered when bot turn has ended, providing that bot turn's audio Args: sample_rate (Optional[int]): Desired output sample rate. If None, uses source rate num_channels (int): Number of channels (1 for mono, 2 for stereo). Defaults to 1 buffer_size (int): Size of buffer before triggering events. 0 for no buffering user_continuous_stream (bool): Whether user audio is continuous or speech-only + enable_turn_audio (bool): Whether turn audio event handlers should be triggered Audio handling: - Mono output (num_channels=1): User and bot audio are mixed @@ -56,18 +63,26 @@ class AudioBufferProcessor(FrameProcessor): num_channels: int = 1, buffer_size: int = 0, user_continuous_stream: bool = True, + enable_turn_audio: bool = False, **kwargs, ): super().__init__(**kwargs) self._init_sample_rate = sample_rate self._sample_rate = 0 + self._audio_buffer_size_1s = 0 self._num_channels = num_channels self._buffer_size = buffer_size self._user_continuous_stream = user_continuous_stream + self._enable_turn_audio = enable_turn_audio self._user_audio_buffer = bytearray() self._bot_audio_buffer = bytearray() + self._user_speaking = False + self._bot_speaking = False + self._user_turn_audio_buffer = bytearray() + self._bot_turn_audio_buffer = bytearray() + # Intermittent (non continous user stream variables) self._last_user_frame_at = 0 self._last_bot_frame_at = 0 @@ -78,6 +93,8 @@ class AudioBufferProcessor(FrameProcessor): self._register_event_handler("on_audio_data") self._register_event_handler("on_track_audio_data") + self._register_event_handler("on_user_turn_audio_data") + self._register_event_handler("on_bot_turn_audio_data") @property def sample_rate(self) -> int: @@ -150,13 +167,9 @@ class AudioBufferProcessor(FrameProcessor): self._update_sample_rate(frame) if self._recording: - if self._user_continuous_stream: - await self._handle_continuous_stream(frame) - else: - await self._handle_intermittent_stream(frame) - - if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size: - await self._call_on_audio_data_handler() + await self._process_recording(frame) + if self._enable_turn_audio: + await self._process_turn_recording(frame) if isinstance(frame, (CancelFrame, EndFrame)): await self.stop_recording() @@ -165,6 +178,50 @@ class AudioBufferProcessor(FrameProcessor): def _update_sample_rate(self, frame: StartFrame): self._sample_rate = self._init_sample_rate or frame.audio_out_sample_rate + self._audio_buffer_size_1s = self._sample_rate * 2 + + async def _process_recording(self, frame: Frame): + if self._user_continuous_stream: + await self._handle_continuous_stream(frame) + else: + await self._handle_intermittent_stream(frame) + + if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size: + await self._call_on_audio_data_handler() + + async def _process_turn_recording(self, frame: Frame): + if isinstance(frame, UserStartedSpeakingFrame): + self._user_speaking = True + elif isinstance(frame, UserStoppedSpeakingFrame): + await self._call_event_handler( + "on_user_turn_audio_data", self._user_turn_audio_buffer, self.sample_rate, 1 + ) + self._user_speaking = False + self._user_turn_audio_buffer = bytearray() + elif isinstance(frame, BotStartedSpeakingFrame): + self._bot_speaking = True + elif isinstance(frame, BotStoppedSpeakingFrame): + await self._call_event_handler( + "on_bot_turn_audio_data", self._bot_turn_audio_buffer, self.sample_rate, 1 + ) + self._bot_speaking = False + self._bot_turn_audio_buffer = bytearray() + + if isinstance(frame, InputAudioRawFrame): + resampled = await self._resample_audio(frame) + self._user_turn_audio_buffer += resampled + # In the case of the user, we need to keep a short buffer of audio + # since VAD notification of when the user starts speaking comes + # later. + if ( + not self._user_speaking + and len(self._user_turn_audio_buffer) > self._audio_buffer_size_1s + ): + discarded = len(self._user_turn_audio_buffer) - self._audio_buffer_size_1s + self._user_turn_audio_buffer = self._user_turn_audio_buffer[discarded:] + elif self._bot_speaking and isinstance(frame, OutputAudioRawFrame): + resampled = await self._resample_audio(frame) + self._bot_turn_audio_buffer += resampled async def _handle_continuous_stream(self, frame: Frame): if isinstance(frame, InputAudioRawFrame): @@ -233,6 +290,8 @@ class AudioBufferProcessor(FrameProcessor): def _reset_audio_buffers(self): self._user_audio_buffer = bytearray() self._bot_audio_buffer = bytearray() + self._user_turn_audio_buffer = bytearray() + self._bot_turn_audio_buffer = bytearray() async def _resample_audio(self, frame: AudioRawFrame) -> bytes: return await self._resampler.resample(frame.audio, frame.sample_rate, self._sample_rate) From d7e93551d24da068f43acb607cfa15648fc01322 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 5 Mar 2025 17:20:19 -0800 Subject: [PATCH 2/3] examples(chatbot-audio-recording): add support for user/bot turn audio --- examples/chatbot-audio-recording/bot.py | 18 ++++++++++++++---- 1 file changed, 14 insertions(+), 4 deletions(-) diff --git a/examples/chatbot-audio-recording/bot.py b/examples/chatbot-audio-recording/bot.py index d37d0414e..2dd87449c 100644 --- a/examples/chatbot-audio-recording/bot.py +++ b/examples/chatbot-audio-recording/bot.py @@ -33,9 +33,11 @@ logger.remove(0) logger.add(sys.stderr, level="DEBUG") -async def save_audio(audio: bytes, sample_rate: int, num_channels: int): +async def save_audio(audio: bytes, sample_rate: int, num_channels: int, name: str): if len(audio) > 0: - filename = f"conversation_recording{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.wav" + filename = ( + f"{name}_conversation_recording{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.wav" + ) with io.BytesIO() as buffer: with wave.open(buffer, "wb") as wf: wf.setsampwidth(2) @@ -110,7 +112,7 @@ async def main(): # NOTE: Watch out! This will save all the conversation in memory. You # can pass `buffer_size` to get periodic callbacks. - audiobuffer = AudioBufferProcessor() + audiobuffer = AudioBufferProcessor(enable_turn_audio=True) pipeline = Pipeline( [ @@ -128,7 +130,15 @@ async def main(): @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) + await save_audio(audio, sample_rate, num_channels, "full") + + @audiobuffer.event_handler("on_user_turn_audio_data") + async def on_user_turn_audio_data(buffer, audio, sample_rate, num_channels): + await save_audio(audio, sample_rate, num_channels, "user") + + @audiobuffer.event_handler("on_bot_turn_audio_data") + async def on_bot_turn_audio_data(buffer, audio, sample_rate, num_channels): + await save_audio(audio, sample_rate, num_channels, "bot") @transport.event_handler("on_first_participant_joined") async def on_first_participant_joined(transport, participant): From a91c26785f7ec407d053e867f65c14a2446256d8 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 6 Mar 2025 18:31:48 -0500 Subject: [PATCH 3/3] Store recording in a folder --- examples/chatbot-audio-recording/bot.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/examples/chatbot-audio-recording/bot.py b/examples/chatbot-audio-recording/bot.py index 2dd87449c..2162ed5e2 100644 --- a/examples/chatbot-audio-recording/bot.py +++ b/examples/chatbot-audio-recording/bot.py @@ -32,11 +32,15 @@ load_dotenv(override=True) logger.remove(0) logger.add(sys.stderr, level="DEBUG") +# Create the recordings directory if it doesn't exist +os.makedirs("recordings", exist_ok=True) + async def save_audio(audio: bytes, sample_rate: int, num_channels: int, name: str): if len(audio) > 0: - filename = ( - f"{name}_conversation_recording{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.wav" + filename = os.path.join( + "recordings", + f"{name}_conversation_recording{datetime.datetime.now().strftime('%Y%m%d_%H%M%S')}.wav", ) with io.BytesIO() as buffer: with wave.open(buffer, "wb") as wf: