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

Fixed audio crackling and popping artifacts in AudioBufferProcessor

changelog/4135.fixed.md

@@ -0,0 +1 @@
+- Fixed audio crackling and popping in recordings when both user and bot are speaking. `AudioBufferProcessor` no longer injects silence into a track's buffer while that track is actively producing audio, preventing mid-utterance interruptions in the recorded output.

src/pipecat/processors/audio/audio_buffer_processor.py

@@ -13,6 +13,8 @@ configurations and event-driven processing.
 
 from typing import Optional
 
+from loguru import logger
+
 from pipecat.audio.utils import create_stream_resampler, interleave_stereo_audio, mix_audio
 from pipecat.frames.frames import (
     BotStartedSpeakingFrame,
@@ -204,6 +206,17 @@ class AudioBufferProcessor(FrameProcessor):
 
     async def _process_recording(self, frame: Frame):
         """Process audio frames for recording."""
+        # Track speaking state here (not just in _process_turn_recording) so the
+        # silence-injection guards below work regardless of enable_turn_audio.
+        if isinstance(frame, UserStartedSpeakingFrame):
+            self._user_speaking = True
+        elif isinstance(frame, UserStoppedSpeakingFrame):
+            self._user_speaking = False
+        elif isinstance(frame, BotStartedSpeakingFrame):
+            self._bot_speaking = True
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            self._bot_speaking = False
+
         resampled = None
         if isinstance(frame, InputAudioRawFrame):
             resampled = await self._resample_input_audio(frame)
@@ -220,15 +233,26 @@ class AudioBufferProcessor(FrameProcessor):
                 # 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))
+                #
+                # Skip silence injection if the bot is actively speaking to avoid
+                # inserting silence in the middle of a bot utterance (causes crackling).
+                if not self._bot_speaking:
+                    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):
+        elif isinstance(frame, OutputAudioRawFrame):
             resampled = await self._resample_output_audio(frame)
             # Ignoring in case we don't have audio
             if len(resampled) > 0:
-                # Sync user buffer to current bot position before adding bot audio
+                # 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))
+                # Skip silence injection if the user is actively speaking to avoid
+                # inserting silence in the middle of a user utterance (causes crackling).
+                if not self._user_speaking:
+                    self._sync_buffer_to_position(
+                        self._user_audio_buffer, len(self._bot_audio_buffer)
+                    )
                 # Add bot audio.
                 self._bot_audio_buffer.extend(resampled)
 
@@ -260,21 +284,17 @@ class AudioBufferProcessor(FrameProcessor):
 
     async def _process_turn_recording(self, frame: Frame, resampled_audio: Optional[bytes] = None):
         """Process frames for turn-based audio recording."""
-        if isinstance(frame, UserStartedSpeakingFrame):
+        # Speaking state (_user_speaking / _bot_speaking) is maintained by
-            self._user_speaking = True
+        # _process_recording so it is always up-to-date here.
-        elif isinstance(frame, UserStoppedSpeakingFrame):
+        if isinstance(frame, UserStoppedSpeakingFrame):
             await self._call_event_handler(
                 "on_user_turn_audio_data", self._user_turn_audio_buffer, self.sample_rate, 1
             )
-            self._user_speaking = False
             self._user_turn_audio_buffer = bytearray()
-        elif isinstance(frame, BotStartedSpeakingFrame):
-            self._bot_speaking = True
         elif isinstance(frame, BotStoppedSpeakingFrame):
             await self._call_event_handler(
                 "on_bot_turn_audio_data", self._bot_turn_audio_buffer, self.sample_rate, 1
             )
-            self._bot_speaking = False
             self._bot_turn_audio_buffer = bytearray()
 
         if isinstance(frame, InputAudioRawFrame) and resampled_audio:

tests/test_audio_buffer_processor.py

@@ -8,8 +8,19 @@ import asyncio
 import struct
 import unittest
 
