diff --git a/CHANGELOG.md b/CHANGELOG.md index 7f7b5e037..a8e536d09 100644 --- a/CHANGELOG.md +++ b/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). diff --git a/src/pipecat/services/gladia/stt.py b/src/pipecat/services/gladia/stt.py index c774c4257..316edd91c 100644 --- a/src/pipecat/services/gladia/stt.py +++ b/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( - f"Reconnecting in {delay} seconds (attempt {self._reconnection_attempts}/{self._max_reconnection_attempts})" + logger.debug( + f"{self} Reconnecting in {delay} seconds (attempt {self._reconnection_attempts}/{self._max_reconnection_attempts})" ) await asyncio.sleep(delay) return True diff --git a/src/pipecat/services/speechmatics/stt.py b/src/pipecat/services/speechmatics/stt.py index 59145742e..d1091b5f0 100644 --- a/src/pipecat/services/speechmatics/stt.py +++ b/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) - yield None + try: + 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( - transcription_config=self._transcription_config, - audio_format=AudioFormat( - encoding=self._params.audio_encoding, - sample_rate=self.sample_rate, - chunk_size=self._params.chunk_size, - ), - ) + try: + await self._client.start_session( + transcription_config=self._transcription_config, + audio_format=AudioFormat( + encoding=self._params.audio_encoding, + 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.