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

markbackman · web-flow · commit 17ab9c425f85 · 2026-02-10T18:03:38.000-05:00
Add silence-based keepalive to WebsocketSTTService
diff --git a/changelog/3675.fixed.md b/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.
diff --git a/src/pipecat/services/cartesia/stt.py b/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 @@ def __init__(
             **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 @@ async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
         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 @@ def _get_websocket(self):
             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 @@ async def _process_messages(self):
             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":
diff --git a/src/pipecat/services/elevenlabs/stt.py b/src/pipecat/services/elevenlabs/stt.py
@@ -459,6 +459,8 @@ def __init__(
         super().__init__(
             sample_rate=sample_rate,
             ttfs_p99_latency=ttfs_p99_latency,
+            keepalive_timeout=10,
+            keepalive_interval=5,
             **kwargs,
         )
 
@@ -611,10 +613,10 @@ async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
 
     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 @@ async def _disconnect(self):
 
         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:
diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py
@@ -231,7 +231,13 @@ def __init__(
                 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 @@ def __init__(
         self.set_model_name(model)
         self._params = params
         self._receive_task = None
-        self._keepalive_task = None
         self._settings = {}
 
         # Session management
@@ -416,8 +421,6 @@ async def _connect(self):
 
         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 @@ async def _connect(self):
 
         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 @@ async def _disconnect(self):
 
         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 @@ async def _receive_messages(self):
             except json.JSONDecodeError:
                 logger.warning(f"{self} Received non-JSON message: {message}")
 
-    async def _keepalive_task_handler(self):
-        """Send periodic empty audio chunks to keep the connection alive."""
-        try:
-            KEEPALIVE_SLEEP = 20
-            while self._connection_active:
-                # Send keepalive (Gladia times out after 30 seconds)
-                await asyncio.sleep(KEEPALIVE_SLEEP)
-                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)
+    async def _send_keepalive(self, silence: bytes):
+        """Send an empty audio chunk to keep the Gladia connection alive.
+
+        Args:
+            silence: Silent PCM audio bytes (unused, Gladia accepts empty chunks).
+        """
+        await self._send_audio(b"")
diff --git a/src/pipecat/services/soniox/stt.py b/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 @@ def __init__(
                 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 @@ def __init__(
         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,27 +273,20 @@ async def _connect(self):
 
         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.
 
         Cleans up tasks and closes websocket connection.
         """
         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 @@ async def send_endpoint_transcript():
             except Exception as e:
                 logger.warning(f"Error processing message: {e}")
 
-    async def _keepalive_task_handler(self):
-        """Connection has to be open all the time."""
-        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)
+    async def _send_keepalive(self, silence: bytes):
+        """Send a Soniox protocol-level keepalive message.
 
-        except Exception as e:
-            logger.debug(f"Keepalive task stopped: {e}")
+        Args:
+            silence: Silent PCM audio bytes (unused, Soniox uses a protocol message).
+        """
+        await self._websocket.send(KEEPALIVE_MESSAGE)
diff --git a/src/pipecat/services/stt_service.py b/src/pipecat/services/stt_service.py

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1 @@`
	`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.