diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index 276051919..fefbafc9b 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -229,9 +229,12 @@ class AudioBufferProcessor(FrameProcessor): # Save time of frame so we can compute silence. self._last_bot_frame_at = time.time() - if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size: + if self._buffer_size > 0 and ( + len(self._user_audio_buffer) >= self._buffer_size + or len(self._bot_audio_buffer) >= self._buffer_size + ): await self._call_on_audio_data_handler() - self._reset_recording() + self._clear_primary_audio_buffers() # Process turn recording with preprocessed data. if self._enable_turn_audio: @@ -272,9 +275,15 @@ class AudioBufferProcessor(FrameProcessor): async def _call_on_audio_data_handler(self): """Call the audio data event handlers with buffered audio.""" - if not self.has_audio() or not self._recording: + if not self._recording: return + if len(self._user_audio_buffer) == 0 and len(self._bot_audio_buffer) == 0: + return + + self._align_track_buffers() + flush_time = time.time() + # Call original handler with merged audio merged_audio = self.merge_audio_buffers() await self._call_event_handler( @@ -290,6 +299,9 @@ 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 @@ -307,6 +319,26 @@ class AudioBufferProcessor(FrameProcessor): self._user_turn_audio_buffer = bytearray() self._bot_turn_audio_buffer = bytearray() + def _clear_primary_audio_buffers(self): + """Clear user and bot buffers while preserving turn buffers and timestamps.""" + self._user_audio_buffer = bytearray() + self._bot_audio_buffer = bytearray() + + def _align_track_buffers(self): + """Pad the shorter track with silence so both tracks stay in sync.""" + user_len = len(self._user_audio_buffer) + bot_len = len(self._bot_audio_buffer) + if user_len == bot_len: + return + + 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) + 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) + async def _resample_input_audio(self, frame: InputAudioRawFrame) -> bytes: """Resample audio frame to the target sample rate.""" return await self._input_resampler.resample(