scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-29 06:33:28 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 267

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

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

* fix(vertex-ai): keep TTS metadata compatible with the cost map guard

* fix(vertex-ai): preserve existing Gemini TTS models and voice mappings

* fix(vertex-ai): recognize provider-stripped TTS model names

* test(vertex-ai): match current speech HTTP error response

* fix(vertex-ai): preserve nested speech locale and voice

* fix(vertex-ai): use provider speech media type in proxy

* fix(gemini): preserve TTS dispatch and nested speech settings

* test(gemini): run TTS regressions in CI shards

* fix(vertex-ai): account for Gemini Cloud TTS audio cost

* fix(vertex-ai): bill Gemini TTS without optional audio decoder

* test(gemini): cover TTS mapping and billing edge cases

* test(vertex-ai): cover Gemini voice normalization in provider shard

* fix(vertex-ai): keep Gemini TTS routing with generic provider metadata

* chore(vertex-ai): remove obsolete lint suppressions

* fix(gemini): strip audio format from speech configuration

* chore(vertex-ai): remove obsolete mutation suppressions

* fix(vertex-ai): preserve FLAC billing and wrapped audio headers

* fix(vertex-ai): reject unmeasurable FLAC before billing

* fix(vertex-ai): reconcile speech support with current provider mappings

* fix(vertex-ai): validate speech mappings and usage types

* fix(vertex-ai): preserve speech mapper override forwarding

* fix(gemini): map voice aliases across speech request forms

* fix(vertex-ai): send only formats the speech bridge cannot serve to Cloud TTS

Gemini TTS detection now goes by model name, so newer Vertex Gemini TTS
models keep working, and only mp3, opus, alaw, mulaw and other container
formats move to Cloud Text-to-Speech. wav and pcm requests stay on
generateContent as on main.

A Cloud TTS response whose duration cannot be measured now returns the
audio unbilled with a warning instead of raising after a paid call. FLAC
handling is removed because Cloud TTS has no FLAC encoding.

The cost map, the google_genai mappers and the proxy error path go back
to main, since those changes were outside this feature.

* fix(vertex_ai): keep Gemini TTS formats, region and instructions on routes that serve them

Only mp3, opus, alaw and mulaw go to Cloud Text-to-Speech. Every other
format gets a local 400 that lists the supported ones, before Google is
called.

Cloud TTS uses the deployment's us or eu regional host. Other regions
and a custom api_base get a 400 that says to request wav or pcm.

instructions is sent as the Cloud TTS style prompt, and a request with
no voice keeps the Gemini model.

get_usage_object is back to main's code, so dict usages still go
through Usage.__init__, and the proxy no longer labels pcm16 or
linear16 output as WAV.

* fix(vertex_ai): send Gemini TTS to every regional Cloud TTS host and keep speak tags as text

Cloud Text-to-Speech serves Gemini TTS on a host named after each Vertex
location, so a non-global vertex_location now uses
{location}-texttospeech.googleapis.com instead of only the us and eu hosts.
Google answers for regions it does not serve.

Gemini TTS voices reject SSML, so input that contains <speak> is sent as
text for Gemini TTS models. Chirp keeps the SSML auto-detect.

Removes the proxy alaw and mulaw media-type entries. Cloud TTS returns
those encodings with a WAV header, so the response is labeled audio/wav
and the entries never applied.

Adds a regression test that custom per-token pricing bills the usage a
speech response reports.

* fix(vertex_ai): map Gemini TTS regions to Google's Cloud TTS locations and keep a bad voice from cooling down the deployment

US and EU Vertex regions go to the us and eu Cloud TTS hosts, northamerica-northeast1 to its own, global or unset to global, and any other region gets a local 400. A Cloud TTS 404 on the Gemini path, such as a misspelled voice, comes back as a 400 so the router keeps the deployment serving wav. Usage reads the sent request from the response instead of the logging object.

* fix(vertex_ai): default Gemini TTS Cloud formats to a Gemini voice and validate speakers

Parse only usage fields from the sent Cloud TTS request, so a numeric speakingRate no longer fails after Google bills it.
Send the Gemini voice Kore when no voice is named, and reject Cloud TTS voice names unless drop_params is set.
Reject malformed speaker entries unless drop_params is set, and translate Cloud TTS speaker fields on wav and chat audio.
Use the Vertex default region when no location is set, so mp3 and wav share a region.

* test(vertex_ai): cover two-speaker Gemini TTS mp3 through aspeech with fake credentials

Replaces two tests that patched the Vertex TTS config's own token and URL methods.

* fix(vertex_ai): reject malformed Gemini TTS voices and keep beta Chirp and Lyria params unmapped

* fix(vertex_ai): drop unknown Gemini TTS voice keys on wav and pcm instead of rejecting them

* fix(vertex_ai): reject unknown Gemini TTS voice keys on wav, pcm, and chat audio unless drop_params is set

* fix(vertex_ai): honor every drop_params setting in the Gemini TTS voice key check

Global drop_params, batch row and deployment drop_params on Vertex batch
file uploads now drop unknown voice keys. A voice object with both
speechConfig and speech_config gets a 400 for the ignored speech_config.

* fix(gemini): drop unknown nested voice fields under drop_params and detect TTS by served model

* fix(gemini): filter unknown speech config fields with fixed-depth helpers and pass audio params through unchanged

* fix(gemini): camel-case snake replicated voice keys so drop_params keeps them

* fix(vertex-ai): reject voice fields Cloud TTS cannot send unless drop_params is set

* fix(vertex-ai): reject wrong-typed Cloud TTS voice values and check speakers as sent

Gemini TTS on mp3, opus, alaw, or mulaw now answers 400 for a voice field whose value Cloud TTS cannot send, such as a list languageCode or a string voiceConfig, and for nested voice fields under a speaker that also names speakerId. drop_params still drops them.

* fix(vertex-ai): reject wrong-typed speech config wrappers and speakerId overriding a prebuilt voice on Cloud TTS

* fix(vertex-ai): keep speakers and voice_config beside a speech_config wrapper on Cloud TTS

* fix(vertex-ai): pick Cloud TTS voices from the same speech config the WAV path sends to Gemini

* fix(files): carry the deployment's drop_params on model= file uploads so Vertex batch rows drop unknown voice keys

* fix(gemini): keep a structured speech voice when the model is a model_alias_map alias for a Gemini TTS model

---------

Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
Emerson Gomes 2026-10-20 23:18:42 -05:00 • committed by GitHub
parent 4ecc5fd13a
commit 045fb7ba84
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
21 changed files with 2878 additions and 126 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
litellm/cost_calculator.py
View file

@ -1673,8 +1673,10 @@ def completion_cost(
verbose_logger.debug("selected model name for cost calculation: %s", model)
if completion_response is not None and (
isinstance(completion_response, BaseModel) or isinstance(completion_response, dict)
): # tts returns a custom class
isinstance(completion_response, (BaseModel, dict))
or isinstance(completion_response, HttpxBinaryResponseContent)
and completion_response.usage is not None
):
if isinstance(completion_response, dict):
usage_obj: dict | Usage | None = completion_response.get("usage", {})
else:

1
litellm/endpoints/speech/speech_to_completion_bridge/handler.py
View file

@ -122,6 +122,7 @@ class SpeechToCompletionBridgeHandler:
if isinstance(result, ModelResponse):
return self.transformation_handler.transform_response(
model_response=result,
model=model,
response_format=requested_response_format if isinstance(requested_response_format, str) else None,
)
else:

67
litellm/endpoints/speech/speech_to_completion_bridge/transformation.py
View file

