diff --git a/CHANGELOG.md b/CHANGELOG.md index a297b8d3f..1ac1af7eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 the Telnyx call when an `EndFrame` or `CancelFrame` is received. It is enabled by default and is configurable via the `auto_hang_up` `InputParam`. +- Added a keepalive task to `GladiaSTTService` to prevent the websocket from + disconnecting after 30 seconds of no audio input. + ### Changed - In `TwilioFrameSerializer`, `call_sid` is Optional so as to avoid a breaking diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py index 55581b3e6..df8038bd3 100644 --- a/src/pipecat/services/gladia/stt.py +++ b/src/pipecat/services/gladia/stt.py @@ -4,6 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # +import asyncio import base64 import json import warnings @@ -224,6 +225,7 @@ class GladiaSTTService(STTService): self._params = params self._websocket = None self._receive_task = None + self._keepalive_task = None def language_to_service_language(self, language: Language) -> Optional[str]: """Convert pipecat Language enum to Gladia's language code.""" @@ -287,14 +289,22 @@ class GladiaSTTService(STTService): self._websocket = await websockets.connect(response["url"]) if self._websocket and not self._receive_task: self._receive_task = self.create_task(self._receive_task_handler()) + if self._websocket and not self._keepalive_task: + self._keepalive_task = self.create_task(self._keepalive_task_handler()) async def stop(self, frame: EndFrame): """Stop the Gladia STT websocket connection.""" await super().stop(frame) await self._send_stop_recording() + + if self._keepalive_task: + await self.cancel_task(self._keepalive_task) + self._keepalive_task = None + if self._websocket: await self._websocket.close() self._websocket = None + if self._receive_task: await self.wait_for_task(self._receive_task) self._receive_task = None @@ -302,7 +312,15 @@ class GladiaSTTService(STTService): async def cancel(self, frame: CancelFrame): """Cancel the Gladia STT websocket connection.""" await super().cancel(frame) - await self._websocket.close() + + if self._keepalive_task: + await self.cancel_task(self._keepalive_task) + self._keepalive_task = None + + if self._websocket: + await self._websocket.close() + self._websocket = None + if self._receive_task: await self.cancel_task(self._receive_task) self._receive_task = None @@ -341,6 +359,24 @@ class GladiaSTTService(STTService): if self._websocket and not self._websocket.closed: await self._websocket.send(json.dumps({"type": "stop_recording"})) + async def _keepalive_task_handler(self): + """Send periodic empty audio chunks to keep the connection alive.""" + try: + while True: + # Send keepalive every 20 seconds (Gladia times out after 30 seconds) + await asyncio.sleep(20) + if self._websocket and not self._websocket.closed: + # Send an empty audio chunk as keepalive + empty_audio = b"" + await self._send_audio(empty_audio) + else: + logger.debug("Websocket closed, stopping keepalive") + break + except websockets.exceptions.ConnectionClosed: + logger.debug("Connection closed during keepalive") + except Exception as e: + logger.error(f"Error in Gladia keepalive task: {e}") + async def _receive_task_handler(self): try: async for message in self._websocket: