mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-05 08:07:05 +00:00
Merge pull request #38740 from BerriAI/litellm_vertex_gemini_35_transcribe
feat(vertex_ai): support gemini-3.5-transcribe on /v1/audio/transcriptions
This commit is contained in:
Mateo Wang
GitHub
commit
6bc8dafa99
7 changed files with 731 additions and 15 deletions
|
|
@ -0,0 +1,216 @@
|
|||
import base64
|
||||
from collections.abc import Mapping, Sequence
|
||||
from typing import Final
|
||||
|
||||
from httpx import Headers, Response
|
||||
|
||||
import litellm
|
||||
from litellm.exceptions import UnsupportedParamsError
|
||||
from litellm.litellm_core_utils.audio_utils.utils import (
|
||||
normalize_transcription_language_to_bcp47,
|
||||
process_audio_file,
|
||||
)
|
||||
from litellm.llms.base_llm.audio_transcription.transformation import (
|
||||
AudioTranscriptionRequestData,
|
||||
BaseAudioTranscriptionConfig,
|
||||
)
|
||||
from litellm.llms.base_llm.chat.transformation import BaseLLMException
|
||||
from litellm.llms.vertex_ai.audio_transcription.transformation import (
|
||||
SUPPORTED_RESPONSE_FORMATS,
|
||||
validate_vertex_transcription_location,
|
||||
validate_vertex_transcription_project_id,
|
||||
)
|
||||
from litellm.llms.vertex_ai.common_utils import VertexAIError, get_vertex_base_url
|
||||
from litellm.llms.vertex_ai.vertex_llm_base import VertexBase
|
||||
from litellm.types.llms.openai import (
|
||||
AllMessageValues,
|
||||
OpenAIAudioTranscriptionOptionalParams,
|
||||
)
|
||||
from litellm.types.llms.vertex_ai_gemini_transcription import (
|
||||
VertexGeminiTranscriptionAudioConfig,
|
||||
VertexGeminiTranscriptionContent,
|
||||
VertexGeminiTranscriptionGenerationConfig,
|
||||
VertexGeminiTranscriptionInlineData,
|
||||
VertexGeminiTranscriptionPart,
|
||||
VertexGeminiTranscriptionRequest,
|
||||
VertexGeminiTranscriptionResponse,
|
||||
)
|
||||
from litellm.types.utils import (
|
||||
FileTypes,
|
||||
TranscriptionResponse,
|
||||
TranscriptionUsageInputTokenDetailsObject,
|
||||
TranscriptionUsageTokensObject,
|
||||
)
|
||||
|
||||
DEFAULT_GEMINI_TRANSCRIBE_LOCATION: Final = "global"
|
||||
AUDIO_MODALITY: Final = "AUDIO"
|
||||
|
||||
|
||||
class VertexGeminiAudioTranscriptionConfig(BaseAudioTranscriptionConfig, VertexBase):
|
||||
def __init__(self) -> None:
|
||||
BaseAudioTranscriptionConfig.__init__(self)
|
||||
VertexBase.__init__(self)
|
||||
|
||||
def get_supported_openai_params(
|
||||
self, model: str
|
||||
) -> list[OpenAIAudioTranscriptionOptionalParams]: # mutable-ok: BaseAudioTranscriptionConfig signature
|
||||
return ["language", "response_format"]
|
||||
|
||||
def map_openai_params(
|
||||
self,
|
||||
non_default_params: Mapping[str, object],
|
||||
optional_params: Mapping[str, object],
|
||||
model: str,
|
||||
drop_params: bool,
|
||||
) -> dict[str, object]: # mutable-ok: BaseAudioTranscriptionConfig signature
|
||||
supported_params: Final = frozenset(self.get_supported_openai_params(model))
|
||||
mapped: Final = {
|
||||
**optional_params,
|
||||
**{k: v for k, v in non_default_params.items() if k in supported_params},
|
||||
}
|
||||
response_format: Final = mapped.get("response_format")
|
||||
if response_format is None or response_format in SUPPORTED_RESPONSE_FORMATS:
|
||||
return mapped
|
||||
if drop_params or litellm.drop_params:
|
||||
return {k: v for k, v in mapped.items() if k != "response_format"}
|
||||
raise UnsupportedParamsError(
|
||||
status_code=400,
|
||||
message=(
|
||||
f"Vertex AI Gemini transcription does not support response_format={response_format!r}. "
|
||||
f"Supported values: {', '.join(SUPPORTED_RESPONSE_FORMATS)}. "
|
||||
"To drop unsupported openai params from the call, set `litellm.drop_params = True`"
|
||||
),
|
||||
)
|
||||
|
||||
def get_error_class(
|
||||
self,
|
||||
error_message: str,
|
||||
status_code: int,
|
||||
headers: dict | Headers, # mutable-ok: base signature and VertexAIError take dict | Headers
|
||||
) -> BaseLLMException:
|
||||
return VertexAIError(status_code=status_code, message=error_message, headers=headers)
|
||||
|
||||
def validate_environment(
|
||||
self,
|
||||
headers: Mapping[str, str],
|
||||
model: str,
|
||||
messages: Sequence[AllMessageValues],
|
||||
optional_params: Mapping[str, object],
|
||||
litellm_params: Mapping[str, object],
|
||||
api_key: str | None = None,
|
||||
api_base: str | None = None,
|
||||
) -> dict[str, str]: # mutable-ok: BaseAudioTranscriptionConfig signature
|
||||
vertex_params: Final = dict(litellm_params)
|
||||
access_token, project_id = self._ensure_access_token(
|
||||
credentials=self.safe_get_vertex_ai_credentials(vertex_params),
|
||||
project_id=self.safe_get_vertex_ai_project(vertex_params),
|
||||
custom_llm_provider="vertex_ai",
|
||||
)
|
||||
return {
|
||||
**headers,
|
||||
"Authorization": f"Bearer {access_token}",
|
||||
"x-goog-user-project": project_id,
|
||||
"Content-Type": "application/json",
|
||||
}
|
||||
|
||||
def get_complete_url(
|
||||
self,
|
||||
api_base: str | None,
|
||||
api_key: str | None,
|
||||
model: str,
|
||||
optional_params: Mapping[str, object],
|
||||
litellm_params: Mapping[str, object],
|
||||
stream: bool | None = None,
|
||||
) -> str:
|
||||
vertex_params: Final = dict(litellm_params)
|
||||
location: Final = validate_vertex_transcription_location(
|
||||
self.safe_get_vertex_ai_location(vertex_params), default_location=DEFAULT_GEMINI_TRANSCRIBE_LOCATION
|
||||
)
|
||||
project_id: Final = validate_vertex_transcription_project_id(
|
||||
self.safe_get_vertex_ai_project(vertex_params) or self._resolve_project_id_from_credentials(vertex_params)
|
||||
)
|
||||
base_url: Final = (api_base or get_vertex_base_url(location)).rstrip("/")
|
||||
bare_model: Final = model.removeprefix("vertex_ai/")
|
||||
model_path: Final = f"projects/{project_id}/locations/{location}/publishers/google/models/{bare_model}"
|
||||
return f"{base_url}/v1/{model_path}:generateContent"
|
||||
|
||||
def _resolve_project_id_from_credentials(self, litellm_params: Mapping[str, object]) -> str:
|
||||
vertex_params: Final = dict(litellm_params)
|
||||
_, project_id = self._ensure_access_token(
|
||||
credentials=self.safe_get_vertex_ai_credentials(vertex_params),
|
||||
project_id=None,
|
||||
custom_llm_provider="vertex_ai",
|
||||
)
|
||||
return project_id
|
||||
|
||||
def transform_audio_transcription_request(
|
||||
self,
|
||||
model: str,
|
||||
audio_file: FileTypes,
|
||||
optional_params: Mapping[str, object],
|
||||
litellm_params: Mapping[str, object],
|
||||
) -> AudioTranscriptionRequestData:
|
||||
processed_audio: Final = process_audio_file(audio_file)
|
||||
request_body: Final = VertexGeminiTranscriptionRequest(
|
||||
contents=(
|
||||
VertexGeminiTranscriptionContent(
|
||||
role="user",
|
||||
parts=(
|
||||
VertexGeminiTranscriptionPart(
|
||||
inlineData=VertexGeminiTranscriptionInlineData(
|
||||
mimeType=processed_audio.content_type,
|
||||
data=base64.b64encode(processed_audio.file_content).decode("utf-8"),
|
||||
)
|
||||
),
|
||||
),
|
||||
),
|
||||
),
|
||||
generationConfig=VertexGeminiTranscriptionGenerationConfig(
|
||||
audioTranscriptionConfig=_audio_transcription_config(optional_params.get("language"))
|
||||
),
|
||||
)
|
||||
return AudioTranscriptionRequestData(data=dict(request_body))
|
||||
|
||||
def transform_audio_transcription_response(
|
||||
self,
|
||||
raw_response: Response,
|
||||
) -> TranscriptionResponse:
|
||||
try:
|
||||
response_json: Final = raw_response.json()
|
||||
except ValueError:
|
||||
raise VertexAIError(
|
||||
status_code=raw_response.status_code,
|
||||
message=f"Received non-JSON response from Vertex AI Gemini transcription: {raw_response.text}",
|
||||
)
|
||||
parsed: Final = VertexGeminiTranscriptionResponse.model_validate(response_json)
|
||||
texts: Final = tuple(
|
||||
part.text
|
||||
for candidate in parsed.candidates
|
||||
if candidate.content is not None
|
||||
for part in candidate.content.parts
|
||||
if part.text
|
||||
)
|
||||
response: Final = TranscriptionResponse(text=" ".join(texts))
|
||||
response["task"] = "transcribe"
|
||||
usage: Final = parsed.usageMetadata
|
||||
if usage is not None:
|
||||
audio_tokens: Final = sum(
|
||||
detail.tokenCount for detail in usage.promptTokensDetails if detail.modality == AUDIO_MODALITY
|
||||
)
|
||||
response.usage = TranscriptionUsageTokensObject(
|
||||
type="tokens",
|
||||
input_tokens=usage.promptTokenCount,
|
||||
output_tokens=usage.candidatesTokenCount,
|
||||
total_tokens=usage.totalTokenCount,
|
||||
input_token_details=TranscriptionUsageInputTokenDetailsObject(
|
||||
audio_tokens=audio_tokens,
|
||||
text_tokens=usage.promptTokenCount - audio_tokens,
|
||||
),
|
||||
)
|
||||
return response
|
||||
|
||||
|
||||
def _audio_transcription_config(language: object) -> VertexGeminiTranscriptionAudioConfig:
|
||||
if not isinstance(language, str) or not language:
|
||||
return VertexGeminiTranscriptionAudioConfig()
|
||||
return VertexGeminiTranscriptionAudioConfig(languageCodes=(normalize_transcription_language_to_bcp47(language),))
|
||||
|
|
@ -35,6 +35,19 @@ SUPPORTED_RESPONSE_FORMATS: Final = ("json", "text")
|
|||
_URL_UNSAFE_PROJECT_CHARS: Final = ("/", "?", "#", "\\", ":", " ", "\t", "\n", "\r")
|
||||
|
||||
|
||||
def validate_vertex_transcription_location(location: str | None, default_location: str) -> str:
|
||||
try:
|
||||
return validate_vertex_location(location or default_location)
|
||||
except ValueError as e:
|
||||
raise VertexAIError(status_code=400, message=str(e)) from e
|
||||
|
||||
|
||||
def validate_vertex_transcription_project_id(project_id: str) -> str:
|
||||
if not project_id or ".." in project_id or any(c in project_id for c in _URL_UNSAFE_PROJECT_CHARS):
|
||||
raise VertexAIError(status_code=400, message=f"Invalid vertex_project format: {project_id!r}")
|
||||
return project_id
|
||||
|
||||
|
||||
class VertexAIAudioTranscriptionConfig(BaseAudioTranscriptionConfig, VertexBase):
|
||||
def __init__(self) -> None:
|
||||
BaseAudioTranscriptionConfig.__init__(self)
|
||||
|
|
@ -103,27 +116,16 @@ class VertexAIAudioTranscriptionConfig(BaseAudioTranscriptionConfig, VertexBase)
|
|||
litellm_params: dict,
|
||||
stream: bool | None = None,
|
||||
) -> str:
|
||||
location: Final = self._validate_location(self.safe_get_vertex_ai_location(litellm_params))
|
||||
project_id: Final = self._validate_project_id(
|
||||
location: Final = validate_vertex_transcription_location(
|
||||
self.safe_get_vertex_ai_location(litellm_params), default_location=DEFAULT_SPEECH_TO_TEXT_LOCATION
|
||||
)
|
||||
project_id: Final = validate_vertex_transcription_project_id(
|
||||
self.safe_get_vertex_ai_project(litellm_params) or self._resolve_project_id_from_credentials(litellm_params)
|
||||
)
|
||||
host: Final = "speech.googleapis.com" if location == "global" else f"{location}-speech.googleapis.com"
|
||||
base_url: Final = (api_base or f"https://{host}").rstrip("/")
|
||||
return f"{base_url}/v2/projects/{project_id}/locations/{location}/recognizers/_:recognize"
|
||||
|
||||
@staticmethod
|
||||
def _validate_location(location: str | None) -> str:
|
||||
try:
|
||||
return validate_vertex_location(location or DEFAULT_SPEECH_TO_TEXT_LOCATION)
|
||||
except ValueError as e:
|
||||
raise VertexAIError(status_code=400, message=str(e)) from e
|
||||
|
||||
@staticmethod
|
||||
def _validate_project_id(project_id: str) -> str:
|
||||
if not project_id or ".." in project_id or any(c in project_id for c in _URL_UNSAFE_PROJECT_CHARS):
|
||||
raise VertexAIError(status_code=400, message=f"Invalid vertex_project format: {project_id!r}")
|
||||
return project_id
|
||||
|
||||
def _resolve_project_id_from_credentials(self, litellm_params: dict) -> str:
|
||||
_, project_id = self._ensure_access_token(
|
||||
credentials=self.safe_get_vertex_ai_credentials(litellm_params),
|
||||
|
|
|
|||
|
|
@ -51844,6 +51844,43 @@
|
|||
"tpm": 250000,
|
||||
"rpm": 10
|
||||
},
|
||||
"vertex_ai/gemini-3.5-transcribe-preview": {
|
||||
"input_cost_per_audio_token": 2.5e-06,
|
||||
"input_cost_per_token": 2.5e-06,
|
||||
"litellm_provider": "vertex_ai",
|
||||
"mode": "audio_transcription",
|
||||
"output_cost_per_token": 1.2e-05,
|
||||
"source": "https://cloud.google.com/gemini-enterprise-agent-platform/generative-ai/pricing",
|
||||
"supported_endpoints": [
|
||||
"/v1/audio/transcriptions"
|
||||
],
|
||||
"supported_modalities": [
|
||||
"text",
|
||||
"audio"
|
||||
],
|
||||
"supported_output_modalities": [
|
||||
"text"
|
||||
],
|
||||
"supports_audio_input": true
|
||||
},
|
||||
"vertex_ai/gemini-3.5-transcribe-live-preview": {
|
||||
"input_cost_per_audio_token": 3.5e-06,
|
||||
"input_cost_per_token": 3.5e-06,
|
||||
"litellm_provider": "vertex_ai",
|
||||
"mode": "audio_transcription",
|
||||
"output_cost_per_token": 2.1e-05,
|
||||
"source": "https://cloud.google.com/gemini-enterprise-agent-platform/generative-ai/pricing",
|
||||
"supported_endpoints": [
|
||||
"/v1/realtime"
|
||||
],
|
||||
"supported_modalities": [
|
||||
"audio"
|
||||
],
|
||||
"supported_output_modalities": [
|
||||
"text"
|
||||
],
|
||||
"supports_audio_input": true
|
||||
},
|
||||
"perplexity/pplx-embed-context-v1-0.6b": {
|
||||
"input_cost_per_token": 8e-09,
|
||||
"litellm_provider": "perplexity",
|
||||
|
|
|
|||
72
litellm/types/llms/vertex_ai_gemini_transcription.py
Normal file
72
litellm/types/llms/vertex_ai_gemini_transcription.py
Normal file
|
|
@ -0,0 +1,72 @@
|
|||
from typing import Literal
|
||||
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
from typing_extensions import ReadOnly, TypedDict
|
||||
|
||||
|
||||
class VertexGeminiTranscriptionInlineData(TypedDict):
|
||||
mimeType: ReadOnly[str]
|
||||
data: ReadOnly[str]
|
||||
|
||||
|
||||
class VertexGeminiTranscriptionPart(TypedDict):
|
||||
inlineData: ReadOnly[VertexGeminiTranscriptionInlineData]
|
||||
|
||||
|
||||
class VertexGeminiTranscriptionContent(TypedDict):
|
||||
role: ReadOnly[Literal["user"]]
|
||||
parts: ReadOnly[tuple[VertexGeminiTranscriptionPart, ...]]
|
||||
|
||||
|
||||
class VertexGeminiTranscriptionAudioConfig(TypedDict, total=False):
|
||||
languageCodes: ReadOnly[tuple[str, ...]]
|
||||
|
||||
|
||||
class VertexGeminiTranscriptionGenerationConfig(TypedDict):
|
||||
audioTranscriptionConfig: ReadOnly[VertexGeminiTranscriptionAudioConfig]
|
||||
|
||||
|
||||
class VertexGeminiTranscriptionRequest(TypedDict):
|
||||
contents: ReadOnly[tuple[VertexGeminiTranscriptionContent, ...]]
|
||||
generationConfig: ReadOnly[VertexGeminiTranscriptionGenerationConfig]
|
||||
|
||||
|
||||
class VertexGeminiTranscriptionResponsePart(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
text: str | None = None
|
||||
|
||||
|
||||
class VertexGeminiTranscriptionResponseContent(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
parts: tuple[VertexGeminiTranscriptionResponsePart, ...] = ()
|
||||
|
||||
|
||||
class VertexGeminiTranscriptionCandidate(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
content: VertexGeminiTranscriptionResponseContent | None = None
|
||||
|
||||
|
||||
class VertexGeminiTranscriptionModalityTokens(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
modality: str | None = None
|
||||
tokenCount: int = 0
|
||||
|
||||
|
||||
class VertexGeminiTranscriptionUsageMetadata(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
promptTokenCount: int = 0
|
||||
candidatesTokenCount: int = 0
|
||||
totalTokenCount: int = 0
|
||||
promptTokensDetails: tuple[VertexGeminiTranscriptionModalityTokens, ...] = ()
|
||||
|
||||
|
||||
class VertexGeminiTranscriptionResponse(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
candidates: tuple[VertexGeminiTranscriptionCandidate, ...] = ()
|
||||
usageMetadata: VertexGeminiTranscriptionUsageMetadata | None = None
|
||||
|
|
@ -8574,6 +8574,13 @@ class ProviderConfigManager:
|
|||
|
||||
return SonioxAudioTranscriptionConfig()
|
||||
elif litellm.LlmProviders.VERTEX_AI == provider:
|
||||
bare_vertex_model: Final = model.removeprefix("vertex_ai/")
|
||||
if bare_vertex_model.startswith("gemini") and "transcribe" in bare_vertex_model:
|
||||
from litellm.llms.vertex_ai.audio_transcription.gemini_transcribe_transformation import (
|
||||
VertexGeminiAudioTranscriptionConfig,
|
||||
)
|
||||
|
||||
return VertexGeminiAudioTranscriptionConfig()
|
||||
from litellm.llms.vertex_ai.audio_transcription.transformation import (
|
||||
VertexAIAudioTranscriptionConfig,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -51844,6 +51844,43 @@
|
|||
"tpm": 250000,
|
||||
"rpm": 10
|
||||
},
|
||||
"vertex_ai/gemini-3.5-transcribe-preview": {
|
||||
"input_cost_per_audio_token": 2.5e-06,
|
||||
"input_cost_per_token": 2.5e-06,
|
||||
"litellm_provider": "vertex_ai",
|
||||
"mode": "audio_transcription",
|
||||
"output_cost_per_token": 1.2e-05,
|
||||
"source": "https://cloud.google.com/gemini-enterprise-agent-platform/generative-ai/pricing",
|
||||
"supported_endpoints": [
|
||||
"/v1/audio/transcriptions"
|
||||
],
|
||||
"supported_modalities": [
|
||||
"text",
|
||||
"audio"
|
||||
],
|
||||
"supported_output_modalities": [
|
||||
"text"
|
||||
],
|
||||
"supports_audio_input": true
|
||||
},
|
||||
"vertex_ai/gemini-3.5-transcribe-live-preview": {
|
||||
"input_cost_per_audio_token": 3.5e-06,
|
||||
"input_cost_per_token": 3.5e-06,
|
||||
"litellm_provider": "vertex_ai",
|
||||
"mode": "audio_transcription",
|
||||
"output_cost_per_token": 2.1e-05,
|
||||
"source": "https://cloud.google.com/gemini-enterprise-agent-platform/generative-ai/pricing",
|
||||
"supported_endpoints": [
|
||||
"/v1/realtime"
|
||||
],
|
||||
"supported_modalities": [
|
||||
"audio"
|
||||
],
|
||||
"supported_output_modalities": [
|
||||
"text"
|
||||
],
|
||||
"supports_audio_input": true
|
||||
},
|
||||
"perplexity/pplx-embed-context-v1-0.6b": {
|
||||
"input_cost_per_token": 8e-09,
|
||||
"litellm_provider": "perplexity",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,345 @@
|
|||
import base64
|
||||
import json
|
||||
import os
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
import litellm
|
||||
from litellm.llms.vertex_ai.audio_transcription.gemini_transcribe_transformation import (
|
||||
VertexGeminiAudioTranscriptionConfig,
|
||||
)
|
||||
from litellm.llms.vertex_ai.audio_transcription.transformation import (
|
||||
VertexAIAudioTranscriptionConfig,
|
||||
)
|
||||
from litellm.llms.vertex_ai.common_utils import VertexAIError
|
||||
from litellm.types.utils import LlmProviders, TranscriptionUsageTokensObject
|
||||
from litellm.utils import ProviderConfigManager, get_optional_params_transcription
|
||||
|
||||
AUDIO_BYTES = b"fake-audio-bytes"
|
||||
TRANSCRIPT_TEXT = (
|
||||
"Four score and seven years ago our fathers brought forth on this continent, a new nation, "
|
||||
"conceived in Liberty, and dedicated to the proposition that all men are created equal. "
|
||||
"Now we are engaged in a great civil war, testing whether that nation, or any nation so "
|
||||
"conceived and so dedicated, can long endure."
|
||||
)
|
||||
GENERATE_CONTENT_RESPONSE = {
|
||||
"candidates": [
|
||||
{
|
||||
"content": {
|
||||
"role": "model",
|
||||
"parts": [
|
||||
{
|
||||
"text": TRANSCRIPT_TEXT,
|
||||
"audioTranscription": {"text": TRANSCRIPT_TEXT},
|
||||
}
|
||||
],
|
||||
},
|
||||
"finishReason": "STOP",
|
||||
}
|
||||
],
|
||||
"usageMetadata": {
|
||||
"promptTokenCount": 440,
|
||||
"candidatesTokenCount": 62,
|
||||
"totalTokenCount": 502,
|
||||
"trafficType": "ON_DEMAND",
|
||||
"promptTokensDetails": [{"modality": "AUDIO", "tokenCount": 440}],
|
||||
"candidatesTokensDetails": [{"modality": "TEXT", "tokenCount": 62}],
|
||||
},
|
||||
"modelVersion": "gemini-3.5-transcribe-preview",
|
||||
"createTime": "2026-08-29T07:25:27.591648Z",
|
||||
"responseId": "Z4mSaqCOJL-O4_UP0aSh4Aw",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config():
|
||||
return VertexGeminiAudioTranscriptionConfig()
|
||||
|
||||
|
||||
class TestProviderRouting:
|
||||
@pytest.mark.parametrize(
|
||||
"model",
|
||||
[
|
||||
"gemini-3.5-transcribe-preview",
|
||||
"gemini-3.5-transcribe-live-preview",
|
||||
"vertex_ai/gemini-3.5-transcribe-preview",
|
||||
],
|
||||
)
|
||||
def test_gemini_transcribe_models_use_generate_content_config(self, model):
|
||||
provider_config = ProviderConfigManager.get_provider_audio_transcription_config(
|
||||
model=model,
|
||||
provider=LlmProviders.VERTEX_AI,
|
||||
)
|
||||
assert isinstance(provider_config, VertexGeminiAudioTranscriptionConfig)
|
||||
|
||||
@pytest.mark.parametrize("model", ["chirp_2", "chirp_3", "long-form", "gemini-2.5-flash"])
|
||||
def test_other_vertex_models_keep_speech_to_text_config(self, model):
|
||||
provider_config = ProviderConfigManager.get_provider_audio_transcription_config(
|
||||
model=model,
|
||||
provider=LlmProviders.VERTEX_AI,
|
||||
)
|
||||
assert isinstance(provider_config, VertexAIAudioTranscriptionConfig)
|
||||
assert not isinstance(provider_config, VertexGeminiAudioTranscriptionConfig)
|
||||
|
||||
|
||||
class TestGetCompleteUrl:
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clear_ambient_vertex_location(self, monkeypatch):
|
||||
monkeypatch.delenv("VERTEXAI_LOCATION", raising=False)
|
||||
monkeypatch.delenv("VERTEX_LOCATION", raising=False)
|
||||
|
||||
def test_defaults_to_global_location(self, config):
|
||||
url = config.get_complete_url(
|
||||
api_base=None,
|
||||
api_key=None,
|
||||
model="gemini-3.5-transcribe-preview",
|
||||
optional_params={},
|
||||
litellm_params={"vertex_project": "test-project"},
|
||||
)
|
||||
assert url == (
|
||||
"https://aiplatform.googleapis.com/v1/projects/test-project/locations/global"
|
||||
"/publishers/google/models/gemini-3.5-transcribe-preview:generateContent"
|
||||
)
|
||||
|
||||
def test_explicit_location_is_honored(self, config):
|
||||
url = config.get_complete_url(
|
||||
api_base=None,
|
||||
api_key=None,
|
||||
model="gemini-3.5-transcribe-preview",
|
||||
optional_params={},
|
||||
litellm_params={"vertex_project": "test-project", "vertex_location": "us-central1"},
|
||||
)
|
||||
assert url == (
|
||||
"https://us-central1-aiplatform.googleapis.com/v1/projects/test-project/locations/us-central1"
|
||||
"/publishers/google/models/gemini-3.5-transcribe-preview:generateContent"
|
||||
)
|
||||
|
||||
def test_model_prefix_is_stripped(self, config):
|
||||
url = config.get_complete_url(
|
||||
api_base=None,
|
||||
api_key=None,
|
||||
model="vertex_ai/gemini-3.5-transcribe-preview",
|
||||
optional_params={},
|
||||
litellm_params={"vertex_project": "test-project"},
|
||||
)
|
||||
assert "/models/gemini-3.5-transcribe-preview:generateContent" in url
|
||||
assert "vertex_ai/" not in url
|
||||
|
||||
def test_api_base_override(self, config):
|
||||
url = config.get_complete_url(
|
||||
api_base="http://localhost:8080/",
|
||||
api_key=None,
|
||||
model="gemini-3.5-transcribe-preview",
|
||||
optional_params={},
|
||||
litellm_params={"vertex_project": "test-project"},
|
||||
)
|
||||
assert url == (
|
||||
"http://localhost:8080/v1/projects/test-project/locations/global"
|
||||
"/publishers/google/models/gemini-3.5-transcribe-preview:generateContent"
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("malicious_location", ["attacker.example/", "evil.com#", "US", "us/../.."])
|
||||
def test_malicious_location_is_rejected(self, config, malicious_location):
|
||||
with pytest.raises(VertexAIError):
|
||||
config.get_complete_url(
|
||||
api_base=None,
|
||||
api_key=None,
|
||||
model="gemini-3.5-transcribe-preview",
|
||||
optional_params={},
|
||||
litellm_params={"vertex_project": "test-project", "vertex_location": malicious_location},
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("malicious_project", ["proj/../../locations", "proj#frag", "proj?a=b", "proj space"])
|
||||
def test_malicious_project_is_rejected(self, config, malicious_project):
|
||||
with pytest.raises(VertexAIError):
|
||||
config.get_complete_url(
|
||||
api_base=None,
|
||||
api_key=None,
|
||||
model="gemini-3.5-transcribe-preview",
|
||||
optional_params={},
|
||||
litellm_params={"vertex_project": malicious_project},
|
||||
)
|
||||
|
||||
|
||||
class TestTransformRequest:
|
||||
def test_request_body_shape(self, config):
|
||||
request_data = config.transform_audio_transcription_request(
|
||||
model="gemini-3.5-transcribe-preview",
|
||||
audio_file=AUDIO_BYTES,
|
||||
optional_params={},
|
||||
litellm_params={},
|
||||
)
|
||||
assert request_data.files is None
|
||||
assert request_data.data == {
|
||||
"contents": (
|
||||
{
|
||||
"role": "user",
|
||||
"parts": (
|
||||
{
|
||||
"inlineData": {
|
||||
"mimeType": "audio/wav",
|
||||
"data": base64.b64encode(AUDIO_BYTES).decode("utf-8"),
|
||||
}
|
||||
},
|
||||
),
|
||||
},
|
||||
),
|
||||
"generationConfig": {"audioTranscriptionConfig": {}},
|
||||
}
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"language,expected_language_codes",
|
||||
[
|
||||
("en", ("en-US",)),
|
||||
("en-US", ("en-US",)),
|
||||
("fr", ("fr-FR",)),
|
||||
],
|
||||
)
|
||||
def test_language_param_maps_to_language_codes(self, config, language, expected_language_codes):
|
||||
request_data = config.transform_audio_transcription_request(
|
||||
model="gemini-3.5-transcribe-preview",
|
||||
audio_file=AUDIO_BYTES,
|
||||
optional_params={"language": language},
|
||||
litellm_params={},
|
||||
)
|
||||
audio_config = request_data.data["generationConfig"]["audioTranscriptionConfig"]
|
||||
assert audio_config["languageCodes"] == expected_language_codes
|
||||
|
||||
def test_body_round_trips_through_json(self, config):
|
||||
request_data = config.transform_audio_transcription_request(
|
||||
model="gemini-3.5-transcribe-preview",
|
||||
audio_file=AUDIO_BYTES,
|
||||
optional_params={"language": "en"},
|
||||
litellm_params={},
|
||||
)
|
||||
round_tripped = json.loads(json.dumps(request_data.data))
|
||||
assert round_tripped["generationConfig"] == {"audioTranscriptionConfig": {"languageCodes": ["en-US"]}}
|
||||
assert round_tripped["contents"][0]["role"] == "user"
|
||||
|
||||
|
||||
class TestTransformResponse:
|
||||
def test_generate_content_response(self, config):
|
||||
raw_response = httpx.Response(status_code=200, json=GENERATE_CONTENT_RESPONSE)
|
||||
response = config.transform_audio_transcription_response(raw_response)
|
||||
assert response.text == TRANSCRIPT_TEXT
|
||||
assert response["task"] == "transcribe"
|
||||
assert isinstance(response.usage, TranscriptionUsageTokensObject)
|
||||
assert response.usage.input_tokens == 440
|
||||
assert response.usage.output_tokens == 62
|
||||
assert response.usage.total_tokens == 502
|
||||
assert response.usage.input_token_details.audio_tokens == 440
|
||||
assert response.usage.input_token_details.text_tokens == 0
|
||||
|
||||
def test_multi_part_texts_are_joined(self, config):
|
||||
raw_response = httpx.Response(
|
||||
status_code=200,
|
||||
json={
|
||||
"candidates": [
|
||||
{"content": {"role": "model", "parts": [{"text": "Hello world."}, {"text": "How are you?"}]}}
|
||||
],
|
||||
"usageMetadata": {"promptTokenCount": 10, "candidatesTokenCount": 5, "totalTokenCount": 15},
|
||||
},
|
||||
)
|
||||
response = config.transform_audio_transcription_response(raw_response)
|
||||
assert response.text == "Hello world. How are you?"
|
||||
|
||||
def test_empty_candidates_returns_empty_text(self, config):
|
||||
raw_response = httpx.Response(status_code=200, json={})
|
||||
response = config.transform_audio_transcription_response(raw_response)
|
||||
assert response.text == ""
|
||||
assert response.usage is None
|
||||
|
||||
def test_non_json_body_raises(self, config):
|
||||
raw_response = httpx.Response(status_code=200, text="<html>not json</html>")
|
||||
with pytest.raises(VertexAIError, match="non-JSON"):
|
||||
config.transform_audio_transcription_response(raw_response)
|
||||
|
||||
|
||||
class TestValidateEnvironment:
|
||||
def test_sets_oauth_headers(self):
|
||||
class StubbedConfig(VertexGeminiAudioTranscriptionConfig):
|
||||
def _ensure_access_token(self, credentials, project_id, custom_llm_provider):
|
||||
return "fake-token", "resolved-project"
|
||||
|
||||
headers = StubbedConfig().validate_environment(
|
||||
headers={},
|
||||
model="gemini-3.5-transcribe-preview",
|
||||
messages=[],
|
||||
optional_params={},
|
||||
litellm_params={"vertex_project": "resolved-project"},
|
||||
)
|
||||
assert headers["Authorization"] == "Bearer fake-token"
|
||||
assert headers["x-goog-user-project"] == "resolved-project"
|
||||
assert headers["Content-Type"] == "application/json"
|
||||
|
||||
|
||||
class TestOptionalParams:
|
||||
def test_language_and_json_response_format_pass_through(self):
|
||||
optional_params = get_optional_params_transcription(
|
||||
model="gemini-3.5-transcribe-preview",
|
||||
custom_llm_provider="vertex_ai",
|
||||
language="fr-FR",
|
||||
response_format="json",
|
||||
)
|
||||
assert optional_params["language"] == "fr-FR"
|
||||
assert optional_params["response_format"] == "json"
|
||||
|
||||
@pytest.mark.parametrize("response_format", ["verbose_json", "srt", "vtt"])
|
||||
def test_unsupported_response_format_raises(self, response_format):
|
||||
with pytest.raises(litellm.utils.UnsupportedParamsError, match="response_format"):
|
||||
get_optional_params_transcription(
|
||||
model="gemini-3.5-transcribe-preview",
|
||||
custom_llm_provider="vertex_ai",
|
||||
response_format=response_format,
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize("response_format", ["verbose_json", "srt", "vtt"])
|
||||
def test_unsupported_response_format_dropped_with_drop_params(self, response_format):
|
||||
optional_params = get_optional_params_transcription(
|
||||
model="gemini-3.5-transcribe-preview",
|
||||
custom_llm_provider="vertex_ai",
|
||||
language="fr-FR",
|
||||
response_format=response_format,
|
||||
drop_params=True,
|
||||
)
|
||||
assert "response_format" not in optional_params
|
||||
assert optional_params["language"] == "fr-FR"
|
||||
|
||||
|
||||
class TestModelCostEntry:
|
||||
REPO_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../.."))
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"cost_map_path",
|
||||
[
|
||||
"model_prices_and_context_window.json",
|
||||
"litellm/model_prices_and_context_window_backup.json",
|
||||
],
|
||||
)
|
||||
def test_transcribe_preview_pricing(self, cost_map_path):
|
||||
with open(os.path.join(self.REPO_ROOT, cost_map_path)) as f:
|
||||
entry = json.load(f)["vertex_ai/gemini-3.5-transcribe-preview"]
|
||||
assert entry["mode"] == "audio_transcription"
|
||||
assert entry["litellm_provider"] == "vertex_ai"
|
||||
assert entry["input_cost_per_audio_token"] == pytest.approx(2.5e-06)
|
||||
assert entry["input_cost_per_token"] == pytest.approx(2.5e-06)
|
||||
assert entry["output_cost_per_token"] == pytest.approx(1.2e-05)
|
||||
assert entry["supported_endpoints"] == ["/v1/audio/transcriptions"]
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"cost_map_path",
|
||||
[
|
||||
"model_prices_and_context_window.json",
|
||||
"litellm/model_prices_and_context_window_backup.json",
|
||||
],
|
||||
)
|
||||
def test_transcribe_live_preview_pricing(self, cost_map_path):
|
||||
with open(os.path.join(self.REPO_ROOT, cost_map_path)) as f:
|
||||
entry = json.load(f)["vertex_ai/gemini-3.5-transcribe-live-preview"]
|
||||
assert entry["mode"] == "audio_transcription"
|
||||
assert entry["litellm_provider"] == "vertex_ai"
|
||||
assert entry["input_cost_per_audio_token"] == pytest.approx(3.5e-06)
|
||||
assert entry["input_cost_per_token"] == pytest.approx(3.5e-06)
|
||||
assert entry["output_cost_per_token"] == pytest.approx(2.1e-05)
|
||||
assert entry["supported_endpoints"] == ["/v1/realtime"]
|
||||
Loading…
Add table
Reference in a new issue