diff --git a/changelog/3889.changed.md b/changelog/3889.changed.md new file mode 100644 index 000000000..c04aa0080 --- /dev/null +++ b/changelog/3889.changed.md @@ -0,0 +1,3 @@ +- Support for Voice Focus 2.0 models. + - Updated `aic-sdk` to `~=2.1.0` to support Voice Focus 2.0 models. + - Cleaned unused `ParameterFixedError` exception handling in `AICFilter` parameter setup. diff --git a/changelog/3889.fixed.md b/changelog/3889.fixed.md new file mode 100644 index 000000000..babe3eb35 --- /dev/null +++ b/changelog/3889.fixed.md @@ -0,0 +1 @@ +- Fixed `BufferError: Existing exports of data: object cannot be re-sized` in `AICFilter` caused by holding a `memoryview` on the mutable audio buffer across async yield points. diff --git a/examples/foundational/07zd-interruptible-aicoustics.py b/examples/foundational/07zd-interruptible-aicoustics.py index 61298a706..1a3c626ef 100644 --- a/examples/foundational/07zd-interruptible-aicoustics.py +++ b/examples/foundational/07zd-interruptible-aicoustics.py @@ -40,7 +40,7 @@ def _create_aic_filter() -> AICFilter: return AICFilter( license_key=license_key, - model_id="quail-vf-l-16khz", + model_id="quail-vf-2.0-l-16khz", ) diff --git a/pyproject.toml b/pyproject.toml index 12687a573..0324c311a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,7 +52,7 @@ Issues = "https://github.com/pipecat-ai/pipecat/issues" Changelog = "https://github.com/pipecat-ai/pipecat/blob/main/CHANGELOG.md" [project.optional-dependencies] -aic = [ "aic-sdk~=2.0.1" ] +aic = [ "aic-sdk~=2.1.0" ] anthropic = [ "anthropic~=0.49.0" ] assemblyai = [ "pipecat-ai[websockets-base]" ] asyncai = [ "pipecat-ai[websockets-base]" ] diff --git a/src/pipecat/audio/filters/aic_filter.py b/src/pipecat/audio/filters/aic_filter.py index 723b3da8f..91f0356d0 100644 --- a/src/pipecat/audio/filters/aic_filter.py +++ b/src/pipecat/audio/filters/aic_filter.py @@ -375,12 +375,7 @@ class AICFilter(BaseAudioFilter): self._vad_ctx = self._processor.get_vad_context() # Apply initial parameters - try: - self._processor_ctx.set_parameter( - ProcessorParameter.Bypass, 1.0 if self._bypass else 0.0 - ) - except ParameterFixedError as e: - logger.error(f"AIC parameter update failed: {e}") + self._processor_ctx.set_parameter(ProcessorParameter.Bypass, 1.0 if self._bypass else 0.0) # Log processor information logger.debug(f"ai-coustics filter started:") @@ -389,7 +384,8 @@ class AICFilter(BaseAudioFilter): logger.debug(f" Frames per chunk: {self._frames_per_block}") logger.debug(f" Optimal sample rate: {self._model.get_optimal_sample_rate()} Hz") logger.debug( - f" Optimal number of frames for {self._sample_rate} Hz: {self._model.get_optimal_num_frames(self._sample_rate)}" + f" Optimal number of frames for {self._sample_rate} Hz: " + f"{self._model.get_optimal_num_frames(self._sample_rate)}" ) logger.debug( f" Output delay: {self._processor_ctx.get_output_delay()} samples " @@ -458,13 +454,16 @@ class AICFilter(BaseAudioFilter): if num_blocks == 0: return b"" - filtered_chunks: List[bytes] = [] - mv = memoryview(self._audio_buffer) block_size = self._frames_per_block * self._bytes_per_sample + total_size = num_blocks * block_size + blocks_data = bytes(self._audio_buffer[:total_size]) + self._audio_buffer = self._audio_buffer[total_size:] + + filtered_chunks: List[bytes] = [] for i in range(num_blocks): start = i * block_size - block_i16 = np.frombuffer(mv[start : start + block_size], dtype=self._dtype) + block_i16 = np.frombuffer(blocks_data[start : start + block_size], dtype=self._dtype) # Reuse input buffer, in-place divide np.copyto(self._in_f32[0], block_i16) @@ -479,5 +478,4 @@ class AICFilter(BaseAudioFilter): filtered_chunks.append(self._out_i16.tobytes()) - self._audio_buffer = self._audio_buffer[num_blocks * block_size :] return b"".join(filtered_chunks) diff --git a/tests/test_aic_filter.py b/tests/test_aic_filter.py index 83eca757c..08c702986 100644 --- a/tests/test_aic_filter.py +++ b/tests/test_aic_filter.py @@ -632,6 +632,84 @@ class TestAICFilter(unittest.IsolatedAsyncioTestCase): for result in results: self.assertIsInstance(result, bytes) + async def test_concurrent_filter_no_buffer_resize_error(self): + """Regression: concurrent filter() must not raise BufferError. + + When process_async yields to the event loop, a second filter() call + runs and calls _audio_buffer.extend(). If the first call still holds + a memoryview on the bytearray, extend() raises: + + BufferError: Existing exports of data: object cannot be re-sized + + The fix snapshots the needed data into immutable bytes and trims the + buffer *before* any await, so no memoryview is held across yield + points. + """ + filter_instance = self._create_filter_with_mocks() + + # Make process_async yield to the event loop so concurrent filter() + # calls can interleave and attempt _audio_buffer.extend(). + async def yielding_process_async(audio_array): + await asyncio.sleep(0) + return audio_array.copy() + + self.mock_processor.process_async = yielding_process_async + await self._start_filter_with_mocks(filter_instance) + + samples = np.random.randint(-32768, 32767, size=160, dtype=np.int16) + input_audio = samples.tobytes() + + async def filter_audio(): + return await filter_instance.filter(input_audio) + + # 20 concurrent calls to reliably trigger the interleaving. + tasks = [filter_audio() for _ in range(20)] + results = await asyncio.gather(*tasks) + + for result in results: + self.assertIsInstance(result, bytes) + + async def test_stop_during_filter_no_buffer_resize_error(self): + """Regression: stop() during filter() must not raise BufferError. + + When filter() holds a memoryview on _audio_buffer across an await + (process_async), a concurrent stop() that calls + _audio_buffer.clear() raises: + + BufferError: Existing exports of data: object cannot be re-sized + + This is the exact error path reported in production (line 414). + The fix removes the memoryview by snapshotting data into immutable + bytes before any await. + """ + filter_instance = self._create_filter_with_mocks() + + # Gate so stop() waits until filter() is inside process_async. + processing_started = asyncio.Event() + + async def yielding_process_async(audio_array): + processing_started.set() + await asyncio.sleep(0) # yield — stop() runs here + return audio_array.copy() + + self.mock_processor.process_async = yielding_process_async + await self._start_filter_with_mocks(filter_instance) + + # Exactly one complete frame so the loop runs once. + samples = np.random.randint(-32768, 32767, size=160, dtype=np.int16) + input_audio = samples.tobytes() + + async def stop_after_filter_enters_process_async(): + await processing_started.wait() + await filter_instance.stop() + + # filter() enters process_async → yields → stop() calls clear() + filter_result, _ = await asyncio.gather( + filter_instance.filter(input_audio), + stop_after_filter_enters_process_async(), + ) + self.assertIsInstance(filter_result, bytes) + async def test_buffer_cleared_on_stop(self): """Test that audio buffer is cleared when stopping.""" filter_instance = self._create_filter_with_mocks()