@ -5,6 +5,7 @@ from typing import TYPE_CHECKING, Final, cast
from typing_extensions import NotRequired, ReadOnly, TypedDict
from litellm.constants import OPENAI_CHAT_COMPLETION_PARAMS
from litellm.litellm_core_utils.core_helpers import drops_unsupported_params
from litellm.litellm_core_utils.hidden_params import get_hidden_params, served_from_cache, set_hidden_param
if TYPE_CHECKING:
@ -38,6 +39,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]
@ -49,6 +55,20 @@ class SpeechToCompletionBridgeTransformationHandler:
return
from litellm.exceptions import BadRequestError
if custom_llm_provider.startswith("vertex_ai"):
from litellm.llms.vertex_ai.text_to_speech.transformation import VertexAITextToSpeechConfig
vertex_supported: Final = ", ".join(
sorted({*GEMINI_TTS_SUPPORTED_RESPONSE_FORMATS, *VertexAITextToSpeechConfig.GEMINI_FORMAT_MAPPINGS})
)
raise BadRequestError(
message=(
f"Vertex AI Gemini TTS does not support response_format='{response_format}'."
f" Supported response formats: {vertex_supported}."
),
model=model,
llm_provider=custom_llm_provider,
)
supported: Final = ", ".join(sorted(GEMINI_TTS_SUPPORTED_RESPONSE_FORMATS))
raise BadRequestError(
message=(
@ -75,8 +95,27 @@ class SpeechToCompletionBridgeTransformationHandler:
return response_format if isinstance(response_format, str) else None
def _chat_audio_param(
self, model: str, voice: str | Mapping[str, object] | None, optional_params: Mapping[str, object]
) -> ChatAudioParam | None:
self,
model: str,
voice: str | Mapping[str, object] | None,
optional_params: Mapping[str, object],
drop_params: bool,
custom_llm_provider: str,
) -> ChatAudioParam | ChatSpeechConfigAudioParam | None:
import litellm
if isinstance(voice, Mapping) and self._is_gemini_tts_model(litellm.model_alias_map.get(model, model)):
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
normalize_gemini_speech_config,
reject_unknown_gemini_tts_voice_keys,
)
reject_unknown_gemini_tts_voice_keys(model, voice, drop_params, custom_llm_provider)
speech_audio: Final[ChatSpeechConfigAudioParam] = {
"speech_config": normalize_gemini_speech_config(voice, drops_unsupported_params(drop_params)),
"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)
@ -90,9 +129,9 @@ class SpeechToCompletionBridgeTransformationHandler:
self,
model: str,
input: str,
voice: str | dict | None,
optional_params: dict,
litellm_params: dict,
voice: str | Mapping[str, object] | None,
optional_params: Mapping[str, object],
litellm_params: Mapping[str, object],
headers: dict,
litellm_logging_obj: "LiteLLMLoggingObj",
custom_llm_provider: str,
@ -104,7 +143,13 @@ class SpeechToCompletionBridgeTransformationHandler:
"messages": [user_message],
"modalities": ["audio"],
**self._chat_completion_params(optional_params),
"audio": self._chat_audio_param(model, voice, optional_params),
"audio": self._chat_audio_param(
model,
voice,
optional_params,
drops_unsupported_params(litellm_params.get("drop_params")),
custom_llm_provider,
),
**litellm_params,
"headers": headers,
"litellm_logging_obj": litellm_logging_obj,
@ -153,8 +198,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:
@ -162,7 +208,7 @@ class SpeechToCompletionBridgeTransformationHandler:
return self._convert_pcm16_to_wav(decoded_audio), "audio/wav"
def transform_response(
self, model_response: "ModelResponse", response_format: str | None
self, model_response: "ModelResponse", model: str, response_format: str | None
) -> "HttpxBinaryResponseContent":
import base64
@ -176,10 +222,9 @@ class SpeechToCompletionBridgeTransformationHandler:
raise ValueError("No audio part found in the response")
decoded_audio: Final = base64.b64decode(audio_part.data)
model: Final = getattr(model_response, "model", "")
content, content_type = (
self._gemini_tts_response_body(decoded_audio, response_format)
if self._is_gemini_tts_model(model)
if self._is_gemini_tts_model(model) or self._is_gemini_tts_model(model_response.model or "")
else (decoded_audio, "audio/mpeg")
)
response: Final = httpx.Response(

8
litellm/llms/vertex_ai/files/transformation.py
View file

@ -8,6 +8,7 @@ import time
from collections.abc import AsyncGenerator, Callable, Iterable, Iterator, Mapping
from contextlib import aclosing
from dataclasses import dataclass
from functools import partial
from types import MappingProxyType
from typing import IO, Any, Final, TypedDict
from urllib.parse import quote, unquote
@ -30,6 +31,7 @@ from litellm.litellm_core_utils.cloud_storage_security import (
split_configured_cloud_bucket_name,
validate_managed_cloud_file_id,
)
from litellm.litellm_core_utils.core_helpers import drops_unsupported_params
from litellm.litellm_core_utils.litellm_logging import Logging
from litellm.litellm_core_utils.prompt_templates.common_utils import (
extract_file_data,
@ -1103,6 +1105,7 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
def _map_openai_to_vertex_params(
self,
openai_request_body: dict[str, Any],
deployment_drop_params: object = None,
) -> dict[str, object]:
"""
wrapper to call VertexGeminiConfig.map_openai_params
@ -1121,7 +1124,7 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
model=model,
non_default_params=openai_request_body,
optional_params={},
drop_params=False,
drop_params=drops_unsupported_params(openai_request_body.get("drop_params", deployment_drop_params)),
allow_integer_enums=False,
)
return vertex_params
@ -1157,11 +1160,12 @@ class VertexAIFilesConfig(VertexBase, BaseFilesConfig):
create_file_data=create_file_data,
content_type=content_type,
):
deployment_drop_params: Final[object] = litellm_params.get("drop_params")
return {
"streaming_media_upload": StreamingMediaUploadConfig(
body_stream=_OpenAIToVertexBatchUploadStream(
file_data,
self._map_openai_to_vertex_params,
partial(self._map_openai_to_vertex_params, deployment_drop_params=deployment_drop_params),
),
content_type="application/json",
)

334
litellm/llms/vertex_ai/gemini/vertex_and_google_ai_studio_gemini.py
View file

@ -12,7 +12,7 @@ from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, Optional, Protocol, Union, cast, get_args
import httpx
from pydantic import JsonValue
from pydantic import JsonValue, TypeAdapter
import litellm
from litellm import verbose_logger
@ -28,7 +28,7 @@ from litellm.constants import (
DEFAULT_REASONING_EFFORT_MINIMAL_THINKING_BUDGET_GEMINI_2_5_PRO,
)
from litellm.exceptions import UnsupportedParamsError
from litellm.litellm_core_utils.core_helpers import normalize_drop_params
from litellm.litellm_core_utils.core_helpers import as_str_mapping, drops_unsupported_params, normalize_drop_params
from litellm.litellm_core_utils.get_model_cost_map import GetModelCostMap
from litellm.litellm_core_utils.json_fragment_accumulator import JSONFragmentAccumulator
from litellm.litellm_core_utils.prompt_templates.factory import (
@ -232,6 +232,274 @@ 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",
"replicated_voice_config": "replicatedVoiceConfig",
"mime_type": "mimeType",
"voice_sample_audio": "voiceSampleAudio",
"language_code": "languageCode",
"speaker_alias": "speakerAlias",
"speaker_id": "speakerId",
}
)
_SPEECH_CONFIG_ITEMS: Final = TypeAdapter(list[object])
def _normalize_gemini_speech_config_item(value: Mapping[str, object]) -> Mapping[str, object]:
normalized: Final[dict[str, object]] = {} # mutable-ok: iterative traversal fills nested output without recursion
pending: Final[ # mutable-ok: the explicit worklist must grow while avoiding recursive traversal
list[tuple[dict[str, object] | list[object], Mapping[str, 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):
continue
normalized_key: str | int = _GEMINI_SPEECH_CONFIG_KEY_MAP.get(key, key)
else:
normalized_key = len(target)
child: object
if (mapping_item := as_str_mapping(item)) is not None:
pending.append((child_mapping := dict[str, object](), mapping_item))
child = child_mapping
elif isinstance(item, list):
pending.append((child_items := list[object](), _SPEECH_CONFIG_ITEMS.validate_python(item)))
child = child_items
else:
child = (
OPENAI_TO_GEMINI_TTS_VOICES.get(item.lower(), item)
if normalized_key == "voiceName" and isinstance(item, str)
else item
)
if isinstance(target, dict):
assert isinstance(normalized_key, str)
target[normalized_key] = child
else:
target.append(child)
return normalized
def _gemini_speaker_voice_config(item: object) -> object:
speaker_config: Final = as_str_mapping(item)
if speaker_config is None:
return item
speaker: Final = speaker_config.get("speakerAlias")
voice_name: Final = speaker_config.get("speakerId")
if not isinstance(speaker, str) or not isinstance(voice_name, str):
return item
return {
**{key: value for key, value in speaker_config.items() if key not in ("speakerAlias", "speakerId")},
"speaker": speaker,
"voiceConfig": {
"prebuiltVoiceConfig": {"voiceName": OPENAI_TO_GEMINI_TTS_VOICES.get(voice_name.lower(), voice_name)}
},
}
def _with_gemini_speaker_voice_configs(config: Mapping[str, object]) -> Mapping[str, object]:
multi_speaker_config: Final = as_str_mapping(config.get("multiSpeakerVoiceConfig"))
if multi_speaker_config is None:
return config
speaker_voice_configs: Final = multi_speaker_config.get("speakerVoiceConfigs")
if not isinstance(speaker_voice_configs, list):
return config
return {
**config,
"multiSpeakerVoiceConfig": {
**multi_speaker_config,
"speakerVoiceConfigs": [
_gemini_speaker_voice_config(item)
for item in _SPEECH_CONFIG_ITEMS.validate_python(speaker_voice_configs)
],
},
}
_GEMINI_SPEECH_CONFIG_FIELDS: Final = frozenset({"voiceConfig", "multiSpeakerVoiceConfig", "languageCode"})
_GEMINI_VOICE_CONFIG_FIELDS: Final = MappingProxyType(
{
"prebuiltVoiceConfig": frozenset({"voiceName"}),
"replicatedVoiceConfig": frozenset({"mimeType", "voiceSampleAudio"}),
"voice": None,
}
)
_GEMINI_SPEAKER_VOICE_CONFIG_FIELDS: Final = frozenset({"speaker", "voiceConfig"})
def _known_leaf_fields(value: object, fields: frozenset[str] | None) -> object:
mapping: Final = as_str_mapping(value)
if fields is None or mapping is None:
return value
return {key: item for key, item in mapping.items() if key in fields}
def _known_voice_config_fields(value: object) -> object:
mapping: Final = as_str_mapping(value)
if mapping is None:
return value
return {
key: _known_leaf_fields(item, _GEMINI_VOICE_CONFIG_FIELDS[key])
for key, item in mapping.items()
if key in _GEMINI_VOICE_CONFIG_FIELDS
}
def _known_speaker_voice_config_fields(value: object) -> object:
mapping: Final = as_str_mapping(value)
if mapping is None:
return value
return {
key: _known_voice_config_fields(item) if key == "voiceConfig" else item
for key, item in mapping.items()
if key in _GEMINI_SPEAKER_VOICE_CONFIG_FIELDS
}
def _known_speaker_voice_configs(value: object) -> object:
if not isinstance(value, list):
return value
return [_known_speaker_voice_config_fields(item) for item in _SPEECH_CONFIG_ITEMS.validate_python(value)]
def _known_multi_speaker_voice_config_fields(value: object) -> object:
mapping: Final = as_str_mapping(value)
if mapping is None:
return value
return {key: _known_speaker_voice_configs(item) for key, item in mapping.items() if key == "speakerVoiceConfigs"}
def _known_speech_config_field(key: str, value: object) -> object:
match key:
case "voiceConfig":
return _known_voice_config_fields(value)
case "multiSpeakerVoiceConfig":
return _known_multi_speaker_voice_config_fields(value)
case _:
return value
def _known_gemini_speech_fields(
config: Mapping[str, object],
) -> dict[str, object]: # mutable-ok: provider request serialization requires a concrete dict
return {
key: _known_speech_config_field(key, item)
for key, item in config.items()
if key in _GEMINI_SPEECH_CONFIG_FIELDS
}
_GEMINI_TTS_VOICE_KEYS: Final = frozenset(
{
"format",
"name",
"voice",
"voiceName",
"voice_name",
"languageCode",
"language_code",
"modelName",
"model_name",
"speechConfig",
"speech_config",
"voiceConfig",
"voice_config",
"multiSpeakerVoiceConfig",
"multi_speaker_voice_config",
}
)
_NAMED_UNKNOWN_VOICE_KEYS: Final = 10
_NAMED_VOICE_KEY_CHARS: Final = 100
def _gemini_speech_config_wrapper(value: Mapping[str, object]) -> Mapping[str, object] | None:
camel_config: Final = as_str_mapping(value.get("speechConfig"))
return camel_config if camel_config is not None else as_str_mapping(value.get("speech_config"))
def named_voice_keys(keys: Sequence[str]) -> str:
named: Final = ", ".join(key[:_NAMED_VOICE_KEY_CHARS] for key in keys[:_NAMED_UNKNOWN_VOICE_KEYS])
unnamed_count: Final = len(keys) - _NAMED_UNKNOWN_VOICE_KEYS
return f"{named}, and {unnamed_count} more" if unnamed_count > 0 else named
def reject_unknown_gemini_tts_voice_keys(
model: str, voice: Mapping[str, object], drop_params: bool, llm_provider: str | None
) -> None:
nested_config: Final = _gemini_speech_config_wrapper(voice) or {}
ignored_wrapper: Final = (
("speech_config",)
if as_str_mapping(voice.get("speechConfig")) is not None and voice.get("speech_config") is not None
else ()
)
unknown_keys: Final = sorted(
{*(key for key in (*voice, *nested_config) if key not in _GEMINI_TTS_VOICE_KEYS), *ignored_wrapper}
)
if not unknown_keys or drops_unsupported_params(drop_params):
return
raise UnsupportedParamsError(
message=(
f"{model} does not support the voice keys {named_voice_keys(unknown_keys)}. Remove them, or set "
"drop_params to ignore them."
),
model=model,
llm_provider=llm_provider,
)
def normalize_gemini_speech_config(
value: Mapping[str, object], drop_unknown_fields: bool = False
) -> dict[str, object]: # mutable-ok: provider request serialization requires a concrete dict
speech_config: Final = {**_with_gemini_speaker_voice_configs(mapped_gemini_speech_config(value))}
return _known_gemini_speech_fields(speech_config) if drop_unknown_fields else speech_config
def _merge_gemini_speech_config_wrapper(value: Mapping[str, object]) -> Mapping[str, object]:
nested_config: Final = _gemini_speech_config_wrapper(value)
if nested_config is None:
return value
return {
**{key: item for key, item in value.items() if key not in ("speechConfig", "speech_config")},
**nested_config,
}
def mapped_gemini_speech_config(
value: Mapping[str, object],
) -> dict[str, object]: # mutable-ok: provider request serialization requires a concrete dict
normalized: Final = _normalize_gemini_speech_config_item(_merge_gemini_speech_config_wrapper(value))
speech_config: Final = {key: item for key, item in normalized.items() if key in _GEMINI_SPEECH_CONFIG_FIELDS}
voice_name: Final = next(
(normalized[key] for key in ("name", "voiceName", "voice") if isinstance(normalized.get(key), str)),
None,
)
if "voiceConfig" in speech_config or "multiSpeakerVoiceConfig" in speech_config or not isinstance(voice_name, str):
return speech_config
return {
**speech_config,
"voiceConfig": {
"prebuiltVoiceConfig": {"voiceName": OPENAI_TO_GEMINI_TTS_VOICES.get(voice_name.lower(), voice_name)}
},
}
class VertexAIBaseConfig:
def get_mapped_special_auth_params(self) -> dict:
"""
@ -1237,7 +1505,9 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
status_code=400,
)
def _map_audio_params(self, value: dict) -> dict:
def _map_audio_params(
self, value: Mapping[str, object], drop_unknown_fields: bool = False
) -> dict[str, object]: # mutable-ok: the public forwarding interface returns a concrete request dictionary
"""
Expected input:
{
@ -1256,12 +1526,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":
@ -1271,22 +1535,46 @@ 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):
return normalize_gemini_speech_config(value, drop_unknown_fields)
if "voice" in value:
voice: Final = value["voice"]
voice_name: Final = (
OPENAI_TO_GEMINI_TTS_VOICES.get(voice.lower(), voice) if isinstance(voice, str) else voice
)
prebuilt_voice_config: Final[PrebuiltVoiceConfig] = {"voiceName": voice_name}
voice_config: Final[VoiceConfig] = {"prebuiltVoiceConfig": prebuilt_voice_config}
speech_config["voiceConfig"] = voice_config
voice: Final = value.get("voice")
speech_config_without_language: Final[ # mutable-ok: provider request serialization requires a concrete dict
dict[str, object]
] = (
{
"voiceConfig": {
"prebuiltVoiceConfig": {
"voiceName": OPENAI_TO_GEMINI_TTS_VOICES.get(voice.lower(), voice)
if isinstance(voice, str)
else voice,
}
}
}
if "voice" in value
else {}
)
if "language_code" in value:
speech_config["languageCode"] = value["language_code"]
if language_code is not None:
return {
**speech_config_without_language,
"languageCode": language_code,
}
return cast(dict, speech_config)
return speech_config_without_language
def _map_chat_audio_params(
self, model: str, value: Mapping[str, object], drop_params: bool
) -> dict[str, object]: # mutable-ok: the public forwarding interface returns a concrete request dictionary
if _gemini_speech_config_wrapper(value) is not None:
reject_unknown_gemini_tts_voice_keys(model, value, drop_params, self.custom_llm_provider)
return self._map_audio_params(value, drop_unknown_fields=drops_unsupported_params(drop_params))
def map_audio_params(
self,
@ -1351,7 +1639,7 @@ class VertexGeminiConfig(VertexAIBaseConfig, BaseConfig):
elif param == "n":
optional_params["candidate_count"] = value
elif param == "audio" and isinstance(value, dict):
optional_params["speechConfig"] = self._map_audio_params(value)
optional_params["speechConfig"] = self._map_chat_audio_params(model, value, drop_params)
elif param == "stop":
if isinstance(value, str):
optional_params["stop_sequences"] = [value]

567
litellm/llms/vertex_ai/text_to_speech/transformation.py
View file

@ -6,20 +6,25 @@ Reference: https://cloud.google.com/text-to-speech/docs/reference/rest/v1/text/s
"""
import base64
from collections.abc import Coroutine, Mapping
import math
from collections.abc import Coroutine, Iterator, Mapping, Sequence
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
from pydantic import ConfigDict, TypeAdapter
from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError
import litellm
from litellm._logging import verbose_logger
from litellm.exceptions import UnsupportedParamsError
from litellm.litellm_core_utils.audio_utils.utils import (
DEFAULT_SPEECH_MEDIA_TYPE,
calculate_request_duration,
speech_media_type_from_audio_bytes,
)
from litellm.litellm_core_utils.core_helpers import as_str_mapping, drops_unsupported_params
from litellm.litellm_core_utils.url_utils import encode_url_path_segment
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.text_to_speech.transformation import (
BaseTextToSpeechConfig,
TextToSpeechRequestData,
@ -33,6 +38,7 @@ from litellm.types.llms.vertex_ai import VERTEX_CREDENTIALS_TYPES
from litellm.types.llms.vertex_ai_text_to_speech import (
VertexTextToSpeechAudioConfig,
VertexTextToSpeechInput,
VertexTextToSpeechSpeakerVoiceConfig,
VertexTextToSpeechVoice,
)
@ -49,6 +55,166 @@ _LyriaVoice: TypeAlias = str | dict | None
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_ARRAY: Final = TypeAdapter(tuple[object, ...], config=ConfigDict(hide_input_in_errors=True))
_STR: Final = TypeAdapter(str, config=ConfigDict(hide_input_in_errors=True))
_GEMINI_CLOUD_TTS_LOCATIONS: Final = MappingProxyType(
{
**dict.fromkeys(
("us", "us-central1", "us-east1", "us-east4", "us-east5", "us-south1", "us-west1", "us-west4"), "us"
),
**dict.fromkeys(
("eu", "europe-central2", "europe-north1", "europe-southwest1", "europe-west1", "europe-west4"), "eu"
),
"northamerica-northeast1": "northamerica-northeast1",
}
)
_DEFAULT_GEMINI_TTS_VOICE: Final = "Kore"
class _GeminiCloudTtsInput(BaseModel):
model_config = ConfigDict(frozen=True)
text: str | None = None
ssml: str | None = None
prompt: str | None = None
class _GeminiCloudTtsAudioConfig(BaseModel):
model_config = ConfigDict(frozen=True)
audioEncoding: str = ""
sampleRateHertz: int | None = None
class _GeminiCloudTtsRequest(BaseModel):
model_config = ConfigDict(frozen=True)
input: _GeminiCloudTtsInput
audioConfig: _GeminiCloudTtsAudioConfig
def _fallback_gemini_tts_audio_duration(audio: bytes, encoding: str, sample_rate: int) -> float | None:
if encoding == "PCM":
return len(audio) / (2 * sample_rate) if sample_rate > 0 else None
if encoding in ("ALAW", "MULAW") and not (audio[:4] == b"RIFF" and audio[8:12] == b"WAVE"):
return len(audio) / sample_rate if sample_rate > 0 else None
if encoding in ("LINEAR16", "ALAW", "MULAW") and audio[:4] == b"RIFF" and audio[8:12] == b"WAVE":
format_offset: Final = audio.find(b"fmt ", 12)
data_offset: Final = audio.find(b"data", 12)
if format_offset >= 0 and data_offset >= 0 and format_offset + 20 <= len(audio):
byte_rate: Final = int.from_bytes(audio[format_offset + 16 : format_offset + 20], "little")
data_size: Final = int.from_bytes(audio[data_offset + 4 : data_offset + 8], "little")
if byte_rate > 0 and data_size <= len(audio) - data_offset - 8:
return data_size / byte_rate
if encoding == "MP3" and speech_media_type_from_audio_bytes(audio) == "audio/mpeg":
return len(audio) / 4000
if encoding == "OGG_OPUS" and audio[:4] == b"OggS":
opus_header: Final = audio.find(b"OpusHead")
last_page: Final = audio.rfind(b"OggS")
if opus_header >= 0 and opus_header + 12 <= len(audio) and last_page + 14 <= len(audio):
pre_skip: Final = int.from_bytes(audio[opus_header + 10 : opus_header + 12], "little")
granule: Final = int.from_bytes(audio[last_page + 6 : last_page + 14], "little")
if pre_skip <= granule < 2**64 - 1:
return (granule - pre_skip) / 48000
return None
_SPEAKER_CONFIG_ITEMS: Final = TypeAdapter(list[object])
_CloudTtsVoiceFields: TypeAlias = "Mapping[str, _CloudTtsVoiceFields]"
_CLOUD_TTS_VOICE_LEAF: Final[_CloudTtsVoiceFields] = MappingProxyType({})
_CLOUD_TTS_PREBUILT_VOICE_FIELDS: Final[_CloudTtsVoiceFields] = MappingProxyType(
{"prebuiltVoiceConfig": MappingProxyType({"voiceName": _CLOUD_TTS_VOICE_LEAF})}
)
_CLOUD_TTS_SPEECH_CONFIG_FIELDS: Final[_CloudTtsVoiceFields] = MappingProxyType(
{
"languageCode": _CLOUD_TTS_VOICE_LEAF,
"voiceConfig": _CLOUD_TTS_PREBUILT_VOICE_FIELDS,
"multiSpeakerVoiceConfig": MappingProxyType(
{
"speakerVoiceConfigs": MappingProxyType(
{
"speaker": _CLOUD_TTS_VOICE_LEAF,
"speakerAlias": _CLOUD_TTS_VOICE_LEAF,
"speakerId": _CLOUD_TTS_VOICE_LEAF,
"voiceConfig": _CLOUD_TTS_PREBUILT_VOICE_FIELDS,
}
)
}
),
}
)
_CLOUD_TTS_LIST_FIELDS: Final = frozenset({"speakerVoiceConfigs"})
_GEMINI_TTS_VOICE_NAME_KEYS: Final = ("name", "voiceName", "voice_name", "voice")
def _fields_cloud_tts_cannot_send(
config: Mapping[str, object], fields: _CloudTtsVoiceFields, path: str
) -> Iterator[str]:
for key, value in config.items():
yield from _field_cloud_tts_cannot_send(f"{path}{key}", value, fields.get(key), key in _CLOUD_TTS_LIST_FIELDS)
def _field_cloud_tts_cannot_send(
path: str, value: object, fields: "_CloudTtsVoiceFields | None", is_list_field: bool
) -> Iterator[str]:
if fields is None:
yield path
return
if value is None:
return
if not fields:
yield from () if isinstance(value, str) else (path,)
return
items: Final = (
_SPEAKER_CONFIG_ITEMS.validate_python(value) if is_list_field and isinstance(value, list) else (value,)
)
for item in items:
mapping = as_str_mapping(item)
yield from (path,) if mapping is None else _fields_cloud_tts_cannot_send(mapping, fields, f"{path}.")
def _non_string_voice_names(config: Mapping[str, object], path: str) -> Iterator[str]:
return (
f"{path}{key}"
for key in _GEMINI_TTS_VOICE_NAME_KEYS
if config.get(key) is not None and not isinstance(config.get(key), str)
)
def _non_mapping_speech_config_wrappers(voice: Mapping[str, object]) -> Iterator[str]:
return (
key
for key in ("speechConfig", "speech_config")
if voice.get(key) is not None and as_str_mapping(voice.get(key)) is None
)
def _speaker_entries(config: Mapping[str, object]) -> Iterator[Mapping[str, object]]:
multi_speaker_config: Final = as_str_mapping(config.get("multiSpeakerVoiceConfig"))
speakers: Final = multi_speaker_config.get("speakerVoiceConfigs") if multi_speaker_config is not None else None
items: Final = _SPEAKER_CONFIG_ITEMS.validate_python(speakers) if isinstance(speakers, list) else ()
return (entry for item in items if (entry := as_str_mapping(item)) is not None)
def _prebuilt_voices_replaced_by_speaker_id(config: Mapping[str, object]) -> Iterator[str]:
return (
"multiSpeakerVoiceConfig.speakerVoiceConfigs.voiceConfig.prebuiltVoiceConfig"
for entry in _speaker_entries(config)
if entry.get("speakerId") is not None
and (voice_config := as_str_mapping(entry.get("voiceConfig"))) is not None
and voice_config.get("prebuiltVoiceConfig") is not None
)
class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
@ -81,14 +247,19 @@ 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(
{"mp3": "MP3", "opus": "OGG_OPUS", "alaw": "ALAW", "mulaw": "MULAW"}
)
def __init__(self) -> None:
BaseTextToSpeechConfig.__init__(self)
@ -96,7 +267,9 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
def _map_voice_to_vertex_format(
self,
voice: str | dict | None,
voice: str | dict[str, object] | None,
model: str | None = None,
drop_params: bool = False,
) -> tuple[str | None, dict | None]:
"""
Map voice to Vertex AI format.
@ -111,6 +284,9 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
- voice_str: Original string voice (for interface compatibility)
- voice_dict: Vertex AI format dict with languageCode and name
"""
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, drop_params=drop_params)
if voice is None:
return None, None
@ -142,11 +318,225 @@ 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)
@staticmethod
def _get_str_value(
source: Mapping[str, object],
*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: Mapping[str, object],
*keys: str,
) -> Mapping[str, object] | None:
for key in keys:
value = source.get(key)
if (mapping := as_str_mapping(value)) is not None:
return mapping
return None
@staticmethod
def _resolve_gemini_tts_voice_name(model: str, voice_name: str, drop_params: bool) -> str:
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import OPENAI_TO_GEMINI_TTS_VOICES
gemini_voice_name: Final = OPENAI_TO_GEMINI_TTS_VOICES.get(voice_name.lower(), voice_name)
if "-" not in gemini_voice_name:
return gemini_voice_name
if drop_params:
return _DEFAULT_GEMINI_TTS_VOICE
raise UnsupportedParamsError(
message=(
f"{model} speaks Gemini voices such as {_DEFAULT_GEMINI_TTS_VOICE}, and {voice_name} is a Cloud "
"Text-to-Speech voice. Use a Gemini voice, or set drop_params to use the default Gemini voice."
),
model=model,
llm_provider="vertex_ai",
)
@staticmethod
def _reject_malformed_gemini_tts_voice(model: str, drop_params: bool) -> None:
if drop_params:
return
raise UnsupportedParamsError(
message=(
f"{model} needs voice to be a Gemini voice name such as {_DEFAULT_GEMINI_TTS_VOICE} or a speech "
"config object. Send one, or set drop_params to use the default Gemini voice."
),
model=model,
llm_provider="vertex_ai",
)
@staticmethod
def _reject_voice_fields_cloud_tts_cannot_send(model: str, voice: Mapping[str, object], drop_params: bool) -> None:
if drop_params:
return
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
mapped_gemini_speech_config,
named_voice_keys,
)
speech_config: Final = VertexAITextToSpeechConfig._get_dict_value(voice, "speechConfig", "speech_config")
mapped_config: Final = mapped_gemini_speech_config(voice)
unsendable: Final = sorted(
{
*_fields_cloud_tts_cannot_send(mapped_config, _CLOUD_TTS_SPEECH_CONFIG_FIELDS, ""),
*_prebuilt_voices_replaced_by_speaker_id(mapped_config),
*_non_mapping_speech_config_wrappers(voice),
*_non_string_voice_names(voice, ""),
*_non_string_voice_names(speech_config or {}, "speechConfig."),
}
)
if not unsendable:
return
raise UnsupportedParamsError(
message=(
f"{model} cannot send the voice fields {named_voice_keys(unsendable)} as mp3, opus, alaw, or mulaw. "
"Request wav or pcm, remove them, or set drop_params to ignore them."
),
model=model,
llm_provider="vertex_ai",
)
@staticmethod
def _reject_malformed_gemini_tts_speaker(model: str, drop_params: bool) -> None:
if drop_params:
return
raise UnsupportedParamsError(
message=(
f"{model} needs each entry of speaker_voice_configs to name a speaker and a voice, as speaker with "
"voice_config.prebuilt_voice_config.voice_name, or speakerAlias with speakerId."
),
model=model,
llm_provider="vertex_ai",
)
def _extract_gemini_tts_speaker_configs(
self,
model: str,
speech_config: Mapping[str, object],
drop_params: bool,
) -> Sequence[VertexTextToSpeechSpeakerVoiceConfig]:
multi_speaker_config: Final = self._get_dict_value(speech_config, "multiSpeakerVoiceConfig")
if multi_speaker_config is None:
return []
raw_speaker_configs: Final = multi_speaker_config.get("speakerVoiceConfigs", [])
try:
validated_speaker_configs: Final = _SPEAKER_CONFIG_ITEMS.validate_python(raw_speaker_configs, strict=True)
except ValidationError:
self._reject_malformed_gemini_tts_speaker(model, drop_params)
return []
speaker_configs: Final[ # mutable-ok: validated payloads are accumulated for serialization
list[VertexTextToSpeechSpeakerVoiceConfig]
] = []
for raw_item in validated_speaker_configs:
if (raw_config := as_str_mapping(raw_item)) is None:
self._reject_malformed_gemini_tts_speaker(model, drop_params)
continue
speaker_alias = self._get_str_value(raw_config, "speakerAlias", "speaker")
speaker_id = self._get_str_value(raw_config, "speakerId") or self._prebuilt_voice_name(raw_config)
if speaker_alias is None or speaker_id is None:
self._reject_malformed_gemini_tts_speaker(model, drop_params)
continue
speaker_configs.append(
{
"speakerAlias": speaker_alias,
"speakerId": self._resolve_gemini_tts_voice_name(model, speaker_id, drop_params),
}
)
return speaker_configs
def _prebuilt_voice_name(self, config: Mapping[str, object]) -> str | None:
voice_config: Final = self._get_dict_value(config, "voiceConfig")
prebuilt_voice_config: Final = (
self._get_dict_value(voice_config, "prebuiltVoiceConfig") if voice_config is not None else None
)
if prebuilt_voice_config is None:
return None
return self._get_str_value(prebuilt_voice_config, "voiceName")
def _map_gemini_tts_voice_to_vertex_format(
self,
model: str,
voice: object,
drop_params: bool = False,
) -> tuple[str | None, dict[str, object]]: # mutable-ok: provider request serialization requires a concrete dict
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
mapped_gemini_speech_config,
reject_unknown_gemini_tts_voice_keys,
)
default_voice: Final[dict[str, object]] = {
"languageCode": self.DEFAULT_LANGUAGE_CODE,
"modelName": model,
"name": _DEFAULT_GEMINI_TTS_VOICE,
}
if voice is None:
return None, default_voice
if isinstance(voice, str) and voice:
return voice, {
"languageCode": self.DEFAULT_LANGUAGE_CODE,
"modelName": model,
"name": self._resolve_gemini_tts_voice_name(model, voice, drop_params),
}
voice_config: Final = as_str_mapping(voice)
if voice_config is None:
self._reject_malformed_gemini_tts_voice(model, drop_params)
return None, default_voice
reject_unknown_gemini_tts_voice_keys(model, voice_config, drop_params, "vertex_ai")
speech_config: Final = mapped_gemini_speech_config(voice_config)
language_code: Final = self._get_str_value(speech_config, "languageCode") or self.DEFAULT_LANGUAGE_CODE
model_name: Final = model
speaker_configs: Final = self._extract_gemini_tts_speaker_configs(model, speech_config, drop_params)
self._reject_voice_fields_cloud_tts_cannot_send(model, voice_config, drop_params)
if speaker_configs:
return None, {
"languageCode": language_code,
"modelName": model_name,
"multiSpeakerVoiceConfig": {
"speakerVoiceConfigs": speaker_configs,
},
}
voice_name: Final = self._prebuilt_voice_name(speech_config)
if voice_name is not None:
return None, {
"languageCode": language_code,
"modelName": model_name,
"name": self._resolve_gemini_tts_voice_name(model, voice_name, drop_params),
}
return None, {
"languageCode": language_code,
"modelName": model_name,
"name": _DEFAULT_GEMINI_TTS_VOICE,
}
@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,
input: str,
voice: str | dict | None,
voice: str | dict[str, object] | None,
optional_params: dict[str, object],
litellm_params_dict: dict[str, object],
logging_obj: "LiteLLMLoggingObj",
@ -175,14 +565,17 @@ 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.map_openai_params(
model=model,
voice=voice,
optional_params=optional_params,
drop_params=drops_unsupported_params(kwargs.get("drop_params")),
kwargs=kwargs,
)
if self._is_gemini_tts_model(model)
else (self._dispatch_voice_name(voice), optional_params)
)
# Store credentials in litellm_params for use in transform methods
litellm_params_dict.update(
@ -198,9 +591,11 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
response: Final = base_llm_http_handler.text_to_speech_handler(
model=model,
input=input,
voice=voice_str,
text_to_speech_provider_config=self,
text_to_speech_optional_params=optional_params,
voice=mapped_voice,
text_to_speech_provider_config=(
VertexAIGeminiCloudTextToSpeechConfig() if self._is_gemini_tts_model(model) else self
),
text_to_speech_optional_params=mapped_params,
custom_llm_provider="vertex_ai",
litellm_params=litellm_params_dict,
logging_obj=logging_obj,
@ -219,13 +614,13 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
Note: Vertex AI also supports additional parameters like audioConfig
which can be passed but are not part of the OpenAI spec
"""
return ["voice", "response_format", "speed"]
return ["voice", "response_format", "speed", *(["instructions"] if self._is_gemini_tts_model(model) else [])]
def map_openai_params(
self,
model: str,
optional_params: dict,
voice: str | dict | None = None,
voice: str | dict[str, object] | None = None,
drop_params: bool = False,
kwargs: dict = {},
) -> tuple[str | None, dict]:
@ -244,19 +639,28 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
Tuple of (mapped_voice_str, mapped_params)
"""
mapped_params: Final[dict[str, object]] = {}
openai_params: Final = _JSON_OBJECT.validate_python(optional_params)
drops_params: Final = drop_params or litellm.drop_params is True
##########################################################
# 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,
drop_params=drops_params,
)
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
@ -265,10 +669,19 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
mapped_params["audioEncoding"] = self.DEFAULT_AUDIO_ENCODING
# Map speed (OpenAI: 0.25-4.0, Vertex AI: speakingRate 0.25-4.0)
if "speed" in optional_params:
speed: Final = optional_params["speed"]
if speed is not None:
mapped_params["speakingRate"] = str(speed)
speed: Final = openai_params.get("speed")
if speed is not None:
mapped_params["speakingRate"] = str(speed)
instructions: Final = openai_params.get("instructions")
if isinstance(instructions, str) and self._is_gemini_tts_model(model):
mapped_params["prompt"] = instructions
elif instructions is not None and self._is_gemini_tts_model(model) and not drops_params:
raise UnsupportedParamsError(
message=f"{model} needs instructions to be a string. Send one, or set drop_params to ignore it.",
model=model,
llm_provider="vertex_ai",
)
# Pass through Vertex AI-specific parameters from kwargs
if "audioConfig" in kwargs:
@ -311,15 +724,48 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
Google Cloud TTS endpoint: https://texttospeech.googleapis.com/v1/text:synthesize
"""
if self._is_gemini_tts_model(model):
return self._get_gemini_tts_url(
model=model, api_base=api_base, litellm_params=_JSON_OBJECT.validate_python(litellm_params)
)
if api_base:
return api_base
return self.TTS_API_URL
def _get_gemini_tts_url(self, model: str, api_base: str | None, litellm_params: Mapping[str, object]) -> str:
if api_base:
raise UnsupportedParamsError(
message=(
f"{model} serves mp3, opus, alaw and mulaw through Cloud Text-to-Speech, which does not use "
"this deployment's api_base. Request wav or pcm, or use a deployment without api_base."
),
model=model,
llm_provider="vertex_ai",
)
location: Final = litellm_params.get("vertex_location")
vertex_location: Final = location if isinstance(location, str) else self.get_default_vertex_location()
if vertex_location == "global":
return self.TTS_API_URL
cloud_tts_location: Final = _GEMINI_CLOUD_TTS_LOCATIONS.get(vertex_location)
if cloud_tts_location is None:
raise UnsupportedParamsError(
message=(
f"{model} serves mp3, opus, alaw and mulaw through Cloud Text-to-Speech, which has no Gemini TTS "
f"location for vertex_location {vertex_location}. Use a deployment in global, us, eu, "
"northamerica-northeast1, or a US or EU Vertex AI region."
),
model=model,
llm_provider="vertex_ai",
)
return f"https://{cloud_tts_location}-texttospeech.googleapis.com/v1/text:synthesize"
def _validate_vertex_input(
self,
input_data: VertexTextToSpeechInput,
optional_params: dict,
detect_ssml: bool,
) -> VertexTextToSpeechInput:
"""
Validate and transform input for Vertex AI TTS
@ -343,7 +789,7 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
else:
# LiteLLM will auto-detect if text is in ssml format
# check if "text" is an ssml - in this case we should pass it as ssml instead of text
if input_data:
if detect_ssml and input_data:
_text: Final = input_data.get("text", None) or ""
if "<speak>" in _text:
input_data["ssml"] = input_data.pop("text")
@ -403,8 +849,13 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
headers["x-goog-user-project"] = vertex_project
####### Build the request ################
vertex_input = VertexTextToSpeechInput(text=input)
vertex_input = self._validate_vertex_input(vertex_input, optional_params)
prompt: Final = _JSON_OBJECT.validate_python(optional_params).get("prompt")
validated_input: Final = self._validate_vertex_input(
VertexTextToSpeechInput(text=input), optional_params, detect_ssml=not self._is_gemini_tts_model(model)
)
vertex_input: Final[VertexTextToSpeechInput] = (
{**validated_input, "prompt": prompt} if isinstance(prompt, str) else validated_input
)
# Build voice configuration
# Check for voice dict stored in:
@ -412,8 +863,8 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
# 2. optional_params by map_openai_params
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):
vertex_voice = _JSON_OBJECT.validate_python(voice_dict)
elif voice is not None:
# Handle string voice (shouldn't normally happen if dispatch was called)
parts: Final = voice.split("-")
if len(parts) >= 2:
@ -485,8 +936,44 @@ class VertexAITextToSpeechConfig(BaseTextToSpeechConfig, VertexBase):
content=binary_data,
)
# Initialize the HttpxBinaryResponseContent instance
return HttpxBinaryResponseContent(response)
binary_response: Final = HttpxBinaryResponseContent(response)
if self._is_gemini_tts_model(model):
from litellm.types.utils import CompletionTokensDetailsWrapper, Usage
request: Final = _GeminiCloudTtsRequest.model_validate_json(raw_response.request.content)
input_data: Final = request.input
audio_config: Final = request.audioConfig
container_duration: Final = calculate_request_duration(binary_data)
sample_rate: Final = audio_config.sampleRateHertz or 24000
duration: Final = (
container_duration
if container_duration is not None
else _fallback_gemini_tts_audio_duration(binary_data, audio_config.audioEncoding, sample_rate)
)
if duration is None:
verbose_logger.warning(
"Could not measure Gemini TTS audio duration for %s, so its audio tokens are not billed", model
)
return binary_response
input_text: Final = " ".join(
value for value in (input_data.text, input_data.ssml, input_data.prompt) if value
)
prompt_tokens: Final = litellm.token_counter(model=model, text=input_text)
audio_tokens: Final = math.ceil(duration * 25)
binary_response.usage = Usage(
prompt_tokens=prompt_tokens,
completion_tokens=audio_tokens,
total_tokens=prompt_tokens + audio_tokens,
completion_tokens_details=CompletionTokensDetailsWrapper(audio_tokens=audio_tokens),
)
return binary_response
class VertexAIGeminiCloudTextToSpeechConfig(VertexAITextToSpeechConfig):
def get_error_class(self, error_message: str, status_code: int, headers: dict[str, str]) -> BaseLLMException:
return super().get_error_class(
error_message=error_message, status_code=400 if status_code == 404 else status_code, headers=headers
)
class VertexAILyriaTextToSpeechConfig(VertexAITextToSpeechConfig):

