wx44wx/pipecat
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add GeminiTTSService

Browse Source
This commit is contained in:
Mark Backman
2025-08-10 07:48:17 -04:00
parent 827a70104d
commit 541a43905b
1 changed files with 263 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

264
src/pipecat/services/google/tts.py
View File

@@ -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)