From 1ca2101e3aa9c05a0881eab33de69db1385a5865 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 26 Feb 2025 10:48:56 -0500 Subject: [PATCH 1/2] Added on_track_audio_data callback to AudioBufferProcessor for track level recording --- CHANGELOG.md | 3 + .../audio/audio_buffer_processor.py | 78 ++++++++++++++++--- 2 files changed, 70 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 341f78a3b..8b8ecbbe8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -132,6 +132,9 @@ stt = DeepgramSTTService(..., live_options=LiveOptions(model="nova-2-general")) ### Added +- Added track-specific audio event `on_track_audio_data` to + `AudioBufferProcessor` for accessing separate input and output audio tracks. + - Added new `AudioContextWordTTSService`. This is a TTS base class for TTS services that handling multiple separate audio requests. diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index 2f9b975e3..1863a0ee6 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -21,20 +21,32 @@ from pipecat.processors.frame_processor import FrameDirection, FrameProcessor class AudioBufferProcessor(FrameProcessor): - """This processor buffers audio raw frames (input and output). The mixed - audio can be obtained by registering an "on_audio_data" event handler. - The event handler will be called every time `buffer_size` is reached. + """Processes and buffers audio frames from both input (user) and output (bot) sources. - 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 manages audio buffering and synchronization, providing both merged and + track-specific audio access through event handlers. It supports various audio configurations + including sample rate conversion and mono/stereo output. - Most of the time, user audio will be a continuous stream but it's possible - that in some cases only the spoken audio is sent. To accomodate for those - cases make sure to set `user_continuous_stream` accordingly. + 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 + 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 + + Audio handling: + - Mono output (num_channels=1): User and bot audio are mixed + - Stereo output (num_channels=2): User audio on left, bot audio on right + - Automatic resampling of incoming audio to match desired sample_rate + - Silence insertion for non-continuous audio streams + - Buffer synchronization between user and bot audio + + Note: + When user_continuous_stream is False, the processor expects only speech + segments and will handle silence insertion between segments automatically. """ def __init__( @@ -65,21 +77,45 @@ class AudioBufferProcessor(FrameProcessor): self._resampler = create_default_resampler() self._register_event_handler("on_audio_data") + self._register_event_handler("on_track_audio_data") @property def sample_rate(self) -> int: + """Current sample rate of the audio processor. + + Returns: + int: The sample rate in Hz + """ return self._sample_rate @property def num_channels(self) -> int: + """Number of channels in the audio output. + + Returns: + int: Number of channels (1 for mono, 2 for stereo) + """ return self._num_channels def has_audio(self) -> bool: + """Check if both user and bot audio buffers contain data. + + Returns: + bool: True if both buffers contain audio data + """ return self._buffer_has_audio(self._user_audio_buffer) and self._buffer_has_audio( self._bot_audio_buffer ) def merge_audio_buffers(self) -> bytes: + """Merge user and bot audio buffers into a single audio stream. + + For mono output, audio is mixed. For stereo output, user audio is placed + on the left channel and bot audio on the right channel. + + Returns: + bytes: Mixed audio data + """ if self._num_channels == 1: return mix_audio(bytes(self._user_audio_buffer), bytes(self._bot_audio_buffer)) elif self._num_channels == 2: @@ -90,14 +126,23 @@ class AudioBufferProcessor(FrameProcessor): return b"" async def start_recording(self): + """Start recording audio from both user and bot. + + Initializes recording state and resets audio buffers. + """ self._recording = True self._reset_recording() async def stop_recording(self): + """Stop recording and trigger final audio data handlers. + + Calls audio handlers with any remaining buffered audio before stopping. + """ await self._call_on_audio_data_handler() self._recording = False async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process incoming audio frames and manage audio buffers.""" await super().process_frame(frame, direction) # Update output sample rate if necessary. @@ -160,10 +205,21 @@ class AudioBufferProcessor(FrameProcessor): if not self.has_audio() or not self._recording: return + # Call original handler with merged audio merged_audio = self.merge_audio_buffers() await self._call_event_handler( "on_audio_data", merged_audio, self._sample_rate, self._num_channels ) + + # Call new handler with separate tracks + await self._call_event_handler( + "on_track_audio_data", + bytes(self._user_audio_buffer), + bytes(self._bot_audio_buffer), + self._sample_rate, + self._num_channels, + ) + self._reset_audio_buffers() def _buffer_has_audio(self, buffer: bytearray) -> bool: From 530bb5233d89584321a42b2e7fa27d137498faf6 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Wed, 26 Feb 2025 10:51:23 -0500 Subject: [PATCH 2/2] example: Added a foundational example (34) for audio recording --- CHANGELOG.md | 9 +- examples/foundational/34-audio-recording.py | 186 ++++++++++++++++++++ 2 files changed, 192 insertions(+), 3 deletions(-) create mode 100644 examples/foundational/34-audio-recording.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 8b8ecbbe8..83b520583 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added track-specific audio event `on_track_audio_data` to + `AudioBufferProcessor` for accessing separate input and output audio tracks. + - Pipecat version will now be logged on every application startup. This will help us identify what version we are running in case of any issues. @@ -128,13 +131,13 @@ stt = DeepgramSTTService(..., live_options=LiveOptions(model="nova-2-general")) - Added Gemini support to `examples/phone-chatbot`. +- Added foundational example `34-audio-recording.py` showing how to use the + AudioBufferProcessor callbacks to save merged and track recordings. + ## [0.0.57] - 2025-02-14 ### Added -- Added track-specific audio event `on_track_audio_data` to - `AudioBufferProcessor` for accessing separate input and output audio tracks. - - Added new `AudioContextWordTTSService`. This is a TTS base class for TTS services that handling multiple separate audio requests. diff --git a/examples/foundational/34-audio-recording.py b/examples/foundational/34-audio-recording.py new file mode 100644 index 000000000..1d0431b83 --- /dev/null +++ b/examples/foundational/34-audio-recording.py @@ -0,0 +1,186 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Audio Recording Example with Pipecat. + +This example demonstrates how to record audio from a conversation between a user and an AI assistant, +saving both merged and individual audio tracks. It showcases the AudioBufferProcessor's capabilities +to handle both combined and separate audio streams. + +The example: + 1. Sets up a basic conversation with an AI assistant + 2. Records the entire conversation + 3. Saves three separate WAV files: + - A merged recording of both participants + - Individual recording of user audio + - Individual recording of assistant audio + +Example usage (run from pipecat root directory): + $ pip install "pipecat-ai[daily,openai,cartesia,silero]" + $ pip install -r dev-requirements.txt + $ python examples/foundational/34-audio-recording.py + +Requirements: + - OpenAI API key (for GPT-4) + - Cartesia API key (for text-to-speech) + - Daily API key (for video/audio transport) + + Environment variables (.env file): + OPENAI_API_KEY=your_openai_key + CARTESIA_API_KEY=your_cartesia_key + DAILY_API_KEY=your_daily_key + +The recordings will be saved in a 'recordings' directory with timestamps: + recordings/ + merged_20240315_123456.wav (Combined audio) + user_20240315_123456.wav (User audio only) + bot_20240315_123456.wav (Bot audio only) + +Note: + This example requires the AudioBufferProcessor with track-specific audio support, + which provides both 'on_audio_data' and 'on_track_audio_data' events for + handling merged and separate audio tracks respectively. +""" + +import asyncio +import datetime +import io +import os +import sys +import wave + +import aiofiles +import aiohttp +from dotenv import load_dotenv +from loguru import logger +from runner import configure + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor +from pipecat.services.cartesia import CartesiaTTSService +from pipecat.services.openai import OpenAILLMService +from pipecat.transports.services.daily import DailyParams, DailyTransport + +load_dotenv(override=True) +logger.remove(0) +logger.add(sys.stderr, level="DEBUG") + + +async def save_audio_file(audio: bytes, filename: str, sample_rate: int, num_channels: int): + """Save audio data to a WAV file.""" + if len(audio) > 0: + 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()) + logger.info(f"Audio saved to {filename}") + + +async def main(): + async with aiohttp.ClientSession() as session: + (room_url, token) = await configure(session) + + transport = DailyTransport( + room_url, + token, + "Recording bot", + DailyParams( + # audio_in_enabled=True, + audio_out_enabled=True, + transcription_enabled=True, + vad_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + vad_audio_passthrough=True, # Enable audio passthrough for recording + ), + ) + + tts = CartesiaTTSService( + api_key=os.getenv("CARTESIA_API_KEY"), + voice_id="79a125e8-cd45-4c13-8a67-188112f4dd22", + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY"), model="gpt-4") + + # Create audio buffer processor + audiobuffer = AudioBufferProcessor() + + messages = [ + { + "role": "system", + "content": "You are a helpful assistant demonstrating audio recording capabilities. Keep your responses brief and clear.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), + context_aggregator.user(), + llm, + tts, + transport.output(), + audiobuffer, # Add audio buffer to pipeline + context_aggregator.assistant(), + ] + ) + + task = PipelineTask(pipeline, params=PipelineParams(allow_interruptions=True)) + + @transport.event_handler("on_first_participant_joined") + async def on_first_participant_joined(transport, participant): + await transport.capture_participant_transcription(participant["id"]) + await audiobuffer.start_recording() + messages.append( + { + "role": "system", + "content": "Greet the user and explain that this conversation will be recorded.", + } + ) + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_participant_left") + async def on_participant_left(transport, participant, reason): + await audiobuffer.stop_recording() + await task.cancel() + + # Handler for merged audio + @audiobuffer.event_handler("on_audio_data") + async def on_audio_data(buffer, audio, sample_rate, num_channels): + timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") + filename = f"recordings/merged_{timestamp}.wav" + os.makedirs("recordings", exist_ok=True) + await save_audio_file(audio, filename, sample_rate, num_channels) + + # Handler for separate tracks + @audiobuffer.event_handler("on_track_audio_data") + async def on_track_audio_data(buffer, user_audio, bot_audio, sample_rate, num_channels): + timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S") + os.makedirs("recordings", exist_ok=True) + + # Save user audio + user_filename = f"recordings/user_{timestamp}.wav" + await save_audio_file(user_audio, user_filename, sample_rate, 1) + + # Save bot audio + bot_filename = f"recordings/bot_{timestamp}.wav" + await save_audio_file(bot_audio, bot_filename, sample_rate, 1) + + runner = PipelineRunner() + await runner.run(task) + + +if __name__ == "__main__": + asyncio.run(main())