From 4f032f5b96349439533b7f62436ccfb2e26fe663 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aleix=20Conchillo=20Flaqu=C3=A9?= Date: Wed, 25 Jun 2025 14:26:50 -0700 Subject: [PATCH] update keepalive times depending on watchdog timers --- src/pipecat/services/elevenlabs/tts.py | 3 ++- src/pipecat/services/gladia/stt.py | 14 ++++++++------ src/pipecat/services/neuphonic/tts.py | 9 +++++++-- 3 files changed, 17 insertions(+), 9 deletions(-) diff --git a/src/pipecat/services/elevenlabs/tts.py b/src/pipecat/services/elevenlabs/tts.py index fdb7bf1a8..7665632fc 100644 --- a/src/pipecat/services/elevenlabs/tts.py +++ b/src/pipecat/services/elevenlabs/tts.py @@ -428,9 +428,10 @@ class ElevenLabsTTSService(AudioContextWordTTSService): self._cumulative_time = word_times[-1][1] async def _keepalive_task_handler(self): + KEEPALIVE_SLEEP = 10 if self.watchdog_timers_enabled else 3 while True: self.reset_watchdog() - await asyncio.sleep(4) + await asyncio.sleep(KEEPALIVE_SLEEP) try: if self._websocket and self._websocket.open: if self._context_id: diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py index a21c26ad5..75e8bbee3 100644 --- a/src/pipecat/services/gladia/stt.py +++ b/src/pipecat/services/gladia/stt.py @@ -392,8 +392,8 @@ class GladiaSTTService(STTService): await self._send_buffered_audio() # Start tasks - self._receive_task = asyncio.create_task(self._receive_task_handler()) - self._keepalive_task = asyncio.create_task(self._keepalive_task_handler()) + self._receive_task = self.create_task(self._receive_task_handler()) + self._keepalive_task = self.create_task(self._keepalive_task_handler()) # Wait for tasks to complete await asyncio.gather(self._receive_task, self._keepalive_task) @@ -404,9 +404,9 @@ class GladiaSTTService(STTService): # Clean up tasks if self._receive_task: - self._receive_task.cancel() + await self.cancel_task(self._receive_task) if self._keepalive_task: - self._keepalive_task.cancel() + await self.cancel_task(self._keepalive_task) # Attempt reconnect using helper if not await self._maybe_reconnect(): @@ -485,9 +485,11 @@ class GladiaSTTService(STTService): async def _keepalive_task_handler(self): """Send periodic empty audio chunks to keep the connection alive.""" try: + KEEPALIVE_SLEEP = 20 if self.watchdog_timers_enabled else 3 while self._connection_active: - # Send keepalive every 20 seconds (Gladia times out after 30 seconds) - await asyncio.sleep(20) + self.reset_watchdog() + # Send keepalive (Gladia times out after 30 seconds) + await asyncio.sleep(KEEPALIVE_SLEEP) if self._websocket and not self._websocket.closed: # Send an empty audio chunk as keepalive empty_audio = b"" diff --git a/src/pipecat/services/neuphonic/tts.py b/src/pipecat/services/neuphonic/tts.py index bfceca50b..85bd9d0cc 100644 --- a/src/pipecat/services/neuphonic/tts.py +++ b/src/pipecat/services/neuphonic/tts.py @@ -30,6 +30,7 @@ from pipecat.processors.frame_processor import FrameDirection from pipecat.services.tts_service import InterruptibleTTSService, TTSService from pipecat.transcriptions.language import Language from pipecat.utils.tracing.service_decorators import traced_tts +from pipecat.utils.watchdog_async_iterator import WatchdogAsyncIterator try: import websockets @@ -221,7 +222,9 @@ class NeuphonicTTSService(InterruptibleTTSService): self._websocket = None async def _receive_messages(self): - async for message in self._websocket: + async for message in WatchdogAsyncIterator( + self._websocket, reseter=self, watchdog_enabled=self.watchdog_timers_enabled + ): if isinstance(message, str): msg = json.loads(message) if msg.get("data", {}).get("audio") is not None: @@ -232,8 +235,10 @@ class NeuphonicTTSService(InterruptibleTTSService): await self.push_frame(frame) async def _keepalive_task_handler(self): + KEEPALIVE_SLEEP = 10 if self.watchdog_timers_enabled else 3 while True: - await asyncio.sleep(10) + self.reset_watchdog() + await asyncio.sleep(KEEPALIVE_SLEEP) await self._send_text("") async def _send_text(self, text: str):