diff --git a/CHANGELOG.md b/CHANGELOG.md index 10cd8738f..6999bd37c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Added +- Added `GeminiVertexMultimodalLiveLLMService`, for accessing Gemini Live via + Google Vertex AI. + - Added some new configuration options to `GeminiMultimodalLiveLLMService`: - `thinking` diff --git a/examples/foundational/26h-gemini-multimodal-live-vertex.py b/examples/foundational/26h-gemini-multimodal-live-vertex.py new file mode 100644 index 000000000..581d2524a --- /dev/null +++ b/examples/foundational/26h-gemini-multimodal-live-vertex.py @@ -0,0 +1,134 @@ +# +# Copyright (c) 2024–2025, 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.runner.types import RunnerArguments +from pipecat.runner.utils import create_transport +from pipecat.services.gemini_multimodal_live.gemini import GeminiMultimodalLiveLLMService +from pipecat.services.gemini_multimodal_live.vertex import GeminiVertexMultimodalLiveLLMService +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 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, + # set stop_secs to something roughly similar to the internal setting + # of the Multimodal Live api, just to align events. + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), + ), + "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. + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), + ), + "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. + vad_analyzer=SileroVADAnalyzer(params=VADParams(stop_secs=0.5)), + ), +} + + +async def run_bot(transport: BaseTransport, runner_args: RunnerArguments): + logger.info(f"Starting bot") + + # Create the Gemini Vertex 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 converted to audio so don't include special characters in your answers. + Respond to what the user said in a creative and helpful way. + """ + + llm = GeminiVertexMultimodalLiveLLMService( + credentials=os.getenv("GOOGLE_VERTEX_TEST_CREDENTIALS"), + project_id=os.getenv("GOOGLE_CLOUD_PROJECT_ID"), + location=os.getenv("GOOGLE_CLOUD_LOCATION"), + system_instruction=system_instruction, + voice_id="Puck", # Aoede, Charon, Fenrir, Kore, Puck + ) + + # Build the pipeline + pipeline = Pipeline( + [ + transport.input(), + 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() diff --git a/src/pipecat/services/gemini_multimodal_live/gemini.py b/src/pipecat/services/gemini_multimodal_live/gemini.py index 65b5eb781..0de76fbf1 100644 --- a/src/pipecat/services/gemini_multimodal_live/gemini.py +++ b/src/pipecat/services/gemini_multimodal_live/gemini.py @@ -601,7 +601,8 @@ class GeminiMultimodalLiveLLMService(LLMService): self._audio_input_paused = start_audio_paused self._video_input_paused = start_video_paused self._context = None - self._create_client(api_key, http_options) + self._api_key = api_key + self._http_options = http_options self._session: AsyncSession = None self._connection_task = None @@ -649,8 +650,8 @@ class GeminiMultimodalLiveLLMService(LLMService): "extra": params.extra if isinstance(params.extra, dict) else {}, } - # Initialize the File API client - self.file_api = GeminiFileAPI(api_key=api_key, base_url=file_api_base_url) + self._file_api_base_url = file_api_base_url + self._file_api: Optional[GeminiFileAPI] = None # Grounding metadata tracking self._search_result_buffer = "" @@ -662,8 +663,23 @@ class GeminiMultimodalLiveLLMService(LLMService): # Bookkeeping for ending gracefully (i.e. after the bot is finished) self._end_frame_pending_bot_turn_finished: Optional[EndFrame] = None - def _create_client(self, api_key: str, http_options: Optional[HttpOptions] = None): - self._client = Client(api_key=api_key, http_options=http_options) + # Initialize the API client. Subclasses can override this if needed. + self.create_client() + + def create_client(self): + """Create the Gemini API client instance. Subclasses can override this.""" + self._client = Client(api_key=self._api_key, http_options=self._http_options) + + @property + def file_api(self) -> GeminiFileAPI: + """Get the Gemini File API client instance. Subclasses can override this. + + Returns: + The Gemini File API client. + """ + if not self._file_api: + self._file_api = GeminiFileAPI(api_key=self._api_key, base_url=self._file_api_base_url) + return self._file_api def can_generate_metrics(self) -> bool: """Check if the service can generate usage metrics. @@ -1282,7 +1298,21 @@ class GeminiMultimodalLiveLLMService(LLMService): inline_data = part.inline_data if not inline_data: return - if inline_data.mime_type != f"audio/pcm;rate={self._sample_rate}": + + # Check if mime type matches expected format + expected_mime_type = f"audio/pcm;rate={self._sample_rate}" + if inline_data.mime_type == expected_mime_type: + # Perfect match, continue processing + pass + elif inline_data.mime_type == "audio/pcm": + # Sample rate not provided in mime type, assume default + if not hasattr(self, "_sample_rate_warning_logged"): + logger.warning( + f"Sample rate not provided in mime type '{inline_data.mime_type}', assuming rate of {self._sample_rate}" + ) + self._sample_rate_warning_logged = True + else: + # Unrecognized format logger.warning(f"Unrecognized server_content format {inline_data.mime_type}") return diff --git a/src/pipecat/services/gemini_multimodal_live/vertex.py b/src/pipecat/services/gemini_multimodal_live/vertex.py new file mode 100644 index 000000000..e07eb474f --- /dev/null +++ b/src/pipecat/services/gemini_multimodal_live/vertex.py @@ -0,0 +1,187 @@ +# +# Copyright (c) 2024–2025, Daily +# +# SPDX-License-Identifier: BSD 2-Clause License +# + +"""Google Vertex AI Gemini Multimodal Live service. + +This module provides integration with Google's Gemini Multimodal Live model via +Vertex AI, supporting both text and audio modalities with voice transcription, +streaming responses, and tool usage. +""" + +import json +import os +from typing import List, Optional, Union + +from loguru import logger + +from pipecat.adapters.schemas.tools_schema import ToolsSchema +from pipecat.services.gemini_multimodal_live.gemini import ( + GeminiMultimodalLiveLLMService, + HttpOptions, + InputParams, +) + +try: + from google.auth import default + from google.auth.exceptions import GoogleAuthError + from google.auth.transport.requests import Request + from google.genai import Client + from google.oauth2 import service_account + +except ModuleNotFoundError as e: + logger.error(f"Exception: {e}") + logger.error( + "In order to use Google Vertex AI, you need to `pip install pipecat-ai[google]`. Also, set up your Google credentials properly." + ) + raise Exception(f"Missing module: {e}") + + +class GeminiVertexMultimodalLiveLLMService(GeminiMultimodalLiveLLMService): + """Google Vertex AI Gemini Multimodal Live service. + + Provides access to Google's Gemini Multimodal Live model via Vertex AI, + supporting both text and audio modalities. It handles voice transcription, + streaming audio responses, and tool usage. + """ + + def __init__( + self, + *, + credentials: Optional[str] = None, + credentials_path: Optional[str] = None, + location: str = "us-east4", + project_id: str, + model="google/gemini-2.0-flash-live-preview-04-09", + voice_id: str = "Charon", + start_audio_paused: bool = False, + start_video_paused: bool = False, + system_instruction: Optional[str] = None, + tools: Optional[Union[List[dict], ToolsSchema]] = None, + params: Optional[InputParams] = None, + inference_on_context_initialization: bool = True, + file_api_base_url: str = "https://generativelanguage.googleapis.com/v1beta/files", + http_options: Optional[HttpOptions] = None, + **kwargs, + ): + """Initialize the Google Vertex AI Gemini Multimodal Live service. + + Args: + credentials: JSON string of service account credentials. + credentials_path: Path to the service account JSON file. + location: GCP region for Vertex AI endpoint (e.g., "us-east4"). + project_id: Google Cloud project ID. + model: Model identifier to use. Defaults to "models/gemini-2.0-flash-live-preview-04-09". + voice_id: TTS voice identifier. Defaults to "Charon". + start_audio_paused: Whether to start with audio input paused. Defaults to False. + start_video_paused: Whether to start with video input paused. Defaults to False. + system_instruction: System prompt for the model. Defaults to None. + tools: Tools/functions available to the model. Defaults to None. + params: Configuration parameters for the model along with Vertex AI + location and project ID. + inference_on_context_initialization: Whether to generate a response when context + is first set. Defaults to True. + file_api_base_url: Base URL for the Gemini File API. Defaults to the official endpoint. + http_options: HTTP options for the client. + **kwargs: Additional arguments passed to parent GeminiMultimodalLiveLLMService. + """ + # Check if user incorrectly passed api_key, which is used by parent + # class but not here. + if "api_key" in kwargs: + logger.error( + "GeminiVertexMultimodalLiveLLMService does not accept 'api_key' parameter. " + "Use 'credentials' or 'credentials_path' instead for Vertex AI authentication." + ) + raise ValueError( + "Invalid parameter 'api_key'. Use 'credentials' or 'credentials_path' for Vertex AI authentication." + ) + + # These need to be set before calling super().__init__() because + # super().__init__() invokes create_client(), which needs these. + self._credentials = self._get_credentials(credentials, credentials_path) + self._project_id = project_id + self._location = location + + # Call parent constructor with the obtained API key + super().__init__( + # api_key is required by parent class, but actually not used with + # Vertex + api_key="dummy", + model=model, + voice_id=voice_id, + start_audio_paused=start_audio_paused, + start_video_paused=start_video_paused, + system_instruction=system_instruction, + tools=tools, + params=params, + inference_on_context_initialization=inference_on_context_initialization, + file_api_base_url=file_api_base_url, + http_options=http_options, + **kwargs, + ) + + def create_client(self): + """Create the Gemini client instance.""" + self._client = Client( + vertexai=True, + credentials=self._credentials, + project=self._project_id, + location=self._location, + ) + + @property + def file_api(self): + """Gemini File API is not supported with Vertex AI.""" + raise NotImplementedError( + "When using Vertex AI, the recommended approach is to use Google Cloud Storage for file handling. The Gemini File API is not directly supported in this context." + ) + + @staticmethod + def _get_credentials(credentials: Optional[str], credentials_path: Optional[str]) -> str: + """Retrieve Credentials using Google service account credentials JSON. + + Supports multiple authentication methods: + 1. Direct JSON credentials string + 2. Path to service account JSON file + 3. Default application credentials (ADC) + + Args: + credentials: JSON string of service account credentials. + credentials_path: Path to the service account JSON file. + + Returns: + OAuth token for API authentication. + + Raises: + ValueError: If no valid credentials are provided or found. + """ + creds: Optional[service_account.Credentials] = None + + if credentials: + # Parse and load credentials from JSON string + creds = service_account.Credentials.from_service_account_info( + json.loads(credentials), + scopes=["https://www.googleapis.com/auth/cloud-platform"], + ) + elif credentials_path: + # Load credentials from JSON file + creds = service_account.Credentials.from_service_account_file( + credentials_path, + scopes=["https://www.googleapis.com/auth/cloud-platform"], + ) + else: + try: + creds, project_id = default( + scopes=["https://www.googleapis.com/auth/cloud-platform"] + ) + except GoogleAuthError: + pass + + if not creds: + raise ValueError("No valid credentials provided.") + + creds.refresh(Request()) # Ensure token is up-to-date, lifetime is 1 hour. + + return creds