diff --git a/changelog/4078.changed.md b/changelog/4078.changed.md new file mode 100644 index 000000000..c0f4fb0a9 --- /dev/null +++ b/changelog/4078.changed.md @@ -0,0 +1 @@ +- Added Gemini 3 support to the Gemini Live service. diff --git a/examples/foundational/26-gemini-live.py b/examples/foundational/26-gemini-live.py index e8ca05407..90b5f0d13 100644 --- a/examples/foundational/26-gemini-live.py +++ b/examples/foundational/26-gemini-live.py @@ -4,6 +4,7 @@ # SPDX-License-Identifier: BSD 2-Clause License # + import os from dotenv import load_dotenv @@ -11,11 +12,17 @@ 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.frames.frames import LLMRunFrame 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.processors.aggregators.llm_context import LLMContext +from pipecat.processors.aggregators.llm_response_universal import ( + AssistantTurnStoppedMessage, + LLMContextAggregatorPair, + LLMUserAggregatorParams, + UserTurnStoppedMessage, +) from pipecat.runner.types import RunnerArguments from pipecat.runner.utils import create_transport from pipecat.services.google.gemini_live.llm import GeminiLiveLLMService @@ -23,7 +30,6 @@ 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) @@ -33,20 +39,14 @@ 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. ), } @@ -54,35 +54,44 @@ transport_params = { 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 + voice="Aoede", # Puck, Charon, Kore, Fenrir, Aoede + # system_instruction="Talk like a pirate." + ), + # inference_on_context_initialization=False, + ) + + context = LLMContext( + [ + { + "role": "user", + "content": "Say hello. Then ask if I want to hear a joke.", + }, + ], + ) + user_aggregator, assistant_aggregator = LLMContextAggregatorPair( + context, + user_params=LLMUserAggregatorParams( + # Set stop_secs to something roughly similar to the internal setting + # of the Multimodal Live api, just to align events. This doesn't + # really matter because we can only use the Multimodal Live API's + # phrase endpointing, for now. + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)) ), ) - vad_processor = VADProcessor(vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5))) - - # Build the pipeline pipeline = Pipeline( [ transport.input(), - vad_processor, + user_aggregator, llm, transport.output(), + assistant_aggregator, ] ) - # Configure the pipeline task task = PipelineTask( pipeline, params=PipelineParams( @@ -92,32 +101,31 @@ async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): 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.", - } - ] - ) - ] - ) + await task.queue_frames([LLMRunFrame()]) - # 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 + @user_aggregator.event_handler("on_user_turn_stopped") + async def on_user_turn_stopped(aggregator, strategy, message: UserTurnStoppedMessage): + timestamp = f"[{message.timestamp}] " if message.timestamp else "" + line = f"{timestamp}user: {message.content}" + logger.info(f"Transcript: {line}") + + @assistant_aggregator.event_handler("on_assistant_turn_stopped") + async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage): + timestamp = f"[{message.timestamp}] " if message.timestamp else "" + line = f"{timestamp}assistant: {message.content}" + logger.info(f"Transcript: {line}") + runner = PipelineRunner(handle_sigint=runner_args.handle_sigint) + await runner.run(task) diff --git a/examples/foundational/26a-gemini-live-transcription.py b/examples/foundational/26a-gemini-live-transcription.py deleted file mode 100644 index 5d9adbc7f..000000000 --- a/examples/foundational/26a-gemini-live-transcription.py +++ /dev/null @@ -1,141 +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 LLMRunFrame -from pipecat.pipeline.pipeline import Pipeline -from pipecat.pipeline.runner import PipelineRunner -from pipecat.pipeline.task import PipelineParams, PipelineTask -from pipecat.processors.aggregators.llm_context import LLMContext -from pipecat.processors.aggregators.llm_response_universal import ( - AssistantTurnStoppedMessage, - LLMContextAggregatorPair, - LLMUserAggregatorParams, - UserTurnStoppedMessage, -) -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_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, - ), - "twilio": lambda: FastAPIWebsocketParams( - audio_in_enabled=True, - audio_out_enabled=True, - ), - "webrtc": lambda: TransportParams( - audio_in_enabled=True, - audio_out_enabled=True, - ), -} - - -async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): - logger.info(f"Starting bot") - - llm = GeminiLiveLLMService( - api_key=os.getenv("GOOGLE_API_KEY"), - settings=GeminiLiveLLMService.Settings( - voice="Aoede", # Puck, Charon, Kore, Fenrir, Aoede - # system_instruction="Talk like a pirate." - # inference_on_context_initialization=False, - ), - ) - - context = LLMContext( - [ - { - "role": "user", - "content": "Say hello. Then ask if I want to hear a joke.", - }, - ], - ) - user_aggregator, assistant_aggregator = LLMContextAggregatorPair( - context, - user_params=LLMUserAggregatorParams( - # Set stop_secs to something roughly similar to the internal setting - # of the Multimodal Live api, just to align events. This doesn't - # really matter because we can only use the Multimodal Live API's - # phrase endpointing, for now. - vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)) - ), - ) - - pipeline = Pipeline( - [ - transport.input(), - user_aggregator, - llm, - transport.output(), - assistant_aggregator, - ] - ) - - 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. - await task.queue_frames([LLMRunFrame()]) - - @transport.event_handler("on_client_disconnected") - async def on_client_disconnected(transport, client): - logger.info(f"Client disconnected") - await task.cancel() - - @user_aggregator.event_handler("on_user_turn_stopped") - async def on_user_turn_stopped(aggregator, strategy, message: UserTurnStoppedMessage): - timestamp = f"[{message.timestamp}] " if message.timestamp else "" - line = f"{timestamp}user: {message.content}" - logger.info(f"Transcript: {line}") - - @assistant_aggregator.event_handler("on_assistant_turn_stopped") - async def on_assistant_turn_stopped(aggregator, message: AssistantTurnStoppedMessage): - timestamp = f"[{message.timestamp}] " if message.timestamp else "" - line = f"{timestamp}assistant: {message.content}" - logger.info(f"Transcript: {line}") - - 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/pyproject.toml b/pyproject.toml index 0416d6367..595571878 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -70,7 +70,7 @@ fal = [] fireworks = [] fish = [ "ormsgpack>=1.7.0,<2", "pipecat-ai[websockets-base]" ] gladia = [ "pipecat-ai[websockets-base]" ] -google = [ "google-cloud-speech>=2.33.0,<3", "google-cloud-texttospeech>=2.31.0,<3", "google-genai>=1.57.0,<2", "pipecat-ai[websockets-base]" ] +google = [ "google-cloud-speech>=2.33.0,<3", "google-cloud-texttospeech>=2.31.0,<3", "google-genai>=1.68.0,<2", "pipecat-ai[websockets-base]" ] gradium = [ "pipecat-ai[websockets-base]" ] grok = [] groq = [ "groq>=0.23.0,<2" ] diff --git a/scripts/evals/run-release-evals.py b/scripts/evals/run-release-evals.py index 0fe66d9a1..f2be2179b 100644 --- a/scripts/evals/run-release-evals.py +++ b/scripts/evals/run-release-evals.py @@ -228,7 +228,6 @@ TESTS_22 = [ TESTS_26 = [ ("26-gemini-live.py", EVAL_SIMPLE_MATH), - ("26a-gemini-live-transcription.py", EVAL_SIMPLE_MATH), ("26b-gemini-live-function-calling.py", EVAL_WEATHER), ("26c-gemini-live-video.py", EVAL_VISION_CAMERA), ("26e-gemini-live-google-search.py", EVAL_ONLINE_SEARCH), diff --git a/src/pipecat/services/google/gemini_live/llm.py b/src/pipecat/services/google/gemini_live/llm.py index 5c9e5f8b3..18b2c2e8a 100644 --- a/src/pipecat/services/google/gemini_live/llm.py +++ b/src/pipecat/services/google/gemini_live/llm.py @@ -98,6 +98,7 @@ try: FunctionResponse, GenerationConfig, GroundingMetadata, + HistoryConfig, HttpOptions, LiveConnectConfig, LiveServerMessage, @@ -648,6 +649,11 @@ class GeminiLiveLLMService(LLMService): # Overriding the default adapter to use the Gemini one. adapter_class = GeminiLLMAdapter + @property + def _is_gemini_3(self) -> bool: + """Check if the current model is a Gemini 3.x model.""" + return "gemini-3" in (self._settings.model or "") + def __init__( self, *, @@ -791,7 +797,7 @@ class GeminiLiveLLMService(LLMService): self._system_instruction_from_init = system_instruction self._tools_from_init = tools self._inference_on_context_initialization = inference_on_context_initialization - self._needs_turn_complete_message = False + self._needs_initial_turn_complete_message = False self._audio_input_paused = start_audio_paused self._video_input_paused = start_video_paused @@ -993,8 +999,8 @@ class GeminiLiveLLMService(LLMService): self._user_is_speaking = False self._user_audio_buffer = bytearray() await self.start_ttfb_metrics() - if self._needs_turn_complete_message: - self._needs_turn_complete_message = False + if self._needs_initial_turn_complete_message: + self._needs_initial_turn_complete_message = False # NOTE: without this, the model ignores the context it's been # seeded with before the user started speaking await self._session.send_client_content(turn_complete=True) @@ -1056,9 +1062,10 @@ class GeminiLiveLLMService(LLMService): elif isinstance(frame, LLMMessagesAppendFrame): # NOTE: handling LLMMessagesAppendFrame here in the LLMService is # unusual - typically this would be handled in the user context - # aggregator. Leaving this handling here so that user code that - # uses this frame *without* a user context aggregator still works - # (we have an example that does just that, actually). + # aggregator. Leaving this handling here so that legacy user code + # that uses this frame *without* a user context aggregator to kick + # off a conversation still works (we used to have an example that + # did that). await self._create_single_response(frame.messages) elif isinstance(frame, LLMSetToolsFrame): # TODO: implement runtime tool updates for Gemini Live. @@ -1204,6 +1211,7 @@ class GeminiLiveLLMService(LLMService): input_audio_transcription=AudioTranscriptionConfig(), output_audio_transcription=AudioTranscriptionConfig(), session_resumption=SessionResumptionConfig(handle=session_resumption_handle), + history_config=HistoryConfig(initial_history_in_client_content=True), ) # Add context window compression to configuration, if enabled @@ -1508,17 +1516,26 @@ class GeminiLiveLLMService(LLMService): await self._session.send_client_content( turns=messages, turn_complete=self._inference_on_context_initialization ) + # Gemini 3.x wants turn_complete=True, but also won't run inference without a realtime input + if self._is_gemini_3 and self._inference_on_context_initialization: + await self._session.send_realtime_input(text=" ") except Exception as e: await self._handle_send_error(e) # If we're generating a response right away upon initializing - # conversation history, set a flag saying that we need a turn complete - # message when the user stops speaking. - if not self._inference_on_context_initialization: - self._needs_turn_complete_message = True + # conversation history, set a flag saying that we'll need a turn + # complete message when the user stops speaking. + # This is a quirky workaround, and not one that Gemini 3 needs. + if not self._inference_on_context_initialization and not self._is_gemini_3: + self._needs_initial_turn_complete_message = True async def _create_single_response(self, messages_list): - """Create a single response from a list of messages.""" + """Create a single response from a list of messages. + + This is only here to support the very specific 'legacy' scenario of + kicking off a conversation using LLMMessagesAppendFrame when there's no + context aggregators in the pipeline (see process_frame for more details). + """ if self._disconnecting or not self._session: return @@ -1537,6 +1554,9 @@ class GeminiLiveLLMService(LLMService): try: await self._session.send_client_content(turns=messages, turn_complete=True) + # Gemini 3.x wants turn_complete=True, but also won't run inference without a realtime input + if self._is_gemini_3: + await self._session.send_realtime_input(text=" ") except Exception as e: await self._handle_send_error(e) diff --git a/uv.lock b/uv.lock index 0089729ef..cd04806a0 100644 --- a/uv.lock +++ b/uv.lock @@ -4877,7 +4877,7 @@ requires-dist = [ { name = "faster-whisper", marker = "extra == 'whisper'", specifier = "~=1.2.1" }, { name = "google-cloud-speech", marker = "extra == 'google'", specifier = ">=2.33.0,<3" }, { name = "google-cloud-texttospeech", marker = "extra == 'google'", specifier = ">=2.31.0,<3" }, - { name = "google-genai", marker = "extra == 'google'", specifier = ">=1.57.0,<2" }, + { name = "google-genai", marker = "extra == 'google'", specifier = ">=1.68.0,<2" }, { name = "groq", marker = "extra == 'groq'", specifier = ">=0.23.0,<2" }, { name = "hume", marker = "extra == 'hume'", specifier = ">=0.11.2,<1" }, { name = "kokoro-onnx", marker = "extra == 'kokoro'", specifier = ">=0.5.0,<1" },