diff --git a/examples/foundational/26a-gemini-live-transcription.py b/examples/foundational/26-gemini-live-transcription.py similarity index 100% rename from examples/foundational/26a-gemini-live-transcription.py rename to examples/foundational/26-gemini-live-transcription.py diff --git a/examples/foundational/26-gemini-live.py b/examples/foundational/26-gemini-live.py deleted file mode 100644 index e8ca05407..000000000 --- a/examples/foundational/26-gemini-live.py +++ /dev/null @@ -1,133 +0,0 @@ -# -# Copyright (c) 2024-2026, Daily -# -# SPDX-License-Identifier: BSD 2-Clause License -# - -import os - -from dotenv import load_dotenv -from loguru import logger - -from pipecat.audio.vad.silero import SileroVADAnalyzer -from pipecat.audio.vad.vad_analyzer import VADParams -from pipecat.frames.frames import LLMMessagesAppendFrame -from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.audio.vad_processor import VADProcessor -from pipecat.runner.types import RunnerArguments -from pipecat.runner.utils import create_transport -from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService -from pipecat.transports.base_transport import BaseTransport, TransportParams -from pipecat.transports.daily.transport import DailyParams -from pipecat.transports.websocket.fastapi import FastAPIWebsocketParams - -# Load environment variables -load_dotenv(override=True) - - -# We use lambdas to defer transport parameter creation until the transport -# type is selected at runtime. -transport_params = { - "daily": lambda: DailyParams( - audio_in_enabled=True, - audio_out_enabled=True, - # set stop_secs to something roughly similar to the internal setting - # of the Multimodal Live api, just to align events. - ), - "twilio": lambda: FastAPIWebsocketParams( - audio_in_enabled=True, - audio_out_enabled=True, - # set stop_secs to something roughly similar to the internal setting - # of the Multimodal Live api, just to align events. - ), - "webrtc": lambda: TransportParams( - audio_in_enabled=True, - audio_out_enabled=True, - # set stop_secs to something roughly similar to the internal setting - # of the Multimodal Live api, just to align events. - ), -} - - -async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): - logger.info(f"Starting bot") - - # Create the Gemini Multimodal Live LLM service - system_instruction = f""" - You are a helpful AI assistant. - Your goal is to demonstrate your capabilities in a helpful and engaging way. - Your output will be spoken aloud, so avoid special characters that can't easily be spoken, such as emojis or bullet points. - Respond to what the user said in a creative and helpful way. - """ - - llm = GeminiLiveLLMService( - api_key=os.getenv("GOOGLE_API_KEY"), - settings=GeminiLiveLLMService.Settings( - system_instruction=system_instruction, - voice="Puck", # Aoede, Charon, Fenrir, Kore, Puck - ), - ) - - vad_processor = VADProcessor(vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5))) - - # Build the pipeline - pipeline = Pipeline( - [ - transport.input(), - vad_processor, - llm, - transport.output(), - ] - ) - - # Configure the pipeline task - task = PipelineTask( - pipeline, - params=PipelineParams( - enable_metrics=True, - enable_usage_metrics=True, - ), - idle_timeout_secs=runner_args.pipeline_idle_timeout_secs, - ) - - # Handle client connection event - @transport.event_handler("on_client_connected") - async def on_client_connected(transport, client): - logger.info(f"Client connected") - # Kick off the conversation. - await task.queue_frames( - [ - LLMMessagesAppendFrame( - messages=[ - { - "role": "user", - "content": f"Greet the user and introduce yourself.", - } - ] - ) - ] - ) - - # Handle client disconnection events - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") - await task.cancel() - - # Run the pipeline - 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()