7
litellm/main.py
View file

@ -8846,8 +8846,11 @@ def speech(
generic_optional_params: Final = GenericLiteLLMParams.model_validate(kwargs)
# Handle Gemini models separately (they use speech_to_completion_bridge)
if "gemini" in model:
needs_cloud_text_to_speech: Final = (
litellm.utils.is_gemini_tts_model(model)
and response_format in VertexAITextToSpeechConfig.GEMINI_FORMAT_MAPPINGS
)
if "gemini" in model and not needs_cloud_text_to_speech:
from .endpoints.speech.speech_to_completion_bridge.handler import (
speech_to_completion_bridge_handler,
)

12
litellm/proxy/openai_files_endpoints/common_utils.py
View file

@ -477,6 +477,18 @@ async def get_authorized_credentials_for_model(
)
def deployment_drop_params(
llm_router: Optional["Router"], model_id: str, request: Mapping[str, object]
) -> Mapping[str, bool | str]:
deployment: Final = (
llm_router.get_credential_deployment(model_id=model_id)
if llm_router is not None and "drop_params" not in request
else None
)
drop_params: Final = deployment.litellm_params.drop_params if deployment is not None else None
return MappingProxyType({}) if drop_params is None else MappingProxyType({"drop_params": drop_params})
def _key_allowlist_covers(public_model_name: str, key_model_allowlist: Sequence[str]) -> bool:
from litellm.proxy.auth.auth_checks import model_matches_any_wildcard_pattern_in_list

3
litellm/proxy/openai_files_endpoints/files_endpoints.py
View file

@ -76,6 +76,7 @@ from litellm.proxy.openai_files_endpoints.common_utils import ( # noqa: F401 #
add_internal_model_credentials,
apply_team_provider_credentials,
authorize_model_for_key,
deployment_drop_params,
encode_file_id_with_model,
extract_file_creation_params,
get_authorized_credentials_for_model,
@ -457,7 +458,7 @@ async def route_create_file(
# Merge credentials into the request
prepare_data_with_credentials(
data=_create_file_request,
credentials=credentials,
credentials={**credentials, **deployment_drop_params(llm_router, model, _create_file_request)},
)
# Create the file with model credentials

10
litellm/types/llms/vertex_ai.py
View file

@ -239,8 +239,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

17
litellm/types/llms/vertex_ai_text_to_speech.py
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):

5
litellm/utils.py
View file

@ -860,6 +860,11 @@ 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) -> bool:
model_name: Final = model.lower()
return "gemini" in model_name and "tts" in model_name
def _remove_thought_signature_from_id(tool_call_id: str, separator: str) -> str:
"""
Remove thought signature from a tool call ID.

BIN
tests/audio_tests/gemini_tts_speech.mp3 Normal file
View file

Binary file not shown.

BIN
tests/audio_tests/gemini_tts_speech.ogg Normal file
View file

Binary file not shown.

121
tests/unit/endpoints/speech/speech_to_completion_bridge/test_transformation.py
View file

@ -11,14 +11,14 @@ from litellm.endpoints.speech.speech_to_completion_bridge.transformation import
)
from litellm.types.utils import ChatCompletionAudioResponse, Choices, Message, ModelResponse
pytestmark = pytest.mark.usefixtures("local_model_cost_map")
GEMINI_TTS_MODEL: Final = "gemini-3.1-flash-tts-preview"
PCM_BYTES: Final = b"\x01\x02\x03\x04" * 6
def _model_response(model: str, pcm: bytes) -> ModelResponse:
audio: Final = ChatCompletionAudioResponse(
data=base64.b64encode(pcm).decode(), expires_at=0, transcript="hello"
)
audio: Final = ChatCompletionAudioResponse(data=base64.b64encode(pcm).decode(), expires_at=0, transcript="hello")
return ModelResponse(model=model, choices=[Choices(message=Message(content=None, audio=audio))])
@ -72,6 +72,103 @@ def test_non_gemini_request_forwards_speech_response_format_as_audio_format() ->
assert request["audio"] == {"voice": "alloy", "format": "wav"}
def test_gemini_request_keeps_voice_beside_nested_speech_config() -> None:
request: Final = SpeechToCompletionBridgeTransformationHandler().transform_request(
model=GEMINI_TTS_MODEL,
input="Bonjour",
voice={"speech_config": {"language_code": "fr-FR"}, "voice": "Kore"},
optional_params={},
litellm_params={},
headers={},
litellm_logging_obj=MagicMock(),
custom_llm_provider="gemini",
)
assert request["audio"]["speech_config"] == {
"languageCode": "fr-FR",
"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Kore"}},
}
def test_gemini_request_keeps_voice_object_for_a_model_alias_map_alias(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(litellm, "model_alias_map", {"voice-alias": GEMINI_TTS_MODEL})
request: Final = SpeechToCompletionBridgeTransformationHandler().transform_request(
model="voice-alias",
input="Hello there",
voice={"voice": "Puck"},
optional_params={},
litellm_params={},
headers={},
litellm_logging_obj=MagicMock(),
custom_llm_provider="gemini",
)
assert request["audio"] == {
"speech_config": {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Puck"}}},
"format": "pcm16",
}
@pytest.mark.parametrize("wrapper", (None, "speech_config", "speechConfig"))
def test_gemini_request_keeps_voice_and_language_without_audio_format(wrapper: str | None) -> None:
voice: Final = {"language_code": "fr-FR", "voice": "Kore", "format": "pcm16"}
request: Final = SpeechToCompletionBridgeTransformationHandler().transform_request(
model=GEMINI_TTS_MODEL,
input="Bonjour",
voice=voice if wrapper is None else {wrapper: voice},
optional_params={},
litellm_params={},
headers={},
litellm_logging_obj=MagicMock(),
custom_llm_provider="gemini",
)
assert request["audio"]["speech_config"] == {
"languageCode": "fr-FR",
"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Kore"}},
}
@pytest.mark.parametrize(
("drop_params", "voice_config"),
[
(True, {"prebuiltVoiceConfig": {"voiceName": "Kore"}}),
(False, {"prebuiltVoiceConfig": {"voiceName": "Kore", "style": "calm"}, "tone": "warm"}),
],
)
def test_gemini_request_drops_unknown_nested_voice_fields_only_with_drop_params(
drop_params: bool, voice_config: dict[str, object]
) -> None:
request: Final = SpeechToCompletionBridgeTransformationHandler().transform_request(
model=GEMINI_TTS_MODEL,
input="Bonjour",
voice={"voice_config": {"prebuilt_voice_config": {"voice_name": "Kore", "style": "calm"}, "tone": "warm"}},
optional_params={},
litellm_params={"drop_params": drop_params},
headers={},
litellm_logging_obj=MagicMock(),
custom_llm_provider="gemini",
)
assert request["audio"]["speech_config"] == {"voiceConfig": voice_config}
def test_gemini_request_reads_snake_case_wrapper_beside_null_camel_case_wrapper() -> None:
request: Final = SpeechToCompletionBridgeTransformationHandler().transform_request(
model=GEMINI_TTS_MODEL,
input="Bonjour",
voice={"speechConfig": None, "speech_config": {"voice_name": "Kore"}},
optional_params={},
litellm_params={},
headers={},
litellm_logging_obj=MagicMock(),
custom_llm_provider="gemini",
)
assert request["audio"]["speech_config"] == {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Kore"}}}
@pytest.mark.parametrize("response_format", ["mp3", "flac", "opus", "aac"])
def test_gemini_tts_request_rejects_formats_gemini_cannot_produce(response_format: str) -> None:
with pytest.raises(litellm.BadRequestError) as excinfo:
@ -86,6 +183,7 @@ def test_gemini_tts_request_rejects_formats_gemini_cannot_produce(response_forma
def test_gemini_tts_pcm_response_returns_raw_pcm_bytes() -> None:
response: Final = SpeechToCompletionBridgeTransformationHandler().transform_response(
model_response=_model_response(GEMINI_TTS_MODEL, PCM_BYTES),
model=GEMINI_TTS_MODEL,
response_format="pcm",
)
@ -97,6 +195,7 @@ def test_gemini_tts_pcm_response_returns_raw_pcm_bytes() -> None:
def test_gemini_tts_wav_and_default_responses_wrap_pcm_in_wav(response_format: str | None) -> None:
response: Final = SpeechToCompletionBridgeTransformationHandler().transform_response(
model_response=_model_response(GEMINI_TTS_MODEL, PCM_BYTES),
model=GEMINI_TTS_MODEL,
response_format=response_format,
)
@ -110,8 +209,24 @@ def test_gemini_tts_wav_and_default_responses_wrap_pcm_in_wav(response_format: s
def test_non_gemini_response_keeps_original_bytes_and_mpeg_content_type() -> None:
response: Final = SpeechToCompletionBridgeTransformationHandler().transform_response(
model_response=_model_response("gpt-4o-audio-preview", PCM_BYTES),
model="gpt-4o-audio-preview",
response_format="mp3",
)
assert response.response.content == PCM_BYTES
assert response.response.headers["content-type"] == "audio/mpeg"
@pytest.mark.parametrize(
("requested_model", "served_model"),
[(GEMINI_TTS_MODEL, "served-speech-alias"), ("requested-speech-alias", GEMINI_TTS_MODEL)],
)
def test_gemini_response_wraps_wav_when_either_model_name_is_gemini_tts(requested_model: str, served_model: str) -> None:
response: Final = SpeechToCompletionBridgeTransformationHandler().transform_response(
model_response=_model_response(served_model, PCM_BYTES),
model=requested_model,
response_format=None,
)
assert response.response.content[:4] == b"RIFF"
assert response.response.headers["content-type"] == "audio/wav"

262
tests/unit/llms/gemini/test_gemini_tts.py
View file

@ -2,14 +2,70 @@
Test Gemini TTS (Text-to-Speech) functionality
"""
import pytest
from typing import Final
from unittest.mock import patch, MagicMock
import pytest
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
pytestmark = pytest.mark.usefixtures("local_model_cost_map")
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 +75,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 +121,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 +130,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(
@ -102,6 +152,16 @@ class TestGeminiTTSTransformation:
assert result["languageCode"] == "de-DE"
assert result["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == "Kore"
def test_map_audio_params_keeps_voice_beside_nested_speech_config(self):
config = GoogleAIStudioGeminiConfig()
result = config._map_audio_params(
{"speech_config": {"language_code": "fr-FR"}, "voice": "Kore", "format": "pcm16"}
)
assert result["languageCode"] == "fr-FR"
assert result["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == "Kore"
def test_map_audio_params_no_language_code(self):
config = GoogleAIStudioGeminiConfig()
@ -110,6 +170,55 @@ class TestGeminiTTSTransformation:
assert "languageCode" not in result
assert result["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == "Kore"
def test_map_audio_params_keeps_nested_speaker_list_and_drops_unknown_fields(self):
config: Final = GoogleAIStudioGeminiConfig()
result: Final = config._map_audio_params(
{
"speech_config": {
"language_code": "fr-FR",
"multi_speaker_voice_config": {
"speaker_voice_configs": [
{"speaker": "Ryan", "voice_config": {"prebuilt_voice_config": {"voice_name": "Umbriel"}}}
]
},
"metadata": ["studio", 2],
},
"format": "pcm16",
}
)
assert result == {
"languageCode": "fr-FR",
"multiSpeakerVoiceConfig": {
"speakerVoiceConfigs": [
{"speaker": "Ryan", "voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Umbriel"}}}
]
},
}
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"]
def test_gemini_tts_audio_parameter_with_existing_modalities(self):
"""Test audio parameter mapping when modalities already exist"""
config = GoogleAIStudioGeminiConfig()
@ -193,9 +302,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,
@ -234,6 +341,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.
@ -250,6 +379,8 @@ 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):
@ -283,12 +414,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",
@ -296,6 +422,8 @@ 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):
@ -342,17 +470,54 @@ 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"]
@pytest.mark.parametrize(
"model,custom_llm_provider",
@ -395,14 +560,35 @@ 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"]
if __name__ == "__main__":
pytest.main([__file__])
@pytest.mark.parametrize("voice_key", ["name", "voiceName", "voice_name", "voice"])
@pytest.mark.parametrize("wrapper", [None, "speechConfig", "speech_config"])
@pytest.mark.parametrize("model", ["gemini-2.5-flash-tts", "gemini-2.5-pro-tts", GEMINI_3_1_FLASH_TTS_MODEL])
def test_speech_bridge_preserves_single_speaker_voice_mapping(voice_key: str, wrapper: str | None, model: str):
handler = SpeechToCompletionBridgeTransformationHandler()
config = {voice_key: "Umbriel", "language_code": "en-US"}
voice = {wrapper: config} if wrapper else config
result = handler.transform_request(
model=model,
input="Hello",
voice=voice,
optional_params={},
litellm_params={},
headers={},
litellm_logging_obj=MagicMock(),
custom_llm_provider="vertex_ai",
)
assert result["audio"] == {
"speech_config": {
"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Umbriel"}},
"languageCode": "en-US",
},
"format": "pcm16",
}

