Merge pull request #3675 from pipecat-ai/mb/elevenlabs-realtime-send-silence

Add silence-based keepalive to WebsocketSTTService

changelog/3675.fixed.md

@@ -0,0 +1 @@
+- Fixed WebSocket STT services (ElevenLabs, Cartesia, Gladia, Soniox) disconnecting due to idle timeout when no audio is being sent (e.g. when inactive behind a `ServiceSwitcher`). `WebsocketSTTService` now provides opt-in silence-based keepalive via `keepalive_timeout` and `keepalive_interval` parameters.

src/pipecat/services/cartesia/stt.py

@@ -129,6 +129,11 @@ class CartesiaSTTService(WebsocketSTTService):
     Provides real-time speech transcription through WebSocket connection
     to Cartesia's Live transcription service. Supports both interim and
     final transcriptions with configurable models and languages.
+
+    Cartesia disconnects WebSocket connections after 3 minutes of inactivity.
+    The timeout resets with each message (audio data or text command) sent to
+    the server. Silence-based keepalive is enabled by default to prevent this.
+    See: https://docs.cartesia.ai/api-reference/stt/stt
     """
 
     def __init__(
@@ -153,7 +158,13 @@ class CartesiaSTTService(WebsocketSTTService):
             **kwargs: Additional arguments passed to parent STTService.
         """
         sample_rate = sample_rate or (live_options.sample_rate if live_options else None)
