Merge pull request #4078 from pipecat-ai/cb/gemini-updates

Updates for Gemini Live

changelog/4078.changed.md

@@ -0,0 +1 @@
+- Added Gemini 3 support to the Gemini Live service.

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="Aoede",  # Puck, Charon, Kore, Fenrir, Aoede
-            voice="Puck",  # Aoede, Charon, Fenrir, Kore, Puck
+            # 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(
+        await task.queue_frames([LLMRunFrame()])
-            [
-                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
+    @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)
 
 

examples/foundational/26a-gemini-live-transcription.py

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

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" ]

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

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:
+        if self._needs_initial_turn_complete_message:
-            self._needs_turn_complete_message = False
+            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
+            # aggregator. Leaving this handling here so that legacy user code
-            # uses this frame *without* a user context aggregator still works
+            # that uses this frame *without* a user context aggregator to kick
-            # (we have an example that does just that, actually).
+            # 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
+        # conversation history, set a flag saying that we'll need a turn
-        # message when the user stops speaking.
+        # complete message when the user stops speaking.
-        if not self._inference_on_context_initialization:
+        # This is a quirky workaround, and not one that Gemini 3 needs.
-            self._needs_turn_complete_message = True
+        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)
 

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" },