Merge pull request #1121 from pipecat-ai/aleix/audio-resamplers

introduce audio resamplers

CHANGELOG.md

@@ -9,6 +9,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Added
 
+- Introduce audio resamplers (`BaseAudioResampler`). This is just a base class
+  to implement audio resamplers. Currently, two implementations are provided
+  `SOXRAudioResampler` and `ResampyResampler`. A new
+  `create_default_resampler()` has been added (replacing the now deprecated
+  `resample_audio()`).
+
 - It is now possible to specify the asyncio event loop that a `PipelineTask` and
   all the processors should run on by passing it as a new argument to the
   `PipelineRunner`. This could allow running pipelines in multiple threads each
@@ -41,6 +47,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- `GatedOpenAILLMContextAggregator` now require keyword arguments. Also, a new
+  `start_open` argument has been added to set the initial state of the gate.
+
 - Added `organization` and `project` level authentication to
   `OpenAILLMService`.
 
@@ -56,8 +65,27 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - `InputDTMFFrame` is now based on `DTMFFrame`. There's also a new
   `OutputDTMFFrame` frame.
 
+### Deprecated
+
+- `resample_audio()` is now deprecated, use `create_default_resampler()`
+  instead.
+
+### Removed
+
+- `AudioBufferProcessor.reset_audio_buffers()` has been removed, use
+  `AudioBufferProcessor.start_recording()` and
+  ``AudioBufferProcessor.stop_recording()` instead.
+
 ### Fixed
 
+- Fixed a `AudioBufferProcessor` that would cause crackling in some recordings.
+
+- Fixed an issue in `AudioBufferProcessor` where user callback would not be
+  called on task cancellation.
+
+- Fixed an issue in `AudioBufferProcessor` that would cause wrong silence
+  padding in some cases.
+
 - Fixed an issue where `ElevenLabsTTSService` messages would return a 1009
   websocket error by increasing the max message size limit to 16MB.
 
@@ -1527,6 +1555,9 @@ async def on_connected(processor):
 
 ### Changed
 
+- `FrameSerializer.serialize()` and `FrameSerializer.deserialize()` are now
+  `async`.
+
 - `Filter` has been renamed to `FrameFilter` and it's now under
   `processors/filters`.
 

examples/canonical-metrics/bot.py

@@ -124,6 +124,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
+            await audio_buffer_processor.start_recording()
             await transport.capture_participant_transcription(participant["id"])
             await task.queue_frames([context_aggregator.user().get_context_frame()])
 

examples/chatbot-audio-recording/bot.py

@@ -109,8 +109,9 @@ async def main():
         context = OpenAILLMContext(messages)
         context_aggregator = llm.create_context_aggregator(context)
 
-        # Save audio every 10 seconds.
+        # NOTE: Watch out! This will save all the conversation in memory. You
-        audiobuffer = AudioBufferProcessor(buffer_size=480000)
+        # can pass `buffer_size` to get periodic callbacks.
+        audiobuffer = AudioBufferProcessor()
 
         pipeline = Pipeline(
             [
@@ -132,6 +133,7 @@ async def main():
 
         @transport.event_handler("on_first_participant_joined")
         async def on_first_participant_joined(transport, participant):
+            await audiobuffer.start_recording()
             await transport.capture_participant_transcription(participant["id"])
             await task.queue_frames([context_aggregator.user().get_context_frame()])
 

examples/foundational/22-natural-conversation.py

@@ -104,8 +104,11 @@ async def main():
         )
 
         # This processor keeps the last context and will let it through once the
-        # notifier is woken up.
+        # notifier is woken up. We start with the gate open because we send an
-        gated_context_aggregator = GatedOpenAILLMContextAggregator(notifier)
+        # initial context frame to start the conversation.
+        gated_context_aggregator = GatedOpenAILLMContextAggregator(
+            notifier=notifier, start_open=True
+        )
 
         # Notify if the user hasn't said anything.
         async def user_idle_notifier(frame):

examples/foundational/22b-natural-conversation-proposal.py

@@ -129,9 +129,9 @@ class CompletenessCheck(FrameProcessor):
 
 
 class OutputGate(FrameProcessor):
-    def __init__(self, notifier: BaseNotifier, **kwargs):
+    def __init__(self, *, notifier: BaseNotifier, start_open: bool = False, **kwargs):
         super().__init__(**kwargs)
-        self._gate_open = False
+        self._gate_open = start_open
         self._frames_buffer = []
         self._notifier = notifier
 
@@ -252,7 +252,9 @@ async def main():
         # sentence, this will wake up the notifier if that happens.
         user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0)
 
-        bot_output_gate = OutputGate(notifier=notifier)
+        # We start with the gate open because we send an initial context frame
+        # to start the conversation.
+        bot_output_gate = OutputGate(notifier=notifier, start_open=True)
 
         async def block_user_stopped_speaking(frame):
             return not isinstance(frame, UserStoppedSpeakingFrame)

examples/foundational/22c-natural-conversation-mixed-llms.py

@@ -333,9 +333,9 @@ class CompletenessCheck(FrameProcessor):
 
 
 class OutputGate(FrameProcessor):
-    def __init__(self, notifier: BaseNotifier, **kwargs):
+    def __init__(self, *, notifier: BaseNotifier, start_open: bool = False, **kwargs):
         super().__init__(**kwargs)
-        self._gate_open = False
+        self._gate_open = start_open
         self._frames_buffer = []
         self._notifier = notifier
 
@@ -461,7 +461,9 @@ async def main():
         # sentence, this will wake up the notifier if that happens.
         user_idle = UserIdleProcessor(callback=user_idle_notifier, timeout=5.0)
 
-        bot_output_gate = OutputGate(notifier=notifier)
+        # We start with the gate open because we send an initial context frame
+        # to start the conversation.
+        bot_output_gate = OutputGate(notifier=notifier, start_open=True)
 
         async def block_user_stopped_speaking(frame):
             return not isinstance(frame, UserStoppedSpeakingFrame)

pyproject.toml

@@ -32,6 +32,7 @@ dependencies = [
     "protobuf~=5.29.3",
     "pydantic~=2.10.5",
     "pyloudnorm~=0.1.1",
+    "resampy~=0.4.3",
     "soxr~=0.5.0"
 ]
 

src/pipecat/audio/resamplers/__init__.py

@@ -0,0 +1 @@
+

src/pipecat/audio/resamplers/base_audio_resampler.py

@@ -0,0 +1,30 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+from abc import ABC, abstractmethod
+
+
+class BaseAudioResampler(ABC):
+    """Abstract base class for audio resampling. This class defines an
+    interface for audio resampling implementations.
+    """
+
+    @abstractmethod
+    async def resample(self, audio: bytes, in_rate: int, out_rate: int) -> bytes:
+        """
+        Resamples the given audio data to a different sample rate.
+
+        This is an abstract method that must be implemented in subclasses.
+
+        Parameters:
+            audio (bytes): The audio data to be resampled, represented as a byte string.
+            in_rate (int): The original sample rate of the audio data (in Hz).
+            out_rate (int): The desired sample rate for the resampled audio data (in Hz).
+
+        Returns:
+            bytes: The resampled audio data as a byte string.
+        """
+        pass

src/pipecat/audio/resamplers/resampy_resampler.py

@@ -0,0 +1,25 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import numpy as np
+import resampy
+
+from pipecat.audio.resamplers.base_audio_resampler import BaseAudioResampler
+
+
+class ResampyResampler(BaseAudioResampler):
+    """Audio resampler implementation using the resampy library."""
+
+    def __init__(self, **kwargs):
+        pass
+
+    async def resample(self, audio: bytes, in_rate: int, out_rate: int) -> bytes:
+        if in_rate == out_rate:
+            return audio
+        audio_data = np.frombuffer(audio, dtype=np.int16)
+        resampled_audio = resampy.resample(audio_data, in_rate, out_rate, filter="kaiser_fast")
+        result = resampled_audio.astype(np.int16).tobytes()
+        return result

src/pipecat/audio/resamplers/soxr_resampler.py

@@ -0,0 +1,25 @@
+#
+# Copyright (c) 2024–2025, Daily
+#
+# SPDX-License-Identifier: BSD 2-Clause License
+#
+
+import numpy as np
+import soxr
+
+from pipecat.audio.resamplers.base_audio_resampler import BaseAudioResampler
+
+
+class SOXRAudioResampler(BaseAudioResampler):
+    """Audio resampler implementation using the SoX resampler library."""
+
+    def __init__(self, **kwargs):
+        pass
+
+    async def resample(self, audio: bytes, in_rate: int, out_rate: int) -> bytes:
+        if in_rate == out_rate:
+            return audio
+        audio_data = np.frombuffer(audio, dtype=np.int16)
+        resampled_audio = soxr.resample(audio_data, in_rate, out_rate, quality="VHQ")
+        result = resampled_audio.astype(np.int16).tobytes()
+        return result

src/pipecat/audio/utils.py

@@ -10,8 +10,24 @@ import numpy as np
 import pyloudnorm as pyln
 import soxr
 
+from pipecat.audio.resamplers.base_audio_resampler import BaseAudioResampler
+from pipecat.audio.resamplers.soxr_resampler import SOXRAudioResampler
+
+
+def create_default_resampler(**kwargs) -> BaseAudioResampler:
+    return SOXRAudioResampler(**kwargs)
+
 
 def resample_audio(audio: bytes, original_rate: int, target_rate: int) -> bytes:
+    import warnings
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("always")
+        warnings.warn(
+            "'resample_audio()' is deprecated, use 'create_default_resampler()' instead.",
+            DeprecationWarning,
+        )
+
     if original_rate == target_rate:
         return audio
     audio_data = np.frombuffer(audio, dtype=np.int16)
@@ -75,41 +91,45 @@ def exp_smoothing(value: float, prev_value: float, factor: float) -> float:
     return prev_value + factor * (value - prev_value)
 
 
-def ulaw_to_pcm(ulaw_bytes: bytes, in_sample_rate: int, out_sample_rate: int):
+async def ulaw_to_pcm(
+    ulaw_bytes: bytes, in_rate: int, out_rate: int, resampler: BaseAudioResampler
+):
     # Convert μ-law to PCM
     in_pcm_bytes = audioop.ulaw2lin(ulaw_bytes, 2)
 
     # Resample
-    out_pcm_bytes = resample_audio(in_pcm_bytes, in_sample_rate, out_sample_rate)
+    out_pcm_bytes = await resampler.resample(in_pcm_bytes, in_rate, out_rate)
 
     return out_pcm_bytes
 
 
-def pcm_to_ulaw(pcm_bytes: bytes, in_sample_rate: int, out_sample_rate: int):
+async def pcm_to_ulaw(pcm_bytes: bytes, in_rate: int, out_rate: int, resampler: BaseAudioResampler):
     # Resample
-    in_pcm_bytes = resample_audio(pcm_bytes, in_sample_rate, out_sample_rate)
+    in_pcm_bytes = await resampler.resample(pcm_bytes, in_rate, out_rate)
 
     # Convert PCM to μ-law
-    ulaw_bytes = audioop.lin2ulaw(in_pcm_bytes, 2)
+    out_ulaw_bytes = audioop.lin2ulaw(in_pcm_bytes, 2)
 
-    return ulaw_bytes
+    return out_ulaw_bytes
 
 
-def alaw_to_pcm(alaw_bytes: bytes, in_sample_rate: int, out_sample_rate: int) -> bytes:
+async def alaw_to_pcm(
+    alaw_bytes: bytes, in_rate: int, out_rate: int, resampler: BaseAudioResampler
+) -> bytes:
     # Convert a-law to PCM
     in_pcm_bytes = audioop.alaw2lin(alaw_bytes, 2)
 
     # Resample
-    out_pcm_bytes = resample_audio(in_pcm_bytes, in_sample_rate, out_sample_rate)
+    out_pcm_bytes = await resampler.resample(in_pcm_bytes, in_rate, out_rate)
 
     return out_pcm_bytes
 
 
-def pcm_to_alaw(pcm_bytes: bytes, in_sample_rate: int, out_sample_rate: int):
+async def pcm_to_alaw(pcm_bytes: bytes, in_rate: int, out_rate: int, resampler: BaseAudioResampler):
     # Resample
-    in_pcm_bytes = resample_audio(pcm_bytes, in_sample_rate, out_sample_rate)
+    in_pcm_bytes = await resampler.resample(pcm_bytes, in_rate, out_rate)
 
     # Convert PCM to μ-law
-    alaw_bytes = audioop.lin2alaw(in_pcm_bytes, 2)
+    out_alaw_bytes = audioop.lin2alaw(in_pcm_bytes, 2)
 
-    return alaw_bytes
+    return out_alaw_bytes

src/pipecat/processors/aggregators/gated_openai_llm_context.py

@@ -16,9 +16,10 @@ class GatedOpenAILLMContextAggregator(FrameProcessor):
 
     """
 
-    def __init__(self, notifier: BaseNotifier, **kwargs):
+    def __init__(self, *, notifier: BaseNotifier, start_open: bool = False, **kwargs):
         super().__init__(**kwargs)
         self._notifier = notifier
+        self._start_open = start_open
         self._last_context_frame = None
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
@@ -31,7 +32,11 @@ class GatedOpenAILLMContextAggregator(FrameProcessor):
             await self._stop()
             await self.push_frame(frame)
         elif isinstance(frame, OpenAILLMContextFrame):
-            self._last_context_frame = frame
+            if self._start_open:
+                self._start_open = False
+                await self.push_frame(frame, direction)
+            else:
+                self._last_context_frame = frame
         else:
             await self.push_frame(frame, direction)
 

src/pipecat/processors/async_generator.py

@@ -30,7 +30,7 @@ class AsyncGeneratorProcessor(FrameProcessor):
         if isinstance(frame, (CancelFrame, EndFrame)):
             await self._data_queue.put(None)
         else:
-            data = self._serializer.serialize(frame)
+            data = await self._serializer.serialize(frame)
             if data:
                 await self._data_queue.put(data)
 

src/pipecat/processors/audio/audio_buffer_processor.py

@@ -4,8 +4,12 @@
 # SPDX-License-Identifier: BSD 2-Clause License
 #
 
-from pipecat.audio.utils import interleave_stereo_audio, mix_audio, resample_audio
+import time
+
+from pipecat.audio.utils import create_default_resampler, interleave_stereo_audio, mix_audio
 from pipecat.frames.frames import (
+    AudioRawFrame,
+    CancelFrame,
     EndFrame,
     Frame,
     InputAudioRawFrame,
@@ -39,6 +43,13 @@ class AudioBufferProcessor(FrameProcessor):
         self._user_audio_buffer = bytearray()
         self._bot_audio_buffer = bytearray()
 
+        self._last_user_frame_at = 0
+        self._last_bot_frame_at = 0
+
+        self._recording = False
+
+        self._resampler = create_default_resampler()
+
         self._register_event_handler("on_audio_data")
 
     @property
@@ -64,43 +75,76 @@ class AudioBufferProcessor(FrameProcessor):
         else:
             return b""
 
-    def reset_audio_buffers(self):
+    async def start_recording(self):
-        self._user_audio_buffer = bytearray()
+        self._recording = True
-        self._bot_audio_buffer = bytearray()
+        self._reset_recording()
+
+    async def stop_recording(self):
+        await self._call_on_audio_data_handler()
+        self._recording = False
 
     async def process_frame(self, frame: Frame, direction: FrameDirection):
         await super().process_frame(frame, direction)
 
-        # Include all audio from the user.
+        if self._recording and isinstance(frame, InputAudioRawFrame):
-        if isinstance(frame, InputAudioRawFrame):
+            # Add silence if we need to.
-            resampled = resample_audio(frame.audio, frame.sample_rate, self._sample_rate)
+            silence = self._compute_silence(self._last_user_frame_at)
+            self._user_audio_buffer.extend(silence)
+            # Add user audio.
+            resampled = await self._resample_audio(frame)
             self._user_audio_buffer.extend(resampled)
-            # Sync the bot's buffer to the user's buffer by adding silence if needed
+            # Save time of frame so we can compute silence.
-            if len(self._user_audio_buffer) > len(self._bot_audio_buffer):
+            self._last_user_frame_at = time.time()
-                silence = b"\x00" * len(resampled)
+        elif self._recording and isinstance(frame, OutputAudioRawFrame):
-                self._bot_audio_buffer.extend(silence)
+            # Add silence if we need to.
-        # If the bot is speaking, include all audio from the bot.
+            silence = self._compute_silence(self._last_bot_frame_at)
-        elif isinstance(frame, OutputAudioRawFrame):
+            self._bot_audio_buffer.extend(silence)
-            resampled = resample_audio(frame.audio, frame.sample_rate, self._sample_rate)
+            # Add bot audio.
+            resampled = await self._resample_audio(frame)
             self._bot_audio_buffer.extend(resampled)
+            # 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:
             await self._call_on_audio_data_handler()
 
-        if isinstance(frame, EndFrame):
+        if isinstance(frame, (CancelFrame, EndFrame)):
-            await self._call_on_audio_data_handler()
+            await self.stop_recording()
 
         await self.push_frame(frame, direction)
 
     async def _call_on_audio_data_handler(self):
-        if not self.has_audio():
+        if not self.has_audio() or not self._recording:
             return
 
         merged_audio = self.merge_audio_buffers()
         await self._call_event_handler(
             "on_audio_data", merged_audio, self._sample_rate, self._num_channels
         )
-        self.reset_audio_buffers()
+        self._reset_audio_buffers()
 
     def _buffer_has_audio(self, buffer: bytearray) -> bool:
         return buffer is not None and len(buffer) > 0
+
+    def _reset_recording(self):
+        self._reset_audio_buffers()
+        self._last_user_frame_at = time.time()
+        self._last_bot_frame_at = time.time()
+
+    def _reset_audio_buffers(self):
+        self._user_audio_buffer = bytearray()
+        self._bot_audio_buffer = bytearray()
+
+    async def _resample_audio(self, frame: AudioRawFrame) -> bytes:
+        return await self._resampler.resample(frame.audio, frame.sample_rate, self._sample_rate)
+
+    def _compute_silence(self, from_time: float) -> bytes:
+        quiet_time = time.time() - from_time
+        # We should get audio frames very frequently. We pick 100ms because
+        # that's big enough, but it could be even a bit slower since we usually
+        # do 20ms audio frames.
+        if from_time == 0 or quiet_time < 0.1:
+            return b""
+        num_bytes = int(quiet_time * self._sample_rate) * 2
+        silence = b"\x00" * num_bytes
+        return silence

src/pipecat/serializers/base_serializer.py

@@ -22,9 +22,9 @@ class FrameSerializer(ABC):
         pass
 
     @abstractmethod
-    def serialize(self, frame: Frame) -> str | bytes | None:
+    async def serialize(self, frame: Frame) -> str | bytes | None:
         pass
 
     @abstractmethod
-    def deserialize(self, data: str | bytes) -> Frame | None:
+    async def deserialize(self, data: str | bytes) -> Frame | None:
         pass

src/pipecat/serializers/livekit.py

@@ -25,7 +25,7 @@ class LivekitFrameSerializer(FrameSerializer):
     def type(self) -> FrameSerializerType:
         return FrameSerializerType.BINARY
 
-    def serialize(self, frame: Frame) -> str | bytes | None:
+    async def serialize(self, frame: Frame) -> str | bytes | None:
         if not isinstance(frame, OutputAudioRawFrame):
             return None
         audio_frame = AudioFrame(
@@ -36,7 +36,7 @@ class LivekitFrameSerializer(FrameSerializer):
         )
         return pickle.dumps(audio_frame)
 
-    def deserialize(self, data: str | bytes) -> Frame | None:
+    async def deserialize(self, data: str | bytes) -> Frame | None:
         audio_frame: AudioFrame = pickle.loads(data)["frame"]
         return InputAudioRawFrame(
             audio=bytes(audio_frame.data),

src/pipecat/serializers/protobuf.py

@@ -41,7 +41,7 @@ class ProtobufFrameSerializer(FrameSerializer):
     def type(self) -> FrameSerializerType:
         return FrameSerializerType.BINARY
 
-    def serialize(self, frame: Frame) -> str | bytes | None:
+    async def serialize(self, frame: Frame) -> str | bytes | None:
         proto_frame = frame_protos.Frame()
         if type(frame) not in self.SERIALIZABLE_TYPES:
             logger.warning(f"Frame type {type(frame)} is not serializable")
@@ -57,26 +57,7 @@ class ProtobufFrameSerializer(FrameSerializer):
 
         return proto_frame.SerializeToString()
 
-    def deserialize(self, data: str | bytes) -> Frame | None:
+    async def deserialize(self, data: str | bytes) -> Frame | None:
-        """Returns a Frame object from a Frame protobuf.
-
-        Used to convert frames
-        passed over the wire as protobufs to Frame objects used in pipelines
-        and frame processors.
-
-        >>> serializer = ProtobufFrameSerializer()
-        >>> serializer.deserialize(
-        ...     serializer.serialize(OutputAudioFrame(data=b'1234567890')))
-        InputAudioFrame(data=b'1234567890')
-
-        >>> serializer.deserialize(
-        ...     serializer.serialize(TextFrame(text='hello world')))
-        TextFrame(text='hello world')
-
-        >>> serializer.deserialize(serializer.serialize(TranscriptionFrame(
-        ...     text="Hello there!", participantId="123", timestamp="2021-01-01")))
-        TranscriptionFrame(text='Hello there!', participantId='123', timestamp='2021-01-01')
-        """
         proto = frame_protos.Frame.FromString(data)
         which = proto.WhichOneof("frame")
         if which not in self.DESERIALIZABLE_FIELDS:

src/pipecat/serializers/telnyx.py

@@ -9,7 +9,13 @@ import json
 
 from pydantic import BaseModel
 
-from pipecat.audio.utils import alaw_to_pcm, pcm_to_alaw, pcm_to_ulaw, ulaw_to_pcm
+from pipecat.audio.utils import (
+    alaw_to_pcm,
+    create_default_resampler,
+    pcm_to_alaw,
+    pcm_to_ulaw,
+    ulaw_to_pcm,
+)
 from pipecat.frames.frames import (
     AudioRawFrame,
     Frame,
@@ -40,21 +46,23 @@ class TelnyxFrameSerializer(FrameSerializer):
         params.inbound_encoding = inbound_encoding
         self._params = params
 
+        self._resampler = create_default_resampler()
+
     @property
     def type(self) -> FrameSerializerType:
         return FrameSerializerType.TEXT
 
-    def serialize(self, frame: Frame) -> str | bytes | None:
+    async def serialize(self, frame: Frame) -> str | bytes | None:
         if isinstance(frame, AudioRawFrame):
             data = frame.audio
 
             if self._params.inbound_encoding == "PCMU":
-                serialized_data = pcm_to_ulaw(
+                serialized_data = await pcm_to_ulaw(
-                    data, frame.sample_rate, self._params.telnyx_sample_rate
+                    data, frame.sample_rate, self._params.telnyx_sample_rate, self._resampler
                 )
             elif self._params.inbound_encoding == "PCMA":
-                serialized_data = pcm_to_alaw(
+                serialized_data = await pcm_to_alaw(
-                    data, frame.sample_rate, self._params.telnyx_sample_rate
+                    data, frame.sample_rate, self._params.telnyx_sample_rate, self._resampler
                 )
             else:
                 raise ValueError(f"Unsupported encoding: {self._params.inbound_encoding}")
@@ -71,7 +79,7 @@ class TelnyxFrameSerializer(FrameSerializer):
             answer = {"event": "clear"}
             return json.dumps(answer)
 
-    def deserialize(self, data: str | bytes) -> Frame | None:
+    async def deserialize(self, data: str | bytes) -> Frame | None:
         message = json.loads(data)
 
         if message["event"] == "media":
@@ -79,12 +87,18 @@ class TelnyxFrameSerializer(FrameSerializer):
             payload = base64.b64decode(payload_base64)
 
             if self._params.outbound_encoding == "PCMU":
-                deserialized_data = ulaw_to_pcm(
+                deserialized_data = await ulaw_to_pcm(
-                    payload, self._params.telnyx_sample_rate, self._params.sample_rate
+                    payload,
+                    self._params.telnyx_sample_rate,
+                    self._params.sample_rate,
+                    self._resampler,
                 )
             elif self._params.outbound_encoding == "PCMA":
-                deserialized_data = alaw_to_pcm(
+                deserialized_data = await alaw_to_pcm(
-                    payload, self._params.telnyx_sample_rate, self._params.sample_rate
+                    payload,
+                    self._params.telnyx_sample_rate,
+                    self._params.sample_rate,
+                    self._resampler,
                 )
             else:
                 raise ValueError(f"Unsupported encoding: {self._params.outbound_encoding}")

src/pipecat/serializers/twilio.py

@@ -9,7 +9,7 @@ import json
 
 from pydantic import BaseModel
 
-from pipecat.audio.utils import pcm_to_ulaw, ulaw_to_pcm
+from pipecat.audio.utils import create_default_resampler, pcm_to_ulaw, ulaw_to_pcm
 from pipecat.frames.frames import (
     AudioRawFrame,
     Frame,
@@ -32,18 +32,22 @@ class TwilioFrameSerializer(FrameSerializer):
         self._stream_sid = stream_sid
         self._params = params
 
+        self._resampler = create_default_resampler()
+
     @property
     def type(self) -> FrameSerializerType:
         return FrameSerializerType.TEXT
 
-    def serialize(self, frame: Frame) -> str | bytes | None:
+    async def serialize(self, frame: Frame) -> str | bytes | None:
         if isinstance(frame, StartInterruptionFrame):
             answer = {"event": "clear", "streamSid": self._stream_sid}
             return json.dumps(answer)
         elif isinstance(frame, AudioRawFrame):
             data = frame.audio
 
-            serialized_data = pcm_to_ulaw(data, frame.sample_rate, self._params.twilio_sample_rate)
+            serialized_data = await pcm_to_ulaw(
+                data, frame.sample_rate, self._params.twilio_sample_rate, self._resampler
+            )
             payload = base64.b64encode(serialized_data).decode("utf-8")
             answer = {
                 "event": "media",
@@ -55,15 +59,15 @@ class TwilioFrameSerializer(FrameSerializer):
         elif isinstance(frame, (TransportMessageFrame, TransportMessageUrgentFrame)):
             return json.dumps(frame.message)
 
-    def deserialize(self, data: str | bytes) -> Frame | None:
+    async def deserialize(self, data: str | bytes) -> Frame | None:
         message = json.loads(data)
 
         if message["event"] == "media":
             payload_base64 = message["media"]["payload"]
             payload = base64.b64decode(payload_base64)
 
-            deserialized_data = ulaw_to_pcm(
+            deserialized_data = await ulaw_to_pcm(
-                payload, self._params.twilio_sample_rate, self._params.sample_rate
+                payload, self._params.twilio_sample_rate, self._params.sample_rate, self._resampler
             )
             audio_frame = InputAudioRawFrame(
                 audio=deserialized_data, num_channels=1, sample_rate=self._params.sample_rate

src/pipecat/services/aws.py

@@ -10,7 +10,7 @@ from typing import AsyncGenerator, Optional
 from loguru import logger
 from pydantic import BaseModel
 
-from pipecat.audio.utils import resample_audio
+from pipecat.audio.utils import create_default_resampler
 from pipecat.frames.frames import (
     ErrorFrame,
     Frame,
@@ -148,6 +148,8 @@ class PollyTTSService(TTSService):
             "volume": params.volume,
         }
 
+        self._resampler = create_default_resampler()
+
         self.set_voice(voice_id)
 
     def can_generate_metrics(self) -> bool:
@@ -193,8 +195,7 @@ class PollyTTSService(TTSService):
             response = self._polly_client.synthesize_speech(**args)
             if "AudioStream" in response:
                 audio_data = response["AudioStream"].read()
-                resampled = resample_audio(audio_data, 16000, self._settings["sample_rate"])
+                return audio_data
-                return resampled
             return None
 
         logger.debug(f"Generating TTS: [{text}]")
@@ -225,6 +226,10 @@ class PollyTTSService(TTSService):
                 yield None
                 return
 
+            audio_data = await self._resampler.resample(
+                audio_data, 16000, self._settings["sample_rate"]
+            )
+
             await self.start_tts_usage_metrics(text)
 
             yield TTSStartedFrame()

src/pipecat/services/canonical.py

@@ -117,7 +117,6 @@ class CanonicalMetricsService(AIService):
         try:
             await self._multipart_upload(filename)
             await aiofiles.os.remove(filename)
-            audio_buffer_processor.reset_audio_buffers()
         except FileNotFoundError:
             pass
         except Exception as e:

src/pipecat/services/tavus.py

@@ -12,7 +12,7 @@ import base64
 import aiohttp
 from loguru import logger
 
-from pipecat.audio.utils import resample_audio
+from pipecat.audio.utils import create_default_resampler
 from pipecat.frames.frames import (
     CancelFrame,
     EndFrame,
@@ -47,6 +47,8 @@ class TavusVideoService(AIService):
 
         self._conversation_id: str
 
+        self._resampler = create_default_resampler()
+
     async def initialize(self) -> str:
         url = "https://tavusapi.com/v2/conversations"
         headers = {"Content-Type": "application/json", "x-api-key": self._api_key}
@@ -89,12 +91,10 @@ class TavusVideoService(AIService):
         async with self._session.post(url, headers=headers) as r:
             r.raise_for_status()
 
-    async def _encode_audio_and_send(
+    async def _encode_audio_and_send(self, audio: bytes, in_rate: int, done: bool) -> None:
-        self, audio: bytes, original_sample_rate: int, done: bool
-    ) -> None:
         """Encodes audio to base64 and sends it to Tavus"""
         if not done:
-            audio = resample_audio(audio, original_sample_rate, 16000)
+            audio = await self._resampler.resample(audio, in_rate, 16000)
         audio_base64 = base64.b64encode(audio).decode("utf-8")
         logger.trace(f"{self}: sending {len(audio)} bytes")
         await self._send_audio_message(audio_base64, done=done)

src/pipecat/services/xtts.py

@@ -9,7 +9,7 @@ from typing import Any, AsyncGenerator, Dict
 import aiohttp
 from loguru import logger
 
-from pipecat.audio.utils import resample_audio
+from pipecat.audio.utils import create_default_resampler
 from pipecat.frames.frames import (
     ErrorFrame,
     Frame,
@@ -89,6 +89,8 @@ class XTTSService(TTSService):
         self._studio_speakers: Dict[str, Any] | None = None
         self._aiohttp_session = aiohttp_session
 
+        self._resampler = create_default_resampler()
+
     def can_generate_metrics(self) -> bool:
         return True
 
@@ -161,7 +163,7 @@ class XTTSService(TTSService):
                         buffer = buffer[48000:]
 
                         # XTTS uses 24000 so we need to resample to our desired rate.
-                        resampled_audio = resample_audio(
+                        resampled_audio = await self._resampler.resample(
                             bytes(process_data), 24000, self._sample_rate
                         )
                         # Create the frame with the resampled audio
@@ -170,7 +172,9 @@ class XTTSService(TTSService):
 
             # Process any remaining data in the buffer.
             if len(buffer) > 0:
-                resampled_audio = resample_audio(bytes(buffer), 24000, self._sample_rate)
+                resampled_audio = await self._resampler.resample(
+                    bytes(buffer), 24000, self._sample_rate
+                )
                 frame = TTSAudioRawFrame(resampled_audio, self._sample_rate, 1)
                 yield frame
 

src/pipecat/transports/network/fastapi_websocket.py

@@ -91,7 +91,7 @@ class FastAPIWebsocketInputTransport(BaseInputTransport):
     async def _receive_messages(self):
         try:
             async for message in self._iter_data():
-                frame = self._params.serializer.deserialize(message)
+                frame = await self._params.serializer.deserialize(message)
 
                 if not frame:
                     continue
@@ -163,7 +163,7 @@ class FastAPIWebsocketOutputTransport(BaseOutputTransport):
 
     async def _write_frame(self, frame: Frame):
         try:
-            payload = self._params.serializer.serialize(frame)
+            payload = await self._params.serializer.serialize(frame)
             if payload and self._websocket.client_state == WebSocketState.CONNECTED:
                 await self._send_data(payload)
         except Exception as e:

src/pipecat/transports/network/websocket_client.py

@@ -138,7 +138,7 @@ class WebsocketClientInputTransport(BaseInputTransport):
         await self._session.disconnect()
 
     async def on_message(self, websocket, message):
-        frame = self._params.serializer.deserialize(message)
+        frame = await self._params.serializer.deserialize(message)
         if not frame:
             return
         if isinstance(frame, InputAudioRawFrame) and self._params.audio_in_enabled:
@@ -200,7 +200,7 @@ class WebsocketClientOutputTransport(BaseOutputTransport):
         await self._write_audio_sleep()
 
     async def _write_frame(self, frame: Frame):
-        payload = self._params.serializer.serialize(frame)
+        payload = await self._params.serializer.serialize(frame)
         if payload:
             await self._session.send(payload)
 

src/pipecat/transports/network/websocket_server.py

@@ -105,7 +105,7 @@ class WebsocketServerInputTransport(BaseInputTransport):
         # Handle incoming messages
         try:
             async for message in websocket:
-                frame = self._params.serializer.deserialize(message)
+                frame = await self._params.serializer.deserialize(message)
 
                 if not frame:
                     continue
@@ -193,7 +193,7 @@ class WebsocketServerOutputTransport(BaseOutputTransport):
 
     async def _write_frame(self, frame: Frame):
         try:
-            payload = self._params.serializer.serialize(frame)
+            payload = await self._params.serializer.serialize(frame)
             if payload and self._websocket:
                 await self._websocket.send(payload)
         except Exception as e:

src/pipecat/transports/services/livekit.py

@@ -11,7 +11,7 @@ from typing import Any, Awaitable, Callable, List, Optional
 from loguru import logger
 from pydantic import BaseModel
 
-from pipecat.audio.utils import resample_audio
+from pipecat.audio.utils import create_default_resampler
 from pipecat.audio.vad.vad_analyzer import VADAnalyzer
 from pipecat.frames.frames import (
     AudioRawFrame,
@@ -349,6 +349,7 @@ class LiveKitInputTransport(BaseInputTransport):
         self._client = client
         self._audio_in_task = None
         self._vad_analyzer: VADAnalyzer | None = params.vad_analyzer
+        self._resampler = create_default_resampler()
 
     async def start(self, frame: StartFrame):
         await super().start(frame)
@@ -384,7 +385,9 @@ class LiveKitInputTransport(BaseInputTransport):
             audio_data = await self._client.get_next_audio_frame()
             if audio_data:
                 audio_frame_event, participant_id = audio_data
-                pipecat_audio_frame = self._convert_livekit_audio_to_pipecat(audio_frame_event)
+                pipecat_audio_frame = await self._convert_livekit_audio_to_pipecat(
+                    audio_frame_event
+                )
                 input_audio_frame = InputAudioRawFrame(
                     audio=pipecat_audio_frame.audio,
                     sample_rate=pipecat_audio_frame.sample_rate,
@@ -392,12 +395,12 @@ class LiveKitInputTransport(BaseInputTransport):
                 )
                 await self.push_audio_frame(input_audio_frame)
 
-    def _convert_livekit_audio_to_pipecat(
+    async def _convert_livekit_audio_to_pipecat(
         self, audio_frame_event: rtc.AudioFrameEvent
     ) -> AudioRawFrame:
         audio_frame = audio_frame_event.frame
 
-        audio_data = resample_audio(
+        audio_data = await self._resampler.resample(
             audio_frame.data.tobytes(), audio_frame.sample_rate, self._params.audio_in_sample_rate
         )
 

tests/test_protobuf_serializer.py

@@ -20,17 +20,19 @@ class TestProtobufFrameSerializer(unittest.IsolatedAsyncioTestCase):
 
     async def test_roundtrip(self):
         text_frame = TextFrame(text="hello world")
-        frame = self.serializer.deserialize(self.serializer.serialize(text_frame))
+        frame = await self.serializer.deserialize(await self.serializer.serialize(text_frame))
         self.assertEqual(text_frame, frame)
 
         transcription_frame = TranscriptionFrame(
             text="Hello there!", user_id="123", timestamp="2021-01-01"
         )
-        frame = self.serializer.deserialize(self.serializer.serialize(transcription_frame))
+        frame = await self.serializer.deserialize(
+            await self.serializer.serialize(transcription_frame)
+        )
         self.assertEqual(frame, transcription_frame)
 
         audio_frame = OutputAudioRawFrame(audio=b"1234567890", sample_rate=16000, num_channels=1)
-        frame = self.serializer.deserialize(self.serializer.serialize(audio_frame))
+        frame = await self.serializer.deserialize(await self.serializer.serialize(audio_frame))
         self.assertEqual(frame.audio, audio_frame.audio)
         self.assertEqual(frame.sample_rate, audio_frame.sample_rate)
         self.assertEqual(frame.num_channels, audio_frame.num_channels)