From 541a43905bf102f45369769e1077132b1e12928d Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 10 Aug 2025 07:48:17 -0400 Subject: [PATCH 1/2] 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) From e720573e602190ba03d57a3a28b9cd2ed24da395 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 10 Aug 2025 08:14:03 -0400 Subject: [PATCH 2/2] Added 07n-interruptible-gemini --- CHANGELOG.md | 7 + .../foundational/07n-interruptible-gemini.py | 163 ++++++++++++++++++ scripts/evals/run-release-evals.py | 1 + 3 files changed, 171 insertions(+) create mode 100644 examples/foundational/07n-interruptible-gemini.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 3f7bd7428..c6a93dff9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `GeminiTTSService` which uses Google Gemini to generate TTS output. The + Gemini model can be prompted to insert styled speech to control the TTS + output. + - Added Exotel support to Pipecat's development runner. You can now connect using the runner with `uv run bot.py -t exotel` and an ngrok connection to HTTP port 7860. @@ -76,6 +80,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 (e.g. `ParallelPipeline`) into a single processor so the main pipeline becomes simpler. +- Added `07n-interruptible-gemini.py`, demonstrating how to use + `GeminiTTSService`. + ## [0.0.79] - 2025-08-07 ### Changed diff --git a/examples/foundational/07n-interruptible-gemini.py b/examples/foundational/07n-interruptible-gemini.py new file mode 100644 index 000000000..0eae45754 --- /dev/null +++ b/examples/foundational/07n-interruptible-gemini.py @@ -0,0 +1,163 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +""" +A conversational AI bot using Gemini for both LLM and TTS. + +This example demonstrates how to use Gemini's TTS capabilities with the new +GeminiTTSService, which uses Gemini's TTS-specific models instead of Google Cloud TTS. + +Features showcased: +- Gemini LLM for conversation +- Gemini TTS with natural voice control +- Support for different voice personalities +- Style and tone control through natural language prompts + +Run with: + python examples/foundational/gemini-tts.py + +Make sure to set your environment variables: + export GOOGLE_API_KEY=your_api_key_here +""" + +import os + +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.google.llm import GoogleLLMService +from pipecat.services.google.stt import GoogleSTTService +from pipecat.services.google.tts import GeminiTTSService +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 with Gemini TTS") + + stt = GoogleSTTService( + params=GoogleSTTService.InputParams(languages=Language.EN_US), + credentials=os.getenv("GOOGLE_TEST_CREDENTIALS"), + ) + + tts = GeminiTTSService( + api_key=os.getenv("GOOGLE_API_KEY"), + model="gemini-2.5-flash-preview-tts", # TTS-specific model + voice_id="Charon", + params=GeminiTTSService.InputParams(language=Language.EN_US), + ) + + llm = GoogleLLMService( + api_key=os.getenv("GOOGLE_API_KEY"), + model="gemini-2.5-flash", + ) + + # System message that instructs the AI on how to speak + messages = [ + { + "role": "system", + "content": """You are a helpful AI assistant in a WebRTC call. Your goal is to demonstrate your capabilities in a succinct way. + + IMPORTANT: Since you're using Gemini TTS which supports natural voice control, you can include speaking instructions in your responses. For example: + - "Say cheerfully: Welcome to our conversation!" + - "Read this in a calm, professional tone: Here are the details you requested." + - "Speak in an excited whisper: I have some great news to share!" + - "Say slowly and clearly: Let me explain this step by step." + + Feel free to use natural language instructions to control your voice style, tone, pace, and emotion. The TTS system will interpret these instructions and adjust the speech accordingly. + + Your output will be converted to audio, so avoid 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, # STT + context_aggregator.user(), # User responses + llm, # LLM + tts, # Gemini 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 with a styled introduction + messages.append( + { + "role": "system", + "content": "Say cheerfully and warmly: Hello! I'm your AI assistant powered by Gemini's new TTS technology. I can speak with different voices, tones, and styles. How can I help you today?", + } + ) + 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/scripts/evals/run-release-evals.py b/scripts/evals/run-release-evals.py index 6b6ac7eda..54985fd16 100644 --- a/scripts/evals/run-release-evals.py +++ b/scripts/evals/run-release-evals.py @@ -68,6 +68,7 @@ TESTS_07 = [ ("07k-interruptible-lmnt.py", PROMPT_SIMPLE_MATH, None), ("07l-interruptible-groq.py", PROMPT_SIMPLE_MATH, None), ("07m-interruptible-aws.py", PROMPT_SIMPLE_MATH, None), + ("07n-interruptible-gemini.py", PROMPT_SIMPLE_MATH, None), ("07n-interruptible-google.py", PROMPT_SIMPLE_MATH, None), ("07o-interruptible-assemblyai.py", PROMPT_SIMPLE_MATH, None), ("07q-interruptible-rime.py", PROMPT_SIMPLE_MATH, None),