diff --git a/changelog/4135.fixed.md b/changelog/4135.fixed.md new file mode 100644 index 000000000..92f57ce83 --- /dev/null +++ b/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. diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index ceadddf95..3181b6e73 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/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 - self._sync_buffer_to_position(self._user_audio_buffer, len(self._bot_audio_buffer)) + # Sync user buffer to current bot position before adding bot audio. + # 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): - self._user_speaking = True - elif isinstance(frame, UserStoppedSpeakingFrame): + # Speaking state (_user_speaking / _bot_speaking) is maintained by + # _process_recording so it is always up-to-date here. + 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: diff --git a/tests/test_audio_buffer_processor.py b/tests/test_audio_buffer_processor.py index 4d3c588bc..27403e6fc 100644 --- a/tests/test_audio_buffer_processor.py +++ b/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._input_resampler = _PassthroughResampler() - self.processor._output_resampler = _PassthroughResampler() - self.processor._update_sample_rate(StartFrame(audio_out_sample_rate=16000)) - await self.processor.start_recording() + self.processor = await _make_processor(buffer_size=4) 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()