-        super().__init__(sample_rate=sample_rate, ttfs_p99_latency=ttfs_p99_latency, **kwargs)
+        super().__init__(
+            sample_rate=sample_rate,
+            ttfs_p99_latency=ttfs_p99_latency,
+            keepalive_timeout=120,
+            keepalive_interval=30,
+            **kwargs,
+        )
 
         default_options = CartesiaLiveOptions(
             model="ink-whisper",
@@ -248,10 +259,10 @@ class CartesiaSTTService(WebsocketSTTService):
         yield None
 
     async def _connect(self):
-        await super()._connect()
-
         await self._connect_websocket()
 
+        await super()._connect()
+
         if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
@@ -295,7 +306,7 @@ class CartesiaSTTService(WebsocketSTTService):
             return self._websocket
         raise Exception("Websocket not connected")
 
-    async def _process_messages(self):
+    async def _receive_messages(self):
         """Process incoming WebSocket messages."""
         async for message in self._get_websocket():
             try:
@@ -306,14 +317,6 @@ class CartesiaSTTService(WebsocketSTTService):
             except Exception as e:
                 logger.error(f"Error processing message: {e}")
 
-    async def _receive_messages(self):
-        while True:
-            await self._process_messages()
-            # Cartesia times out after 5 minutes of innactivity (no keepalive
-            # mechanism is available). So, we try to reconnect.
-            logger.debug(f"{self} Cartesia connection was disconnected (timeout?), reconnecting")
-            await self._connect_websocket()
-
     async def _process_response(self, data):
         if "type" in data:
             if data["type"] == "transcript":

src/pipecat/services/elevenlabs/stt.py

@@ -459,6 +459,8 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
         super().__init__(
             sample_rate=sample_rate,
             ttfs_p99_latency=ttfs_p99_latency,
+            keepalive_timeout=10,
+            keepalive_interval=5,
             **kwargs,
         )
 
@@ -611,10 +613,10 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
 
     async def _connect(self):
         """Establish WebSocket connection to ElevenLabs Realtime STT."""
-        await super()._connect()
-
         await self._connect_websocket()
 
+        await super()._connect()
+
         if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
@@ -628,6 +630,21 @@ class ElevenLabsRealtimeSTTService(WebsocketSTTService):
 
         await self._disconnect_websocket()
 
+    async def _send_keepalive(self, silence: bytes):
+        """Send silent audio wrapped in ElevenLabs' JSON protocol.
+
+        Args:
+            silence: Silent 16-bit mono PCM audio bytes.
+        """
+        audio_base64 = base64.b64encode(silence).decode("utf-8")
+        message = {
+            "message_type": "input_audio_chunk",
+            "audio_base_64": audio_base64,
+            "commit": False,
+            "sample_rate": self.sample_rate,
+        }
+        await self._websocket.send(json.dumps(message))
+
     async def _connect_websocket(self):
         """Connect to ElevenLabs Realtime STT WebSocket endpoint."""
         try:

src/pipecat/services/gladia/stt.py

@@ -231,7 +231,13 @@ class GladiaSTTService(WebsocketSTTService):
                 Override for your deployment. See https://github.com/pipecat-ai/stt-benchmark
             **kwargs: Additional arguments passed to the STTService parent class.
         """
-        super().__init__(sample_rate=sample_rate, ttfs_p99_latency=ttfs_p99_latency, **kwargs)
+        super().__init__(
+            sample_rate=sample_rate,
+            ttfs_p99_latency=ttfs_p99_latency,
+            keepalive_timeout=20,
+            keepalive_interval=5,
+            **kwargs,
+        )
 
         params = params or GladiaInputParams()
 
@@ -261,7 +267,6 @@ class GladiaSTTService(WebsocketSTTService):
         self.set_model_name(model)
         self._params = params
         self._receive_task = None
-        self._keepalive_task = None
         self._settings = {}
 
         # Session management
@@ -416,8 +421,6 @@ class GladiaSTTService(WebsocketSTTService):
 
         Initializes the session if needed and establishes websocket connection.
         """
-        await super()._connect()
-
         # Initialize session if needed
         if not self._session_url:
             settings = self._prepare_settings()
@@ -428,12 +431,11 @@ class GladiaSTTService(WebsocketSTTService):
 
         await self._connect_websocket()
 
+        await super()._connect()
+
         if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
-        if self._websocket and not self._keepalive_task:
-            self._keepalive_task = self.create_task(self._keepalive_task_handler())
-
     async def _disconnect(self):
         """Disconnect from the Gladia service.
 
@@ -443,10 +445,6 @@ class GladiaSTTService(WebsocketSTTService):
 
         self._connection_active = False
 
-        if self._keepalive_task:
-            await self.cancel_task(self._keepalive_task)
-            self._keepalive_task = None
-
         if self._receive_task:
             await self.cancel_task(self._receive_task)
             self._receive_task = None
@@ -644,21 +642,10 @@ class GladiaSTTService(WebsocketSTTService):
             except json.JSONDecodeError:
                 logger.warning(f"{self} Received non-JSON message: {message}")
 
-    async def _keepalive_task_handler(self):
+    async def _send_keepalive(self, silence: bytes):
-        """Send periodic empty audio chunks to keep the connection alive."""
+        """Send an empty audio chunk to keep the Gladia connection alive.
-        try:
+
-            KEEPALIVE_SLEEP = 20
+        Args:
-            while self._connection_active:
+            silence: Silent PCM audio bytes (unused, Gladia accepts empty chunks).
-                # Send keepalive (Gladia times out after 30 seconds)
+        """
-                await asyncio.sleep(KEEPALIVE_SLEEP)
+        await self._send_audio(b"")
-                if self._websocket and self._websocket.state is State.OPEN:
-                    # Send an empty audio chunk as keepalive
-                    empty_audio = b""
-                    await self._send_audio(empty_audio)
-                else:
-                    logger.debug(f"{self} Websocket closed, stopping keepalive")
-                    break
-        except websockets.exceptions.ConnectionClosed:
-            logger.debug(f"{self} Connection closed during keepalive")
-        except Exception as e:
-            await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e)

src/pipecat/services/soniox/stt.py

@@ -6,7 +6,6 @@
 
 """Soniox speech-to-text service implementation."""
 
-import asyncio
 import json
 import time
 from typing import AsyncGenerator, List, Optional
@@ -170,7 +169,13 @@ class SonioxSTTService(WebsocketSTTService):
                 Override for your deployment. See https://github.com/pipecat-ai/stt-benchmark
             **kwargs: Additional arguments passed to the STTService.
         """
-        super().__init__(sample_rate=sample_rate, ttfs_p99_latency=ttfs_p99_latency, **kwargs)
+        super().__init__(
+            sample_rate=sample_rate,
+            ttfs_p99_latency=ttfs_p99_latency,
+            keepalive_timeout=1,
+            keepalive_interval=5,
+            **kwargs,
+        )
         params = params or SonioxInputParams()
 
         self._api_key = api_key
@@ -183,7 +188,6 @@ class SonioxSTTService(WebsocketSTTService):
         self._last_tokens_received: Optional[float] = None
 
         self._receive_task = None
-        self._keepalive_task = None
 
     async def start(self, frame: StartFrame):
         """Start the Soniox STT websocket connection.
@@ -269,16 +273,13 @@ class SonioxSTTService(WebsocketSTTService):
 
         Establishes websocket connection and starts receive and keepalive tasks.
         """
-        await super()._connect()
-
         await self._connect_websocket()
 
+        await super()._connect()
+
         if self._websocket and not self._receive_task:
             self._receive_task = self.create_task(self._receive_task_handler(self._report_error))
 
-        if self._websocket and not self._keepalive_task:
-            self._keepalive_task = self.create_task(self._keepalive_task_handler())
-
     async def _disconnect(self):
         """Disconnect from the Soniox service.
 
@@ -286,10 +287,6 @@ class SonioxSTTService(WebsocketSTTService):
         """
         await super()._disconnect()
 
-        if self._keepalive_task:
-            await self.cancel_task(self._keepalive_task)
-            self._keepalive_task = None
-
         if self._receive_task:
             await self.cancel_task(self._receive_task)
             self._receive_task = None
@@ -462,17 +459,10 @@ class SonioxSTTService(WebsocketSTTService):
             except Exception as e:
                 logger.warning(f"Error processing message: {e}")
 
-    async def _keepalive_task_handler(self):
+    async def _send_keepalive(self, silence: bytes):
-        """Connection has to be open all the time."""
+        """Send a Soniox protocol-level keepalive message.
-        try:
-            while True:
-                logger.trace("Sending keepalive message")
-                if self._websocket and self._websocket.state is State.OPEN:
-                    await self._websocket.send(KEEPALIVE_MESSAGE)
-                else:
-                    logger.debug("WebSocket connection closed.")
-                    break
-                await asyncio.sleep(5)
 
-        except Exception as e:
+        Args:
-            logger.debug(f"Keepalive task stopped: {e}")
+            silence: Silent PCM audio bytes (unused, Soniox uses a protocol message).
+        """
+        await self._websocket.send(KEEPALIVE_MESSAGE)

src/pipecat/services/stt_service.py

@@ -14,6 +14,7 @@ from abc import abstractmethod
 from typing import Any, AsyncGenerator, Dict, Mapping, Optional
 
 from loguru import logger
+from websockets.protocol import State
 
 from pipecat.frames.frames import (
     AudioRawFrame,
@@ -37,6 +38,9 @@ from pipecat.services.stt_latency import DEFAULT_TTFS_P99
 from pipecat.services.websocket_service import WebsocketService
 from pipecat.transcriptions.language import Language
 
+# Duration in seconds of silent audio sent for WebSocket keepalive (100ms).
+_KEEPALIVE_SILENCE_DURATION = 0.1
+
 
 class STTService(AIService):
     """Base class for speech-to-text services.
@@ -543,18 +547,120 @@ class WebsocketSTTService(STTService, WebsocketService):
     """Base class for websocket-based STT services.
 
     Combines STT functionality with websocket connectivity, providing automatic
-    error handling and reconnection capabilities.
+    error handling, reconnection capabilities, and optional silence-based keepalive.
+
+    The keepalive feature sends silent audio when no real audio has been sent for
+    a configurable timeout, preventing servers from closing idle connections (e.g.
+    when behind a ServiceSwitcher). Subclasses can override ``_send_keepalive()``
+    to wrap the silence in a service-specific protocol.
     """
 
-    def __init__(self, *, reconnect_on_error: bool = True, **kwargs):
+    def __init__(
+        self,
+        *,
+        reconnect_on_error: bool = True,
+        keepalive_timeout: Optional[float] = None,
+        keepalive_interval: float = 5.0,
+        **kwargs,
+    ):
         """Initialize the Websocket STT service.
 
         Args:
             reconnect_on_error: Whether to automatically reconnect on websocket errors.
+            keepalive_timeout: Seconds of no audio before sending silence to keep the
+                connection alive. None disables keepalive. Useful for services that
+                close idle connections (e.g. behind a ServiceSwitcher).
+            keepalive_interval: Seconds between idle checks when keepalive is enabled.
             **kwargs: Additional arguments passed to parent classes.
         """
         STTService.__init__(self, **kwargs)
         WebsocketService.__init__(self, reconnect_on_error=reconnect_on_error, **kwargs)
+        self._keepalive_timeout = keepalive_timeout
+        self._keepalive_interval = keepalive_interval
+        self._keepalive_task: Optional[asyncio.Task] = None
+        self._last_audio_time: float = 0
+
+    async def process_audio_frame(self, frame: AudioRawFrame, direction: FrameDirection):
+        """Process an audio frame, tracking the last audio time for keepalive.
+
+        Args:
+            frame: The audio frame to process.
+            direction: The direction of frame processing.
+        """
+        self._last_audio_time = time.monotonic()
+        await super().process_audio_frame(frame, direction)
+
+    async def _connect(self):
+        """Connect and start keepalive task if enabled."""
+        await super()._connect()
+        self._create_keepalive_task()
+
+    async def _disconnect(self):
+        """Disconnect and cancel keepalive task."""
+        await super()._disconnect()
+        await self._cancel_keepalive_task()
+
+    async def _reconnect_websocket(self, attempt_number: int) -> bool:
+        """Reconnect and restart keepalive task.
+
+        The keepalive task breaks out of its loop on send errors, so it may
+        be dead after the websocket failure that triggered this reconnect.
+        """
+        result = await super()._reconnect_websocket(attempt_number)
+        if result:
+            await self._cancel_keepalive_task()
+            self._create_keepalive_task()
+        return result
+
+    def _create_keepalive_task(self):
+        """Start the keepalive task if keepalive is enabled."""
+        if self._keepalive_timeout is not None:
+            self._last_audio_time = time.monotonic()
+            self._keepalive_task = self.create_task(
+                self._keepalive_task_handler(), name="keepalive"
+            )
+
+    async def _cancel_keepalive_task(self):
+        """Stop the keepalive task if running."""
+        if self._keepalive_task:
+            await self.cancel_task(self._keepalive_task)
+            self._keepalive_task = None
+
+    async def _keepalive_task_handler(self):
+        """Send periodic silent audio to prevent the server from closing the connection.
+
+        When keepalive is enabled, this task checks periodically if the connection
+        has been idle (no audio sent) for longer than keepalive_timeout seconds.
+        If so, it generates silent 16-bit mono PCM audio and passes it to
+        _send_keepalive() for service-specific formatting and sending.
+        """
+        while True:
+            await asyncio.sleep(self._keepalive_interval)
+            try:
+                if not self._websocket or self._websocket.state is not State.OPEN:
+                    continue
+                elapsed = time.monotonic() - self._last_audio_time
+                if elapsed < self._keepalive_timeout:
+                    continue
+                num_samples = int(self.sample_rate * _KEEPALIVE_SILENCE_DURATION)
+                silence = b"\x00" * (num_samples * 2)
+                await self._send_keepalive(silence)
+                self._last_audio_time = time.monotonic()
+                logger.trace(f"{self} sent keepalive silence")
+            except Exception as e:
+                logger.warning(f"{self} keepalive error: {e}")
+                break
+
+    async def _send_keepalive(self, silence: bytes):
+        """Send silent audio over the websocket to keep the connection alive.
+
+        The default implementation sends raw PCM bytes directly. Subclasses
+        can override this to wrap the silence in a service-specific protocol.
+
+        Args:
+            silence: Silent 16-bit mono PCM audio bytes.
+        """
+        await self._websocket.send(silence)
 
     async def _report_error(self, error: ErrorFrame):
         await self._call_event_handler("on_connection_error", error.error)