feat(vertex-ai): support gemini 3.1 flash tts

This commit is contained in:
Emerson Gomes 2026-09-15 12:12:58 -05:00
parent 238f434153
commit a5403573ca
No known key found for this signature in database
GPG key ID: D3DF28AB5D1B5E17
20 changed files with 1271 additions and 188 deletions

View file

@ -288,6 +288,7 @@ def string_key_schemas(modes: tuple) -> dict[str, JsonSchema]:
"type": "string",
"enum": ["lyria_predict", "lyria_interactions"],
},
"health_check_voice": STRING,
}

View file

@ -30,6 +30,11 @@ class ChatAudioParam(TypedDict):
format: ReadOnly[NotRequired[str]]
class ChatSpeechConfigAudioParam(TypedDict):
speech_config: ReadOnly[Mapping[str, object]]
format: ReadOnly[str]
class SpeechToCompletionBridgeTransformationHandler:
def _validate_response_format(
self, model: str, custom_llm_provider: str, optional_params: Mapping[str, object]
@ -68,7 +73,17 @@ class SpeechToCompletionBridgeTransformationHandler:
def _chat_audio_param(
self, model: str, voice: str | Mapping[str, object] | None, optional_params: Mapping[str, object]
) -> ChatAudioParam | None:
) -> ChatAudioParam | ChatSpeechConfigAudioParam | None:
if isinstance(voice, Mapping) and self._is_gemini_tts_model(model):
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
normalize_gemini_speech_config,
)
speech_audio: Final[ChatSpeechConfigAudioParam] = {
"speech_config": normalize_gemini_speech_config(voice),
"format": GEMINI_TTS_CHAT_AUDIO_FORMAT,
}
return speech_audio
if not isinstance(voice, str):
return None
audio_format: Final = self._chat_audio_format(model, optional_params)
@ -145,8 +160,9 @@ class SpeechToCompletionBridgeTransformationHandler:
return wav_header + pcm_data
def _is_gemini_tts_model(self, model: str) -> bool:
"""Check if the model is a Gemini TTS model that returns PCM16 data."""
return "gemini" in model.lower() and ("tts" in model.lower() or "preview-tts" in model.lower())
from litellm.utils import is_gemini_tts_model
return is_gemini_tts_model(model)
def _gemini_tts_response_body(self, decoded_audio: bytes, response_format: str | None) -> tuple[bytes, str]:
if response_format == GEMINI_TTS_RAW_RESPONSE_FORMAT:

View file

