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

Merge pull request #2407 from pipecat-ai/mb/add-gemini-tts

Browse Source 
Add GeminiTTSService
This commit is contained in:
Mark Backman
2025-08-12 11:56:45 -07:00
committed by GitHub
parent 6aa80c0b8e e720573e60
commit 38af514d95
4 changed files with 434 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
CHANGELOG.md
View File

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

163
examples/foundational/07n-interruptible-gemini.py Normal file
View File

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

1
scripts/evals/run-release-evals.py
View File

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

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)