Merge pull request #2490 from pipecat-ai/aleix/speechmatics-exceptions

Speechmatics exception handling

CHANGELOG.md

@@ -60,6 +60,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Added `SpeechmaticsSTTService` exception handling on connection and sending.
+
 - Replaced `asyncio.wait_for()` for `wait_for2.wait_for()` for Python <
   3.12. because of issues regarding task cancellation (i.e. cancellation is
   never propagated).

src/pipecat/services/gladia/stt.py

@@ -15,7 +15,6 @@ import base64
 import json
 import warnings
 from typing import Any, AsyncGenerator, Dict, Literal, Optional
-from urllib.parse import urlencode
 
 import aiohttp
 from loguru import logger
@@ -432,7 +431,7 @@ class GladiaSTTService(STTService):
                     try:
                         self._websocket = websocket
                         self._connection_active = True
-                        logger.info("Connected to Gladia WebSocket")
+                        logger.debug(f"{self} Connected to Gladia WebSocket")
 
                         # Send buffered audio if any
                         await self._send_buffered_audio()
@@ -525,7 +524,7 @@ class GladiaSTTService(STTService):
         """Send any buffered audio after reconnection."""
         async with self._buffer_lock:
             if self._audio_buffer:
-                logger.info(f"Sending {len(self._audio_buffer)} bytes of buffered audio")
+                logger.debug(f"{self} Sending {len(self._audio_buffer)} bytes of buffered audio")
                 await self._send_audio(bytes(self._audio_buffer))
 
     async def _send_stop_recording(self):
@@ -627,8 +626,8 @@ class GladiaSTTService(STTService):
             self._should_reconnect = False
             return False
         delay = self._reconnection_delay * (2 ** (self._reconnection_attempts - 1))
-        logger.info(
+        logger.debug(
-            f"Reconnecting in {delay} seconds (attempt {self._reconnection_attempts}/{self._max_reconnection_attempts})"
+            f"{self} Reconnecting in {delay} seconds (attempt {self._reconnection_attempts}/{self._max_reconnection_attempts})"
         )
         await asyncio.sleep(delay)
         return True

src/pipecat/services/speechmatics/stt.py

@@ -23,6 +23,7 @@ from pipecat.frames.frames import (
     BotInterruptionFrame,
     CancelFrame,
     EndFrame,
+    ErrorFrame,
     Frame,
     InterimTranscriptionFrame,
     StartFrame,
@@ -463,8 +464,14 @@ class SpeechmaticsSTTService(STTService):
 
     async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]:
         """Adds audio to the audio buffer and yields None."""
-        await self._client.send_audio(audio)
+        try:
-        yield None
+            if self._client:
+                await self._client.send_audio(audio)
+            yield None
+        except Exception as e:
+            logger.error(f"Speechmatics error: {e}")
+            yield ErrorFrame(f"Speechmatics error: {e}", fatal=False)
+            await self._disconnect()
 
     def update_params(
         self,
@@ -520,7 +527,7 @@ class SpeechmaticsSTTService(STTService):
         )
 
         # Log the event
-        logger.debug("Connected to Speechmatics STT service")
+        logger.debug(f"{self} Connecting to Speechmatics STT service")
 
         # Recognition started event
         @self._client.on(ServerMessageType.RECOGNITION_STARTED)
@@ -562,31 +569,36 @@ class SpeechmaticsSTTService(STTService):
                 )
 
         # Start session
-        await self._client.start_session(
+        try:
-            transcription_config=self._transcription_config,
+            await self._client.start_session(
-            audio_format=AudioFormat(
+                transcription_config=self._transcription_config,
-                encoding=self._params.audio_encoding,
+                audio_format=AudioFormat(
-                sample_rate=self.sample_rate,
+                    encoding=self._params.audio_encoding,
-                chunk_size=self._params.chunk_size,
+                    sample_rate=self.sample_rate,
-            ),
+                    chunk_size=self._params.chunk_size,
-        )
+                ),
+            )
+            logger.debug(f"{self} Connected to Speechmatics STT service")
+        except Exception as e:
+            logger.error(f"{self} Error connecting to Speechmatics: {e}")
+        finally:
+            self._client = None
 
     async def _disconnect(self) -> None:
         """Disconnect from the STT service."""
         # Disconnect the client
+        logger.debug(f"{self} Disconnecting from Speechmatics STT service")
         try:
             if self._client:
-                await asyncio.wait_for(self._client.close(), timeout=1.0)
+                await asyncio.wait_for(self._client.close(), timeout=5.0)
+                logger.debug(f"{self} Disconnected from Speechmatics STT service")
         except asyncio.TimeoutError:
-            logger.warning("Timeout while closing Speechmatics client connection")
+            logger.warning(f"{self} Timeout while closing Speechmatics client connection")
         except Exception as e:
-            logger.error(f"Error closing Speechmatics client: {e}")
+            logger.error(f"{self} Error closing Speechmatics client: {e}")
         finally:
             self._client = None
 
-        # Log the event
-        logger.debug("Disconnected from Speechmatics STT service")
-
     def _process_config(self) -> None:
         """Create a formatted STT transcription config.