diff --git a/changelog/3671.added.2.md b/changelog/3671.added.2.md new file mode 100644 index 000000000..db4caeba5 --- /dev/null +++ b/changelog/3671.added.2.md @@ -0,0 +1 @@ +- Added `bulbul:v3-beta` TTS model support for Sarvam AI with temperature control and 25 new speaker voices. diff --git a/changelog/3671.added.md b/changelog/3671.added.md new file mode 100644 index 000000000..625d443b7 --- /dev/null +++ b/changelog/3671.added.md @@ -0,0 +1 @@ +- Added `saaras:v3` STT model support for Sarvam AI with new `mode` parameter (transcribe, translate, verbatim, translit, codemix) and prompt support. diff --git a/changelog/3671.fixed.md b/changelog/3671.fixed.md new file mode 100644 index 000000000..6e53ea864 --- /dev/null +++ b/changelog/3671.fixed.md @@ -0,0 +1 @@ +- Fixed issues in Sarvam STT and TTS services: missing event handler registration for VAD signals, `Optional[bool]` type annotations, WebSocket state cleanup on API errors, and TTS disconnect/reconnection state management. diff --git a/src/pipecat/services/sarvam/stt.py b/src/pipecat/services/sarvam/stt.py index e0da8520d..998597956 100644 --- a/src/pipecat/services/sarvam/stt.py +++ b/src/pipecat/services/sarvam/stt.py @@ -1,3 +1,9 @@ +# +# Copyright (c) 2024–2026, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + """Sarvam AI Speech-to-Text service implementation. This module provides a streaming Speech-to-Text service using Sarvam AI's WebSocket-based @@ -7,7 +13,7 @@ can handle multiple audio formats for Indian language speech recognition. import base64 from dataclasses import dataclass -from typing import Dict, Literal, Optional +from typing import AsyncGenerator, Dict, Literal, Optional from loguru import logger from pydantic import BaseModel @@ -149,8 +155,8 @@ class SarvamSTTService(STTService): language: Optional[Language] = None prompt: Optional[str] = None mode: Optional[Literal["transcribe", "translate", "verbatim", "translit", "codemix"]] = None - vad_signals: bool = None - high_vad_sensitivity: bool = None + vad_signals: Optional[bool] = None + high_vad_sensitivity: Optional[bool] = None def __init__( self, @@ -233,6 +239,12 @@ class SarvamSTTService(STTService): self._websocket_context = None self._socket_client = None self._receive_task = None + + if self._vad_signals: + self._register_event_handler("on_speech_started") + self._register_event_handler("on_speech_stopped") + self._register_event_handler("on_utterance_end") + logger.info(f"Sarvam STT initialized with SDK headers: {self._sdk_headers}") def language_to_service_language(self, language: Language) -> str: @@ -337,7 +349,7 @@ class SarvamSTTService(STTService): await super().cancel(frame) await self._disconnect() - async def run_stt(self, audio: bytes): + async def run_stt(self, audio: bytes) -> AsyncGenerator[Frame, None]: """Send audio data to Sarvam for transcription. Args: @@ -385,18 +397,25 @@ class SarvamSTTService(STTService): logger.debug("Connecting to Sarvam") try: - # Convert boolean parameters to string for SDK - vad_signals_str = "true" if self._vad_signals else "false" - high_vad_sensitivity_str = "true" if self._high_vad_sensitivity else "false" - # Build common connection parameters connect_kwargs = { "model": self.model_name, - "vad_signals": vad_signals_str, - "high_vad_sensitivity": high_vad_sensitivity_str, "sample_rate": str(self.sample_rate), } + # Enable flush signal when using Pipecat's VAD (not Sarvam's) so that + # the flush() call on user-stopped-speaking is honored by the server. + if not self._vad_signals: + connect_kwargs["flush_signal"] = "true" + + # Only send vad parameters when explicitly set (avoid overriding server defaults) + if self._vad_signals is not None: + connect_kwargs["vad_signals"] = "true" if self._vad_signals else "false" + if self._high_vad_sensitivity is not None: + connect_kwargs["high_vad_sensitivity"] = ( + "true" if self._high_vad_sensitivity else "false" + ) + # Add language_code for models that support it if self._language_string is not None: connect_kwargs["language_code"] = self._language_string @@ -447,6 +466,8 @@ class SarvamSTTService(STTService): logger.info("Connected to Sarvam successfully") except ApiError as e: + self._socket_client = None + self._websocket_context = None await self.push_error(error_msg=f"Sarvam API error: {e}", exception=e) except Exception as e: self._socket_client = None diff --git a/src/pipecat/services/sarvam/tts.py b/src/pipecat/services/sarvam/tts.py index 2e767eb6a..778a29005 100644 --- a/src/pipecat/services/sarvam/tts.py +++ b/src/pipecat/services/sarvam/tts.py @@ -525,7 +525,7 @@ class SarvamHttpTTSService(TTSService): audio_data = base64.b64decode(base64_audio) # Strip WAV header (first 44 bytes) if present - if audio_data.startswith(b"RIFF"): + if len(audio_data) > 44 and audio_data.startswith(b"RIFF"): logger.debug("Stripping WAV header from Sarvam audio data") audio_data = audio_data[44:] @@ -792,7 +792,6 @@ class SarvamTTSService(InterruptibleTTSService): self._started = False self._receive_task = None self._keepalive_task = None - self._disconnecting = False def can_generate_metrics(self) -> bool: """Check if this service can generate processing metrics. @@ -844,10 +843,13 @@ class SarvamTTSService(InterruptibleTTSService): await self._disconnect() async def flush_audio(self): - """Flush any pending audio synthesis by sending stop command.""" - if self._websocket: - msg = {"type": "flush"} - await self._websocket.send(json.dumps(msg)) + """Flush any pending audio synthesis by sending flush command.""" + try: + if self._websocket: + msg = {"type": "flush"} + await self._websocket.send(json.dumps(msg)) + except Exception as e: + await self.push_error(error_msg=f"Error sending flush to Sarvam: {e}", exception=e) async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): """Push a frame downstream with special handling for stop conditions. @@ -894,29 +896,15 @@ class SarvamTTSService(InterruptibleTTSService): """Disconnect from Sarvam WebSocket and clean up tasks.""" await super()._disconnect() - try: - # First, set a flag to prevent new operations - self._disconnecting = True + if self._receive_task: + await self.cancel_task(self._receive_task) + self._receive_task = None - # Cancel background tasks BEFORE closing websocket - if self._receive_task: - await self.cancel_task(self._receive_task, timeout=2.0) - self._receive_task = None + if self._keepalive_task: + await self.cancel_task(self._keepalive_task) + self._keepalive_task = None - if self._keepalive_task: - await self.cancel_task(self._keepalive_task, timeout=2.0) - self._keepalive_task = None - - # Now close the websocket - await self._disconnect_websocket() - - except Exception as e: - await self.push_error(error_msg=f"Unknown error occurred: {e}", exception=e) - finally: - # Reset state only after everything is cleaned up - self._started = False - self._websocket = None - self._disconnecting = False + await self._disconnect_websocket() async def _connect_websocket(self): """Establish WebSocket connection to Sarvam API.""" @@ -996,6 +984,7 @@ class SarvamTTSService(InterruptibleTTSService): if "too long" in error_msg.lower() or "timeout" in error_msg.lower(): logger.warning("Connection timeout detected, service may need restart") + self._started = False await self.push_frame(ErrorFrame(error=f"TTS Error: {error_msg}")) async def _keepalive_task_handler(self): @@ -1007,19 +996,12 @@ class SarvamTTSService(InterruptibleTTSService): async def _send_keepalive(self): """Send keepalive message to maintain connection.""" - if self._disconnecting: - return - if self._websocket and self._websocket.state == State.OPEN: msg = {"type": "ping"} await self._websocket.send(json.dumps(msg)) async def _send_text(self, text: str): """Send text to Sarvam WebSocket for synthesis.""" - if self._disconnecting: - logger.warning("Service is disconnecting, ignoring text send") - return - if self._websocket and self._websocket.state == State.OPEN: msg = {"type": "text", "data": {"text": text}} await self._websocket.send(json.dumps(msg))