40
tests/unit/llms/vertex_ai/files/test_vertex_ai_files_streaming.py
View file

@ -278,6 +278,46 @@ class TestGetObjectNameLazyParse:
assert exc_info.value.message.startswith("Batch input file line 1 could not be converted")
def _tts_batch_row(**body_extra: object) -> bytes:
body: Final = {
"model": "gemini-2.5-flash-preview-tts",
"messages": [{"role": "user", "content": "Say hi"}],
"audio": {"format": "pcm16", "speech_config": {"voice_name": "Kore", "style": "calm"}},
**body_extra,
}
return json.dumps({"custom_id": "r-0", "method": "POST", "url": "/v1/chat/completions", "body": body}).encode()
def _upload_tts_batch_row(row: bytes, litellm_params: dict[str, object]) -> bytes:
request: Final[CreateFileRequest] = {"file": ("batch.jsonl", row + b"\n", "application/jsonl"), "purpose": "batch"}
return _join_upload_body(
VertexAIFilesConfig().transform_create_file_request(
model="", create_file_data=request, optional_params={}, litellm_params=litellm_params
)
)
class TestBatchRowUnknownVoiceKeys:
def test_unknown_voice_keys_are_a_400_without_drop_params(self) -> None:
with pytest.raises(VertexAIError, match="voice keys style") as exc_info:
_upload_tts_batch_row(_tts_batch_row(), {})
assert exc_info.value.status_code == 400
@pytest.mark.parametrize(
"row, litellm_params",
[(_tts_batch_row(drop_params=True), {}), (_tts_batch_row(), {"drop_params": True})],
ids=["row", "deployment"],
)
def test_drop_params_on_the_row_or_deployment_drops_unknown_voice_keys(
self, row: bytes, litellm_params: dict[str, object]
) -> None:
uploaded: Final = json.loads(_upload_tts_batch_row(row, litellm_params))
assert uploaded["request"]["generationConfig"]["speechConfig"] == {
"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Kore"}}
}
class TestStreamingLaziness:
def test_upload_stream_transforms_rows_only_as_they_are_pulled(self) -> None:
cfg: Final = VertexAIFilesConfig()

