Merge pull request #1299 from pipecat-ai/mb/add-track-level-recording

Added on_track_audio_data callback to AudioBufferProcessor for track level recording

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.
 
@@ -131,6 +134,9 @@ 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

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

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
+    """Processes and buffers audio frames from both input (user) and output (bot) sources.
-    audio can be obtained 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
+    This processor manages audio buffering and synchronization, providing both merged and
-    will resampled to match it. Also, you can provide the number of channels, 1
+    track-specific audio access through event handlers. It supports various audio configurations
-    for mono and 2 for stereo. With mono audio user and bot audio will be mixed,
+    including sample rate conversion and mono/stereo output.
-    in the case of stereo the left channel will be used for the user's audio and
-    the right channel for the bot.
 
-    Most of the time, user audio will be a continuous stream but it's possible
+    Events:
-    that in some cases only the spoken audio is sent. To accomodate for those
+        on_audio_data: Triggered when buffer_size is reached, providing merged audio
-    cases make sure to set `user_continuous_stream` accordingly.
+        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: