From e720573e602190ba03d57a3a28b9cd2ed24da395 Mon Sep 17 00:00:00 2001 From: Mark Backman Date: Sun, 10 Aug 2025 08:14:03 -0400 Subject: [PATCH] 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),