Merge pull request #1272 from pipecat-ai/aleix/tts-websocket-interruptions

services: fix some TTS websocket service interruption handling

CHANGELOG.md

@@ -42,6 +42,10 @@ stt = DeepgramSTTService(..., live_options=LiveOptions(model="nova-2-general"))
 
 ### Fixed
 
+- Fixed a `ElevenLabsTTSService`, `FishAudioTTSService`, `LMNTTTSService` and
+  `PlayHTTTSService` issue that was resulting in audio requested before an
+  interruption being played after an interruption.
+
 - Fixed `match_endofsentence` support for ellipses.
 
 - Fixed an issue that would cause undesired interruptions via

src/pipecat/services/ai_services.py

@@ -15,6 +15,7 @@ from loguru import logger
 from pipecat.audio.utils import calculate_audio_volume, exp_smoothing
 from pipecat.frames.frames import (
     AudioRawFrame,
+    BotStartedSpeakingFrame,
     BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
@@ -40,6 +41,7 @@ from pipecat.frames.frames import (
 from pipecat.metrics.metrics import MetricsData
 from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext
 from pipecat.processors.frame_processor import FrameDirection, FrameProcessor
+from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
 from pipecat.utils.string import match_endofsentence
 from pipecat.utils.text.base_text_filter import BaseTextFilter
@@ -434,6 +436,12 @@ class TTSService(AIService):
 
 
 class WordTTSService(TTSService):
+    """This is a base class for TTS services that support word timestamps. Word
+    timestamps are useful to synchronize audio with text of the spoken
+    words. This way only the spoken words are added to the conversation context.
+
+    """
+
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self._initial_word_timestamp = -1
@@ -503,11 +511,93 @@ class WordTTSService(TTSService):
             self._words_queue.task_done()
 
 
-class AudioContextWordTTSService(WordTTSService):
+class WebsocketTTSService(TTSService, WebsocketService):
-    """This services allow us to send multiple TTS request to the services. Each
+    """This is a base class for websocket-based TTS services."""
-    request could be multiple sentences long which are grouped by context. For
+
-    this to work, the TTS service needs to support handling multiple requests at
+    def __init__(self, **kwargs):
-    once (i.e. multiple simultaneous contexts).
+        TTSService.__init__(self, **kwargs)
+        WebsocketService.__init__(self)
+
+
+class InterruptibleTTSService(WebsocketTTSService):
+    """This is a base class for websocket-based TTS services that don't support
+    word timestamps and that don't offer a way to correlate the generated audio
+    to the requested text.
+
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+        # Indicates if the bot is speaking. If the bot is not speaking we don't
+        # need to reconnect when the user speaks. If the bot is speaking and the
+        # user interrupts we need to reconnect.
+        self._bot_speaking = False
+
+    async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
+        await super()._handle_interruption(frame, direction)
+        if self._bot_speaking:
+            await self._disconnect()
+            await self._connect()
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, BotStartedSpeakingFrame):
+            self._bot_speaking = True
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            self._bot_speaking = False
+
+
+class WebsocketWordTTSService(WordTTSService, WebsocketService):
+    """This is a base class for websocket-based TTS services that support word
+    timestamps.
+
+    """
+
+    def __init__(self, **kwargs):
+        WordTTSService.__init__(self, **kwargs)
+        WebsocketService.__init__(self)
+
+
+class InterruptibleWordTTSService(WebsocketWordTTSService):
+    """This is a base class for websocket-based TTS services that support word
+    timestamps but don't offer a way to correlate the generated audio to the
+    requested text.
+
+    """
+
+    def __init__(self, **kwargs):
+        super().__init__(**kwargs)
+
+        # Indicates if the bot is speaking. If the bot is not speaking we don't
+        # need to reconnect when the user speaks. If the bot is speaking and the
+        # user interrupts we need to reconnect.
+        self._bot_speaking = False
+
+    async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
+        await super()._handle_interruption(frame, direction)
+        if self._bot_speaking:
+            await self._disconnect()
+            await self._connect()
+
+    async def process_frame(self, frame: Frame, direction: FrameDirection):
+        await super().process_frame(frame, direction)
+
+        if isinstance(frame, BotStartedSpeakingFrame):
+            self._bot_speaking = True
+        elif isinstance(frame, BotStoppedSpeakingFrame):
+            self._bot_speaking = False
+
+
+class AudioContextWordTTSService(WebsocketWordTTSService):
+    """This is a base class for websocket-based TTS services that support word
+    timestamps and also allow correlating the generated audio with the requested
+    text.
+
+    Each request could be multiple sentences long which are grouped by
+    context. For this to work, the TTS service needs to support handling
+    multiple requests at once (i.e. multiple simultaneous contexts).
 
     The audio received from the TTS will be played in context order. That is, if
     we requested audio for a context "A" and then audio for context "B", the

src/pipecat/services/cartesia.py

@@ -13,22 +13,18 @@ from loguru import logger
 from pydantic import BaseModel
 
 from pipecat.frames.frames import (
-    BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
     ErrorFrame,
     Frame,
-    LLMFullResponseEndFrame,
     StartFrame,
     StartInterruptionFrame,
     TTSAudioRawFrame,
-    TTSSpeakFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import AudioContextWordTTSService, TTSService
-from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
 
 # See .env.example for Cartesia configuration needed
@@ -75,7 +71,7 @@ def language_to_cartesia_language(language: Language) -> Optional[str]:
     return result
 
 
-class CartesiaTTSService(AudioContextWordTTSService, WebsocketService):
+class CartesiaTTSService(AudioContextWordTTSService):
     class InputParams(BaseModel):
         language: Optional[Language] = Language.EN
         speed: Optional[Union[str, float]] = ""
@@ -105,15 +101,13 @@ class CartesiaTTSService(AudioContextWordTTSService, WebsocketService):
         # if we're interrupted. Cartesia gives us word-by-word timestamps. We
         # can use those to generate text frames ourselves aligned with the
         # playout timing of the audio!
-        AudioContextWordTTSService.__init__(
+        super().__init__(
-            self,
             aggregate_sentences=True,
             push_text_frames=False,
             pause_frame_processing=True,
             sample_rate=sample_rate,
             **kwargs,
         )
-        WebsocketService.__init__(self)
 
         self._api_key = api_key
         self._cartesia_version = cartesia_version

src/pipecat/services/elevenlabs.py

@@ -14,22 +14,18 @@ from loguru import logger
 from pydantic import BaseModel, model_validator
 
 from pipecat.frames.frames import (
-    BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
     ErrorFrame,
     Frame,
-    LLMFullResponseEndFrame,
     StartFrame,
     StartInterruptionFrame,
     TTSAudioRawFrame,
-    TTSSpeakFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import TTSService, WordTTSService
+from pipecat.services.ai_services import InterruptibleWordTTSService, TTSService
-from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
 
 # See .env.example for ElevenLabs configuration needed
@@ -141,7 +137,7 @@ def calculate_word_times(
     return word_times
 
 
-class ElevenLabsTTSService(WordTTSService, WebsocketService):
+class ElevenLabsTTSService(InterruptibleWordTTSService):
     class InputParams(BaseModel):
         language: Optional[Language] = None
         optimize_streaming_latency: Optional[str] = None
@@ -186,8 +182,7 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
         # Finally, ElevenLabs doesn't provide information on when the bot stops
         # speaking for a while, so we want the parent class to send TTSStopFrame
         # after a short period not receiving any audio.
-        WordTTSService.__init__(
+        super().__init__(
-            self,
             aggregate_sentences=True,
             push_text_frames=False,
             push_stop_frames=True,
@@ -195,7 +190,6 @@ class ElevenLabsTTSService(WordTTSService, WebsocketService):
             sample_rate=sample_rate,
             **kwargs,
         )
-        WebsocketService.__init__(self)
 
         self._api_key = api_key
         self._url = url

src/pipecat/services/fish.py

@@ -22,8 +22,7 @@ from pipecat.frames.frames import (
     TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import TTSService
+from pipecat.services.ai_services import InterruptibleTTSService
-from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
 
 try:
@@ -40,7 +39,7 @@ except ModuleNotFoundError as e:
 FishAudioOutputFormat = Literal["opus", "mp3", "pcm", "wav"]
 
 
-class FishAudioTTSService(TTSService, WebsocketService):
+class FishAudioTTSService(InterruptibleTTSService):
     class InputParams(BaseModel):
         language: Optional[Language] = Language.EN
         latency: Optional[str] = "normal"  # "normal" or "balanced"
@@ -110,11 +109,12 @@ class FishAudioTTSService(TTSService, WebsocketService):
         self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
 
     async def _disconnect(self):
-        await self._disconnect_websocket()
         if self._receive_task:
             await self.cancel_task(self._receive_task)
             self._receive_task = None
 
+        await self._disconnect_websocket()
+
     async def _connect_websocket(self):
         try:
             logger.debug("Connecting to Fish Audio")
@@ -149,6 +149,11 @@ class FishAudioTTSService(TTSService, WebsocketService):
             return self._websocket
         raise Exception("Websocket not connected")
 
+    async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
+        await super()._handle_interruption(frame, direction)
+        await self.stop_all_metrics()
+        self._request_id = None
+
     async def _receive_messages(self):
         async for message in self._get_websocket():
             try:
@@ -168,11 +173,6 @@ class FishAudioTTSService(TTSService, WebsocketService):
             except Exception as e:
                 logger.error(f"Error processing message: {e}")
 
-    async def _handle_interruption(self, frame: StartInterruptionFrame, direction: FrameDirection):
-        await super()._handle_interruption(frame, direction)
-        await self.stop_all_metrics()
-        self._request_id = None
-
     async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]:
         logger.debug(f"Generating Fish TTS: [{text}]")
         try:

src/pipecat/services/lmnt.py

@@ -21,8 +21,7 @@ from pipecat.frames.frames import (
     TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import TTSService
+from pipecat.services.ai_services import InterruptibleTTSService
-from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
 
 # See .env.example for LMNT configuration needed
@@ -60,7 +59,7 @@ def language_to_lmnt_language(language: Language) -> Optional[str]:
     return result
 
 
-class LmntTTSService(TTSService, WebsocketService):
+class LmntTTSService(InterruptibleTTSService):
     def __init__(
         self,
         *,
@@ -70,14 +69,12 @@ class LmntTTSService(TTSService, WebsocketService):
         language: Language = Language.EN,
         **kwargs,
     ):
-        TTSService.__init__(
+        super().__init__(
-            self,
             push_stop_frames=True,
             pause_frame_processing=True,
             sample_rate=sample_rate,
             **kwargs,
         )
-        WebsocketService.__init__(self)
 
         self._api_key = api_key
         self._voice_id = voice_id
@@ -116,12 +113,12 @@ class LmntTTSService(TTSService, WebsocketService):
         self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
 
     async def _disconnect(self):
-        await self._disconnect_websocket()
-
         if self._receive_task:
             await self.cancel_task(self._receive_task)
             self._receive_task = None
 
+        await self._disconnect_websocket()
+
     async def _connect_websocket(self):
         """Connect to LMNT websocket."""
         try:
@@ -153,8 +150,9 @@ class LmntTTSService(TTSService, WebsocketService):
 
             if self._websocket:
                 logger.debug("Disconnecting from LMNT")
-                # Send EOF message before closing
+                # NOTE(aleix): sending EOF message before closing is causing
-                await self._websocket.send(json.dumps({"eof": True}))
+                # errors on the websocket, so we just skip it for now.
+                # await self._websocket.send(json.dumps({"eof": True}))
                 await self._websocket.close()
                 self._websocket = None
 

src/pipecat/services/playht.py

@@ -16,22 +16,18 @@ from loguru import logger
 from pydantic import BaseModel
 
 from pipecat.frames.frames import (
-    BotStoppedSpeakingFrame,
     CancelFrame,
     EndFrame,
     ErrorFrame,
     Frame,
-    LLMFullResponseEndFrame,
     StartFrame,
     StartInterruptionFrame,
     TTSAudioRawFrame,
-    TTSSpeakFrame,
     TTSStartedFrame,
     TTSStoppedFrame,
 )
 from pipecat.processors.frame_processor import FrameDirection
-from pipecat.services.ai_services import TTSService
+from pipecat.services.ai_services import InterruptibleTTSService, TTSService
-from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
 
 try:
@@ -100,7 +96,7 @@ def language_to_playht_language(language: Language) -> Optional[str]:
     return result
 
 
-class PlayHTTTSService(TTSService, WebsocketService):
+class PlayHTTTSService(InterruptibleTTSService):
     class InputParams(BaseModel):
         language: Optional[Language] = Language.EN
         speed: Optional[float] = 1.0
@@ -118,13 +114,11 @@ class PlayHTTTSService(TTSService, WebsocketService):
         params: InputParams = InputParams(),
         **kwargs,
     ):
-        TTSService.__init__(
+        super().__init__(
-            self,
             pause_frame_processing=True,
             sample_rate=sample_rate,
             **kwargs,
         )
-        WebsocketService.__init__(self)
 
         self._api_key = api_key
         self._user_id = user_id
@@ -168,12 +162,12 @@ class PlayHTTTSService(TTSService, WebsocketService):
         self._receive_task = self.create_task(self._receive_task_handler(self.push_error))
 
     async def _disconnect(self):
-        await self._disconnect_websocket()
-
         if self._receive_task:
             await self.cancel_task(self._receive_task)
             self._receive_task = None
 
+        await self._disconnect_websocket()
+
     async def _connect_websocket(self):
         try:
             logger.debug("Connecting to PlayHT")

src/pipecat/services/rime.py

@@ -26,7 +26,6 @@ from pipecat.frames.frames import (
 )
 from pipecat.processors.frame_processor import FrameDirection
 from pipecat.services.ai_services import AudioContextWordTTSService, TTSService
-from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
 
 try:
@@ -55,7 +54,7 @@ def language_to_rime_language(language: Language) -> str:
     return LANGUAGE_MAP.get(language, "eng")
 
 
-class RimeTTSService(AudioContextWordTTSService, WebsocketService):
+class RimeTTSService(AudioContextWordTTSService):
     """Text-to-Speech service using Rime's websocket API.
 
     Uses Rime's websocket JSON API to convert text to speech with word-level timing
@@ -92,8 +91,7 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService):
             params: Additional configuration parameters.
         """
         # Initialize with parent class settings for proper frame handling
-        AudioContextWordTTSService.__init__(
+        super().__init__(
-            self,
             aggregate_sentences=True,
             push_text_frames=False,
             push_stop_frames=True,
@@ -101,7 +99,6 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService):
             sample_rate=sample_rate,
             **kwargs,
         )
-        WebsocketService.__init__(self)
 
         # Store service configuration
         self._api_key = api_key
@@ -172,11 +169,12 @@ class RimeTTSService(AudioContextWordTTSService, WebsocketService):
 
     async def _disconnect(self):
         """Close websocket connection and clean up tasks."""
-        await self._disconnect_websocket()
         if self._receive_task:
             await self.cancel_task(self._receive_task)
             self._receive_task = None
 
+        await self._disconnect_websocket()
+
     async def _connect_websocket(self):
         """Connect to Rime websocket API with configured settings."""
         try:

src/pipecat/services/websocket_service.py

@@ -90,14 +90,32 @@ class WebsocketService(ABC):
                     logger.error(f"{self} reconnection failed: {reconnect_error}")
                     continue
 
+    @abstractmethod
+    async def _connect(self):
+        """Implement service-specific connection logic. This function will
+        connect to the websocket via _connect_websocket() among other connection
+        logic."""
+        pass
+
+    @abstractmethod
+    async def _disconnect(self):
+        """Implement service-specific disconnection logic. This function will
+        disconnect to the websocket via _connect_websocket() among other
+        connection logic.
+
+        """
+        pass
+
     @abstractmethod
     async def _connect_websocket(self):
-        """Implement service-specific websocket connection logic."""
+        """Implement service-specific websocket connection logic. This function
+        should only connect to the websocket."""
         pass
 
     @abstractmethod
     async def _disconnect_websocket(self):
-        """Implement service-specific websocket disconnection logic."""
+        """Implement service-specific websocket disconnection logic. This
+        function should only disconnect from the websocket."""
         pass
 
     @abstractmethod