-from pipecat.frames.frames import InputAudioRawFrame, OutputAudioRawFrame, StartFrame
+from pipecat.clocks.system_clock import SystemClock
+from pipecat.frames.frames import (
+    BotStartedSpeakingFrame,
+    BotStoppedSpeakingFrame,
+    InputAudioRawFrame,
+    OutputAudioRawFrame,
+    StartFrame,
+    UserStartedSpeakingFrame,
+    UserStoppedSpeakingFrame,
+)
 from pipecat.processors.audio.audio_buffer_processor import AudioBufferProcessor
+from pipecat.processors.frame_processor import FrameDirection, FrameProcessorSetup
+from pipecat.utils.asyncio.task_manager import TaskManager, TaskManagerParams
 
 
 class _PassthroughResampler:
@@ -19,13 +30,48 @@ class _PassthroughResampler:
         return audio
 
 
+async def _make_processor(*, buffer_size: int = 0) -> AudioBufferProcessor:
+    """Create and start a processor ready to record.
+
+    Calls setup() and sends a StartFrame through the public process_frame path so that
+    the processor is fully initialised (task manager set, sample rate configured,
+    __started flag set) without needing a full pipeline.
+    """
+    processor = AudioBufferProcessor(sample_rate=16000, num_channels=2, buffer_size=buffer_size)
+    processor._input_resampler = _PassthroughResampler()
+    processor._output_resampler = _PassthroughResampler()
+
+    loop = asyncio.get_event_loop()
+    task_manager = TaskManager()
+    task_manager.setup(TaskManagerParams(loop=loop))
+    await processor.setup(FrameProcessorSetup(clock=SystemClock(), task_manager=task_manager))
+
+    await processor.process_frame(
+        StartFrame(audio_out_sample_rate=16000), FrameDirection.DOWNSTREAM
+    )
+    await processor.start_recording()
+    return processor
+
+
+async def _capture_track_audio(processor: AudioBufferProcessor) -> tuple[bytes, bytes]:
+    """Flush the processor and return (user_track, bot_track) from on_track_audio_data."""
+    captured = {}
+    event = asyncio.Event()
+
+    async def on_track_audio_data(_, user, bot, sample_rate, num_channels):
+        captured["user"] = user
+        captured["bot"] = bot
+        event.set()
+
+    processor.add_event_handler("on_track_audio_data", on_track_audio_data)
+    await processor.stop_recording()
+    await asyncio.wait_for(event.wait(), timeout=1)
+    return captured["user"], captured["bot"]
+
+
 class TestAudioBufferProcessor(unittest.IsolatedAsyncioTestCase):
     async def asyncSetUp(self):
-        self.processor = AudioBufferProcessor(sample_rate=16000, num_channels=2, buffer_size=4)
+        self.processor = await _make_processor(buffer_size=4)
-        self.processor._input_resampler = _PassthroughResampler()
-        self.processor._output_resampler = _PassthroughResampler()
-        self.processor._update_sample_rate(StartFrame(audio_out_sample_rate=16000))
-        await self.processor.start_recording()
 
     async def asyncTearDown(self):
         if getattr(self.processor, "_recording", False):
@@ -52,7 +98,7 @@ class TestAudioBufferProcessor(unittest.IsolatedAsyncioTestCase):
         self.processor.add_event_handler("on_track_audio_data", on_track_audio_data)
 
         frame = InputAudioRawFrame(audio=user_audio, sample_rate=16000, num_channels=1)
-        await self.processor._process_recording(frame)
+        await self.processor.process_frame(frame, FrameDirection.DOWNSTREAM)
 
         await asyncio.wait_for(audio_event.wait(), timeout=1)
         await asyncio.wait_for(track_event.wait(), timeout=1)
@@ -94,7 +140,7 @@ class TestAudioBufferProcessor(unittest.IsolatedAsyncioTestCase):
         self.processor.add_event_handler("on_track_audio_data", on_track_audio_data)
 
         frame = OutputAudioRawFrame(audio=bot_audio, sample_rate=16000, num_channels=1)
-        await self.processor._process_recording(frame)
+        await self.processor.process_frame(frame, FrameDirection.DOWNSTREAM)
 
         await asyncio.wait_for(audio_event.wait(), timeout=1)
         await asyncio.wait_for(track_event.wait(), timeout=1)