239
tests/unit/llms/vertex_ai/gemini/test_vertex_and_google_ai_studio_gemini.py
View file

@ -24,8 +24,11 @@ from litellm.llms.gemini.chat.transformation import GoogleAIStudioGeminiConfig
from litellm.llms.vertex_ai.common_utils import GEMINI_MAX_ENUM_CHARACTERS, VertexAIError
from litellm.llms.vertex_ai.vertex_llm_base import VertexBase
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
OPENAI_TO_GEMINI_TTS_VOICES,
VertexGeminiConfig,
_log_gemini_sampling_params_deprecated,
normalize_gemini_speech_config,
reject_unknown_gemini_tts_voice_keys,
)
from litellm.types.llms.vertex_ai import GeminiFinishReason, RequestBody, UsageMetadata
from litellm.types.utils import ChoiceLogprobs, Usage
@ -104,6 +107,242 @@ GEMINI_3_IMAGE_SIZE_MAPPINGS: Final[tuple[tuple[str, str, str], ...]] = (
)
@pytest.mark.parametrize(
"speech_config",
(
{"name": "aLlOy"},
{"speech_config": {"voice_name": "aLlOy"}},
{"voice_config": {"prebuilt_voice_config": {"voice_name": "aLlOy"}}},
),
)
def test_gemini_structured_speech_config_maps_openai_voice_names(speech_config: Mapping[str, object]) -> None:
assert normalize_gemini_speech_config(speech_config) == {
"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": OPENAI_TO_GEMINI_TTS_VOICES["alloy"]}}
}
@pytest.mark.parametrize(
"speaker",
(
{"speakerAlias": "Ryan", "speakerId": "alloy"},
{"speaker_alias": "Ryan", "speaker_id": "alloy"},
),
)
def test_gemini_speech_config_accepts_cloud_tts_speaker_entries(speaker: Mapping[str, object]) -> None:
speech_config: Final = {
"multi_speaker_voice_config": {
"speaker_voice_configs": [speaker, {"speaker": "Katie", "voice_config": {"voice_name": "Leda"}}]
}
}
assert normalize_gemini_speech_config(speech_config) == {
"multiSpeakerVoiceConfig": {
"speakerVoiceConfigs": [
{
"speaker": "Ryan",
"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": OPENAI_TO_GEMINI_TTS_VOICES["alloy"]}},
},
{"speaker": "Katie", "voiceConfig": {"voiceName": "Leda"}},
]
}
}
@pytest.mark.parametrize(
("voice", "expected"),
(
(
{
"languageCode": "en-US",
"modelName": "gemini-2.5-flash-tts",
"multiSpeakerVoiceConfig": {"speakerVoiceConfigs": [{"speakerAlias": "Ryan", "speakerId": "Umbriel"}]},
},
{
"languageCode": "en-US",
"multiSpeakerVoiceConfig": {
"speakerVoiceConfigs": [
{"speaker": "Ryan", "voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Umbriel"}}}
]
},
},
),
(
{"model_name": "gemini-2.5-flash-tts", "voice_config": {"prebuilt_voice_config": {"voice_name": "Leda"}}},
{"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Leda"}}},
),
({"speech_config": {"modelName": "gemini-2.5-flash-tts", "language_code": "fr-FR"}}, {"languageCode": "fr-FR"}),
),
)
def test_gemini_speech_config_drops_cloud_tts_model_name(
voice: Mapping[str, object], expected: Mapping[str, object]
) -> None:
assert normalize_gemini_speech_config(voice) == expected
def test_gemini_speech_config_drops_non_string_keys_without_adding_voice():
speech_config: Final = {
"speech_config": {
"language_code": "fr-FR",
"voice_config": {"prebuilt_voice_config": {"voice_name": "Leda", 1: "discard"}},
},
"format": "pcm16",
}
assert normalize_gemini_speech_config(speech_config) == {
"languageCode": "fr-FR",
"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Leda"}},
}
def test_gemini_speech_config_keeps_only_speech_config_fields():
speech_config: Final = {
"voice": "Kore",
"metadata": {"source": "studio"},
"speech_config": {"language_code": "fr-FR", "style": "calm"},
}
assert normalize_gemini_speech_config(speech_config) == {
"languageCode": "fr-FR",
"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Kore"}},
}
@pytest.mark.parametrize(
"voice, unknown_keys",
[
({"name": "Kore", "style": "calm"}, "style"),
({"speech_config": {"voice": "Kore", "pitch": 2}}, "pitch"),
({"speechConfig": {"voiceConfig": {}}, "tone": "x", "pitch": 2}, "pitch, tone"),
({"speechConfig": {"voiceName": "Kore"}, "speech_config": {"style": "calm"}}, "speech_config"),
],
)
def test_unknown_gemini_tts_voice_keys_are_rejected_without_drop_params(voice, unknown_keys):
with pytest.raises(litellm.UnsupportedParamsError, match=f"voice keys {unknown_keys}\\."):
reject_unknown_gemini_tts_voice_keys("gemini-3.1-flash-tts-preview", voice, False, "vertex_ai")
reject_unknown_gemini_tts_voice_keys("gemini-3.1-flash-tts-preview", voice, True, "vertex_ai")
def test_chat_audio_speech_config_unknown_keys_are_rejected_unless_dropped():
audio: Final = {"format": "pcm16", "speech_config": {"voice_name": "Kore", "style": "calm"}}
with pytest.raises(litellm.UnsupportedParamsError, match="voice keys style\\."):
VertexGeminiConfig().map_openai_params(
non_default_params={"audio": audio},
optional_params={},
model="gemini-2.5-flash-preview-tts",
drop_params=False,
)
dropped: Final = VertexGeminiConfig().map_openai_params(
non_default_params={"audio": audio},
optional_params={},
model="gemini-2.5-flash-preview-tts",
drop_params=True,
)
assert dropped["speechConfig"] == {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Kore"}}}
voice_only: Final = VertexGeminiConfig().map_openai_params(
non_default_params={"audio": {"voice": "Kore", "format": "pcm16", "style": "calm"}},
optional_params={},
model="gemini-2.5-flash-preview-tts",
drop_params=False,
)
assert voice_only["speechConfig"] == {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Kore"}}}
def test_chat_audio_speech_config_unknown_keys_follow_global_drop_params(monkeypatch):
monkeypatch.setattr(litellm, "drop_params", True)
dropped: Final = VertexGeminiConfig().map_openai_params(
non_default_params={"audio": {"format": "pcm16", "speech_config": {"voice_name": "Kore", "style": "calm"}}},
optional_params={},
model="gemini-2.5-flash-preview-tts",
drop_params=False,
)
assert dropped["speechConfig"] == {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Kore"}}}
def _chat_speech_config(speech_config: Mapping[str, object], drop_params: bool) -> object:
return VertexGeminiConfig().map_openai_params(
non_default_params={"audio": {"format": "pcm16", "speech_config": speech_config}},
optional_params={},
model="gemini-2.5-flash-preview-tts",
drop_params=drop_params,
)["speechConfig"]
def test_chat_audio_drop_params_strips_unknown_fields_inside_speaker_voices():
speech_config: Final = {
"multi_speaker_voice_config": {
"speaker_voice_configs": [
{"speaker": "Joe", "voice_config": {"prebuilt_voice_config": {"voice_name": "Puck", "pitch": 2}}},
{"speaker_alias": "Jane", "speaker_id": "Kore", "style": "calm"},
],
"turn_order": "alternate",
}
}
assert _chat_speech_config(speech_config, drop_params=True) == {
"multiSpeakerVoiceConfig": {
"speakerVoiceConfigs": [
{"speaker": "Joe", "voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Puck"}}},
{"speaker": "Jane", "voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Kore"}}},
]
}
}
GEMINI_VOICE_CONFIG: Final = {
"voice": "Kore",
"prebuiltVoiceConfig": {"voiceName": "Kore"},
"replicatedVoiceConfig": {"mimeType": "audio/wav", "voiceSampleAudio": "UklGRg=="},
}
@pytest.mark.parametrize(
"speech_config",
[
{"voiceConfig": GEMINI_VOICE_CONFIG, "languageCode": "fr-FR"},
{
"voice_config": {
"voice": "Kore",
"prebuilt_voice_config": {"voice_name": "Kore"},
"replicated_voice_config": {"mime_type": "audio/wav", "voice_sample_audio": "UklGRg=="},
},
"language_code": "fr-FR",
},
],
)
def test_chat_audio_drop_params_keeps_every_voice_config_field_gemini_accepts(speech_config: dict[str, object]):
assert _chat_speech_config(speech_config, drop_params=True) == {
"voiceConfig": GEMINI_VOICE_CONFIG,
"languageCode": "fr-FR",
}
def test_chat_audio_reads_snake_case_wrapper_beside_null_camel_case_wrapper():
mapped: Final = VertexGeminiConfig().map_openai_params(
non_default_params={"audio": {"format": "pcm16", "speechConfig": None, "speech_config": {"voice_name": "Kore"}}},
optional_params={},
model="gemini-2.5-flash-preview-tts",
drop_params=False,
)
assert mapped["speechConfig"] == {"voiceConfig": {"prebuiltVoiceConfig": {"voiceName": "Kore"}}}
def test_unknown_voice_key_error_stays_bounded_for_large_voice_objects():
voice: Final = {"name": "Kore", **{f"{index:04d}{'x' * 1000}": index for index in range(2000)}}
with pytest.raises(litellm.UnsupportedParamsError) as excinfo:
reject_unknown_gemini_tts_voice_keys("gemini-3.1-flash-tts-preview", voice, False, "vertex_ai")
assert "and 1990 more" in excinfo.value.message
assert len(excinfo.value.message) < 1500
def test_top_logprobs():
non_default_params = {
"top_logprobs": 2,

1233
tests/unit/llms/vertex_ai/text_to_speech/test_transformation.py
View file

File diff suppressed because it is too large Load diff

43
tests/unit/proxy/openai_files_endpoint/test_files_endpoint.py
View file

@ -722,6 +722,16 @@ def _router_with_executed_batch_model() -> Router:
"litellm_params": {"model": "gemini/gemini-2.0-flash"},
"model_info": {"id": "gemini-2.0-flash-id"},
},
{
"model_name": "vertex-batch-drop",
"litellm_params": {"model": "vertex_ai/gemini-2.5-flash", "drop_params": True},
"model_info": {"id": "vertex-batch-drop-id"},
},
{
"model_name": "vertex-batch-keep",
"litellm_params": {"model": "vertex_ai/gemini-2.5-flash", "drop_params": False},
"model_info": {"id": "vertex-batch-keep-id"},
},
]
)
@ -939,6 +949,39 @@ def test_batch_upload_for_a_provider_model_still_goes_to_the_provider(batch_uplo
assert provider_upload.call_args.kwargs["custom_llm_provider"] == "gemini"
@pytest.mark.parametrize(
"model, expected_drop_params",
[("vertex-batch-drop", True), ("vertex-batch-keep", False)],
)
def test_batch_upload_for_a_provider_model_carries_the_deployment_drop_params(
batch_upload_seams, model: str, expected_drop_params: bool
):
_, provider_upload, _ = batch_upload_seams
response = _upload_batch_file({"x-litellm-model": model}, {})
assert response.status_code == 200, response.text
assert provider_upload.call_args.kwargs["drop_params"] is expected_drop_params
def test_batch_upload_for_a_provider_model_without_deployment_drop_params_sends_no_drop_params(batch_upload_seams):
_, provider_upload, _ = batch_upload_seams
response = _upload_batch_file({"x-litellm-model": "gemini-2.0-flash"}, {})
assert response.status_code == 200, response.text
assert "drop_params" not in provider_upload.call_args.kwargs
def test_batch_upload_keeps_the_callers_drop_params_over_the_deployment(batch_upload_seams):
_, provider_upload, _ = batch_upload_seams
response = _upload_batch_file({"x-litellm-model": "vertex-batch-keep", "user-agent": "claude-cli/2.1.0"}, {})
assert response.status_code == 200, response.text
assert provider_upload.call_args.kwargs["drop_params"] is True
@pytest.mark.skip(reason="mock respx fails on ci/cd - unclear why")
def test_create_file_and_call_chat_completion_e2e(
mocker: MockerFixture, monkeypatch, llm_router: Router

29
tests/unit/test_cost_calculator.py
View file

@ -31,7 +31,7 @@ from litellm.llms.custom_httpx.http_handler import HTTPHandler
from litellm.llms.fireworks_ai.cost_calculator import get_base_model_for_pricing
from litellm.llms.together_ai.cost_calculator import get_model_params_and_category
from litellm.types.llms.base import CachedTokensDetails
from litellm.types.llms.openai import OpenAIRealtimeStreamList, ResponseAPIUsage, ResponsesAPIResponse
from litellm.types.llms.openai import HttpxBinaryResponseContent, OpenAIRealtimeStreamList, ResponseAPIUsage, ResponsesAPIResponse
from litellm.types.llms.xai import RealtimeMeteredUsage
from litellm.types.rerank import RerankResponse
from litellm.types.utils import CallTypes, Choices, CompletionTokensDetailsWrapper, EmbeddingResponse, ImageObject, ImageResponse, ImageUsage, ImageUsageInputTokensDetails, LiteLLMRealtimeStreamLoggingObject, Message, ModelInfo, ModelResponse, PromptTokensDetails, PromptTokensDetailsWrapper, Usage
@ -9134,6 +9134,33 @@ def test_azure_tts_cost_uses_test_owned_character_rate(monkeypatch: pytest.Monke
assert cost == pytest.approx(0.05)
def test_speech_cost_with_custom_pricing_bills_the_reported_usage() -> None:
usage: Final = Usage(
prompt_tokens=10,
completion_tokens=1000,
total_tokens=1010,
completion_tokens_details=CompletionTokensDetailsWrapper(audio_tokens=1000),
)
response: Final = HttpxBinaryResponseContent(
httpx.Response(200, content=b"RIFF....WAVE", headers={"content-type": "audio/wav"})
)
response.usage = usage
rates: Final = MappingProxyType({"input_cost_per_token": 1e-6, "output_cost_per_token": 2e-5})
cost: Final = completion_cost(
completion_response=response,
model="vertex_ai/gemini-3.1-flash-tts-preview",
call_type="speech",
prompt="Say hello",
custom_cost_per_token=dict(rates),
)
assert cost == pytest.approx(
usage.prompt_tokens * rates["input_cost_per_token"]
+ usage.completion_tokens * rates["output_cost_per_token"]
)
def test_moderation_cost_uses_test_owned_token_rate() -> None:
prompt: Final = "moderation input"
rates: Final = MappingProxyType(