@ -16,7 +16,10 @@ from litellm.llms.vertex_ai.common_utils import (
_build_vertex_schema,
supports_response_json_schema,
)
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import VertexLLM
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
VertexLLM,
normalize_gemini_speech_config,
)
from litellm.types.router import GenericLiteLLMParams
if TYPE_CHECKING:
@ -140,7 +143,10 @@ class GoogleGenAIConfig(BaseGoogleGenAIGenerateContentConfig, VertexLLM):
if is_supported:
# Always output in camelCase for Google GenAI API
output_key = param_camel if param != param_camel else param
_generate_content_config_dict[output_key] = value
if param_snake == "speech_config" and isinstance(value, dict):
_generate_content_config_dict[output_key] = normalize_gemini_speech_config(value)
else:
_generate_content_config_dict[output_key] = value
return _generate_content_config_dict
def validate_environment(

View file

@ -7,6 +7,7 @@ import time
from collections.abc import Callable, Mapping, Sequence
from copy import deepcopy
from functools import partial
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, Optional, Union, cast, get_args
import httpx
@ -133,6 +134,73 @@ def _served_model_name(model_version: object) -> str | None:
return model_version.split("@", 1)[0]
_GEMINI_SPEECH_CONFIG_KEY_MAP: Final[Mapping[str, str]] = MappingProxyType(
{
"multi_speaker_voice_config": "multiSpeakerVoiceConfig",
"speaker_voice_configs": "speakerVoiceConfigs",
"voice_config": "voiceConfig",
"prebuilt_voice_config": "prebuiltVoiceConfig",
"voice_name": "voiceName",
"language_code": "languageCode",
}
)
def _normalize_gemini_speech_config_item(value: object) -> object:
if isinstance(value, Mapping):
normalized: ( # mutable-ok: the iterative walk fills exactly one normalized provider payload
dict[str, object] | list[object]
) = {} # mutable-ok: the iterative walk fills the normalized tree; rebind-ok: each shape branch initializes the shared result
elif isinstance(value, list):
normalized = [] # mutable-ok: the iterative walk fills the normalized tree; rebind-ok: each shape branch initializes the shared result
else:
return value
pending: Final[ # mutable-ok: the explicit worklist must grow while avoiding recursive traversal
list[tuple[dict[str, object] | list[object], Mapping[object, object] | list[object]]]
] = [(normalized, value)]
while pending:
target, source = pending.pop()
items = source.items() if isinstance(source, Mapping) else enumerate(source)
for key, item in items:
if isinstance(source, Mapping):
if not isinstance(key, str) or not isinstance(target, dict):
continue
normalized_key: str | int = _GEMINI_SPEECH_CONFIG_KEY_MAP.get(key, key)
else:
if not isinstance(target, list):
continue
normalized_key = len(target)
child: object
if isinstance(item, Mapping):
child = {} # mutable-ok: nested mappings are filled by the worklist
elif isinstance(item, list):
child = [] # mutable-ok: nested lists are filled by the worklist
else:
child = item
if isinstance(target, dict) and isinstance(normalized_key, str):
target[normalized_key] = child
elif isinstance(target, list):
target.append(child)
if isinstance(child, (dict, list)) and isinstance(item, (Mapping, list)):
pending.append((child, item))
return normalized
def normalize_gemini_speech_config(
value: Mapping[str, object],
) -> dict[str, object]: # mutable-ok: provider request serialization requires a concrete dict
normalized_value: Final = _normalize_gemini_speech_config_item(value)
if isinstance(normalized_value, dict):
return cast( # cast-ok: the runtime dict check establishes the normalizer's concrete result shape
dict[str, object], normalized_value
)
return {} # mutable-ok: provider request serialization requires a concrete empty dict
class VertexAIBaseConfig:
def get_mapped_special_auth_params(self) -> dict:
"""
@ -1032,12 +1100,6 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
languageCode: "en-US",
}
"""
from litellm.types.llms.vertex_ai import (
PrebuiltVoiceConfig,
SpeechConfig,
VoiceConfig,
)
# Validate audio format - Gemini TTS only supports pcm16
audio_format: Final = value.get("format")
if audio_format is not None and audio_format != "pcm16":
@ -1047,18 +1109,42 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
f"Please set audio format to 'pcm16'."
)
# Map OpenAI audio parameter to Gemini speech config
speech_config: Final[SpeechConfig] = {}
camel_speech_config: Final = value.get("speechConfig")
snake_speech_config: Final = value.get("speech_config")
speech_config_value: Final = (
camel_speech_config if isinstance(camel_speech_config, Mapping) else snake_speech_config
)
language_code: Final = value.get("language_code", value.get("languageCode"))
if isinstance(speech_config_value, Mapping):
speech_config: Final = normalize_gemini_speech_config(speech_config_value)
if language_code is not None and "languageCode" not in speech_config:
return { # mutable-ok: provider request serialization requires a concrete dict
**speech_config,
"languageCode": language_code,
}
return speech_config
if "voice" in value:
prebuilt_voice_config: Final[PrebuiltVoiceConfig] = {"voiceName": value["voice"]}
voice_config: Final[VoiceConfig] = {"prebuiltVoiceConfig": prebuilt_voice_config}
speech_config["voiceConfig"] = voice_config
speech_config_without_language: Final[ # mutable-ok: provider request serialization requires a concrete dict
dict[str, object] # mutable-ok: provider request serialization requires a concrete dict
] = (
{ # mutable-ok: provider request serialization requires a concrete dict
"voiceConfig": { # mutable-ok: nested provider payload is serialized as a dict
"prebuiltVoiceConfig": { # mutable-ok: nested provider payload is serialized as a dict
"voiceName": value["voice"],
}
}
}
if "voice" in value
else {} # mutable-ok: provider request serialization requires a concrete empty dict
)
if "language_code" in value:
speech_config["languageCode"] = value["language_code"]
if language_code is not None:
return { # mutable-ok: provider request serialization requires a concrete dict
**speech_config_without_language,
"languageCode": language_code,
}
return cast(dict, speech_config)
return speech_config_without_language
@staticmethod
def _apply_include_server_side_tool_invocations(
@ -1225,7 +1311,10 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
optional_params["responseModalities"].append("AUDIO")
# Set default temperature to 1.0 for Gemini 3 models if not specified
if VertexGeminiConfig._is_gemini_3_or_newer(model):
if VertexGeminiConfig._is_gemini_3_or_newer(model) and not litellm.utils.is_gemini_tts_model(
model,
custom_llm_provider="vertex_ai",
):
if "temperature" not in optional_params:
optional_params["temperature"] = 1.0

View file

@ -38,34 +38,15 @@ class VertexAIGoogleGenAIConfig(GoogleGenAIConfig):
return default_headers
def _camel_to_snake(self, camel_str: str) -> str:
"""Convert camelCase to snake_case"""
import re
return re.sub(r"(?<!^)(?=[A-Z])", "_", camel_str).lower()
def map_generate_content_optional_params(
self,
generate_content_config_dict,
generate_content_config_dict: dict[str, Any], # mutable-ok: inherited provider interface accepts a request dict
model: str,
):
"""
Map Google GenAI parameters to provider-specific format.
Args:
generate_content_optional_params: Optional parameters for generate content
model: The model name
Returns:
Mapped parameters for the provider
"""
_generate_content_config_dict: Final[dict] = {}
for param, value in generate_content_config_dict.items():
camel_case_key = self._camel_to_snake(param)
_generate_content_config_dict[camel_case_key] = value
return _generate_content_config_dict
) -> dict[str, Any]: # mutable-ok: inherited provider interface returns a request dict
return super().map_generate_content_optional_params(
generate_content_config_dict=generate_content_config_dict,
model=model,
)
def transform_generate_content_request(
self,

View file

@ -6,9 +6,9 @@ Reference: https://cloud.google.com/text-to-speech/docs/reference/rest/v1/text/s
"""
import base64
from collections.abc import Coroutine
from collections.abc import Coroutine, Mapping
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, TypeAlias, Union
from typing import TYPE_CHECKING, Any, ClassVar, Final, TypeAlias, Union
import httpx
@ -78,14 +78,27 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
}
# Response format mappings from OpenAI to Google Cloud audio encoding
FORMAT_MAPPINGS = {
"mp3": "MP3",
"opus": "OGG_OPUS",
"aac": "MP3", # Google doesn't have AAC, use MP3
"flac": "FLAC",
"wav": "LINEAR16",
"pcm": "LINEAR16",
}
FORMAT_MAPPINGS: ClassVar[Mapping[str, str]] = MappingProxyType(
{
"mp3": "MP3",
"opus": "OGG_OPUS",
"aac": "MP3", # Google doesn't have AAC, use MP3
"flac": "FLAC",
"wav": "LINEAR16",
"pcm": "LINEAR16",
}
)
GEMINI_FORMAT_MAPPINGS: ClassVar[Mapping[str, str]] = MappingProxyType(
{
**FORMAT_MAPPINGS,
"alaw": "ALAW",
"mulaw": "MULAW",
"ogg_opus": "OGG_OPUS",
"pcm": "LINEAR16",
"pcm16": "LINEAR16",
"linear16": "LINEAR16",
}
)
def __init__(self) -> None:
BaseTextToSpeechConfig.__init__(self)
@ -94,6 +107,7 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
def _map_voice_to_vertex_format(
self,
voice: str | dict | None,
model: str | None = None,
) -> tuple[str | None, dict | None]:
"""
Map voice to Vertex AI format.
@ -111,6 +125,9 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
if voice is None:
return None, None
if model is not None and self._is_gemini_tts_model(model):
return self._map_gemini_tts_voice_to_vertex_format(model=model, voice=voice)
if isinstance(voice, dict):
# Already in Vertex AI format
return None, voice
@ -139,6 +156,156 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
return voice_str, voice_dict
@staticmethod
def _is_gemini_tts_model(model: str) -> bool:
from litellm.utils import is_gemini_tts_model
return is_gemini_tts_model(model, custom_llm_provider="vertex_ai")
@staticmethod
def _get_str_value(
source: dict, # mutable-ok: provider voice payloads arrive as concrete dictionaries
*keys: str,
) -> str | None:
for key in keys:
value = source.get(key)
if isinstance(value, str):
return value
return None
@staticmethod
def _get_dict_value(
source: dict, # mutable-ok: provider voice payloads arrive as concrete dictionaries
*keys: str,
) -> dict | None: # mutable-ok: nested provider payloads remain concrete dictionaries
for key in keys:
value = source.get(key)
if isinstance(value, dict):
return value
return None
def _extract_gemini_tts_speaker_configs(
self,
voice: dict, # mutable-ok: provider voice payload arrives as a concrete dictionary
) -> list[dict[str, str]]: # mutable-ok: provider request serialization requires a concrete list
speech_config: Final = self._get_dict_value(voice, "speechConfig", "speech_config") or voice
multi_speaker_config: Final = self._get_dict_value(
speech_config,
"multiSpeakerVoiceConfig",
"multi_speaker_voice_config",
)
if multi_speaker_config is None:
return [] # mutable-ok: provider request serialization requires a concrete empty list
raw_speaker_configs: Final = multi_speaker_config.get(
"speakerVoiceConfigs",
multi_speaker_config.get(
"speaker_voice_configs",
[], # mutable-ok: missing speaker configuration uses a concrete empty-list sentinel
),
)
if not isinstance(raw_speaker_configs, list):
return [] # mutable-ok: malformed speaker configuration produces a concrete empty list
speaker_configs: Final[ # mutable-ok: validated payloads are accumulated for serialization
list[dict[str, str]]
] = []
for raw_config in raw_speaker_configs:
if not isinstance(raw_config, dict):
continue
speaker_alias = self._get_str_value(
raw_config,
"speakerAlias",
"speaker_alias",
"speaker",
)
speaker_id = self._get_str_value(raw_config, "speakerId", "speaker_id")
if speaker_id is None:
voice_config = self._get_dict_value(raw_config, "voiceConfig", "voice_config")
if voice_config is not None:
prebuilt_voice_config = self._get_dict_value(
voice_config,
"prebuiltVoiceConfig",
"prebuilt_voice_config",
)
if prebuilt_voice_config is not None:
speaker_id = self._get_str_value(
prebuilt_voice_config,
"voiceName",
"voice_name",
)
if speaker_alias is not None and speaker_id is not None:
speaker_configs.append(
{ # mutable-ok: provider request serialization requires a concrete dict
"speakerAlias": speaker_alias,
"speakerId": speaker_id,
}
)
return speaker_configs
def _extract_gemini_tts_voice_name(
self,
voice: dict, # mutable-ok: provider voice payload arrives as a concrete dictionary
) -> str | None:
voice_name: Final = self._get_str_value(voice, "name", "voiceName", "voice_name")
if voice_name is not None:
return voice_name
speech_config: Final = self._get_dict_value(voice, "speechConfig", "speech_config") or voice
voice_config: Final = self._get_dict_value(speech_config, "voiceConfig", "voice_config")
if voice_config is None:
return None
prebuilt_voice_config: Final = self._get_dict_value(
voice_config,
"prebuiltVoiceConfig",
"prebuilt_voice_config",
)
if prebuilt_voice_config is None:
return None
return self._get_str_value(prebuilt_voice_config, "voiceName", "voice_name")
def _map_gemini_tts_voice_to_vertex_format(
self,
model: str,
voice: str | dict, # mutable-ok: provider voice payload arrives as a concrete dictionary
) -> tuple[str | None, dict]: # mutable-ok: provider request serialization requires a concrete dict
if isinstance(voice, str):
return voice, { # mutable-ok: provider request serialization requires a concrete dict
"languageCode": self.DEFAULT_LANGUAGE_CODE,
"modelName": model,
"name": voice,
}
language_code: Final = self._get_str_value(voice, "languageCode", "language_code") or self.DEFAULT_LANGUAGE_CODE
model_name: Final = model
speaker_configs: Final = self._extract_gemini_tts_speaker_configs(voice)
if speaker_configs:
return None, { # mutable-ok: provider request serialization requires a concrete dict
"languageCode": language_code,
"modelName": model_name,
"multiSpeakerVoiceConfig": { # mutable-ok: nested provider payload is serialized as a dict
"speakerVoiceConfigs": speaker_configs,
},
}
voice_name: Final = self._extract_gemini_tts_voice_name(voice)
if voice_name is not None:
return None, { # mutable-ok: provider request serialization requires a concrete dict
"languageCode": language_code,
"modelName": model_name,
"name": voice_name,
}
return None, { # mutable-ok: provider request serialization requires a concrete dict
**voice,
"languageCode": language_code,
"modelName": model_name,
}
@staticmethod
def _dispatch_voice_name(voice: str | Mapping[str, object] | None) -> str | None:
if isinstance(voice, str):
return voice
if not isinstance(voice, Mapping):
return None
name: Final = voice.get("name")
return name if isinstance(name, str) else None
def dispatch_text_to_speech(
self,
model: str,
@ -172,14 +339,16 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
vertex_project: Final = self.safe_get_vertex_ai_project(litellm_params_dict)
vertex_location: Final = self.safe_get_vertex_ai_location(litellm_params_dict)
# Convert voice to string if it's a dict (extract name)
# Actual voice mapping happens in map_openai_params
voice_str: str | None = None
if isinstance(voice, str):
voice_str = voice
elif isinstance(voice, dict):
# Extract voice name from dict if needed
voice_str = voice.get("name") if voice else None
mapped_voice, mapped_params = (
(self._dispatch_voice_name(voice), optional_params)
if "audioEncoding" in optional_params
else self.map_openai_params(
model=model,
voice=voice,
optional_params=optional_params,
kwargs=kwargs,
)
)
# Store credentials in litellm_params for use in transform methods
litellm_params_dict.update(
@ -195,9 +364,9 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
response: Final = base_llm_http_handler.text_to_speech_handler(
model=model,
input=input,
voice=voice_str,
voice=mapped_voice,
text_to_speech_provider_config=self,
text_to_speech_optional_params=optional_params,
text_to_speech_optional_params=mapped_params,
custom_llm_provider="vertex_ai",
litellm_params=litellm_params_dict,
logging_obj=logging_obj,
@ -245,15 +414,21 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
##########################################################
# Map voice using helper
##########################################################
mapped_voice_str, voice_dict = self._map_voice_to_vertex_format(voice)
mapped_voice_str, voice_dict = self._map_voice_to_vertex_format(
voice=voice,
model=model,
)
if voice_dict is not None:
mapped_params["vertex_voice_dict"] = voice_dict
# Map response format
if "response_format" in optional_params:
format_name: Final = optional_params["response_format"]
if format_name in self.FORMAT_MAPPINGS:
mapped_params["audioEncoding"] = self.FORMAT_MAPPINGS[format_name]
format_mappings: Final = (
self.GEMINI_FORMAT_MAPPINGS if self._is_gemini_tts_model(model) else self.FORMAT_MAPPINGS
)
if format_name in format_mappings:
mapped_params["audioEncoding"] = format_mappings[format_name]
else:
# Try to use it directly as Google Cloud format
mapped_params["audioEncoding"] = format_name
@ -410,7 +585,7 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
voice_dict: Final = litellm_params.get("vertex_voice_dict") or optional_params.get("vertex_voice_dict")
if voice_dict is not None and isinstance(voice_dict, dict):
vertex_voice = VertexTextToSpeechVoice(**voice_dict)
elif voice is not None and isinstance(voice, str):
elif voice is not None:
# Handle string voice (shouldn't normally happen if dispatch was called)
parts: Final = voice.split("-")
if len(parts) >= 2:

View file

@ -8198,6 +8198,7 @@ def speech(
kwargs.pop("tags", [])
optional_params = {}
original_voice: Final = voice
if response_format is not None:
optional_params["response_format"] = response_format
if speed is not None:
@ -8450,7 +8451,13 @@ def speech(
generic_optional_params: Final = GenericLiteLLMParams(**kwargs)
# Handle Gemini models separately (they use speech_to_completion_bridge)
if "gemini" in model:
if (
litellm.utils.is_gemini_tts_model(
model,
custom_llm_provider=custom_llm_provider,
)
and response_format is None
):
from .endpoints.speech.speech_to_completion_bridge.handler import (
speech_to_completion_bridge_handler,
)
@ -8458,7 +8465,7 @@ def speech(
return speech_to_completion_bridge_handler.speech(
model=model,
input=input,
voice=voice,
voice=original_voice,
optional_params=optional_params,
litellm_params=litellm_params_dict,
headers=headers or {},

View file

@ -47679,6 +47679,34 @@
"web_search_billing_unit": "per_query",
"google_maps_grounding_cost_per_query": 0.014
},
"vertex_ai/gemini-3.1-flash-tts-preview": {
"input_cost_per_token": 1e-06,
"input_cost_per_token_batches": 5e-07,
"litellm_provider": "vertex_ai-language-models",
"mode": "audio_speech",
"output_cost_per_audio_token": 2e-05,
"output_cost_per_token": 2e-05,
"source": "https://cloud.google.com/text-to-speech/pricing",
"health_check_voice": "Kore",
"max_input_tokens": 8192,
"max_output_tokens": 16384,
"max_tokens": 16384,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
],
"supports_audio_input": false,
"supports_audio_output": true,
"supports_function_calling": false,
"supports_prompt_caching": false,
"tpm": 4000000,
"rpm": 10
},
"vertex_ai/gemini-3.1-flash-lite": {
"cache_read_input_audio_token_cost": 5e-08,
"deprecation_date": "2027-05-07",
@ -54463,6 +54491,33 @@
"supports_response_schema": false,
"supports_web_search": false
},
"gemini-3.1-flash-tts-preview": {
"health_check_voice": "Kore",
"input_cost_per_token": 1e-06,
"litellm_provider": "gemini",
"max_input_tokens": 8192,
"max_output_tokens": 16384,
"max_tokens": 16384,
"mode": "audio_speech",
"output_cost_per_audio_token": 2e-05,
"output_cost_per_token": 2e-05,
"source": "https://cloud.google.com/text-to-speech/pricing",
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
],
"supports_audio_input": false,
"supports_audio_output": true,
"supports_function_calling": false,
"supports_prompt_caching": false,
"tpm": 4000000,
"rpm": 10
},
"gemini-flash-latest": {
"cache_read_input_token_cost": 7.5e-08,
"input_cost_per_token": 7.5e-07,
@ -65768,13 +65823,14 @@
"output_cost_per_token": 1e-05,
"source": "https://cloud.google.com/gemini-enterprise-agent-platform/generative-ai/pricing"
},
"vertex_ai/gemini-3.1-flash-tts-preview": {
"input_cost_per_token": 1e-06,
"input_cost_per_token_batches": 5e-07,
"vertex_ai/gemini-3.1-flash-live-preview": {
"input_cost_per_audio_token": 3e-06,
"input_cost_per_second": 8.33333333333e-05,
"input_cost_per_token": 7.5e-07,
"litellm_provider": "vertex_ai",
"mode": "audio_speech",
"output_cost_per_audio_token": 2e-05,
"output_cost_per_token": 2e-05,
"mode": "realtime",
"output_cost_per_audio_token": 1.2e-05,
"output_cost_per_token": 4.5e-06,
"source": "https://cloud.google.com/gemini-enterprise-agent-platform/generative-ai/pricing"
},
"vertex_ai/gemini-3.5-transcribe": {

View file

@ -12121,12 +12121,40 @@ async def audio_speech(
custom_headers.update(callback_headers)
requested_format: Final = data.get("response_format")
requested_format_str: Final = requested_format if isinstance(requested_format, str) else None
extra_response_format_media_types: Final = MappingProxyType(
{
"pcm16": "audio/wav",
"linear16": "audio/wav",
"alaw": "audio/x-alaw-basic",
"mulaw": "audio/basic",
}
)
extra_media_type: Final = (
extra_response_format_media_types.get(requested_format_str.lower())
if requested_format_str is not None
else None
)
upstream_content_type: Final = (
response.response.headers.get("content-type") if isinstance(response, HttpxBinaryResponseContent) else None
)
media_type: Final = resolve_speech_media_type(
resolved_media_type: Final = resolve_speech_media_type(
upstream_content_type=upstream_content_type,
response_format=requested_format if isinstance(requested_format, str) else None,
response_format=requested_format_str,
)
request_model: Final = data.get("model", "")
media_type: Final = (
extra_media_type
if extra_media_type is not None
else (
"audio/wav"
if (
requested_format_str is None
and isinstance(request_model, str)
and litellm.utils.is_gemini_tts_model(request_model)
)
else resolved_media_type
)
)
return StreamingResponse(
@ -12150,13 +12178,16 @@ async def audio_speech(
detail=e.detail,
headers=headers_with_litellm_call_id(e.headers, litellm_call_id),
)
raw_status_code: Final = getattr(e, "status_code", None)
if not isinstance(raw_status_code, int):
raise
raise ProxyException(
message=getattr(e, "message", f"{e}"),
type=getattr(e, "type", "None"),
param=getattr(e, "param", "None"),
headers=litellm_call_id_headers(litellm_call_id),
openai_code=getattr(e, "code", None),
code=getattr(e, "status_code", 500),
code=raw_status_code,
)

View file

@ -1,7 +1,9 @@
from collections.abc import Sequence
from enum import Enum
from typing import Any, Final, Literal, Protocol
from typing_extensions import (
ReadOnly,
Required,
TypedDict,
)
@ -226,8 +228,18 @@ class VoiceConfig(TypedDict):
prebuiltVoiceConfig: PrebuiltVoiceConfig
class SpeakerVoiceConfig(TypedDict):
speaker: ReadOnly[str]
voiceConfig: ReadOnly[VoiceConfig]
class MultiSpeakerVoiceConfig(TypedDict):
speakerVoiceConfigs: ReadOnly[Sequence[SpeakerVoiceConfig]]
class SpeechConfig(TypedDict, total=False):
voiceConfig: VoiceConfig
multiSpeakerVoiceConfig: ReadOnly[MultiSpeakerVoiceConfig]
languageCode: str

View file

@ -4,7 +4,9 @@ Type definitions for Vertex AI Text-to-Speech API
Reference: https://cloud.google.com/text-to-speech/docs/reference/rest/v1/text/synthesize
"""
from typing_extensions import TypedDict
from collections.abc import Sequence
from typing_extensions import ReadOnly, TypedDict
class VertexTextToSpeechInput(TypedDict, total=False):
@ -16,6 +18,16 @@ class VertexTextToSpeechInput(TypedDict, total=False):
text: str | None
ssml: str | None
prompt: ReadOnly[str | None]
class VertexTextToSpeechSpeakerVoiceConfig(TypedDict):
speakerAlias: ReadOnly[str]
speakerId: ReadOnly[str]
class VertexTextToSpeechMultiSpeakerVoiceConfig(TypedDict):
speakerVoiceConfigs: ReadOnly[Sequence[VertexTextToSpeechSpeakerVoiceConfig]]
class VertexTextToSpeechVoice(TypedDict, total=False):
@ -29,6 +41,8 @@ class VertexTextToSpeechVoice(TypedDict, total=False):
languageCode: str
name: str
modelName: ReadOnly[str]
multiSpeakerVoiceConfig: ReadOnly[VertexTextToSpeechMultiSpeakerVoiceConfig]
class VertexTextToSpeechAudioConfig(TypedDict, total=False):
@ -42,6 +56,7 @@ class VertexTextToSpeechAudioConfig(TypedDict, total=False):
audioEncoding: str
speakingRate: str
sampleRateHertz: ReadOnly[int]
class VertexTextToSpeechRequest(TypedDict, total=False):

View file

@ -723,6 +723,36 @@ def _is_gemini_model(model: str | None, custom_llm_provider: str | None) -> bool
return model is not None and "gemini" in model.lower()
def is_gemini_tts_model(model: str, custom_llm_provider: str | None = None) -> bool:
try:
runtime_model_info: Final = get_model_info(
model=model,
custom_llm_provider=custom_llm_provider,
)
except Exception: # noqa: BLE001 # model lookup failures intentionally fall back to the bundled registry
provider_model: Final = (
f"{custom_llm_provider}/{model}"
if custom_llm_provider is not None and not model.startswith(f"{custom_llm_provider}/")
else model
)
bundled_model_cost: Final = _get_bundled_model_cost_map()
bundled_model_info: Final = bundled_model_cost.get(provider_model) or bundled_model_cost.get(model)
return (
bundled_model_info is not None
and bundled_model_info.get("mode") == "audio_speech"
and bundled_model_info.get("litellm_provider")
in (
"gemini",
"vertex_ai-language-models",
)
)
return runtime_model_info.get("mode") == "audio_speech" and runtime_model_info.get("litellm_provider") in (
"gemini",
"vertex_ai-language-models",
)
def _remove_thought_signature_from_id(tool_call_id: str, separator: str) -> str:
"""
Remove thought signature from a tool call ID.

View file

@ -47679,6 +47679,34 @@
"web_search_billing_unit": "per_query",
"google_maps_grounding_cost_per_query": 0.014
},
"vertex_ai/gemini-3.1-flash-tts-preview": {
"input_cost_per_token": 1e-06,
"input_cost_per_token_batches": 5e-07,
"litellm_provider": "vertex_ai-language-models",
"mode": "audio_speech",
"output_cost_per_audio_token": 2e-05,
"output_cost_per_token": 2e-05,
"source": "https://cloud.google.com/text-to-speech/pricing",
"health_check_voice": "Kore",
"max_input_tokens": 8192,
"max_output_tokens": 16384,
"max_tokens": 16384,
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
],
"supports_audio_input": false,
"supports_audio_output": true,
"supports_function_calling": false,
"supports_prompt_caching": false,
"tpm": 4000000,
"rpm": 10
},
"vertex_ai/gemini-3.1-flash-lite": {
"cache_read_input_audio_token_cost": 5e-08,
"deprecation_date": "2027-05-07",
@ -54463,6 +54491,33 @@
"supports_response_schema": false,
"supports_web_search": false
},
"gemini-3.1-flash-tts-preview": {
"health_check_voice": "Kore",
"input_cost_per_token": 1e-06,
"litellm_provider": "gemini",
"max_input_tokens": 8192,
"max_output_tokens": 16384,
"max_tokens": 16384,
"mode": "audio_speech",
"output_cost_per_audio_token": 2e-05,
"output_cost_per_token": 2e-05,
"source": "https://cloud.google.com/text-to-speech/pricing",
"supported_endpoints": [
"/v1/audio/speech"
],
"supported_modalities": [
"text"
],
"supported_output_modalities": [
"audio"
],
"supports_audio_input": false,
"supports_audio_output": true,
"supports_function_calling": false,
"supports_prompt_caching": false,
"tpm": 4000000,
"rpm": 10
},
"gemini-flash-latest": {
"cache_read_input_token_cost": 7.5e-08,
"input_cost_per_token": 7.5e-07,
@ -65768,13 +65823,14 @@
"output_cost_per_token": 1e-05,
"source": "https://cloud.google.com/gemini-enterprise-agent-platform/generative-ai/pricing"
},
"vertex_ai/gemini-3.1-flash-tts-preview": {
"input_cost_per_token": 1e-06,
"input_cost_per_token_batches": 5e-07,
"vertex_ai/gemini-3.1-flash-live-preview": {
"input_cost_per_audio_token": 3e-06,
"input_cost_per_second": 8.33333333333e-05,
"input_cost_per_token": 7.5e-07,
"litellm_provider": "vertex_ai",
"mode": "audio_speech",
"output_cost_per_audio_token": 2e-05,
"output_cost_per_token": 2e-05,
"mode": "realtime",
"output_cost_per_audio_token": 1.2e-05,
"output_cost_per_token": 4.5e-06,
"source": "https://cloud.google.com/gemini-enterprise-agent-platform/generative-ai/pricing"
},
"vertex_ai/gemini-3.5-transcribe": {

View file

@ -262,6 +262,9 @@
"minimum": 0
}
},
"health_check_voice": {
"type": "string"
},
"input_cost_per_audio_per_second": {
"type": "number",
"minimum": 0

View file

@ -3,10 +3,10 @@
Test to verify the Google GenAI transformation logic for generateContent parameters
"""
import pytest
from litellm.llms.gemini.google_genai.transformation import GoogleGenAIConfig
from litellm.llms.vertex_ai.google_genai.transformation import VertexAIGoogleGenAIConfig
from litellm.responses.litellm_completion_transformation.transformation import (
LiteLLMCompletionResponsesConfig,
)
@ -31,10 +31,7 @@ def test_map_generate_content_optional_params_response_json_schema_camelcase():
# responseJsonSchema should be in the result (camelCase format for Google GenAI API)
assert "responseJsonSchema" in result
assert (
result["responseJsonSchema"]
== generate_content_config_dict["responseJsonSchema"]
)
assert result["responseJsonSchema"] == generate_content_config_dict["responseJsonSchema"]
assert "temperature" in result
assert result["temperature"] == 1.0
@ -58,10 +55,7 @@ def test_map_generate_content_optional_params_response_schema_snakecase():
# response_schema should be converted to responseJsonSchema (camelCase)
assert "responseJsonSchema" in result
assert (
result["responseJsonSchema"]
== generate_content_config_dict["response_json_schema"]
)
assert result["responseJsonSchema"] == generate_content_config_dict["response_json_schema"]
assert "temperature" in result
@ -157,6 +151,68 @@ def test_map_generate_content_optional_params_response_mime_type():
assert "responseJsonSchema" in result
@pytest.mark.parametrize("config_class", [GoogleGenAIConfig, VertexAIGoogleGenAIConfig])
def test_map_generate_content_optional_params_speech_config(config_class):
config = config_class()
result = config.map_generate_content_optional_params(
generate_content_config_dict={
"response_modalities": ["AUDIO"],
"speech_config": {
"multi_speaker_voice_config": {
"speaker_voice_configs": [
{
"speaker": "Ryan",
"voice_config": {
"prebuilt_voice_config": {
"voice_name": "Umbriel",
},
},
},
{
"speaker": "Katie",
"voice_config": {
"prebuilt_voice_config": {
"voice_name": "Leda",
},
},
},
],
},
"language_code": "en-US",
},
},
model="gemini-3.1-flash-tts-preview",
)
assert result["responseModalities"] == ["AUDIO"]
assert "response_modalities" not in result
assert "speech_config" not in result
assert result["speechConfig"] == {
"multiSpeakerVoiceConfig": {
"speakerVoiceConfigs": [
{
"speaker": "Ryan",
"voiceConfig": {
"prebuiltVoiceConfig": {
"voiceName": "Umbriel",
},
},
},
{
"speaker": "Katie",
"voiceConfig": {
"prebuiltVoiceConfig": {
"voiceName": "Leda",
},
},
},
],
},
"languageCode": "en-US",
}
def test_responses_api_reasoning_dict_format():
"""Test that reasoning parameter with dict format is mapped to reasoning_effort"""
from litellm.types.llms.openai import ResponsesAPIOptionalRequestParams
@ -254,9 +310,7 @@ def test_transform_generate_content_request_with_system_instruction():
# Verify that systemInstruction is in the request
assert "systemInstruction" in result, "systemInstruction should be in request body"
assert (
result["systemInstruction"] == system_instruction
), "systemInstruction should match input"
assert result["systemInstruction"] == system_instruction, "systemInstruction should match input"
assert result["model"] == "gemini-3-flash-preview"
assert result["contents"] == contents
@ -279,9 +333,7 @@ def test_transform_generate_content_request_without_system_instruction():
)
# Verify that systemInstruction is NOT in the request when not provided
assert (
"systemInstruction" not in result
), "systemInstruction should not be in request when None"
assert "systemInstruction" not in result, "systemInstruction should not be in request when None"
assert result["model"] == "gemini-3-flash-preview"
assert result["contents"] == contents
@ -290,9 +342,7 @@ def test_transform_generate_content_request_system_instruction_with_tools():
"""Test that systemInstruction works correctly alongside tools"""
config = GoogleGenAIConfig()
system_instruction = {
"parts": [{"text": "You are a helpful assistant that uses tools"}]
}
system_instruction = {"parts": [{"text": "You are a helpful assistant that uses tools"}]}
contents = [{"role": "user", "parts": [{"text": "What's the weather?"}]}]
@ -374,9 +424,7 @@ def test_transform_generate_content_request_normalizes_response_schema_2_5():
assert "responseJsonSchema" in gen_config
normalized = gen_config["responseJsonSchema"]
assert "$defs" in normalized
assert normalized["properties"]["highlights"]["items"] == {
"$ref": "#/$defs/Highlight"
}
assert normalized["properties"]["highlights"]["items"] == {"$ref": "#/$defs/Highlight"}
def test_transform_generate_content_request_flattens_response_schema_1_5():
@ -581,12 +629,8 @@ def test_validate_environment_with_dict_api_key():
# The dict should be merged into headers, not set as a value
assert "x-goog-api-key" in result, "x-goog-api-key should be in headers"
assert (
result["x-goog-api-key"] == "sk-test-key-123"
), "API key should be the string value, not a dict"
assert isinstance(
result["x-goog-api-key"], str
), "Header value should be a string, not a dict"
assert result["x-goog-api-key"] == "sk-test-key-123", "API key should be the string value, not a dict"
assert isinstance(result["x-goog-api-key"], str), "Header value should be a string, not a dict"
assert "Content-Type" in result, "Content-Type should be in headers"
assert result["Content-Type"] == "application/json"
@ -632,9 +676,7 @@ def test_validate_environment_with_extra_headers():
# Both the auth dict and extra headers should be merged
assert "x-goog-api-key" in result, "x-goog-api-key should be in headers"
assert (
result["x-goog-api-key"] == "sk-test-key-789"
), "API key should be correctly set"
assert result["x-goog-api-key"] == "sk-test-key-789", "API key should be correctly set"
assert isinstance(result["x-goog-api-key"], str), "Header value should be a string"
assert "X-Custom-Header" in result, "Extra headers should be merged"
assert result["X-Custom-Header"] == "custom-value"

View file

@ -5,11 +5,63 @@ Test Gemini TTS (Text-to-Speech) functionality
import pytest
from unittest.mock import patch, MagicMock
import litellm
from litellm.endpoints.speech.speech_to_completion_bridge.transformation import (
SpeechToCompletionBridgeTransformationHandler,
)
from litellm.llms.gemini.chat.transformation import GoogleAIStudioGeminiConfig
from litellm.utils import get_supported_openai_params
GEMINI_3_1_FLASH_TTS_MODEL = "gemini-3.1-flash-tts-preview"
MULTI_SPEAKER_SPEECH_CONFIG = {
"multi_speaker_voice_config": {
"speaker_voice_configs": [
{
"speaker": "Ryan",
"voice_config": {
"prebuilt_voice_config": {
"voice_name": "Umbriel",
},
},
},
{
"speaker": "Katie",
"voice_config": {
"prebuilt_voice_config": {
"voice_name": "Leda",
},
},
},
],
},
"language_code": "en-US",
}
NORMALIZED_MULTI_SPEAKER_SPEECH_CONFIG = {
"multiSpeakerVoiceConfig": {
"speakerVoiceConfigs": [
{
"speaker": "Ryan",
"voiceConfig": {
"prebuiltVoiceConfig": {
"voiceName": "Umbriel",
},
},
},
{
"speaker": "Katie",
"voiceConfig": {
"prebuiltVoiceConfig": {
"voiceName": "Leda",
},
},
},
],
},
"languageCode": "en-US",
}
class TestGeminiTTSTransformation:
"""Test Gemini TTS transformation functionality"""
@ -19,17 +71,16 @@ class TestGeminiTTSTransformation:
config = GoogleAIStudioGeminiConfig()
# Test TTS models (both preview and non-preview versions)
assert (
config.is_model_gemini_audio_model("gemini-2.5-flash-preview-tts") == True
)
assert config.is_model_gemini_audio_model("gemini-2.5-pro-preview-tts") == True
assert config.is_model_gemini_audio_model("gemini-2.5-flash-tts") == True
assert config.is_model_gemini_audio_model("gemini-2.5-pro-tts") == True
assert config.is_model_gemini_audio_model("gemini-2.5-flash-preview-tts")
assert config.is_model_gemini_audio_model("gemini-2.5-pro-preview-tts")
assert config.is_model_gemini_audio_model("gemini-2.5-flash-tts")
assert config.is_model_gemini_audio_model("gemini-2.5-pro-tts")
assert config.is_model_gemini_audio_model(GEMINI_3_1_FLASH_TTS_MODEL)
# Test non-TTS models
assert config.is_model_gemini_audio_model("gemini-2.5-flash") == False
assert config.is_model_gemini_audio_model("gemini-2.5-pro") == False
assert config.is_model_gemini_audio_model("gpt-4o-audio-preview") == False
assert not config.is_model_gemini_audio_model("gemini-2.5-flash")
assert not config.is_model_gemini_audio_model("gemini-2.5-pro")
assert not config.is_model_gemini_audio_model("gpt-4o-audio-preview")
def test_gemini_tts_supported_params(self):
"""Test that audio parameter is included for TTS models"""
@ -66,10 +117,7 @@ class TestGeminiTTSTransformation:
assert "speechConfig" in result
assert "voiceConfig" in result["speechConfig"]
assert "prebuiltVoiceConfig" in result["speechConfig"]["voiceConfig"]
assert (
result["speechConfig"]["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"]
== "Kore"
)
assert result["speechConfig"]["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == "Kore"
# Check response modalities
assert "responseModalities" in result
@ -78,9 +126,7 @@ class TestGeminiTTSTransformation:
def test_gemini_tts_audio_parameter_mapping_with_language_code(self):
config = GoogleAIStudioGeminiConfig()
non_default_params = {
"audio": {"voice": "Kore", "format": "pcm16", "language_code": "en-US"}
}
non_default_params = {"audio": {"voice": "Kore", "format": "pcm16", "language_code": "en-US"}}
optional_params = {}
result = config.map_openai_params(
@ -92,17 +138,12 @@ class TestGeminiTTSTransformation:
assert "speechConfig" in result
assert result["speechConfig"]["languageCode"] == "en-US"
assert (
result["speechConfig"]["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"]
== "Kore"
)
assert result["speechConfig"]["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == "Kore"
def test_map_audio_params_language_code(self):
config = GoogleAIStudioGeminiConfig()
result = config._map_audio_params(
{"voice": "Kore", "format": "pcm16", "language_code": "de-DE"}
)
result = config._map_audio_params({"voice": "Kore", "format": "pcm16", "language_code": "de-DE"})
assert result["languageCode"] == "de-DE"
assert result["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == "Kore"
@ -115,6 +156,29 @@ class TestGeminiTTSTransformation:
assert "languageCode" not in result
assert result["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == "Kore"
def test_gemini_tts_multi_speaker_audio_parameter_mapping(self):
"""Test multi-speaker audio parameter mapping for Gemini 3.1 TTS models"""
config = GoogleAIStudioGeminiConfig()
non_default_params = {
"audio": {
"speech_config": MULTI_SPEAKER_SPEECH_CONFIG,
"format": "pcm16",
}
}
optional_params = {}
result = config.map_openai_params(
non_default_params=non_default_params,
optional_params=optional_params,
model=GEMINI_3_1_FLASH_TTS_MODEL,
drop_params=False,
)
assert result["speechConfig"] == NORMALIZED_MULTI_SPEAKER_SPEECH_CONFIG
assert result["responseModalities"] == ["AUDIO"]
assert "temperature" not in result
def test_gemini_tts_audio_parameter_with_existing_modalities(self):
"""Test audio parameter mapping when modalities already exist"""
config = GoogleAIStudioGeminiConfig()
@ -198,9 +262,7 @@ class TestGeminiTTSTransformation:
}
optional_params = {}
with pytest.raises(
ValueError, match="Unsupported audio format for Gemini TTS models"
):
with pytest.raises(ValueError, match="Unsupported audio format for Gemini TTS models"):
config.map_openai_params(
non_default_params=non_default_params,
optional_params=optional_params,
@ -239,6 +301,28 @@ def test_gemini_tts_completion_mock():
assert response.choices[0].message.content is not None
def test_gemini_tts_speech_bridge_accepts_multi_speaker_voice_dict():
handler = SpeechToCompletionBridgeTransformationHandler()
result = handler.transform_request(
model=f"vertex_ai/{GEMINI_3_1_FLASH_TTS_MODEL}",
input="Ryan: How are you doing today Katie?\nKatie: Not too bad.",
voice=MULTI_SPEAKER_SPEECH_CONFIG,
optional_params={"response_format": "pcm"},
litellm_params={},
headers={},
litellm_logging_obj=MagicMock(),
custom_llm_provider="vertex_ai",
)
assert result["modalities"] == ["audio"]
assert result["audio"] == {
"speech_config": NORMALIZED_MULTI_SPEAKER_SPEECH_CONFIG,
"format": "pcm16",
}
assert "response_format" not in result
class TestGeminiTTSSpeechConfigInRequestBody:
"""Test that speechConfig is properly included in the final request body.
@ -255,11 +339,11 @@ class TestGeminiTTSSpeechConfigInRequestBody:
("gemini-2.5-flash-preview-tts", "vertex_ai"),
("gemini-2.5-flash-preview-tts", "gemini"),
("gemini-2.5-pro-tts", "vertex_ai"),
(GEMINI_3_1_FLASH_TTS_MODEL, "vertex_ai"),
(GEMINI_3_1_FLASH_TTS_MODEL, "gemini"),
],
)
def test_speechconfig_in_generation_config_transform_request_body(
self, model, custom_llm_provider
):
def test_speechconfig_in_generation_config_transform_request_body(self, model, custom_llm_provider):
"""Test that speechConfig is included in generationConfig after _transform_request_body()"""
from litellm.llms.vertex_ai.gemini.transformation import (
_transform_request_body,
@ -267,9 +351,7 @@ class TestGeminiTTSSpeechConfigInRequestBody:
# Simulate optional_params after map_openai_params() has run
optional_params = {
"speechConfig": {
"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Kore"}}
},
"speechConfig": {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Kore"}}},
"responseModalities": ["AUDIO"],
}
@ -292,12 +374,7 @@ class TestGeminiTTSSpeechConfigInRequestBody:
f"speechConfig was filtered out of generationConfig for model={model}, provider={custom_llm_provider}. "
"Ensure speechConfig is in the GenerationConfig TypedDict."
)
assert (
generation_config["speechConfig"]["voiceConfig"]["prebuiltVoiceConfig"][
"voiceName"
]
== "Kore"
)
assert generation_config["speechConfig"]["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == "Kore"
@pytest.mark.parametrize(
"model,custom_llm_provider",
@ -305,16 +382,18 @@ class TestGeminiTTSSpeechConfigInRequestBody:
("gemini-2.5-flash-tts", "vertex_ai"),
("gemini-2.5-flash-tts", "gemini"),
("gemini-2.5-flash-preview-tts", "vertex_ai"),
(GEMINI_3_1_FLASH_TTS_MODEL, "vertex_ai"),
(GEMINI_3_1_FLASH_TTS_MODEL, "gemini"),
],
)
def test_speechconfig_end_to_end_mapping(self, model, custom_llm_provider):
"""Test full pipeline: audio param -> map_openai_params -> _transform_request_body"""
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
VertexGeminiConfig,
)
from litellm.llms.vertex_ai.gemini.transformation import (
_transform_request_body,
)
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
VertexGeminiConfig,
)
config = VertexGeminiConfig()
@ -351,17 +430,55 @@ class TestGeminiTTSSpeechConfigInRequestBody:
f"speechConfig was filtered out during _transform_request_body() for model={model}, provider={custom_llm_provider}. "
"This breaks Gemini TTS - speechConfig must be in GenerationConfig TypedDict."
)
assert (
generation_config["speechConfig"]["voiceConfig"]["prebuiltVoiceConfig"][
"voiceName"
]
== "Puck"
)
assert generation_config["speechConfig"]["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == "Puck"
# Also verify responseModalities is present
assert "responseModalities" in generation_config
assert "AUDIO" in generation_config["responseModalities"]
@pytest.mark.parametrize(
"custom_llm_provider",
[
"vertex_ai",
"gemini",
],
)
def test_multi_speaker_speechconfig_end_to_end_mapping(self, custom_llm_provider):
"""Test full pipeline for Gemini 3.1 multi-speaker TTS speechConfig"""
from litellm.llms.vertex_ai.gemini.transformation import (
_transform_request_body,
)
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
VertexGeminiConfig,
)
config = VertexGeminiConfig()
mapped_params = config.map_openai_params(
non_default_params={
"audio": {
"speech_config": MULTI_SPEAKER_SPEECH_CONFIG,
"format": "pcm16",
}
},
optional_params={},
model=GEMINI_3_1_FLASH_TTS_MODEL,
drop_params=False,
)
request_body = _transform_request_body(
messages=[{"role": "user", "content": "Ryan: Hi.\nKatie: Hello."}],
model=GEMINI_3_1_FLASH_TTS_MODEL,
optional_params=mapped_params,
custom_llm_provider=custom_llm_provider,
litellm_params={},
cached_content=None,
)
generation_config = request_body["generationConfig"]
assert generation_config["speechConfig"] == NORMALIZED_MULTI_SPEAKER_SPEECH_CONFIG
assert generation_config["responseModalities"] == ["AUDIO"]
assert "temperature" not in generation_config
@pytest.mark.parametrize(
"model,custom_llm_provider",
@ -372,18 +489,16 @@ class TestGeminiTTSSpeechConfigInRequestBody:
],
)
def test_language_code_end_to_end_mapping(self, model, custom_llm_provider):
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
VertexGeminiConfig,
)
from litellm.llms.vertex_ai.gemini.transformation import (
_transform_request_body,
)
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
VertexGeminiConfig,
)
config = VertexGeminiConfig()
non_default_params = {
"audio": {"voice": "Puck", "format": "pcm16", "language_code": "pt-BR"}
}
non_default_params = {"audio": {"voice": "Puck", "format": "pcm16", "language_code": "pt-BR"}}
optional_params = {}
mapped_params = config.map_openai_params(
@ -406,12 +521,7 @@ class TestGeminiTTSSpeechConfigInRequestBody:
generation_config = request_body["generationConfig"]
assert generation_config["speechConfig"]["languageCode"] == "pt-BR"
assert (
generation_config["speechConfig"]["voiceConfig"]["prebuiltVoiceConfig"][
"voiceName"
]
== "Puck"
)
assert generation_config["speechConfig"]["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == "Puck"
assert "AUDIO" in generation_config["responseModalities"]

View file

@ -14,6 +14,7 @@ from unittest.mock import AsyncMock, MagicMock
import httpx
import pytest
from fastapi import HTTPException
from litellm.proxy import proxy_server
from litellm.types.llms.openai import HttpxBinaryResponseContent
@ -85,7 +86,7 @@ def patched_speech_error(monkeypatch):
def patched_speech_provider_rejection(monkeypatch, patched_speech_error):
import litellm
async def _raise(*args, **kwargs):
async def _raise_rejection(*args, **kwargs):
raise litellm.BadRequestError(
message=(
"Gemini TTS only produces raw PCM16 audio, so response_format='mp3' is not supported."
@ -95,6 +96,29 @@ def patched_speech_provider_rejection(monkeypatch, patched_speech_error):
llm_provider="gemini",
)
monkeypatch.setattr(proxy_server, "route_request", _raise_rejection)
yield
@pytest.fixture
def patched_speech_proxy_error(monkeypatch, patched_speech):
async def _raise(*args, **kwargs):
raise proxy_server.ProxyException(
message="Unsupported audio encoding.",
type="invalid_request_error",
param=None,
code=400,
)
monkeypatch.setattr(proxy_server, "route_request", _raise)
yield
@pytest.fixture
def patched_speech_http_error(monkeypatch, patched_speech):
async def _raise(*args, **kwargs):
raise HTTPException(status_code=422, detail="Provider rejected the speech request.")
monkeypatch.setattr(proxy_server, "route_request", _raise)
yield
@ -119,9 +143,7 @@ def patched_transcription(monkeypatch):
return data
monkeypatch.setattr(proxy_server, "add_litellm_data_to_request", _add_data)
monkeypatch.setattr(
proxy_server, "check_file_size_under_limit", lambda **kwargs: True
)
monkeypatch.setattr(proxy_server, "check_file_size_under_limit", lambda **kwargs: True)
async def _form_data(request):
from starlette.datastructures import FormData, UploadFile
@ -201,6 +223,35 @@ def test_audio_speech_content_type_matches_audio_format(
assert response.headers.get("content-type", "").split(";")[0] == expected_content_type
@pytest.mark.parametrize("path", ["/v1/audio/speech", "/audio/speech"])
@pytest.mark.parametrize(
"response_format,expected_media_type",
[
("pcm16", "audio/wav"),
("alaw", "audio/x-alaw-basic"),
("mulaw", "audio/basic"),
],
)
def test_audio_speech_uses_requested_media_type(
client,
auth_as,
patched_speech,
path,
response_format,
expected_media_type,
):
payload = {
"model": "gemini-3.1-flash-tts-preview",
"input": "Hi",
"voice": "Umbriel",
"response_format": response_format,
}
with auth_as():
response = client.post(path, json=payload)
assert response.status_code == 200
assert response.headers.get("content-type", "") == expected_media_type
@pytest.mark.parametrize("path", ["/v1/audio/speech", "/audio/speech"])
def test_audio_speech_error(client, auth_as, patched_speech_error, path):
"""Pins ``POST /v1/audio/speech`` and ``POST /audio/speech`` (error)."""
@ -208,7 +259,13 @@ def test_audio_speech_error(client, auth_as, patched_speech_error, path):
with auth_as():
response = client.post(path, json=payload)
assert response.status_code == 500
assert len(response.content) > 0
assert response.json() == {
"error": {
"message": "Internal server error",
"type": "internal_server_error",
}
}
assert "speech boom" not in response.text
def test_audio_speech_bad_request_maps_to_400(client, auth_as, patched_speech_provider_rejection):
@ -223,6 +280,29 @@ def test_audio_speech_bad_request_maps_to_400(client, auth_as, patched_speech_pr
assert "wav" in error["message"]
@pytest.mark.parametrize("path", ["/v1/audio/speech", "/audio/speech"])
def test_audio_speech_proxy_error_preserves_status_and_message(
client,
auth_as,
patched_speech_proxy_error,
path,
):
payload = {"model": "tts-1", "input": "Hi", "voice": "alloy"}
with auth_as():
response = client.post(path, json=payload)
assert response.status_code == 400
assert response.json()["error"]["message"] == "Unsupported audio encoding."
@pytest.mark.parametrize("path", ["/v1/audio/speech", "/audio/speech"])
def test_audio_speech_http_error_preserves_detail(client, auth_as, patched_speech_http_error, path):
payload = {"model": "tts-1", "input": "Hi", "voice": "alloy"}
with auth_as():
response = client.post(path, json=payload)
assert response.status_code == 422
assert response.json()["error"]["message"] == "Provider rejected the speech request."
@pytest.mark.parametrize("path", ["/v1/audio/transcriptions", "/audio/transcriptions"])
def test_audio_transcription_happy_path(client, auth_as, patched_transcription, path):
"""Pins ``POST /v1/audio/transcriptions`` / ``POST /audio/transcriptions`` (happy)."""

View file

@ -0,0 +1,75 @@
import json
from pathlib import Path
import pytest
import litellm
from litellm.utils import _invalidate_model_cost_lowercase_map, is_gemini_tts_model
EXPECTED_MODELS = {
"gemini-3.1-flash-tts-preview": "gemini",
"gemini/gemini-3.1-flash-tts-preview": "gemini",
"vertex_ai/gemini-3.1-flash-tts-preview": "vertex_ai-language-models",
}
def _load_model_cost_map(path: Path) -> dict:
with open(path, encoding="utf-8") as model_cost_file:
return json.load(model_cost_file)
@pytest.mark.parametrize("model,provider", EXPECTED_MODELS.items())
def test_gemini_3_1_flash_tts_model_metadata(model, provider):
model_cost = _load_model_cost_map(Path(__file__).parents[2] / "model_prices_and_context_window.json")
info = model_cost.get(model)
assert info is not None, f"{model} not found in model cost map"
assert info["litellm_provider"] == provider
assert info["mode"] == "audio_speech"
assert info["input_cost_per_token"] == 1e-06
assert info["output_cost_per_token"] == 2e-05
assert info["max_input_tokens"] == 8192
assert info["max_output_tokens"] == 16384
assert info["max_tokens"] == 16384
assert info["supported_endpoints"] == ["/v1/audio/speech"]
if model == "gemini/gemini-3.1-flash-tts-preview":
assert info["rpm"] == 10
assert info["tpm"] == 4000000
else:
assert info["output_cost_per_audio_token"] == 2e-05
assert info["supported_modalities"] == ["text"]
assert info["supported_output_modalities"] == ["audio"]
assert info["supports_audio_input"] is False
assert info["supports_audio_output"] is True
assert info["supports_function_calling"] is False
assert info["supports_prompt_caching"] is False
assert info["health_check_voice"] == "Kore"
def test_gemini_3_1_flash_tts_backup_matches_main():
repo_root = Path(__file__).parents[2]
main_cost = _load_model_cost_map(repo_root / "model_prices_and_context_window.json")
backup_cost = _load_model_cost_map(repo_root / "litellm" / "model_prices_and_context_window_backup.json")
for model in EXPECTED_MODELS:
assert backup_cost.get(model) == main_cost.get(model), (
f"{model} differs between main and backup model cost maps"
)
def test_gemini_tts_detection_uses_model_metadata(monkeypatch):
model: str = "gemini/future-generative-speech"
monkeypatch.setitem(
litellm.model_cost,
model,
{
"litellm_provider": "gemini",
"mode": "audio_speech",
},
)
litellm.get_model_info.cache_clear()
_invalidate_model_cost_lowercase_map()
assert is_gemini_tts_model(model)

View file

@ -817,6 +817,7 @@ def test_aaamodel_prices_and_context_window_json_is_valid():
"ocr_cost_per_page_batches": {"type": "number"},
"ocr_cost_per_credit": {"type": "number"},
"code_interpreter_cost_per_session": {"type": "number"},
"health_check_voice": {"type": "string"},
"inference_geo": {"type": "string"},
"litellm_provider": {"type": "string"},
"max_input_tokens": {"type": "number"},

View file

@ -42,8 +42,12 @@ class TestVertexAITextToSpeechConfig:
assert url == custom_url
@patch.object(VertexAITextToSpeechConfig, "_ensure_access_token")
@patch.object(VertexAITextToSpeechConfig, "_get_token_and_url")
@patch.object( # test-quality-ok: isolates provider credentials while testing request serialization
VertexAITextToSpeechConfig, "_ensure_access_token"
)
@patch.object( # test-quality-ok: fixes the provider URL at the authentication boundary
VertexAITextToSpeechConfig, "_get_token_and_url"
)
def test_transform_text_to_speech_request_body(self, mock_get_token, mock_ensure_token):
"""Test that transform_text_to_speech_request generates correct request body"""
# Mock authentication
@ -125,6 +129,222 @@ class TestVertexAITextToSpeechConfig:
assert voice_str is None
assert voice_dict == voice_input
def test_gemini_tts_multi_speaker_voice_mapping(self):
config = VertexAITextToSpeechConfig()
voice = {
"multi_speaker_voice_config": {
"speaker_voice_configs": [
{
"speaker": "Ryan",
"voice_config": {
"prebuilt_voice_config": {
"voice_name": "Umbriel",
},
},
},
{
"speaker": "Katie",
"voice_config": {
"prebuilt_voice_config": {
"voice_name": "Leda",
},
},
},
],
},
}
expected_voice = {
"languageCode": "en-US",
"modelName": "gemini-3.1-flash-tts-preview",
"multiSpeakerVoiceConfig": {
"speakerVoiceConfigs": [
{
"speakerAlias": "Ryan",
"speakerId": "Umbriel",
},
{
"speakerAlias": "Katie",
"speakerId": "Leda",
},
],
},
}
voice_str, optional_params = config.map_openai_params(
model="gemini-3.1-flash-tts-preview",
optional_params={"response_format": "mp3"},
voice=voice,
)
assert voice_str is None
assert optional_params["audioEncoding"] == "MP3"
assert optional_params["vertex_voice_dict"] == expected_voice
voice_str, optional_params = config.map_openai_params(
model="gemini-3.1-flash-tts-preview",
optional_params={"response_format": "pcm16"},
voice=voice,
)
assert voice_str is None
assert optional_params["audioEncoding"] == "LINEAR16"
assert optional_params["vertex_voice_dict"] == expected_voice
@pytest.mark.parametrize(
"voice",
[
{"name": "Umbriel", "modelName": "gemini-2.5-flash-tts"},
{"name": "Umbriel", "model_name": "gemini-2.5-flash-tts"},
{"modelName": "gemini-2.5-flash-tts", "model_name": "chirp-3"},
{
"modelName": "gemini-2.5-flash-tts",
"model_name": "chirp-3",
"multi_speaker_voice_config": {
"speaker_voice_configs": [
{
"speaker": "Ryan",
"voice_config": {
"prebuilt_voice_config": {
"voice_name": "Umbriel",
},
},
},
],
},
},
],
)
def test_gemini_tts_ignores_voice_model_name_override(self, voice):
config = VertexAITextToSpeechConfig()
routed_model = "gemini-3.1-flash-tts-preview"
_, optional_params = config.map_openai_params(
model=routed_model,
optional_params={"response_format": "mp3"},
voice=voice,
)
assert optional_params["vertex_voice_dict"]["modelName"] == routed_model
def test_dispatch_maps_gemini_cloud_tts_params_when_provider_config_skipped(self):
config = VertexAITextToSpeechConfig()
handler = MagicMock()
handler.text_to_speech_handler.return_value = "ok"
result = config.dispatch_text_to_speech(
model="gemini-3.1-flash-tts-preview",
input="Hi",
voice={
"name": "Umbriel",
"modelName": "gemini-2.5-flash-tts",
},
optional_params={"response_format": "mp3"},
litellm_params_dict={},
logging_obj=MagicMock(),
timeout=10,
extra_headers=None,
base_llm_http_handler=handler,
aspeech=False,
api_base=None,
api_key=None,
)
assert result == "ok"
mapped_params = handler.text_to_speech_handler.call_args.kwargs["text_to_speech_optional_params"]
assert mapped_params["audioEncoding"] == "MP3"
assert mapped_params["vertex_voice_dict"]["name"] == "Umbriel"
assert mapped_params["vertex_voice_dict"]["modelName"] == "gemini-3.1-flash-tts-preview"
def test_dispatch_keeps_pre_mapped_cloud_tts_params(self):
config = VertexAITextToSpeechConfig()
handler = MagicMock()
handler.text_to_speech_handler.return_value = "ok"
optional_params = {
"audioEncoding": "OGG_OPUS",
"vertex_voice_dict": {"languageCode": "en-US", "name": "en-US-Chirp3-HD-Charon"},
}
config.dispatch_text_to_speech(
model="chirp",
input="Hi",
voice="en-US-Chirp3-HD-Charon",
optional_params=optional_params,
litellm_params_dict={},
logging_obj=MagicMock(),
timeout=10,
extra_headers=None,
base_llm_http_handler=handler,
aspeech=False,
api_base=None,
api_key=None,
)
call_kwargs = handler.text_to_speech_handler.call_args.kwargs
assert call_kwargs["voice"] == "en-US-Chirp3-HD-Charon"
assert call_kwargs["text_to_speech_optional_params"] is optional_params
assert call_kwargs["text_to_speech_optional_params"]["audioEncoding"] == "OGG_OPUS"
@patch.object( # test-quality-ok: isolates provider credentials while testing Gemini request serialization
VertexAITextToSpeechConfig, "_ensure_access_token"
)
@patch.object( # test-quality-ok: fixes the Gemini provider URL at the authentication boundary
VertexAITextToSpeechConfig, "_get_token_and_url"
)
def test_gemini_tts_mp3_request_body(self, mock_get_token, mock_ensure_token):
mock_ensure_token.return_value = ("mock-token", "test-project")
mock_get_token.return_value = ("mock-token", "mock-url")
config = VertexAITextToSpeechConfig()
result = config.transform_text_to_speech_request(
model="gemini-3.1-flash-tts-preview",
input="Ryan: Hi.\nKatie: Hello.",
voice=None,
optional_params={
"audioEncoding": "MP3",
"vertex_voice_dict": {
"languageCode": "en-US",
"modelName": "gemini-3.1-flash-tts-preview",
"multiSpeakerVoiceConfig": {
"speakerVoiceConfigs": [
{
"speakerAlias": "Ryan",
"speakerId": "Umbriel",
},
{
"speakerAlias": "Katie",
"speakerId": "Leda",
},
],
},
},
},
litellm_params={
"vertex_credentials": None,
"vertex_project": "test-project",
"vertex_location": "global",
},
headers={},
)
request_body = result["dict_body"]
assert request_body["input"] == {"text": "Ryan: Hi.\nKatie: Hello."}
assert request_body["voice"] == {
"languageCode": "en-US",
"modelName": "gemini-3.1-flash-tts-preview",
"multiSpeakerVoiceConfig": {
"speakerVoiceConfigs": [
{
"speakerAlias": "Ryan",
"speakerId": "Umbriel",
},
{
"speakerAlias": "Katie",
"speakerId": "Leda",
},
],
},
}
assert request_body["audioConfig"]["audioEncoding"] == "MP3"
@pytest.mark.parametrize(
("audio", "expected_content_type"),
@ -553,9 +773,15 @@ class TestVertexAILyriaTextToSpeechConfig:
assert mock_post.call_args.kwargs["json"] == expected_body
@patch("litellm.llms.custom_httpx.llm_http_handler.HTTPHandler.post")
@patch.object(VertexAITextToSpeechConfig, "_ensure_access_token")
@patch.object(VertexAITextToSpeechConfig, "_get_token_and_url")
@patch( # test-quality-ok: exercises public speech dispatch up to the outbound HTTP boundary
"litellm.llms.custom_httpx.llm_http_handler.HTTPHandler.post"
)
@patch.object( # test-quality-ok: isolates provider credentials in the public API test
VertexAITextToSpeechConfig, "_ensure_access_token"
)
@patch.object( # test-quality-ok: fixes the provider URL for deterministic dispatch assertions
VertexAITextToSpeechConfig, "_get_token_and_url"
)
def test_litellm_speech_vertex_ai_chirp(mock_get_token, mock_ensure_token, mock_post):
"""
Test that litellm.speech(model="vertex_ai/chirp") sends the correct URL and request body
@ -607,3 +833,74 @@ def test_litellm_speech_vertex_ai_chirp(mock_get_token, mock_ensure_token, mock_
assert "headers" in call_kwargs
assert "Authorization" in call_kwargs["headers"]
assert call_kwargs["headers"]["Authorization"] == "Bearer mock-token"
@patch( # test-quality-ok: exercises public Gemini speech dispatch up to the outbound HTTP boundary
"litellm.llms.custom_httpx.llm_http_handler.HTTPHandler.post"
)
@patch.object( # test-quality-ok: isolates provider credentials in the public Gemini API test
VertexAITextToSpeechConfig, "_ensure_access_token"
)
@patch.object( # test-quality-ok: fixes the Gemini provider URL for deterministic dispatch assertions
VertexAITextToSpeechConfig, "_get_token_and_url"
)
def test_litellm_speech_vertex_ai_gemini_tts_mp3_uses_cloud_tts(mock_get_token, mock_ensure_token, mock_post):
mock_ensure_token.return_value = ("mock-token", "test-project")
mock_get_token.return_value = ("mock-token", "mock-url")
mock_response = Mock(spec=httpx.Response)
mock_response.status_code = 200
mock_response.headers = {"content-type": "application/json"}
mock_response.json.return_value = {"audioContent": "SGVsbG8gV29ybGQ="}
mock_post.return_value = mock_response
litellm.speech(
model="vertex_ai/gemini-3.1-flash-tts-preview",
input="Ryan: Hi.\nKatie: Hello.",
voice={
"multi_speaker_voice_config": {
"speaker_voice_configs": [
{
"speaker": "Ryan",
"voice_config": {
"prebuilt_voice_config": {
"voice_name": "Umbriel",
},
},
},
{
"speaker": "Katie",
"voice_config": {
"prebuilt_voice_config": {
"voice_name": "Leda",
},
},
},
],
},
},
response_format="mp3",
vertex_project="test-project",
vertex_location="global",
)
mock_post.assert_called_once()
call_kwargs = mock_post.call_args.kwargs
assert call_kwargs["url"] == "https://texttospeech.googleapis.com/v1/text:synthesize"
request_body = call_kwargs["json"]
assert request_body["audioConfig"]["audioEncoding"] == "MP3"
assert request_body["voice"] == {
"languageCode": "en-US",
"modelName": "gemini-3.1-flash-tts-preview",
"multiSpeakerVoiceConfig": {
"speakerVoiceConfigs": [
{
"speakerAlias": "Ryan",
"speakerId": "Umbriel",
},
{
"speakerAlias": "Katie",
"speakerId": "Leda",
},
],
},
}