a4375274b2
- Add dedicated Settings subclasses to 20 LLM services that were borrowing parent Settings classes (e.g. AzureLLMSettings, GroqLLMSettings) so users don't need cross-module imports - Fix field defaults to NOT_GIVEN in BaseWhisperSTTSettings, OpenAIRealtimeSTTSettings, and NvidiaSegmentedSTTSettings for delta-mode safety - Fix incomplete default_settings in AWS, Cartesia, ElevenLabs, Fish, and Whisper services so validate_complete() passes - Add auto-discovered tests that verify all Settings classes default to NOT_GIVEN (delta safety) and all services initialize with complete settings (store completeness)
272 lines
11 KiB
Python
272 lines
11 KiB
Python
#
|
|
# Copyright (c) 2024-2026, Daily
|
|
#
|
|
# SPDX-License-Identifier: BSD 2-Clause License
|
|
#
|
|
|
|
"""Service for accessing Gemini Live via Google Vertex AI.
|
|
|
|
This module provides integration with Google's Gemini Live model via
|
|
Vertex AI, supporting both text and audio modalities with voice transcription,
|
|
streaming responses, and tool usage.
|
|
"""
|
|
|
|
import json
|
|
from dataclasses import dataclass
|
|
from typing import List, Optional, Union
|
|
|
|
from loguru import logger
|
|
|
|
from pipecat.adapters.schemas.tools_schema import ToolsSchema
|
|
from pipecat.services.google.gemini_live.llm import (
|
|
GeminiLiveLLMService,
|
|
GeminiLiveLLMSettings,
|
|
GeminiMediaResolution,
|
|
GeminiModalities,
|
|
HttpOptions,
|
|
InputParams,
|
|
language_to_gemini_language,
|
|
)
|
|
from pipecat.services.settings import _warn_deprecated_param
|
|
|
|
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]`.")
|
|
raise Exception(f"Missing module: {e}")
|
|
|
|
|
|
@dataclass
|
|
class GeminiLiveVertexLLMSettings(GeminiLiveLLMSettings):
|
|
"""Settings for Gemini Live Vertex LLM service."""
|
|
|
|
pass
|
|
|
|
|
|
class GeminiLiveVertexLLMService(GeminiLiveLLMService):
|
|
"""Provides access to Google's Gemini Live model via Vertex AI.
|
|
|
|
This service enables real-time conversations with Gemini, 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,
|
|
project_id: str,
|
|
model: Optional[str] = None,
|
|
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,
|
|
settings: Optional[GeminiLiveVertexLLMSettings] = 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 service for accessing Gemini Live via Google Vertex AI.
|
|
|
|
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.
|
|
|
|
.. deprecated::
|
|
Use ``settings=GeminiLiveLLMSettings(model=...)`` instead.
|
|
|
|
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.
|
|
|
|
.. deprecated::
|
|
Use ``settings=GeminiLiveLLMSettings(...)`` instead.
|
|
|
|
settings: Gemini Live LLM settings. If provided together with deprecated
|
|
top-level parameters, the ``settings`` values take precedence.
|
|
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 GeminiLiveLLMService.
|
|
"""
|
|
# Check if user incorrectly passed api_key, which is used by parent
|
|
# class but not here.
|
|
if "api_key" in kwargs:
|
|
logger.error(
|
|
"GeminiLiveVertexLLMService 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
|
|
|
|
# Build default_settings from deprecated args, then apply settings delta.
|
|
# We pass settings= to super() instead of model=/params= to avoid
|
|
# double deprecation warnings from the parent.
|
|
|
|
# 1. Initialize default_settings with hardcoded defaults
|
|
default_settings = GeminiLiveVertexLLMSettings(
|
|
model="google/gemini-live-2.5-flash-native-audio",
|
|
frequency_penalty=None,
|
|
max_tokens=4096,
|
|
presence_penalty=None,
|
|
temperature=None,
|
|
top_k=None,
|
|
top_p=None,
|
|
seed=None,
|
|
filter_incomplete_user_turns=False,
|
|
user_turn_completion_config=None,
|
|
modalities=GeminiModalities.AUDIO,
|
|
language="en-US",
|
|
media_resolution=GeminiMediaResolution.UNSPECIFIED,
|
|
vad=None,
|
|
context_window_compression={},
|
|
thinking={},
|
|
enable_affective_dialog=False,
|
|
proactivity={},
|
|
extra={},
|
|
)
|
|
|
|
# 2. Apply direct init arg overrides (deprecated)
|
|
if model is not None:
|
|
_warn_deprecated_param("model", GeminiLiveVertexLLMSettings, "model")
|
|
default_settings.model = model
|
|
|
|
# 3. Apply params overrides — only if settings not provided
|
|
if params is not None:
|
|
_warn_deprecated_param("params", GeminiLiveVertexLLMSettings)
|
|
if not settings:
|
|
default_settings.frequency_penalty = params.frequency_penalty
|
|
default_settings.max_tokens = params.max_tokens
|
|
default_settings.presence_penalty = params.presence_penalty
|
|
default_settings.temperature = params.temperature
|
|
default_settings.top_k = params.top_k
|
|
default_settings.top_p = params.top_p
|
|
default_settings.modalities = params.modalities
|
|
default_settings.language = (
|
|
language_to_gemini_language(params.language) if params.language else "en-US"
|
|
)
|
|
default_settings.media_resolution = params.media_resolution
|
|
default_settings.vad = params.vad
|
|
default_settings.context_window_compression = (
|
|
params.context_window_compression.model_dump()
|
|
if params.context_window_compression
|
|
else {}
|
|
)
|
|
default_settings.thinking = params.thinking or {}
|
|
default_settings.enable_affective_dialog = params.enable_affective_dialog or False
|
|
default_settings.proactivity = params.proactivity or {}
|
|
if isinstance(params.extra, dict):
|
|
default_settings.extra = params.extra
|
|
|
|
# 4. Apply settings delta (canonical API, always wins)
|
|
if settings is not None:
|
|
default_settings.apply_update(settings)
|
|
|
|
# Call parent constructor with the obtained settings
|
|
super().__init__(
|
|
# api_key is required by parent class, but actually not used with
|
|
# Vertex
|
|
api_key="dummy",
|
|
voice_id=voice_id,
|
|
start_audio_paused=start_audio_paused,
|
|
start_video_paused=start_video_paused,
|
|
system_instruction=system_instruction,
|
|
tools=tools,
|
|
settings=default_settings,
|
|
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,
|
|
http_options=self._http_options,
|
|
)
|
|
|
|
@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
|