diff --git a/changelog/3675.fixed.md b/changelog/3675.fixed.md new file mode 100644 index 000000000..12e19efac --- /dev/null +++ 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 index 2c76412f2..c4429226f 100644 --- 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 @@ 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": diff --git a/src/pipecat/services/elevenlabs/stt.py b/src/pipecat/services/elevenlabs/stt.py index bf853ac98..388f7146b 100644 --- a/src/pipecat/services/elevenlabs/stt.py +++ b/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: diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py index 243152d68..475a7213e 100644 --- a/src/pipecat/services/gladia/stt.py +++ b/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): - """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 index fb5b78e4a..c9184ba4c 100644 --- 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 @@ 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): - """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 index 7e9ed4c9f..b556bb23a 100644 --- a/src/pipecat/services/stt_service.py +++ b/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)