From fc89aad469959411f675ef8cfe35af5bb95567bf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 31 Jan 2025 11:04:18 -0800 Subject: [PATCH 01/11] introduce audio resamplers --- CHANGELOG.md | 11 ++++++ pyproject.toml | 1 + src/pipecat/audio/resamplers/__init__.py | 1 + .../audio/resamplers/base_audio_resampler.py | 30 ++++++++++++++++ .../audio/resamplers/resampy_resampler.py | 24 +++++++++++++ .../audio/resamplers/soxr_resampler.py | 24 +++++++++++++ src/pipecat/audio/utils.py | 34 ++++++++++++++----- .../audio/audio_buffer_processor.py | 13 +++++-- src/pipecat/serializers/telnyx.py | 25 +++++++++++--- src/pipecat/serializers/twilio.py | 10 ++++-- src/pipecat/services/aws.py | 8 +++-- src/pipecat/services/tavus.py | 10 +++--- src/pipecat/services/xtts.py | 8 +++-- src/pipecat/transports/services/livekit.py | 5 +-- 14 files changed, 173 insertions(+), 31 deletions(-) create mode 100644 src/pipecat/audio/resamplers/__init__.py create mode 100644 src/pipecat/audio/resamplers/base_audio_resampler.py create mode 100644 src/pipecat/audio/resamplers/resampy_resampler.py create mode 100644 src/pipecat/audio/resamplers/soxr_resampler.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 1d9a402eb..f9e1e4c34 100644 --- a/CHANGELOG.md +++ b/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 @@ -56,6 +62,11 @@ 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. + ### Fixed - Fixed an issue where `ElevenLabsTTSService` messages would return a 1009 diff --git a/pyproject.toml b/pyproject.toml index 801948c8f..ee911098f 100644 --- a/pyproject.toml +++ b/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" ] diff --git a/src/pipecat/audio/resamplers/__init__.py b/src/pipecat/audio/resamplers/__init__.py new file mode 100644 index 000000000..8b1378917 --- /dev/null +++ b/src/pipecat/audio/resamplers/__init__.py @@ -0,0 +1 @@ + diff --git a/src/pipecat/audio/resamplers/base_audio_resampler.py b/src/pipecat/audio/resamplers/base_audio_resampler.py new file mode 100644 index 000000000..0d7c3afef --- /dev/null +++ b/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 + 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 diff --git a/src/pipecat/audio/resamplers/resampy_resampler.py b/src/pipecat/audio/resamplers/resampy_resampler.py new file mode 100644 index 000000000..be194b767 --- /dev/null +++ b/src/pipecat/audio/resamplers/resampy_resampler.py @@ -0,0 +1,24 @@ +# +# 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 + + 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") + return resampled_audio.astype(np.int16).tobytes() diff --git a/src/pipecat/audio/resamplers/soxr_resampler.py b/src/pipecat/audio/resamplers/soxr_resampler.py new file mode 100644 index 000000000..eaa06ad4e --- /dev/null +++ b/src/pipecat/audio/resamplers/soxr_resampler.py @@ -0,0 +1,24 @@ +# +# 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 + + 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") + return resampled_audio.astype(np.int16).tobytes() diff --git a/src/pipecat/audio/utils.py b/src/pipecat/audio/utils.py index 8e95ebc31..9a9d442dd 100644 --- a/src/pipecat/audio/utils.py +++ b/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,19 +91,19 @@ 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): +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 = 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): +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 = resampler.resample(pcm_bytes, in_rate, out_rate) # Convert PCM to μ-law ulaw_bytes = audioop.lin2ulaw(in_pcm_bytes, 2) @@ -95,19 +111,21 @@ def pcm_to_ulaw(pcm_bytes: bytes, in_sample_rate: int, out_sample_rate: int): return ulaw_bytes -def alaw_to_pcm(alaw_bytes: bytes, in_sample_rate: int, out_sample_rate: int) -> bytes: +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 = 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): +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 = resampler.resample(pcm_bytes, in_rate, out_rate) # Convert PCM to μ-law alaw_bytes = audioop.lin2alaw(in_pcm_bytes, 2) diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index fed4db3c8..7500a3f9e 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -4,8 +4,10 @@ # SPDX-License-Identifier: BSD 2-Clause License # -from pipecat.audio.utils import interleave_stereo_audio, mix_audio, resample_audio +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 +41,8 @@ class AudioBufferProcessor(FrameProcessor): self._user_audio_buffer = bytearray() self._bot_audio_buffer = bytearray() + self._resampler = create_default_resampler() + self._register_event_handler("on_audio_data") @property @@ -73,7 +77,7 @@ class AudioBufferProcessor(FrameProcessor): # Include all audio from the user. if isinstance(frame, InputAudioRawFrame): - resampled = resample_audio(frame.audio, frame.sample_rate, self._sample_rate) + resampled = 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 if len(self._user_audio_buffer) > len(self._bot_audio_buffer): @@ -81,7 +85,7 @@ class AudioBufferProcessor(FrameProcessor): self._bot_audio_buffer.extend(silence) # If the bot is speaking, include all audio from the bot. elif isinstance(frame, OutputAudioRawFrame): - resampled = resample_audio(frame.audio, frame.sample_rate, self._sample_rate) + resampled = self._resample_audio(frame) self._bot_audio_buffer.extend(resampled) if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size: @@ -104,3 +108,6 @@ class AudioBufferProcessor(FrameProcessor): def _buffer_has_audio(self, buffer: bytearray) -> bool: return buffer is not None and len(buffer) > 0 + + def _resample_audio(self, frame: AudioRawFrame) -> bytes: + return self._resampler.resample(frame.audio, frame.sample_rate, self._sample_rate) diff --git a/src/pipecat/serializers/telnyx.py b/src/pipecat/serializers/telnyx.py index 14482531d..aacb6b646 100644 --- a/src/pipecat/serializers/telnyx.py +++ b/src/pipecat/serializers/telnyx.py @@ -6,10 +6,17 @@ import base64 import json +from typing import Optional 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,6 +47,8 @@ class TelnyxFrameSerializer(FrameSerializer): params.inbound_encoding = inbound_encoding self._params = params + self._resampler = create_default_resampler() + @property def type(self) -> FrameSerializerType: return FrameSerializerType.TEXT @@ -50,11 +59,11 @@ class TelnyxFrameSerializer(FrameSerializer): if self._params.inbound_encoding == "PCMU": serialized_data = 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( - 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}") @@ -80,11 +89,17 @@ class TelnyxFrameSerializer(FrameSerializer): if self._params.outbound_encoding == "PCMU": deserialized_data = 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( - 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}") diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py index 40c8c726f..7da7a80c7 100644 --- a/src/pipecat/serializers/twilio.py +++ b/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,6 +32,8 @@ class TwilioFrameSerializer(FrameSerializer): self._stream_sid = stream_sid self._params = params + self._resampler = create_default_resampler() + @property def type(self) -> FrameSerializerType: return FrameSerializerType.TEXT @@ -43,7 +45,9 @@ class TwilioFrameSerializer(FrameSerializer): elif isinstance(frame, AudioRawFrame): data = frame.audio - serialized_data = pcm_to_ulaw(data, frame.sample_rate, self._params.twilio_sample_rate) + serialized_data = 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", @@ -63,7 +67,7 @@ class TwilioFrameSerializer(FrameSerializer): payload = base64.b64decode(payload_base64) deserialized_data = 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 diff --git a/src/pipecat/services/aws.py b/src/pipecat/services/aws.py index 3061e13a1..eb03bbc59 100644 --- a/src/pipecat/services/aws.py +++ b/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,7 +195,9 @@ 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"]) + resampled = self._resampler.resample( + audio_data, 16000, self._settings["sample_rate"] + ) return resampled return None diff --git a/src/pipecat/services/tavus.py b/src/pipecat/services/tavus.py index b1dfef9e4..e6693ecc7 100644 --- a/src/pipecat/services/tavus.py +++ b/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( - self, audio: bytes, original_sample_rate: int, done: bool - ) -> None: + async def _encode_audio_and_send(self, audio: bytes, in_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 = 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) diff --git a/src/pipecat/services/xtts.py b/src/pipecat/services/xtts.py index ba8803f51..b26660175 100644 --- a/src/pipecat/services/xtts.py +++ b/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 = self._resampler.resample( bytes(process_data), 24000, self._sample_rate ) # Create the frame with the resampled audio @@ -170,7 +172,7 @@ 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 = self._resampler.resample(bytes(buffer), 24000, self._sample_rate) frame = TTSAudioRawFrame(resampled_audio, self._sample_rate, 1) yield frame diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index 49f177179..a80e2fa32 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/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) @@ -397,7 +398,7 @@ class LiveKitInputTransport(BaseInputTransport): ) -> AudioRawFrame: audio_frame = audio_frame_event.frame - audio_data = resample_audio( + audio_data = self._resampler.resample( audio_frame.data.tobytes(), audio_frame.sample_rate, self._params.audio_in_sample_rate ) From 13611fd8e19b79ada8506dd8625179b49f90c5b7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 31 Jan 2025 11:04:43 -0800 Subject: [PATCH 02/11] AudioBufferProcessor: call callback on CancelFrame --- CHANGELOG.md | 3 +++ src/pipecat/processors/audio/audio_buffer_processor.py | 2 +- 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f9e1e4c34..ca876830f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -69,6 +69,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Fixed +- Fixed an issue in `AudioBufferProcessor` where user callback would not be + called on task cancellation. + - Fixed an issue where `ElevenLabsTTSService` messages would return a 1009 websocket error by increasing the max message size limit to 16MB. diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index 7500a3f9e..eb92c8109 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -91,7 +91,7 @@ class AudioBufferProcessor(FrameProcessor): 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.push_frame(frame, direction) From cda34a1320051884bf4a95cc4686e4deb3e6d3bd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 31 Jan 2025 11:06:27 -0800 Subject: [PATCH 03/11] AudioBufferProcessor: fix user/bot audio buffers silence padding --- CHANGELOG.md | 3 +++ src/pipecat/processors/audio/audio_buffer_processor.py | 10 ++++++++-- 2 files changed, 11 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ca876830f..154918835 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -72,6 +72,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - 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. diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index eb92c8109..376f6bf74 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -79,14 +79,20 @@ class AudioBufferProcessor(FrameProcessor): if isinstance(frame, InputAudioRawFrame): resampled = 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 + # Sync the bot's buffer to the user's buffer by adding silence if needed. if len(self._user_audio_buffer) > len(self._bot_audio_buffer): - silence = b"\x00" * len(resampled) + missing = len(self._user_audio_buffer) - len(self._bot_audio_buffer) + silence = b"\x00" * missing self._bot_audio_buffer.extend(silence) # If the bot is speaking, include all audio from the bot. elif isinstance(frame, OutputAudioRawFrame): resampled = self._resample_audio(frame) self._bot_audio_buffer.extend(resampled) + # Sync the user's buffer to the bot's buffer by adding silence if needed. + if len(self._bot_audio_buffer) > len(self._user_audio_buffer): + missing = len(self._bot_audio_buffer) - len(self._user_audio_buffer) + silence = b"\x00" * missing + self._user_audio_buffer.extend(silence) if self._buffer_size > 0 and len(self._user_audio_buffer) > self._buffer_size: await self._call_on_audio_data_handler() From dcfb86583deba92fdc88d0e04c69cd9ffd25b3de Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 31 Jan 2025 12:47:49 -0800 Subject: [PATCH 04/11] serializers: serialize()/deserialize() are now async --- CHANGELOG.md | 3 +++ src/pipecat/processors/async_generator.py | 2 +- src/pipecat/serializers/base_serializer.py | 4 ++-- src/pipecat/serializers/livekit.py | 4 ++-- src/pipecat/serializers/protobuf.py | 23 ++----------------- src/pipecat/serializers/telnyx.py | 4 ++-- src/pipecat/serializers/twilio.py | 4 ++-- .../transports/network/fastapi_websocket.py | 4 ++-- .../transports/network/websocket_client.py | 4 ++-- .../transports/network/websocket_server.py | 4 ++-- src/pipecat/transports/services/livekit.py | 8 ++++--- tests/test_protobuf_serializer.py | 8 ++++--- 12 files changed, 30 insertions(+), 42 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 154918835..8734221cf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1544,6 +1544,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`. diff --git a/src/pipecat/processors/async_generator.py b/src/pipecat/processors/async_generator.py index e13026cdb..142d9cb47 100644 --- a/src/pipecat/processors/async_generator.py +++ b/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) diff --git a/src/pipecat/serializers/base_serializer.py b/src/pipecat/serializers/base_serializer.py index 428649131..8a0efc49b 100644 --- a/src/pipecat/serializers/base_serializer.py +++ b/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 diff --git a/src/pipecat/serializers/livekit.py b/src/pipecat/serializers/livekit.py index 760097ffe..d856a7a56 100644 --- a/src/pipecat/serializers/livekit.py +++ b/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), diff --git a/src/pipecat/serializers/protobuf.py b/src/pipecat/serializers/protobuf.py index 13d7ded34..125f2037f 100644 --- a/src/pipecat/serializers/protobuf.py +++ b/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: - """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') - """ + async def deserialize(self, data: str | bytes) -> Frame | None: proto = frame_protos.Frame.FromString(data) which = proto.WhichOneof("frame") if which not in self.DESERIALIZABLE_FIELDS: diff --git a/src/pipecat/serializers/telnyx.py b/src/pipecat/serializers/telnyx.py index aacb6b646..2e0246ea7 100644 --- a/src/pipecat/serializers/telnyx.py +++ b/src/pipecat/serializers/telnyx.py @@ -53,7 +53,7 @@ class TelnyxFrameSerializer(FrameSerializer): 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 @@ -80,7 +80,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": diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py index 7da7a80c7..c6d4e1e13 100644 --- a/src/pipecat/serializers/twilio.py +++ b/src/pipecat/serializers/twilio.py @@ -38,7 +38,7 @@ class TwilioFrameSerializer(FrameSerializer): 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) @@ -59,7 +59,7 @@ 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": diff --git a/src/pipecat/transports/network/fastapi_websocket.py b/src/pipecat/transports/network/fastapi_websocket.py index d4d458927..b53daa3b4 100644 --- a/src/pipecat/transports/network/fastapi_websocket.py +++ b/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: diff --git a/src/pipecat/transports/network/websocket_client.py b/src/pipecat/transports/network/websocket_client.py index f2e26c4b5..56a7b57ac 100644 --- a/src/pipecat/transports/network/websocket_client.py +++ b/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) diff --git a/src/pipecat/transports/network/websocket_server.py b/src/pipecat/transports/network/websocket_server.py index c0dd299f6..684e030ad 100644 --- a/src/pipecat/transports/network/websocket_server.py +++ b/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: diff --git a/src/pipecat/transports/services/livekit.py b/src/pipecat/transports/services/livekit.py index a80e2fa32..11926f86c 100644 --- a/src/pipecat/transports/services/livekit.py +++ b/src/pipecat/transports/services/livekit.py @@ -385,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, @@ -393,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 = self._resampler.resample( + audio_data = await self._resampler.resample( audio_frame.data.tobytes(), audio_frame.sample_rate, self._params.audio_in_sample_rate ) diff --git a/tests/test_protobuf_serializer.py b/tests/test_protobuf_serializer.py index 88fd7867e..7d45800a8 100644 --- a/tests/test_protobuf_serializer.py +++ b/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) From 06449eff2c0db1f02e244e629e1ff5813955d01d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 31 Jan 2025 12:58:55 -0800 Subject: [PATCH 05/11] BaseAudioResampler: make resample() async --- .../audio/resamplers/base_audio_resampler.py | 2 +- .../audio/resamplers/resampy_resampler.py | 5 +++-- src/pipecat/audio/resamplers/soxr_resampler.py | 5 +++-- src/pipecat/audio/utils.py | 18 ++++++++++-------- .../processors/audio/audio_buffer_processor.py | 8 ++++---- src/pipecat/serializers/telnyx.py | 9 ++++----- src/pipecat/serializers/twilio.py | 4 ++-- src/pipecat/services/aws.py | 9 +++++---- src/pipecat/services/tavus.py | 2 +- src/pipecat/services/xtts.py | 6 ++++-- 10 files changed, 37 insertions(+), 31 deletions(-) diff --git a/src/pipecat/audio/resamplers/base_audio_resampler.py b/src/pipecat/audio/resamplers/base_audio_resampler.py index 0d7c3afef..6afbbcbfe 100644 --- a/src/pipecat/audio/resamplers/base_audio_resampler.py +++ b/src/pipecat/audio/resamplers/base_audio_resampler.py @@ -13,7 +13,7 @@ class BaseAudioResampler(ABC): """ @abstractmethod - def resample(self, audio: bytes, in_rate: int, out_rate: int) -> bytes: + async def resample(self, audio: bytes, in_rate: int, out_rate: int) -> bytes: """ Resamples the given audio data to a different sample rate. diff --git a/src/pipecat/audio/resamplers/resampy_resampler.py b/src/pipecat/audio/resamplers/resampy_resampler.py index be194b767..8c053fc3b 100644 --- a/src/pipecat/audio/resamplers/resampy_resampler.py +++ b/src/pipecat/audio/resamplers/resampy_resampler.py @@ -16,9 +16,10 @@ class ResampyResampler(BaseAudioResampler): def __init__(self, **kwargs): pass - def resample(self, audio: bytes, in_rate: int, out_rate: int) -> bytes: + 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") - return resampled_audio.astype(np.int16).tobytes() + result = resampled_audio.astype(np.int16).tobytes() + return result diff --git a/src/pipecat/audio/resamplers/soxr_resampler.py b/src/pipecat/audio/resamplers/soxr_resampler.py index eaa06ad4e..88edb84eb 100644 --- a/src/pipecat/audio/resamplers/soxr_resampler.py +++ b/src/pipecat/audio/resamplers/soxr_resampler.py @@ -16,9 +16,10 @@ class SOXRAudioResampler(BaseAudioResampler): def __init__(self, **kwargs): pass - def resample(self, audio: bytes, in_rate: int, out_rate: int) -> bytes: + 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") - return resampled_audio.astype(np.int16).tobytes() + result = resampled_audio.astype(np.int16).tobytes() + return result diff --git a/src/pipecat/audio/utils.py b/src/pipecat/audio/utils.py index 9a9d442dd..0cb5d4f06 100644 --- a/src/pipecat/audio/utils.py +++ b/src/pipecat/audio/utils.py @@ -91,19 +91,21 @@ 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_rate: int, out_rate: int, resampler: BaseAudioResampler): +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 = resampler.resample(in_pcm_bytes, in_rate, out_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_rate: int, out_rate: int, resampler: BaseAudioResampler): +async def pcm_to_ulaw(pcm_bytes: bytes, in_rate: int, out_rate: int, resampler: BaseAudioResampler): # Resample - in_pcm_bytes = resampler.resample(pcm_bytes, in_rate, out_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) @@ -111,21 +113,21 @@ def pcm_to_ulaw(pcm_bytes: bytes, in_rate: int, out_rate: int, resampler: BaseAu return ulaw_bytes -def alaw_to_pcm( +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 = resampler.resample(in_pcm_bytes, in_rate, out_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_rate: int, out_rate: int, resampler: BaseAudioResampler): +async def pcm_to_alaw(pcm_bytes: bytes, in_rate: int, out_rate: int, resampler: BaseAudioResampler): # Resample - in_pcm_bytes = resampler.resample(pcm_bytes, in_rate, out_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) diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index 376f6bf74..f00da6eb8 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -77,7 +77,7 @@ class AudioBufferProcessor(FrameProcessor): # Include all audio from the user. if isinstance(frame, InputAudioRawFrame): - resampled = self._resample_audio(frame) + 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. if len(self._user_audio_buffer) > len(self._bot_audio_buffer): @@ -86,7 +86,7 @@ class AudioBufferProcessor(FrameProcessor): self._bot_audio_buffer.extend(silence) # If the bot is speaking, include all audio from the bot. elif isinstance(frame, OutputAudioRawFrame): - resampled = self._resample_audio(frame) + resampled = await self._resample_audio(frame) self._bot_audio_buffer.extend(resampled) # Sync the user's buffer to the bot's buffer by adding silence if needed. if len(self._bot_audio_buffer) > len(self._user_audio_buffer): @@ -115,5 +115,5 @@ class AudioBufferProcessor(FrameProcessor): def _buffer_has_audio(self, buffer: bytearray) -> bool: return buffer is not None and len(buffer) > 0 - def _resample_audio(self, frame: AudioRawFrame) -> bytes: - return self._resampler.resample(frame.audio, frame.sample_rate, self._sample_rate) + async def _resample_audio(self, frame: AudioRawFrame) -> bytes: + return await self._resampler.resample(frame.audio, frame.sample_rate, self._sample_rate) diff --git a/src/pipecat/serializers/telnyx.py b/src/pipecat/serializers/telnyx.py index 2e0246ea7..79c6f42ff 100644 --- a/src/pipecat/serializers/telnyx.py +++ b/src/pipecat/serializers/telnyx.py @@ -6,7 +6,6 @@ import base64 import json -from typing import Optional from pydantic import BaseModel @@ -58,11 +57,11 @@ class TelnyxFrameSerializer(FrameSerializer): 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, 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, self._resampler ) else: @@ -88,14 +87,14 @@ 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, 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, diff --git a/src/pipecat/serializers/twilio.py b/src/pipecat/serializers/twilio.py index c6d4e1e13..7862fa0db 100644 --- a/src/pipecat/serializers/twilio.py +++ b/src/pipecat/serializers/twilio.py @@ -45,7 +45,7 @@ class TwilioFrameSerializer(FrameSerializer): elif isinstance(frame, AudioRawFrame): data = frame.audio - serialized_data = pcm_to_ulaw( + 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") @@ -66,7 +66,7 @@ class TwilioFrameSerializer(FrameSerializer): 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, self._resampler ) audio_frame = InputAudioRawFrame( diff --git a/src/pipecat/services/aws.py b/src/pipecat/services/aws.py index eb03bbc59..4444f07b7 100644 --- a/src/pipecat/services/aws.py +++ b/src/pipecat/services/aws.py @@ -195,10 +195,7 @@ class PollyTTSService(TTSService): response = self._polly_client.synthesize_speech(**args) if "AudioStream" in response: audio_data = response["AudioStream"].read() - resampled = self._resampler.resample( - audio_data, 16000, self._settings["sample_rate"] - ) - return resampled + return audio_data return None logger.debug(f"Generating TTS: [{text}]") @@ -229,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() diff --git a/src/pipecat/services/tavus.py b/src/pipecat/services/tavus.py index e6693ecc7..b0ca699cb 100644 --- a/src/pipecat/services/tavus.py +++ b/src/pipecat/services/tavus.py @@ -94,7 +94,7 @@ class TavusVideoService(AIService): async def _encode_audio_and_send(self, audio: bytes, in_rate: int, done: bool) -> None: """Encodes audio to base64 and sends it to Tavus""" if not done: - audio = self._resampler.resample(audio, in_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) diff --git a/src/pipecat/services/xtts.py b/src/pipecat/services/xtts.py index b26660175..a10247acd 100644 --- a/src/pipecat/services/xtts.py +++ b/src/pipecat/services/xtts.py @@ -163,7 +163,7 @@ class XTTSService(TTSService): buffer = buffer[48000:] # XTTS uses 24000 so we need to resample to our desired rate. - resampled_audio = self._resampler.resample( + resampled_audio = await self._resampler.resample( bytes(process_data), 24000, self._sample_rate ) # Create the frame with the resampled audio @@ -172,7 +172,9 @@ class XTTSService(TTSService): # Process any remaining data in the buffer. if len(buffer) > 0: - resampled_audio = self._resampler.resample(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 From 1f14f6269698c11c806ebe90f4b7938fa65be04e Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 31 Jan 2025 14:27:25 -0800 Subject: [PATCH 06/11] AudioBufferProcessor: fix audio buffer silence computation --- CHANGELOG.md | 2 + .../audio/audio_buffer_processor.py | 48 ++++++++++++++----- 2 files changed, 38 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8734221cf..83dd4b49a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -69,6 +69,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### 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. diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index f00da6eb8..86a41cbd4 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -4,6 +4,8 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import time + from pipecat.audio.utils import create_default_resampler, interleave_stereo_audio, mix_audio from pipecat.frames.frames import ( AudioRawFrame, @@ -12,6 +14,7 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, OutputAudioRawFrame, + StartFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -41,6 +44,9 @@ 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._resampler = create_default_resampler() self._register_event_handler("on_audio_data") @@ -72,27 +78,34 @@ class AudioBufferProcessor(FrameProcessor): self._user_audio_buffer = bytearray() self._bot_audio_buffer = bytearray() + self._last_user_frame_at = time.time() + self._last_bot_frame_at = time.time() + async def process_frame(self, frame: Frame, direction: FrameDirection): await super().process_frame(frame, direction) - # Include all audio from the user. + if isinstance(frame, StartFrame): + self._last_user_frame_at = time.time() + self._last_bot_frame_at = time.time() + if isinstance(frame, InputAudioRawFrame): + # Add silence if we need to. + 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. - if len(self._user_audio_buffer) > len(self._bot_audio_buffer): - missing = len(self._user_audio_buffer) - len(self._bot_audio_buffer) - silence = b"\x00" * missing - self._bot_audio_buffer.extend(silence) - # If the bot is speaking, include all audio from the bot. + # Save time of frame so we can compute silence. + self._last_user_frame_at = time.time() elif isinstance(frame, OutputAudioRawFrame): + # Add silence if we need to. + silence = self._compute_silence(self._last_bot_frame_at) + self._bot_audio_buffer.extend(silence) + # Add bot audio. resampled = await self._resample_audio(frame) self._bot_audio_buffer.extend(resampled) - # Sync the user's buffer to the bot's buffer by adding silence if needed. - if len(self._bot_audio_buffer) > len(self._user_audio_buffer): - missing = len(self._bot_audio_buffer) - len(self._user_audio_buffer) - silence = b"\x00" * missing - self._user_audio_buffer.extend(silence) + # 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() @@ -117,3 +130,14 @@ class AudioBufferProcessor(FrameProcessor): 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 From 371c2f3704354206be8e42982a6e255d38c2ceb9 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 31 Jan 2025 16:22:48 -0800 Subject: [PATCH 07/11] canonical: do not reset audio buffers --- src/pipecat/services/canonical.py | 1 - 1 file changed, 1 deletion(-) diff --git a/src/pipecat/services/canonical.py b/src/pipecat/services/canonical.py index 884171b10..31bd6e89a 100644 --- a/src/pipecat/services/canonical.py +++ b/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: From f3f22cf61c7c646308c23fffd6bef871be7a0da3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Fri, 31 Jan 2025 16:23:54 -0800 Subject: [PATCH 08/11] AudioBufferProcessor: add start_recording()/stop_recording() --- CHANGELOG.md | 6 +++ examples/canonical-metrics/bot.py | 1 + examples/chatbot-audio-recording/bot.py | 6 ++- .../audio/audio_buffer_processor.py | 37 +++++++++++-------- 4 files changed, 33 insertions(+), 17 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 83dd4b49a..73e188ced 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -67,6 +67,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - `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. diff --git a/examples/canonical-metrics/bot.py b/examples/canonical-metrics/bot.py index 945792b25..e2673bb3e 100644 --- a/examples/canonical-metrics/bot.py +++ b/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()]) diff --git a/examples/chatbot-audio-recording/bot.py b/examples/chatbot-audio-recording/bot.py index 7cf750c62..62b7b29c8 100644 --- a/examples/chatbot-audio-recording/bot.py +++ b/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. - audiobuffer = AudioBufferProcessor(buffer_size=480000) + # NOTE: Watch out! This will save all the conversation in memory. You + # 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()]) diff --git a/src/pipecat/processors/audio/audio_buffer_processor.py b/src/pipecat/processors/audio/audio_buffer_processor.py index 86a41cbd4..d56604330 100644 --- a/src/pipecat/processors/audio/audio_buffer_processor.py +++ b/src/pipecat/processors/audio/audio_buffer_processor.py @@ -14,7 +14,6 @@ from pipecat.frames.frames import ( Frame, InputAudioRawFrame, OutputAudioRawFrame, - StartFrame, ) from pipecat.processors.frame_processor import FrameDirection, FrameProcessor @@ -47,6 +46,8 @@ class AudioBufferProcessor(FrameProcessor): 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") @@ -74,21 +75,18 @@ class AudioBufferProcessor(FrameProcessor): else: return b"" - def reset_audio_buffers(self): - self._user_audio_buffer = bytearray() - self._bot_audio_buffer = bytearray() + async def start_recording(self): + self._recording = True + self._reset_recording() - self._last_user_frame_at = time.time() - self._last_bot_frame_at = time.time() + 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) - if isinstance(frame, StartFrame): - self._last_user_frame_at = time.time() - self._last_bot_frame_at = time.time() - - if isinstance(frame, InputAudioRawFrame): + if self._recording and isinstance(frame, InputAudioRawFrame): # Add silence if we need to. silence = self._compute_silence(self._last_user_frame_at) self._user_audio_buffer.extend(silence) @@ -97,7 +95,7 @@ class AudioBufferProcessor(FrameProcessor): self._user_audio_buffer.extend(resampled) # Save time of frame so we can compute silence. self._last_user_frame_at = time.time() - elif isinstance(frame, OutputAudioRawFrame): + elif self._recording and isinstance(frame, OutputAudioRawFrame): # Add silence if we need to. silence = self._compute_silence(self._last_bot_frame_at) self._bot_audio_buffer.extend(silence) @@ -111,23 +109,32 @@ class AudioBufferProcessor(FrameProcessor): await self._call_on_audio_data_handler() 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) From aafb2db62058d031d145e3f5de996b7176e1eeb5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 3 Feb 2025 09:15:24 -0800 Subject: [PATCH 09/11] GatedOpenAILLMContextAggregator: use keyword argument and add start_open --- CHANGELOG.md | 3 +++ .../processors/aggregators/gated_openai_llm_context.py | 9 +++++++-- 2 files changed, 10 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 73e188ced..7ef8af2b3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,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`. diff --git a/src/pipecat/processors/aggregators/gated_openai_llm_context.py b/src/pipecat/processors/aggregators/gated_openai_llm_context.py index 8c3f496eb..f3ae78121 100644 --- a/src/pipecat/processors/aggregators/gated_openai_llm_context.py +++ b/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) From 5dcdc68ef50e4b30eb6e37fb9ec03134ca59dea6 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 3 Feb 2025 09:16:58 -0800 Subject: [PATCH 10/11] examples: fix 22 series initial gate state --- examples/foundational/22-natural-conversation.py | 7 +++++-- .../foundational/22b-natural-conversation-proposal.py | 8 +++++--- .../foundational/22c-natural-conversation-mixed-llms.py | 8 +++++--- 3 files changed, 15 insertions(+), 8 deletions(-) diff --git a/examples/foundational/22-natural-conversation.py b/examples/foundational/22-natural-conversation.py index 9b6471f7c..5e6b0e501 100644 --- a/examples/foundational/22-natural-conversation.py +++ b/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. - gated_context_aggregator = GatedOpenAILLMContextAggregator(notifier) + # notifier is woken up. We start with the gate open because we send an + # 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): diff --git a/examples/foundational/22b-natural-conversation-proposal.py b/examples/foundational/22b-natural-conversation-proposal.py index 8f19618a8..57639fe20 100644 --- a/examples/foundational/22b-natural-conversation-proposal.py +++ b/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) diff --git a/examples/foundational/22c-natural-conversation-mixed-llms.py b/examples/foundational/22c-natural-conversation-mixed-llms.py index e4509da9c..1a24af21c 100644 --- a/examples/foundational/22c-natural-conversation-mixed-llms.py +++ b/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) From 57dadb63595ba65d3db5e281d4d273829fa76e20 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Mon, 3 Feb 2025 09:33:04 -0800 Subject: [PATCH 11/11] audio(utils): some variable renames --- src/pipecat/audio/utils.py | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) diff --git a/src/pipecat/audio/utils.py b/src/pipecat/audio/utils.py index 0cb5d4f06..7c4c25fcc 100644 --- a/src/pipecat/audio/utils.py +++ b/src/pipecat/audio/utils.py @@ -108,9 +108,9 @@ async def pcm_to_ulaw(pcm_bytes: bytes, in_rate: int, out_rate: int, resampler: 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 async def alaw_to_pcm( @@ -130,6 +130,6 @@ async def pcm_to_alaw(pcm_bytes: bytes, in_rate: int, out_rate: int, resampler: 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