Merge pull request #3541 from pipecat-ai/filipi/audio_buffer

Refactoring AudioBufferProcessor to fix audio track synchronization.

changelog/3541.fixed.md

@@ -0,0 +1 @@
+- Fixed how audio tracks are synchronized inside the `AudioBufferProcessor` to fix timing issues where silence and audio were misaligned between user and bot buffers.

src/pipecat/processors/audio/audio_buffer_processor.py

@@ -11,7 +11,6 @@ of audio from both user input and bot output sources, with support for various a
 configurations and event-driven processing.
 """
 
-import time
 from typing import Optional
 
 from pipecat.audio.utils import create_stream_resampler, interleave_stereo_audio, mix_audio
@@ -104,10 +103,6 @@ class AudioBufferProcessor(FrameProcessor):
         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
-
         self._recording = False
 
         self._input_resampler = create_stream_resampler()
@@ -211,23 +206,31 @@ class AudioBufferProcessor(FrameProcessor):
         """Process audio frames for recording."""
         resampled = None
         if isinstance(frame, InputAudioRawFrame):
-            # Add silence if we need to.
-            silence = self._compute_silence(self._last_user_frame_at)
-            self._user_audio_buffer.extend(silence)
-            # Add user audio.
             resampled = await self._resample_input_audio(frame)
-            self._user_audio_buffer.extend(resampled)
-            # Save time of frame so we can compute silence.
-            self._last_user_frame_at = time.time()
+            # Ignoring in case we don't have audio
+            if len(resampled) > 0:
+                # Sync bot buffer to current user position before adding user audio.
+                # We sync BEFORE extending to align both buffers at the same starting timestamp.
+                # For example, user buffer is at 100 bytes, and you receive 20 bytes of new audio
+                #  - Bot buffer sees User is at 100. Bot pads itself to 100.
+                #  - User buffer adds 20. User is now at 120.
+                #  - Outcome: At index 100-120, we have User Audio and (potentially) Bot Audio or silence. They are aligned
+                # This gives the opportunity to the bot to send audio.
+                #
+                # If we synced AFTER, we'd pad the bot buffer with silence for the same
+                # window we just gave to the user, effectively "overwriting" that time slot
+                # with silence and causing the bot's audio to flicker or cut out.
+                self._sync_buffer_to_position(self._bot_audio_buffer, len(self._user_audio_buffer))
+                # Add user audio.
+                self._user_audio_buffer.extend(resampled)
         elif self._recording and isinstance(frame, OutputAudioRawFrame):
-            # Add silence if we need to.
-            silence = self._compute_silence(self._last_bot_frame_at)
-            self._bot_audio_buffer.extend(silence)
-            # Add bot audio.
             resampled = await self._resample_output_audio(frame)
-            self._bot_audio_buffer.extend(resampled)
-            # Save time of frame so we can compute silence.
-            self._last_bot_frame_at = time.time()
+            # Ignoring in case we don't have audio
+            if len(resampled) > 0:
+                # Sync user buffer to current bot position before adding bot audio
+                self._sync_buffer_to_position(self._user_audio_buffer, len(self._bot_audio_buffer))
+                # Add bot audio.
+                self._bot_audio_buffer.extend(resampled)
 
         if self._buffer_size > 0 and (
             len(self._user_audio_buffer) >= self._buffer_size
@@ -240,6 +243,21 @@ class AudioBufferProcessor(FrameProcessor):
         if self._enable_turn_audio:
             await self._process_turn_recording(frame, resampled)
 
+    def _sync_buffer_to_position(self, buffer: bytearray, target_position: int):
+        """Pad buffer with silence if it's behind the target position.
+
+        This ensures both buffers stay synchronized by padding the lagging
+        buffer before new audio is added to the other buffer.
+
+        Args:
+            buffer: The buffer to potentially pad.
+            target_position: The position (in bytes) the buffer should reach.
+        """
+        current_len = len(buffer)
+        if current_len < target_position:
+            silence_needed = target_position - current_len
+            buffer.extend(b"\x00" * silence_needed)
+
     async def _process_turn_recording(self, frame: Frame, resampled_audio: Optional[bytes] = None):
         """Process frames for turn-based audio recording."""
         if isinstance(frame, UserStartedSpeakingFrame):
@@ -281,8 +299,8 @@ class AudioBufferProcessor(FrameProcessor):
         if len(self._user_audio_buffer) == 0 and len(self._bot_audio_buffer) == 0:
             return
 
+        # Final alignment before we send the audio
         self._align_track_buffers()
-        flush_time = time.time()
 
         # Call original handler with merged audio
         merged_audio = self.merge_audio_buffers()
@@ -299,9 +317,6 @@ class AudioBufferProcessor(FrameProcessor):
             self._num_channels,
         )
 
-        self._last_user_frame_at = flush_time
-        self._last_bot_frame_at = flush_time
-
     def _buffer_has_audio(self, buffer: bytearray) -> bool:
         """Check if a buffer contains audio data."""
         return buffer is not None and len(buffer) > 0
@@ -309,8 +324,6 @@ class AudioBufferProcessor(FrameProcessor):
     def _reset_recording(self):
         """Reset recording state and buffers."""
         self._reset_all_audio_buffers()
-        self._last_user_frame_at = time.time()
-        self._last_bot_frame_at = time.time()
 
     def _reset_all_audio_buffers(self):
         """Reset all audio buffers to empty state."""
@@ -336,11 +349,9 @@ class AudioBufferProcessor(FrameProcessor):
 
         target_len = max(user_len, bot_len)
         if user_len < target_len:
-            self._user_audio_buffer.extend(b"\x00" * (target_len - user_len))
-            self._last_user_frame_at = max(self._last_user_frame_at, self._last_bot_frame_at)
+            self._sync_buffer_to_position(self._user_audio_buffer, target_len)
         if bot_len < target_len:
-            self._bot_audio_buffer.extend(b"\x00" * (target_len - bot_len))
-            self._last_bot_frame_at = max(self._last_bot_frame_at, self._last_user_frame_at)
+            self._sync_buffer_to_position(self._bot_audio_buffer, target_len)
 
     async def _resample_input_audio(self, frame: InputAudioRawFrame) -> bytes:
         """Resample audio frame to the target sample rate."""
@@ -353,14 +364,3 @@ class AudioBufferProcessor(FrameProcessor):
         return await self._output_resampler.resample(
             frame.audio, frame.sample_rate, self._sample_rate
         )
-
-    def _compute_silence(self, from_time: float) -> bytes:
-        """Compute silence to insert based on time gap."""
-        quiet_time = time.time() - from_time
-        # We should get audio frames very frequently. We introduce silence only
-        # if there's a big enough gap of 1s.
-        if from_time == 0 or quiet_time < 1.0:
-            return b""
-        num_bytes = int(quiet_time * self._sample_rate) * 2
-        silence = b"\x00" * num_bytes
-        return silence