From 541a43905bf102f45369769e1077132b1e12928d Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 10 Aug 2025 07:48:17 -0400 Subject: [PATCH] Add GeminiTTSService --- src/pipecat/services/google/tts.py | 264 ++++++++++++++++++++++++++++- 1 file changed, 263 insertions(+), 1 deletion(-) diff --git a/src/pipecat/services/google/tts.py b/src/pipecat/services/google/tts.py index 0bd201c4b..6fb99a728 100644 --- a/src/pipecat/services/google/tts.py +++ b/src/pipecat/services/google/tts.py @@ -9,6 +9,9 @@ This module provides integration with Google Cloud Text-to-Speech API, offering both HTTP-based synthesis with SSML support and streaming synthesis for real-time applications. + +It also includes GeminiTTSService which uses Gemini's TTS-specific models +for natural voice control and multi-speaker conversations. """ import json @@ -19,7 +22,7 @@ from pipecat.utils.tracing.service_decorators import traced_tts # Suppress gRPC fork warnings os.environ["GRPC_ENABLE_FORK_SUPPORT"] = "false" -from typing import AsyncGenerator, Literal, Optional +from typing import AsyncGenerator, List, Literal, Optional from loguru import logger from pydantic import BaseModel @@ -27,6 +30,7 @@ from pydantic import BaseModel from pipecat.frames.frames import ( ErrorFrame, Frame, + StartFrame, TTSAudioRawFrame, TTSStartedFrame, TTSStoppedFrame, @@ -47,6 +51,15 @@ except ModuleNotFoundError as e: ) raise Exception(f"Missing module: {e}") +try: + from google import genai + from google.genai import types + +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error("In order to use Gemini TTS, you need to `pip install pipecat-ai[google]`.") + raise Exception(f"Missing module: {e}") + def language_to_google_tts_language(language: Language) -> Optional[str]: """Convert a Language enum to Google TTS language code. @@ -642,3 +655,252 @@ class GoogleTTSService(TTSService): logger.exception(f"{self} error generating TTS: {e}") error_message = f"TTS generation error: {str(e)}" yield ErrorFrame(error=error_message) + + +class GeminiTTSService(TTSService): + """Gemini Text-to-Speech service using Gemini TTS models. + + Provides text-to-speech synthesis using Gemini's TTS-specific models + (gemini-2.5-flash-preview-tts and gemini-2.5-pro-preview-tts) with + support for natural voice control, multiple speakers, and voice styles. + + Note: + Requires Google AI API key. This uses the Gemini API, not Google Cloud TTS. + Audio-out is currently a preview feature. + + Example:: + + tts = GeminiTTSService( + api_key="your-google-ai-api-key", + model="gemini-2.5-flash-preview-tts", + voice_id="Kore", + params=GeminiTTSService.InputParams( + language=Language.EN_US, + ) + ) + """ + + GOOGLE_SAMPLE_RATE = 24000 # Google TTS always outputs at 24kHz + + # List of available Gemini TTS voices + AVAILABLE_VOICES = [ + "Zephyr", + "Puck", + "Charon", + "Kore", + "Fenrir", + "Leda", + "Orus", + "Aoede", + "Callirhoe", + "Autonoe", + "Enceladus", + "Iapetus", + "Umbriel", + "Algieba", + "Despina", + "Erinome", + "Algenib", + "Rasalgethi", + "Laomedeia", + "Achernar", + "Alnilam", + "Schedar", + "Gacrux", + "Pulcherrima", + "Achird", + "Zubenelgenubi", + "Vindemiatrix", + "Sadachbia", + "Sadaltager", + "Sulafar", + ] + + class InputParams(BaseModel): + """Input parameters for Gemini TTS configuration. + + Parameters: + language: Language for synthesis. Defaults to English. + multi_speaker: Whether to enable multi-speaker support. + speaker_configs: List of speaker configurations for multi-speaker mode. + """ + + language: Optional[Language] = Language.EN + multi_speaker: bool = False + speaker_configs: Optional[List[dict]] = None + + def __init__( + self, + *, + api_key: str, + model: str = "gemini-2.5-flash-preview-tts", + voice_id: str = "Kore", + sample_rate: Optional[int] = None, + params: Optional[InputParams] = None, + **kwargs, + ): + """Initializes the Gemini TTS service. + + Args: + api_key: Google AI API key for authentication. + model: Gemini TTS model to use. Must be a TTS model like + "gemini-2.5-flash-preview-tts" or "gemini-2.5-pro-preview-tts". + voice_id: Voice name from the available Gemini voices. + sample_rate: Audio sample rate in Hz. If None, uses Google's default 24kHz. + params: TTS configuration parameters. + **kwargs: Additional arguments passed to parent TTSService. + """ + if sample_rate and sample_rate != self.GOOGLE_SAMPLE_RATE: + logger.warning( + f"Google TTS only supports {self.GOOGLE_SAMPLE_RATE}Hz sample rate. " + f"Current rate of {sample_rate}Hz may cause issues." + ) + super().__init__(sample_rate=sample_rate, **kwargs) + + params = params or GeminiTTSService.InputParams() + + if voice_id not in self.AVAILABLE_VOICES: + logger.warning(f"Voice '{voice_id}' not in known voices list. Using anyway.") + + self._api_key = api_key + self._model = model + self._voice_id = voice_id + self._settings = { + "language": self.language_to_service_language(params.language) + if params.language + else "en-US", + "multi_speaker": params.multi_speaker, + "speaker_configs": params.speaker_configs, + } + + self._client = genai.Client(api_key=api_key) + + def can_generate_metrics(self) -> bool: + """Check if this service can generate processing metrics. + + Returns: + True, as Gemini TTS service supports metrics generation. + """ + return True + + def language_to_service_language(self, language: Language) -> Optional[str]: + """Convert a Language enum to Gemini TTS language format. + + Args: + language: The language to convert. + + Returns: + The Gemini TTS-specific language code, or None if not supported. + """ + return language_to_google_tts_language(language) + + def set_voice(self, voice_id: str): + """Set the voice for TTS generation. + + Args: + voice_id: Name of the voice to use from AVAILABLE_VOICES. + """ + if voice_id not in self.AVAILABLE_VOICES: + logger.warning(f"Voice '{voice_id}' not in known voices list. Using anyway.") + self._voice_id = voice_id + + async def start(self, frame: StartFrame): + """Start the Gemini TTS service. + + Args: + frame: The start frame containing initialization parameters. + """ + await super().start(frame) + if self.sample_rate != self.GOOGLE_SAMPLE_RATE: + logger.warning( + f"Google TTS requires {self.GOOGLE_SAMPLE_RATE}Hz sample rate. " + f"Current rate of {self.sample_rate}Hz may cause issues." + ) + + @traced_tts + async def run_tts(self, text: str) -> AsyncGenerator[Frame, None]: + """Generate speech from text using Gemini TTS models. + + Args: + text: The text to synthesize into speech. Can include natural language + instructions for style, tone, etc. + + Yields: + Frame: Audio frames containing the synthesized speech. + """ + logger.debug(f"{self}: Generating TTS [{text}]") + + try: + await self.start_ttfb_metrics() + + # Build the speech config + if self._settings["multi_speaker"] and self._settings["speaker_configs"]: + # Multi-speaker mode + speaker_voice_configs = [] + for speaker_config in self._settings["speaker_configs"]: + speaker_voice_configs.append( + types.SpeakerVoiceConfig( + speaker=speaker_config["speaker"], + voice_config=types.VoiceConfig( + prebuilt_voice_config=types.PrebuiltVoiceConfig( + voice_name=speaker_config.get("voice_id", self._voice_id) + ) + ), + ) + ) + + speech_config = types.SpeechConfig( + multi_speaker_voice_config=types.MultiSpeakerVoiceConfig( + speaker_voice_configs=speaker_voice_configs + ) + ) + else: + # Single speaker mode + speech_config = types.SpeechConfig( + voice_config=types.VoiceConfig( + prebuilt_voice_config=types.PrebuiltVoiceConfig(voice_name=self._voice_id) + ) + ) + + # Create the generation config + generation_config = types.GenerateContentConfig( + response_modalities=["AUDIO"], + speech_config=speech_config, + ) + + # Generate the content + response = await self._client.aio.models.generate_content( + model=self._model, + contents=text, + config=generation_config, + ) + + await self.start_tts_usage_metrics(text) + + yield TTSStartedFrame() + + # Extract audio data from response + if response.candidates and len(response.candidates) > 0: + candidate = response.candidates[0] + if candidate.content and candidate.content.parts: + for part in candidate.content.parts: + if part.inline_data and part.inline_data.mime_type.startswith("audio/"): + audio_data = part.inline_data.data + await self.stop_ttfb_metrics() + + # Gemini TTS returns PCM audio data, chunk it appropriately + CHUNK_SIZE = self.chunk_size + + for i in range(0, len(audio_data), CHUNK_SIZE): + chunk = audio_data[i : i + CHUNK_SIZE] + if not chunk: + break + frame = TTSAudioRawFrame(chunk, self.sample_rate, 1) + yield frame + + yield TTSStoppedFrame() + + except Exception as e: + logger.exception(f"{self} error generating TTS: {e}") + error_message = f"Gemini TTS generation error: {str(e)}" + yield ErrorFrame(error=error_message)