From a592b7fdf0941bb06df863e02309ef44b69d7928 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Thu, 29 Jan 2026 00:06:09 -0500 Subject: [PATCH] Update per PR 1789, align with ErrorFrame norms --- changelog/3134.added.md | 1 + src/pipecat/services/resembleai/tts.py | 105 +++++++++++++++++++++---- 2 files changed, 92 insertions(+), 14 deletions(-) create mode 100644 changelog/3134.added.md diff --git a/changelog/3134.added.md b/changelog/3134.added.md new file mode 100644 index 000000000..adddde398 --- /dev/null +++ b/changelog/3134.added.md @@ -0,0 +1 @@ +- Added `ResembleAITTSService` for text-to-speech using Resemble AI's streaming WebSocket API with word-level timestamps and jitter buffering for smooth audio playback. diff --git a/src/pipecat/services/resembleai/tts.py b/src/pipecat/services/resembleai/tts.py index ae97982b7..21a8c7a78 100644 --- a/src/pipecat/services/resembleai/tts.py +++ b/src/pipecat/services/resembleai/tts.py @@ -87,6 +87,18 @@ class ResembleAITTSService(AudioContextWordTTSService): self._current_request_id = None self._receive_task = None + # Per-request audio buffers to handle concurrent TTS requests + # ResembleAI may send odd-length data even for PCM_16, so buffering helps us + # create properly aligned frames while maintaining smooth audio output + self._audio_buffers: dict[str, bytearray] = {} + self._buffer_threshold_bytes = 2208 + + # Jitter buffer: accumulate audio before starting playback to absorb network latency + # ResembleAI sends audio in bursts with 300-450ms gaps between them + # We need to buffer enough to cover these gaps before starting playback + self._jitter_buffer_bytes = 44100 # ~1000ms at 22050Hz to handle 400ms+ network gaps + self._playback_started: dict[str, bool] = {} # Track if we've started playback per request + self.set_voice(voice_id) def can_generate_metrics(self) -> bool: @@ -173,8 +185,7 @@ class ResembleAITTSService(AudioContextWordTTSService): self._websocket = await websocket_connect(self._url, additional_headers=headers) await self._call_event_handler("on_connected") except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) self._websocket = None await self._call_event_handler("on_connection_error", f"{e}") @@ -185,13 +196,16 @@ class ResembleAITTSService(AudioContextWordTTSService): if self._websocket: logger.debug("Disconnecting from Resemble AI") + # ResembleAI doesn't send disconnect acknowledgement, set close_timeout to 0 + self._websocket.close_timeout = 0 await self._websocket.close() except Exception as e: - logger.error(f"{self} exception: {e}") - await self.push_error(ErrorFrame(error=f"{self} error: {e}")) + await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) finally: self._current_request_id = None self._websocket = None + self._audio_buffers.clear() + self._playback_started.clear() await self._call_event_handler("on_disconnected") def _get_websocket(self): @@ -235,7 +249,7 @@ class ResembleAITTSService(AudioContextWordTTSService): try: msg = json.loads(message) except json.JSONDecodeError: - logger.error(f"{self} received invalid JSON: {message}") + await self.push_error(error_msg=f"Received invalid JSON: {message}") continue if not msg: @@ -257,10 +271,44 @@ class ResembleAITTSService(AudioContextWordTTSService): # Decode base64 audio content audio_content = msg.get("audio_content", "") - if audio_content: - audio_data = base64.b64decode(audio_content) + if not audio_content: + continue + + audio_bytes = base64.b64decode(audio_content) + if len(audio_bytes) == 0: + continue + + # Get or create buffer for this request + if request_id_str not in self._audio_buffers: + self._audio_buffers[request_id_str] = bytearray() + self._playback_started[request_id_str] = False + buffer = self._audio_buffers[request_id_str] + + # Add to buffer + buffer.extend(audio_bytes) + + # Wait for jitter buffer to fill before starting playback + # This absorbs network latency gaps (ResembleAI sends in bursts) + if not self._playback_started.get(request_id_str, False): + if len(buffer) < self._jitter_buffer_bytes: + continue + self._playback_started[request_id_str] = True + + # Send complete (even-byte) chunks for PCM_16 alignment + while len(buffer) >= self._buffer_threshold_bytes: + chunk_size = self._buffer_threshold_bytes + if chunk_size % 2 != 0: + chunk_size -= 1 + + chunk_to_send = bytes(buffer[:chunk_size]) + self._audio_buffers[request_id_str] = buffer[chunk_size:] + buffer = self._audio_buffers[request_id_str] + + if len(chunk_to_send) == 0: + continue + frame = TTSAudioRawFrame( - audio=audio_data, + audio=chunk_to_send, sample_rate=self.sample_rate, num_channels=1, ) @@ -284,6 +332,28 @@ class ResembleAITTSService(AudioContextWordTTSService): elif msg_type == "audio_end": await self.stop_ttfb_metrics() + + # Flush remaining buffer, ensuring even length for PCM_16 + buffer = self._audio_buffers.get(request_id_str, bytearray()) + if buffer: + remaining = bytes(buffer) + # PCM_16 requires even number of bytes + if len(remaining) % 2 != 0: + remaining = remaining[:-1] + if remaining: + frame = TTSAudioRawFrame( + audio=remaining, + sample_rate=self.sample_rate, + num_channels=1, + ) + await self.append_to_audio_context(request_id_str, frame) + + # Clean up buffer and playback tracking for this request + if request_id_str in self._audio_buffers: + del self._audio_buffers[request_id_str] + if request_id_str in self._playback_started: + del self._playback_started[request_id_str] + await self.add_word_timestamps([("TTSStoppedFrame", 0), ("Reset", 0)]) await self.remove_audio_context(request_id_str) # Clear current request if this was it @@ -294,7 +364,16 @@ class ResembleAITTSService(AudioContextWordTTSService): error_name = msg.get("error_name", "Unknown") error_msg = msg.get("message", "Unknown error") status_code = msg.get("status_code", 0) - logger.error(f"{self} error: {error_name} (status {status_code}): {error_msg}") + await self.push_error( + error_msg=f"Error: {error_name} (status {status_code}): {error_msg}" + ) + + # Clean up buffer and playback tracking for this request + if request_id_str in self._audio_buffers: + del self._audio_buffers[request_id_str] + if request_id_str in self._playback_started: + del self._playback_started[request_id_str] + await self.push_frame(TTSStoppedFrame()) await self.stop_all_metrics() await self.push_error(ErrorFrame(error=f"{self} error: {error_name} - {error_msg}")) @@ -317,7 +396,7 @@ class ResembleAITTSService(AudioContextWordTTSService): try: await self._process_messages() except Exception as e: - logger.error(f"{self} error in receive loop: {e}") + await self.push_error(error_msg=f"Error in receive loop: {e}", exception=e) # Try to reconnect logger.debug(f"{self} Resemble AI connection lost, reconnecting") await self._connect_websocket() @@ -354,13 +433,11 @@ class ResembleAITTSService(AudioContextWordTTSService): await self._get_websocket().send(msg) await self.start_tts_usage_metrics(text) except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}") yield TTSStoppedFrame() await self._disconnect() await self._connect() return yield None except Exception as e: - logger.error(f"{self} exception: {e}") - yield ErrorFrame(error=f"{self} error: {e}") + yield ErrorFrame(error=f"Unknown error occurred: {e}")