@@ -117,5 +163,166 @@ class TestAudioBufferProcessor(unittest.IsolatedAsyncioTestCase):
         self.assertEqual(len(self.processor._bot_audio_buffer), 0)
 
 
+class TestSilenceInjectionGuards(unittest.IsolatedAsyncioTestCase):
+    """Tests that silence is not injected mid-utterance (fix for crackling artifacts).
+
+    Each test verifies the audio alignment in the flushed tracks to confirm that
+    silence is only added by _align_track_buffers at flush time (end of the buffer),
+    never injected mid-stream while the affected track is actively producing audio.
+    """
+
+    async def test_no_silence_injected_into_bot_buffer_while_bot_speaking(self):
+        """Bot audio must appear at the start of the bot track, not after mid-stream silence.
+
+        Timeline:
+          1. User sends 4 bytes  (bot not speaking → normal sync, no-op since bot is at 0)
+          2. Bot starts speaking
+          3. User sends 4 more bytes  (bot speaking → sync skipped; bot stays at 0)
+          4. Bot sends 4 bytes of known audio
+
+        Expected final bot track (8 bytes total after _align_track_buffers at flush):
+          [bot_audio][silence_padding]  ← audio first, silence only at the end
+
+        With the bug the bot track would be:
+          [silence_injected_mid_stream][bot_audio]  ← silence inserted before the audio
+        """
+        p = await _make_processor()
+
+        bot_audio = b"\xaa\xbb\xcc\xdd"
+
+        await p.process_frame(
+            InputAudioRawFrame(audio=b"\x01\x02\x03\x04", sample_rate=16000, num_channels=1),
+            FrameDirection.DOWNSTREAM,
+        )
+        await p.process_frame(BotStartedSpeakingFrame(), FrameDirection.DOWNSTREAM)
+        await p.process_frame(
+            InputAudioRawFrame(audio=b"\x05\x06\x07\x08", sample_rate=16000, num_channels=1),
+            FrameDirection.DOWNSTREAM,
+        )
+        await p.process_frame(
+            OutputAudioRawFrame(audio=bot_audio, sample_rate=16000, num_channels=1),
+            FrameDirection.DOWNSTREAM,
+        )
+
+        _, bot_track = await _capture_track_audio(p)
+        await p.cleanup()
+
+        # Audio must appear at the beginning of the bot track (not after injected silence).
+        self.assertEqual(bot_track[:4], bot_audio)
+        self.assertEqual(bot_track[4:], b"\x00" * 4)
+
+    async def test_no_silence_injected_into_user_buffer_while_user_speaking(self):
+        """User audio must appear at the start of the user track, not after mid-stream silence.
+
+        Timeline:
+          1. Bot sends 4 bytes  (user not speaking → normal sync, no-op since user is at 0)
+          2. User starts speaking
+          3. Bot sends 4 more bytes  (user speaking → sync skipped; user stays at 0)
+          4. User sends 4 bytes of known audio
+
+        Expected final user track (8 bytes total after _align_track_buffers at flush):
+          [user_audio][silence_padding]  ← audio first, silence only at the end
+
+        With the bug the user track would be:
+          [silence_injected_mid_stream][user_audio]
+        """
+        p = await _make_processor()
+
+        user_audio = b"\xaa\xbb\xcc\xdd"
+
+        await p.process_frame(
+            OutputAudioRawFrame(audio=b"\x01\x02\x03\x04", sample_rate=16000, num_channels=1),
+            FrameDirection.DOWNSTREAM,
+        )
+        await p.process_frame(UserStartedSpeakingFrame(), FrameDirection.DOWNSTREAM)
+        await p.process_frame(
+            OutputAudioRawFrame(audio=b"\x05\x06\x07\x08", sample_rate=16000, num_channels=1),
+            FrameDirection.DOWNSTREAM,
+        )
+        await p.process_frame(
+            InputAudioRawFrame(audio=user_audio, sample_rate=16000, num_channels=1),
+            FrameDirection.DOWNSTREAM,
+        )
+
+        user_track, _ = await _capture_track_audio(p)
+        await p.cleanup()
+
+        self.assertEqual(user_track[:4], user_audio)
+        self.assertEqual(user_track[4:], b"\x00" * 4)
+
+    async def test_silence_resumes_into_bot_buffer_after_bot_stops_speaking(self):
+        """After bot stops speaking, the bot buffer is synced again on user audio arrival.
+
+        Timeline:
+          1. User sends 4 bytes  (user=4, bot=0)
+          2. Bot starts speaking
+          3. User sends 4 more bytes  (sync skipped; user=8, bot=0)
+          4. Bot stops speaking
+          5. User sends 4 more bytes  (sync resumes; bot gets 8 bytes silence, user=12)
+
+        Expected final bot track (12 bytes): 8 bytes silence then no more audio (bot never
+        sent audio, _align_track_buffers pads bot to 12).
+        The key assertion: bot has 8 bytes of silence at positions 0-7, confirming that
+        the sync at step 5 did inject 8 bytes (positions 0-7 of the bot buffer).
+        """
+        p = await _make_processor()
+
+        await p.process_frame(
+            InputAudioRawFrame(audio=b"\x01\x02\x03\x04", sample_rate=16000, num_channels=1),
+            FrameDirection.DOWNSTREAM,
+        )
+        await p.process_frame(BotStartedSpeakingFrame(), FrameDirection.DOWNSTREAM)
+        await p.process_frame(
+            InputAudioRawFrame(audio=b"\x05\x06\x07\x08", sample_rate=16000, num_channels=1),
+            FrameDirection.DOWNSTREAM,
+        )
+        await p.process_frame(BotStoppedSpeakingFrame(), FrameDirection.DOWNSTREAM)
+        await p.process_frame(
+            InputAudioRawFrame(audio=b"\x09\x0a\x0b\x0c", sample_rate=16000, num_channels=1),
+            FrameDirection.DOWNSTREAM,
+        )
+
+        _, bot_track = await _capture_track_audio(p)
+        await p.cleanup()
+
+        # The sync at step 5 targets len(user)=8, so bot must have 8 bytes of silence
+        # written before user's third chunk was added.
+        self.assertEqual(bot_track[:8], b"\x00" * 8)
+
+    async def test_silence_resumes_into_user_buffer_after_user_stops_speaking(self):
+        """After user stops speaking, the user buffer is synced again on bot audio arrival.
+
+        Timeline:
+          1. Bot sends 4 bytes  (user=0, bot=4)
+          2. User starts speaking
+          3. Bot sends 4 more bytes  (sync skipped; user=0, bot=8)
+          4. User stops speaking
+          5. Bot sends 4 more bytes  (sync resumes; user gets 8 bytes silence, bot=12)
+
+        Expected: user track has 8 bytes of silence at positions 0-7.
+        """
+        p = await _make_processor()
+
+        await p.process_frame(
+            OutputAudioRawFrame(audio=b"\x01\x02\x03\x04", sample_rate=16000, num_channels=1),
+            FrameDirection.DOWNSTREAM,
+        )
+        await p.process_frame(UserStartedSpeakingFrame(), FrameDirection.DOWNSTREAM)
+        await p.process_frame(
+            OutputAudioRawFrame(audio=b"\x05\x06\x07\x08", sample_rate=16000, num_channels=1),
+            FrameDirection.DOWNSTREAM,
+        )
+        await p.process_frame(UserStoppedSpeakingFrame(), FrameDirection.DOWNSTREAM)
+        await p.process_frame(
+            OutputAudioRawFrame(audio=b"\x09\x0a\x0b\x0c", sample_rate=16000, num_channels=1),
+            FrameDirection.DOWNSTREAM,
+        )
+
+        user_track, _ = await _capture_track_audio(p)
+        await p.cleanup()
+
+        self.assertEqual(user_track[:8], b"\x00" * 8)
+
+
 if __name__ == "__main__":
     unittest.main()