From 7921bce4af3044f667c681c443d73b6c598fdc36 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Fri, 23 Jan 2026 16:15:48 -0300 Subject: [PATCH 1/2] Refactoring AudioBufferProcessor to fix audio track synchronization. --- .../audio/audio_buffer_processor.py | 80 +++++++++---------- 1 file changed, 40 insertions(+), 40 deletions(-) diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index 0d3ed76d4..ceadddf95 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/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 From f128cdd19a207148db1269569f030cb724bfd073 Mon Sep 17 00:00:00 2001 From: filipi87 Date: Fri, 23 Jan 2026 16:16:01 -0300 Subject: [PATCH 2/2] Adding a changelog entry to the AudioBufferProcessor fix. --- changelog/3541.fixed.md | 1 + 1 file changed, 1 insertion(+) create mode 100644 changelog/3541.fixed.md diff --git a/changelog/3541.fixed.md b/changelog/3541.fixed.md new file mode 100644 index 000000000..ac5b59529 --- /dev/null +++ b/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.