diff --git a/examples/foundational/07z-interruptible-sarvam-http.py b/examples/foundational/07z-interruptible-sarvam-http.py new file mode 100644 index 000000000..e75ef7909 --- /dev/null +++ b/examples/foundational/07z-interruptible-sarvam-http.py @@ -0,0 +1,126 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + + +import os + +import aiohttp +from dotenv import load_dotenv +from loguru import logger + +from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.pipeline.pipeline import Pipeline +from pipecat.pipeline.runner import PipelineRunner +from pipecat.pipeline.task import PipelineParams, PipelineTask +from pipecat.processors.aggregators.openai_llm_context import OpenAILLMContext +from pipecat.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.deepgram.stt import DeepgramSTTService +from pipecat.services.openai.llm import OpenAILLMService +from pipecat.services.sarvam.tts import SarvamHttpTTSService +from pipecat.transcriptions.language import Language +from pipecat.transports.base_transport import BaseTransport, TransportParams +from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams +from pipecat.transports.services.daily import DailyParams + +load_dotenv(override=True) + + +# We store functions so objects (e.g. SileroVADAnalyzer) don't get +# instantiated. The function will be called when the desired transport gets +# selected. +transport_params = { + "daily": lambda: DailyParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + "twilio": lambda: FastAPIWebsocketParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), + "webrtc": lambda: TransportParams( + audio_in_enabled=True, + audio_out_enabled=True, + vad_analyzer=SileroVADAnalyzer(), + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + # Create an HTTP session + async with aiohttp.ClientSession() as session: + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + + tts = SarvamHttpTTSService( + api_key=os.getenv("SARVAM_API_KEY"), + aiohttp_session=session, + params=SarvamHttpTTSService.InputParams(language=Language.EN), + ) + + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] + + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses + ] + ) + + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, + ) + + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([context_aggregator.user().get_context_frame()]) + + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() + + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + + await runner.run(task) + + +async def bot(runner_args: RunnerArguments): + """Main bot entry point compatible with Pipecat Cloud.""" + transport = await create_transport(runner_args, transport_params) + await run_bot(transport, runner_args) + + +if __name__ == "__main__": + from pipecat.runner.run import main + + main() diff --git a/examples/foundational/07z-interruptible-sarvam.py b/examples/foundational/07z-interruptible-sarvam.py index b3d13a690..91cb0587b 100644 --- a/examples/foundational/07z-interruptible-sarvam.py +++ b/examples/foundational/07z-interruptible-sarvam.py @@ -5,6 +5,7 @@ # +import asyncio import os import aiohttp @@ -12,6 +13,7 @@ from dotenv import load_dotenv from loguru import logger from pipecat.audio.vad.silero import SileroVADAnalyzer +from pipecat.frames.frames import TTSUpdateSettingsFrame from pipecat.pipeline.pipeline import Pipeline from pipecat.pipeline.runner import PipelineRunner from pipecat.pipeline.task import PipelineParams, PipelineTask @@ -21,7 +23,6 @@ from pipecat.runner.utils import create_transport from pipecat.services.deepgram.stt import DeepgramSTTService from pipecat.services.openai.llm import OpenAILLMService from pipecat.services.sarvam.tts import SarvamTTSService -from pipecat.transcriptions.language import Language from pipecat.transports.base_transport import BaseTransport, TransportParams from pipecat.transports.network.fastapi_websocket import FastAPIWebsocketParams from pipecat.transports.services.daily import DailyParams @@ -54,64 +55,64 @@ transport_params = { async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): logger.info(f"Starting bot") - # Create an HTTP session - async with aiohttp.ClientSession() as session: - stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) + stt = DeepgramSTTService(api_key=os.getenv("DEEPGRAM_API_KEY")) - tts = SarvamTTSService( - api_key=os.getenv("SARVAM_API_KEY"), - aiohttp_session=session, - params=SarvamTTSService.InputParams(language=Language.EN), - ) + tts = SarvamTTSService( + api_key=os.getenv("SARVAM_API_KEY"), + model="bulbul:v2", + voice_id="manisha", + ) + llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) - llm = OpenAILLMService(api_key=os.getenv("OPENAI_API_KEY")) + messages = [ + { + "role": "system", + "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", + }, + ] - messages = [ - { - "role": "system", - "content": "You are a helpful LLM in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. Your output will be converted to audio so don't include special characters in your answers. Respond to what the user said in a creative and helpful way.", - }, + context = OpenAILLMContext(messages) + context_aggregator = llm.create_context_aggregator(context) + + pipeline = Pipeline( + [ + transport.input(), # Transport user input + stt, + context_aggregator.user(), # User responses + llm, # LLM + tts, # TTS + transport.output(), # Transport bot output + context_aggregator.assistant(), # Assistant spoken responses ] + ) - context = OpenAILLMContext(messages) - context_aggregator = llm.create_context_aggregator(context) + task = PipelineTask( + pipeline, + params=PipelineParams( + enable_metrics=True, + enable_usage_metrics=True, + ), + ) - pipeline = Pipeline( - [ - transport.input(), # Transport user input - stt, - context_aggregator.user(), # User responses - llm, # LLM - tts, # TTS - transport.output(), # Transport bot output - context_aggregator.assistant(), # Assistant spoken responses - ] - ) + @transport.event_handler("on_client_connected") + async def on_client_connected(transport, client): + logger.info(f"Client connected") + # Kick off the conversation. + messages.append({"role": "system", "content": "Please introduce yourself to the user."}) + await task.queue_frames([context_aggregator.user().get_context_frame()]) - task = PipelineTask( - pipeline, - params=PipelineParams( - enable_metrics=True, - enable_usage_metrics=True, - ), - idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, - ) + # Optionally, you can wait for 30 seconds and then change the voice. + # await asyncio.sleep(30) + # await task.queue_frame(TTSUpdateSettingsFrame(settings={"voice": "anushka"})) - @transport.event_handler("on_client_connected") - async def on_client_connected(transport, client): - logger.info(f"Client connected") - # Kick off the conversation. - messages.append({"role": "system", "content": "Please introduce yourself to the user."}) - await task.queue_frames([context_aggregator.user().get_context_frame()]) + @transport.event_handler("on_client_disconnected") + async def on_client_disconnected(transport, client): + logger.info(f"Client disconnected") + await task.cancel() - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") - await task.cancel() + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) - runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) - - await runner.run(task) + await runner.run(task) async def bot(runner_args: RunnerArguments): diff --git a/pyproject.toml b/pyproject.toml index dbfd105c5..0912c02c1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -90,6 +90,7 @@ rime = [ "websockets>=13.1,<15.0" ] riva = [ "nvidia-riva-client~=2.21.1" ] runner = [ "python-dotenv>=1.0.0,<2.0.0", "uvicorn>=0.32.0,<1.0.0", "fastapi>=0.115.6,<0.117.0", "pipecat-ai-small-webrtc-prebuilt>=1.0.0"] sambanova = [] +sarvam = [ "websockets>=13.1,<15.0" ] sentry = [ "sentry-sdk~=2.23.1" ] local-smart-turn = [ "coremltools>=8.0", "transformers", "torch>=2.5.0,<3", "torchaudio>=2.5.0,<3" ] remote-smart-turn = [] diff --git a/src/pipecat/services/sarvam/tts.py b/src/pipecat/services/sarvam/tts.py index 51fd4f07b..c0b162594 100644 --- a/src/pipecat/services/sarvam/tts.py +++ b/src/pipecat/services/sarvam/tts.py @@ -6,25 +6,42 @@ """Sarvam AI text-to-speech service implementation.""" +import asyncio import base64 -from typing import AsyncGenerator, Optional +import json +import warnings +from typing import Any, AsyncGenerator, Mapping, Optional import aiohttp from loguru import logger from pydantic import BaseModel, Field from pipecat.frames.frames import ( + CancelFrame, + EndFrame, ErrorFrame, Frame, + LLMFullResponseEndFrame, StartFrame, + StartInterruptionFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, ) -from pipecat.services.tts_service import TTSService +from pipecat.processors.frame_processor import FrameDirection +from pipecat.services.tts_service import InterruptibleTTSService, TTSService from pipecat.transcriptions.language import Language +from pipecat.utils.asyncio.watchdog_async_iterator import WatchdogAsyncIterator from pipecat.utils.tracing.service_decorators import traced_tts +try: + from websockets.asyncio.client import connect as websocket_connect + from websockets.protocol import State +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error("In order to use Sarvam, you need to `pip install pipecat-ai[sarvam]`.") + raise Exception(f"Missing module: {e}") + def language_to_sarvam_language(language: Language) -> Optional[str]: """Convert Pipecat Language enum to Sarvam AI language codes. @@ -52,7 +69,7 @@ def language_to_sarvam_language(language: Language) -> Optional[str]: return LANGUAGE_MAP.get(language) -class SarvamTTSService(TTSService): +class SarvamHttpTTSService(TTSService): """Text-to-Speech service using Sarvam AI's API. Converts text to speech using Sarvam AI's TTS models with support for multiple @@ -95,9 +112,9 @@ class SarvamTTSService(TTSService): self, *, api_key: str, + aiohttp_session: aiohttp.ClientSession, voice_id: str = "anushka", model: str = "bulbul:v2", - aiohttp_session: aiohttp.ClientSession, base_url: str = "https://api.sarvam.ai", sample_rate: Optional[int] = None, params: Optional[InputParams] = None, @@ -107,9 +124,9 @@ class SarvamTTSService(TTSService): Args: api_key: Sarvam AI API subscription key. + aiohttp_session: Shared aiohttp session for making requests. voice_id: Speaker voice ID (e.g., "anushka", "meera"). Defaults to "anushka". model: TTS model to use ("bulbul:v1" or "bulbul:v2"). Defaults to "bulbul:v2". - aiohttp_session: Shared aiohttp session for making requests. base_url: Sarvam AI API base URL. Defaults to "https://api.sarvam.ai". sample_rate: Audio sample rate in Hz (8000, 16000, 22050, 24000). If None, uses default. params: Additional voice and preprocessing parameters. If None, uses defaults. @@ -117,16 +134,16 @@ class SarvamTTSService(TTSService): """ super().__init__(sample_rate=sample_rate, **kwargs) - params = params or SarvamTTSService.InputParams() + params = params or SarvamHttpTTSService.InputParams() self._api_key = api_key self._base_url = base_url self._session = aiohttp_session self._settings = { - "language": self.language_to_service_language(params.language) - if params.language - else "en-IN", + "language": ( + self.language_to_service_language(params.language) if params.language else "en-IN" + ), "pitch": params.pitch, "pace": params.pace, "loudness": params.loudness, @@ -186,7 +203,7 @@ class SarvamTTSService(TTSService): "pitch": self._settings["pitch"], "pace": self._settings["pace"], "loudness": self._settings["loudness"], - "speech_sample_rate": self.sample_rate, + "sample_rate": self.sample_rate, "enable_preprocessing": self._settings["enable_preprocessing"], "model": self._model_name, } @@ -240,3 +257,391 @@ class SarvamTTSService(TTSService): finally: await self.stop_ttfb_metrics() yield TTSStoppedFrame() + + +class SarvamTTSService(InterruptibleTTSService): + """WebSocket-based text-to-speech service using Sarvam AI. + + Provides streaming TTS with real-time audio generation for multiple Indian languages. + Supports voice control parameters like pitch, pace, and loudness adjustment. + + Example:: + + tts = SarvamTTSService( + api_key="your-api-key", + voice_id="anushka", + model="bulbul:v2", + params=SarvamTTSService.InputParams( + language=Language.HI, + pitch=0.1, + pace=1.2 + ) + ) + """ + + class InputParams(BaseModel): + """Configuration parameters for Sarvam TTS. + + Parameters: + pitch: Voice pitch adjustment (-0.75 to 0.75). Defaults to 0.0. + pace: Speech pace multiplier (0.3 to 3.0). Defaults to 1.0. + loudness: Volume multiplier (0.1 to 3.0). Defaults to 1.0. + enable_preprocessing: Enable text preprocessing. Defaults to False. + min_buffer_size: Minimum number of characters to buffer before generating audio. + Lower values reduce latency but may affect quality. Defaults to 50. + max_chunk_length: Maximum number of characters processed in a single chunk. + Controls memory usage and processing efficiency. Defaults to 200. + output_audio_codec: Audio codec format. Defaults to "linear16". + output_audio_bitrate: Audio bitrate. Defaults to "128k". + language: Target language for synthesis. Supports Bengali (bn-IN), English (en-IN), + Gujarati (gu-IN), Hindi (hi-IN), Kannada (kn-IN), Malayalam (ml-IN), + Marathi (mr-IN), Odia (od-IN), Punjabi (pa-IN), Tamil (ta-IN), + Telugu (te-IN). Defaults to en-IN. + + Available Speakers: + Female: anushka, manisha, vidya, arya + Male: abhilash, karun, hitesh + """ + + pitch: Optional[float] = Field(default=0.0, ge=-0.75, le=0.75) + pace: Optional[float] = Field(default=1.0, ge=0.3, le=3.0) + loudness: Optional[float] = Field(default=1.0, ge=0.1, le=3.0) + enable_preprocessing: Optional[bool] = False + min_buffer_size: Optional[int] = 50 + max_chunk_length: Optional[int] = 200 + output_audio_codec: Optional[str] = "linear16" + output_audio_bitrate: Optional[str] = "128k" + language: Optional[Language] = Language.EN + + def __init__( + self, + *, + api_key: str, + model: str = "bulbul:v2", + voice_id: str = "anushka", + url: str = "wss://api.sarvam.ai/text-to-speech/ws", + aiohttp_session: Optional[aiohttp.ClientSession] = None, + aggregate_sentences: Optional[bool] = True, + sample_rate: Optional[int] = None, + params: Optional[InputParams] = None, + **kwargs, + ): + """Initialize the Sarvam TTS service with voice and transport configuration. + + Args: + api_key: Sarvam API key for authenticating TTS requests. + model: Identifier of the Sarvam speech model (default "bulbul:v2"). + voice_id: Voice identifier for synthesis (default "anushka"). + url: WebSocket URL for connecting to the TTS backend (default production URL). + aiohttp_session: Optional shared aiohttp session. To maintain backward compatibility. + .. deprecated:: 0.0.81 + aiohttp_session is no longer used. This parameter will be removed in a future version. + aggregate_sentences: Whether to merge multiple sentences into one audio chunk (default True). + sample_rate: Desired sample rate for the output audio in Hz (overrides default if set). + params: Optional input parameters to override global configuration. + **kwargs: Optional keyword arguments forwarded to InterruptibleTTSService (such as + `push_stop_frames`, `sample_rate`, task manager parameters, event hooks, etc.) + to customize transport behavior or enable metrics support. + + This method sets up the internal TTS configuration mapping, constructs the WebSocket + URL based on the chosen model, and initializes state flags before connecting. + """ + # Initialize parent class first + super().__init__( + aggregate_sentences=aggregate_sentences, + push_text_frames=True, + pause_frame_processing=True, + push_stop_frames=True, + sample_rate=sample_rate, + **kwargs, + ) + params = params or SarvamTTSService.InputParams() + if aiohttp_session is not None: + warnings.warn( + "The 'aiohttp_session' parameter is deprecated and will be removed in a future version. ", + DeprecationWarning, + stacklevel=2, + ) + # WebSocket endpoint URL + self._websocket_url = f"{url}?model={model}" + self._api_key = api_key + self.set_model_name(model) + self.set_voice(voice_id) + # Configuration parameters + self._settings = { + "target_language_code": ( + self.language_to_service_language(params.language) if params.language else "en-IN" + ), + "pitch": params.pitch, + "pace": params.pace, + "speaker": voice_id, + "loudness": params.loudness, + "speech_sample_rate": 0, + "enable_preprocessing": params.enable_preprocessing, + "min_buffer_size": params.min_buffer_size, + "max_chunk_length": params.max_chunk_length, + "output_audio_codec": params.output_audio_codec, + "output_audio_bitrate": params.output_audio_bitrate, + } + 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. + + Returns: + True, as Sarvam service supports metrics generation. + """ + return True + + def language_to_service_language(self, language: Language) -> Optional[str]: + """Convert a Language enum to Sarvam AI language format. + + Args: + language: The language to convert. + + Returns: + The Sarvam AI-specific language code, or None if not supported. + """ + return language_to_sarvam_language(language) + + async def start(self, frame: StartFrame): + """Start the Sarvam TTS service. + + Args: + frame: The start frame containing initialization parameters. + """ + await super().start(frame) + + self._settings["speech_sample_rate"] = self.sample_rate + await self._connect() + + async def stop(self, frame: EndFrame): + """Stop the Sarvam TTS service. + + Args: + frame: The end frame. + """ + await super().stop(frame) + await self._disconnect() + + async def cancel(self, frame: CancelFrame): + """Cancel the Sarvam TTS service. + + Args: + frame: The cancel frame. + """ + await super().cancel(frame) + 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)) + + async def push_frame(self, frame: Frame, direction: FrameDirection = FrameDirection.DOWNSTREAM): + """Push a frame downstream with special handling for stop conditions. + + Args: + frame: The frame to push. + direction: The direction to push the frame. + """ + await super().push_frame(frame, direction) + if isinstance(frame, (TTSStoppedFrame, StartInterruptionFrame)): + self._started = False + + async def process_frame(self, frame: Frame, direction: FrameDirection): + """Process a frame and flush audio if it's the end of a full response.""" + if isinstance(frame, LLMFullResponseEndFrame): + await self.flush_audio() + return await super().process_frame(frame, direction) + + async def _update_settings(self, settings: Mapping[str, Any]): + """Update service settings and reconnect if voice changed.""" + prev_voice = self._voice_id + await super()._update_settings(settings) + if not prev_voice == self._voice_id: + logger.info(f"Switching TTS voice to: [{self._voice_id}]") + await self._send_config() + + async def _connect(self): + """Connect to Sarvam WebSocket and start background tasks.""" + await self._connect_websocket() + + if self._websocket and not self._receive_task: + self._receive_task = self.create_task(self._receive_task_handler(self._report_error)) + + if self._websocket and not self._keepalive_task: + self._keepalive_task = self.create_task( + self._keepalive_task_handler(), + watchdog_timeout_secs=25, + ) + + async def _disconnect(self): + """Disconnect from Sarvam WebSocket and clean up tasks.""" + try: + # First, set a flag to prevent new operations + self._disconnecting = True + + # 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, timeout=2.0) + self._keepalive_task = None + + # Now close the websocket + await self._disconnect_websocket() + + except Exception as e: + logger.error(f"Error during disconnect: {e}") + finally: + # Reset state only after everything is cleaned up + self._started = False + self._websocket = None + self._disconnecting = False + + async def _connect_websocket(self): + """Establish WebSocket connection to Sarvam API.""" + try: + if self._websocket and self._websocket.state is State.OPEN: + return + + self._websocket = await websocket_connect( + self._websocket_url, + additional_headers={ + "api-subscription-key": self._api_key, + }, + ) + logger.debug("Connected to Sarvam TTS Websocket") + await self._send_config() + + except Exception as e: + logger.error(f"{self} initialization error: {e}") + self._websocket = None + await self._call_event_handler("on_connection_error", f"{e}") + + async def _send_config(self): + """Send initial configuration message.""" + if not self._websocket: + raise Exception("WebSocket not connected") + self._settings["speaker"] = self._voice_id + logger.debug(f"Config being sent is {self._settings}") + config_message = {"type": "config", "data": self._settings} + + try: + await self._websocket.send(json.dumps(config_message)) + logger.debug("Configuration sent successfully") + except Exception as e: + logger.error(f"Failed to send config: {str(e)}") + await self.push_frame(ErrorFrame(f"Failed to send config: {str(e)}")) + raise + + async def _disconnect_websocket(self): + """Close WebSocket connection and clean up state.""" + try: + await self.stop_all_metrics() + + if self._websocket: + logger.debug("Disconnecting from Sarvam") + await self._websocket.close() + except Exception as e: + logger.error(f"{self} error closing websocket: {e}") + + def _get_websocket(self): + if self._websocket: + return self._websocket + raise Exception("Websocket not connected") + + async def _receive_messages(self): + """Receive and process messages from Sarvam WebSocket.""" + async for message in WatchdogAsyncIterator( + self._get_websocket(), manager=self.task_manager + ): + if isinstance(message, str): + msg = json.loads(message) + if msg.get("type") == "audio": + # Check for interruption before processing audio + await self.stop_ttfb_metrics() + audio = base64.b64decode(msg["data"]["audio"]) + frame = TTSAudioRawFrame(audio, self.sample_rate, 1) + await self.push_frame(frame) + elif msg.get("type") == "error": + error_msg = msg["data"]["message"] + logger.error(f"TTS Error: {error_msg}") + + # If it's a timeout error, the connection might need to be reset + if "too long" in error_msg.lower() or "timeout" in error_msg.lower(): + logger.warning("Connection timeout detected, service may need restart") + + await self.push_frame(ErrorFrame(f"TTS Error: {error_msg}")) + + async def _keepalive_task_handler(self): + """Handle keepalive messages to maintain WebSocket connection.""" + KEEPALIVE_SLEEP = 20 + while True: + self.reset_watchdog() + await asyncio.sleep(KEEPALIVE_SLEEP) + await self._send_keepalive() + + 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)) + else: + logger.warning("WebSocket not ready, cannot send text") + + @traced_tts + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + """Generate speech audio frames from input text using Sarvam TTS. + + Sends text over WebSocket for synthesis and yields corresponding audio or status frames. + + Args: + text: The text input to synthesize. + + Yields: + Frame objects including TTSStartedFrame, TTSAudioRawFrame(s), or TTSStoppedFrame. + """ + logger.debug(f"Generating TTS: [{text}]") + + try: + if not self._websocket or self._websocket.state is State.CLOSED: + await self._connect() + + try: + if not self._started: + await self.start_ttfb_metrics() + yield TTSStartedFrame() + self._started = True + await self._send_text(text) + await self.start_tts_usage_metrics(text) + except Exception as e: + logger.error(f"{self} error sending message: {e}") + yield TTSStoppedFrame() + await self._disconnect() + await self._connect() + return + yield None + except Exception as e: + logger.error(f"{self} exception: {e}") diff --git a/uv.lock b/uv.lock index 06daa64d3..2cb829240 100644 --- a/uv.lock +++ b/uv.lock @@ -4337,6 +4337,9 @@ runner = [ { name = "python-dotenv" }, { name = "uvicorn" }, ] +sarvam = [ + { name = "websockets" }, +] sentry = [ { name = "sentry-sdk" }, ] @@ -4491,10 +4494,11 @@ requires-dist = [ { name = "websockets", marker = "extra == 'openai'", specifier = ">=13.1,<15.0" }, { name = "websockets", marker = "extra == 'playht'", specifier = ">=13.1,<15.0" }, { name = "websockets", marker = "extra == 'rime'", specifier = ">=13.1,<15.0" }, + { name = "websockets", marker = "extra == 'sarvam'", specifier = ">=13.1,<15.0" }, { name = "websockets", marker = "extra == 'soniox'", specifier = ">=13.1,<15.0" }, { name = "websockets", marker = "extra == 'websocket'", specifier = ">=13.1,<15.0" }, ] -provides-extras = ["anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "deepseek", "daily", "deepgram", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "inworld", "krisp", "koala", "langchain", "livekit", "lmnt", "local", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "nim", "neuphonic", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "rime", "riva", "runner", "sambanova", "sentry", "local-smart-turn", "remote-smart-turn", "silero", "simli", "soniox", "soundfile", "speechmatics", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "whisper"] +provides-extras = ["anthropic", "assemblyai", "asyncai", "aws", "aws-nova-sonic", "azure", "cartesia", "cerebras", "deepseek", "daily", "deepgram", "elevenlabs", "fal", "fireworks", "fish", "gladia", "google", "grok", "groq", "gstreamer", "heygen", "inworld", "krisp", "koala", "langchain", "livekit", "lmnt", "local", "mcp", "mem0", "mistral", "mlx-whisper", "moondream", "nim", "neuphonic", "noisereduce", "openai", "openpipe", "openrouter", "perplexity", "playht", "qwen", "rime", "riva", "runner", "sambanova", "sarvam", "sentry", "local-smart-turn", "remote-smart-turn", "silero", "simli", "soniox", "soundfile", "speechmatics", "tavus", "together", "tracing", "ultravox", "webrtc", "websocket", "whisper"] [package.metadata.requires-dev] dev = [