mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-07 02:59:05 +00:00
feat(gemini): day-0 support for gemini-3.5-transcribe and transcribe-live
Adds a Gemini audio transcription config that maps /v1/audio/transcriptions onto the Interactions API (speaker attribution and word timestamps land on the OpenAI verbose_json shape), registers both models with published pricing, routes text-only Live sessions to TEXT responseModalities so gemini-3.5-transcribe-live sessions survive, and makes the token-priced transcription cost path provider-aware instead of hardcoding OpenAI.
This commit is contained in:
parent
cd63c7e5a7
commit
ef4c84dc36
12 changed files with 786 additions and 14 deletions
|
|
@ -557,9 +557,10 @@ def cost_per_token(
|
|||
)
|
||||
elif call_type == "atranscription" or call_type == "transcription":
|
||||
if _transcription_usage_has_token_details(usage_block):
|
||||
return openai_cost_per_token(
|
||||
return generic_cost_per_token(
|
||||
model=model_without_prefix,
|
||||
usage=usage_block,
|
||||
custom_llm_provider=custom_llm_provider,
|
||||
service_tier=service_tier,
|
||||
data_residency=data_residency,
|
||||
)
|
||||
|
|
|
|||
0
litellm/llms/gemini/audio_transcription/__init__.py
Normal file
0
litellm/llms/gemini/audio_transcription/__init__.py
Normal file
250
litellm/llms/gemini/audio_transcription/transformation.py
Normal file
250
litellm/llms/gemini/audio_transcription/transformation.py
Normal file
|
|
@ -0,0 +1,250 @@
|
|||
import base64
|
||||
from collections.abc import Mapping, Sequence
|
||||
from typing import Final
|
||||
|
||||
from httpx import Headers, Response
|
||||
|
||||
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.gemini.common_utils import GeminiError, GeminiModelInfo
|
||||
from litellm.types.llms.gemini_audio_transcription import (
|
||||
GeminiTranscriptionAudioInput,
|
||||
GeminiTranscriptionConfig,
|
||||
GeminiTranscriptionInteractionRequest,
|
||||
GeminiTranscriptionInteractionResponse,
|
||||
GeminiTranscriptionWordAnnotation,
|
||||
)
|
||||
from litellm.types.llms.openai import (
|
||||
AllMessageValues,
|
||||
OpenAIAudioTranscriptionOptionalParams,
|
||||
)
|
||||
from litellm.types.utils import (
|
||||
FileTypes,
|
||||
TranscriptionResponse,
|
||||
TranscriptionUsageInputTokenDetailsObject,
|
||||
TranscriptionUsageTokensObject,
|
||||
)
|
||||
|
||||
INTERACTIONS_API_REVISION: Final = "2026-05-20"
|
||||
WORD_INFO_ANNOTATION_TYPE: Final = "word_info"
|
||||
|
||||
|
||||
class GeminiAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
|
||||
"""
|
||||
Maps OpenAI /v1/audio/transcriptions onto the Gemini Interactions API
|
||||
(POST /v1beta/interactions) for transcription models like
|
||||
gemini-3.5-transcribe. https://ai.google.dev/gemini-api/docs/transcribe
|
||||
"""
|
||||
|
||||
def get_supported_openai_params(
|
||||
self, model: str
|
||||
) -> list[OpenAIAudioTranscriptionOptionalParams]: # mutable-ok: BaseAudioTranscriptionConfig signature
|
||||
return ["language", "response_format", "timestamp_granularities"] # mutable-ok: base contract returns a list
|
||||
|
||||
def map_openai_params(
|
||||
self,
|
||||
non_default_params: Mapping[str, object],
|
||||
optional_params: Mapping[str, object],
|
||||
model: str,
|
||||
drop_params: bool,
|
||||
) -> dict: # mutable-ok: BaseAudioTranscriptionConfig signature
|
||||
supported_params: Final = frozenset(self.get_supported_openai_params(model))
|
||||
accepted: Final = tuple((k, v) for k, v in non_default_params.items() if k in supported_params)
|
||||
return dict((*optional_params.items(), *accepted)) # mutable-ok: base contract returns a plain dict
|
||||
|
||||
def get_error_class(
|
||||
self,
|
||||
error_message: str,
|
||||
status_code: int,
|
||||
headers: dict | Headers, # mutable-ok: base signature and BaseLLMException take dict | Headers
|
||||
) -> BaseLLMException:
|
||||
return GeminiError(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: # mutable-ok: BaseAudioTranscriptionConfig signature
|
||||
resolved_api_key: Final = GeminiModelInfo.get_api_key(api_key)
|
||||
if not resolved_api_key:
|
||||
raise GeminiError(
|
||||
status_code=401,
|
||||
message="Google API key is required. Set GOOGLE_API_KEY or GEMINI_API_KEY environment variable.",
|
||||
)
|
||||
return { # mutable-ok: the http handler passes these headers straight to httpx
|
||||
**headers,
|
||||
"Content-Type": "application/json",
|
||||
"x-goog-api-key": resolved_api_key,
|
||||
"Api-Revision": INTERACTIONS_API_REVISION,
|
||||
}
|
||||
|
||||
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:
|
||||
resolved_api_base: Final = GeminiModelInfo.get_api_base(api_base)
|
||||
return f"{resolved_api_base}/v1beta/interactions"
|
||||
|
||||
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)
|
||||
audio_input: Final = GeminiTranscriptionAudioInput(
|
||||
type="audio",
|
||||
data=base64.b64encode(processed_audio.file_content).decode("utf-8"),
|
||||
mime_type=processed_audio.content_type,
|
||||
)
|
||||
request: Final = _build_interaction_request(
|
||||
model=model,
|
||||
audio_input=audio_input,
|
||||
transcription_config=_build_transcription_config(optional_params),
|
||||
)
|
||||
return AudioTranscriptionRequestData(data=dict(request)) # mutable-ok: AudioTranscriptionRequestData wants dict
|
||||
|
||||
def transform_audio_transcription_response(
|
||||
self,
|
||||
raw_response: Response,
|
||||
) -> TranscriptionResponse:
|
||||
try:
|
||||
response_json: Final = raw_response.json()
|
||||
except ValueError:
|
||||
raise GeminiError(
|
||||
status_code=raw_response.status_code,
|
||||
message=f"Received non-JSON response from Gemini Interactions API: {raw_response.text}",
|
||||
)
|
||||
parsed: Final = GeminiTranscriptionInteractionResponse.model_validate(response_json)
|
||||
if parsed.status != "completed":
|
||||
raise GeminiError(
|
||||
status_code=raw_response.status_code,
|
||||
message=f"Gemini transcription interaction did not complete (status={parsed.status}): {raw_response.text}",
|
||||
)
|
||||
text_contents: Final = tuple(
|
||||
content
|
||||
for step in parsed.steps
|
||||
for content in step.content
|
||||
if content.type == "text" and content.text is not None
|
||||
)
|
||||
response: Final = TranscriptionResponse(text=" ".join(content.text or "" for content in text_contents))
|
||||
response["task"] = "transcribe"
|
||||
words: Final = tuple(
|
||||
word
|
||||
for content in text_contents
|
||||
for annotation in content.annotations
|
||||
if (word := _annotation_to_word(annotation)) is not None
|
||||
)
|
||||
if words:
|
||||
response["words"] = list(words) # mutable-ok: verbose_json words is a JSON array
|
||||
last_word_end: Final = words[-1].get("end")
|
||||
if last_word_end is not None:
|
||||
response["duration"] = last_word_end
|
||||
if parsed.usage is not None:
|
||||
audio_tokens: Final = sum(
|
||||
by_modality.tokens
|
||||
for by_modality in parsed.usage.input_tokens_by_modality
|
||||
if by_modality.modality == "audio"
|
||||
)
|
||||
response.usage = TranscriptionUsageTokensObject(
|
||||
type="tokens",
|
||||
input_tokens=parsed.usage.total_input_tokens,
|
||||
output_tokens=parsed.usage.total_output_tokens,
|
||||
total_tokens=parsed.usage.total_tokens,
|
||||
input_token_details=TranscriptionUsageInputTokenDetailsObject(
|
||||
audio_tokens=audio_tokens,
|
||||
text_tokens=parsed.usage.total_input_tokens - audio_tokens,
|
||||
),
|
||||
)
|
||||
return response
|
||||
|
||||
|
||||
_EMPTY_TRANSCRIPTION_CONFIG: Final[GeminiTranscriptionConfig] = {}
|
||||
_WORD_TIMESTAMP_CONFIG: Final[GeminiTranscriptionConfig] = {
|
||||
"mode": {
|
||||
"type": "verbatim",
|
||||
"timestamp_granularities": ("word",),
|
||||
"diarization_mode": "speaker",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _build_interaction_request(
|
||||
model: str,
|
||||
audio_input: GeminiTranscriptionAudioInput,
|
||||
transcription_config: GeminiTranscriptionConfig,
|
||||
) -> GeminiTranscriptionInteractionRequest:
|
||||
if not transcription_config:
|
||||
bare_request: Final[GeminiTranscriptionInteractionRequest] = {
|
||||
"model": model.removeprefix("gemini/"),
|
||||
"input": (audio_input,),
|
||||
}
|
||||
return bare_request
|
||||
configured_request: Final[GeminiTranscriptionInteractionRequest] = {
|
||||
"model": model.removeprefix("gemini/"),
|
||||
"input": (audio_input,),
|
||||
"generation_config": {"transcription_config": transcription_config},
|
||||
}
|
||||
return configured_request
|
||||
|
||||
|
||||
def _language_config(language: object) -> GeminiTranscriptionConfig:
|
||||
if not isinstance(language, str) or not language:
|
||||
return _EMPTY_TRANSCRIPTION_CONFIG
|
||||
language_config: Final[GeminiTranscriptionConfig] = {
|
||||
"language_codes": (normalize_transcription_language_to_bcp47(language),),
|
||||
}
|
||||
return language_config
|
||||
|
||||
|
||||
def _timestamp_config(timestamp_granularities: object) -> GeminiTranscriptionConfig:
|
||||
if isinstance(timestamp_granularities, list) and "word" in timestamp_granularities:
|
||||
return _WORD_TIMESTAMP_CONFIG
|
||||
return _EMPTY_TRANSCRIPTION_CONFIG
|
||||
|
||||
|
||||
def _build_transcription_config(optional_params: Mapping[str, object]) -> GeminiTranscriptionConfig:
|
||||
transcription_config: Final[GeminiTranscriptionConfig] = {
|
||||
**_language_config(optional_params.get("language")),
|
||||
**_timestamp_config(optional_params.get("timestamp_granularities")),
|
||||
}
|
||||
return transcription_config
|
||||
|
||||
|
||||
def _annotation_to_word(annotation: GeminiTranscriptionWordAnnotation) -> Mapping[str, str | float] | None:
|
||||
if annotation.type != WORD_INFO_ANNOTATION_TYPE or annotation.text is None:
|
||||
return None
|
||||
entries: Final = (
|
||||
("word", annotation.text),
|
||||
("start", _parse_offset_seconds(annotation.start_offset)),
|
||||
("end", _parse_offset_seconds(annotation.end_offset)),
|
||||
("speaker", annotation.speaker),
|
||||
)
|
||||
return {key: value for key, value in entries if value is not None} # mutable-ok: word entries serialize to JSON
|
||||
|
||||
|
||||
def _parse_offset_seconds(offset: str | None) -> float | None:
|
||||
if offset is None or not offset.endswith("s"):
|
||||
return None
|
||||
try:
|
||||
return float(offset[:-1])
|
||||
except ValueError:
|
||||
return None
|
||||
|
|
@ -4,7 +4,7 @@ This file contains the transformation logic for the Gemini realtime API.
|
|||
|
||||
import json
|
||||
from collections import OrderedDict
|
||||
from collections.abc import Mapping
|
||||
from collections.abc import Mapping, Sequence
|
||||
from typing import Any, Final, cast
|
||||
|
||||
import litellm
|
||||
|
|
@ -384,17 +384,25 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
|
|||
return bool(entry.get("gemini_native_audio") or entry.get("gemini_audio_only_live"))
|
||||
|
||||
@staticmethod
|
||||
def _coerce_response_modalities(model: str, modalities: list[Any]) -> list[str]:
|
||||
"""Map unsupported TEXT responseModalities to AUDIO for audio-only Live models."""
|
||||
normalized: Final = [
|
||||
def _is_text_only_live_model(model: str) -> bool:
|
||||
return GeminiRealtimeConfig._model_cost_entry(model).get("mode") == "audio_transcription"
|
||||
|
||||
@staticmethod
|
||||
def _default_response_modality(model: str) -> GeminiResponseModalities:
|
||||
return "TEXT" if GeminiRealtimeConfig._is_text_only_live_model(model) else "AUDIO"
|
||||
|
||||
@staticmethod
|
||||
def _coerce_response_modalities(model: str, modalities: Sequence[Any]) -> tuple[str, ...]:
|
||||
"""Swap responseModalities a Live model cannot produce: TEXT to AUDIO for
|
||||
audio-only models, AUDIO to TEXT for text-only ones (e.g. transcribe-live)."""
|
||||
normalized: Final = tuple(
|
||||
modality.upper() if isinstance(modality, str) else str(modality).upper() for modality in modalities
|
||||
]
|
||||
if not GeminiRealtimeConfig._is_audio_only_live_model(model):
|
||||
return normalized
|
||||
if "TEXT" not in normalized:
|
||||
return normalized
|
||||
without_text: Final = [modality for modality in normalized if modality != "TEXT"]
|
||||
return without_text if without_text else ["AUDIO"]
|
||||
)
|
||||
if GeminiRealtimeConfig._is_audio_only_live_model(model) and "TEXT" in normalized:
|
||||
return tuple(modality for modality in normalized if modality != "TEXT") or ("AUDIO",)
|
||||
if GeminiRealtimeConfig._is_text_only_live_model(model) and "AUDIO" in normalized:
|
||||
return tuple(modality for modality in normalized if modality != "AUDIO") or ("TEXT",)
|
||||
return normalized
|
||||
|
||||
@staticmethod
|
||||
def _finalize_gemini_live_setup(model: str, setup: dict[str, Any]) -> dict[str, Any]:
|
||||
|
|
@ -436,7 +444,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
|
|||
|
||||
if session_configuration_request is None:
|
||||
generation_config: Final = new_overrides.setdefault("generationConfig", {})
|
||||
generation_config.setdefault("responseModalities", ["AUDIO"])
|
||||
generation_config.setdefault("responseModalities", [GeminiRealtimeConfig._default_response_modality(model)])
|
||||
new_overrides.setdefault("inputAudioTranscription", {})
|
||||
new_overrides["model"] = f"models/{model}"
|
||||
verbose_logger.debug("Gemini Realtime: Sending initial setup with tools to backend")
|
||||
|
|
@ -1583,7 +1591,9 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
|
|||
```
|
||||
"""
|
||||
|
||||
response_modalities: Final[list[GeminiResponseModalities]] = ["AUDIO"]
|
||||
response_modalities: Final[list[GeminiResponseModalities]] = [
|
||||
GeminiRealtimeConfig._default_response_modality(model)
|
||||
]
|
||||
output_audio_transcription: Final = False
|
||||
# if "audio" in model: ## UNCOMMENT THIS WHEN AUDIO IS SUPPORTED
|
||||
# output_audio_transcription = True
|
||||
|
|
|
|||
|
|
@ -51340,6 +51340,43 @@
|
|||
"supports_audio_output": true,
|
||||
"tpm": 250000
|
||||
},
|
||||
"gemini/gemini-3.5-transcribe": {
|
||||
"input_cost_per_audio_token": 2e-06,
|
||||
"input_cost_per_token": 2e-06,
|
||||
"litellm_provider": "gemini",
|
||||
"mode": "audio_transcription",
|
||||
"output_cost_per_token": 1.2e-05,
|
||||
"source": "https://ai.google.dev/gemini-api/docs/pricing",
|
||||
"supported_endpoints": [
|
||||
"/v1/audio/transcriptions"
|
||||
],
|
||||
"supported_modalities": [
|
||||
"text",
|
||||
"audio"
|
||||
],
|
||||
"supported_output_modalities": [
|
||||
"text"
|
||||
],
|
||||
"supports_audio_input": true
|
||||
},
|
||||
"gemini/gemini-3.5-transcribe-live": {
|
||||
"input_cost_per_audio_token": 3.5e-06,
|
||||
"input_cost_per_token": 3.5e-06,
|
||||
"litellm_provider": "gemini",
|
||||
"mode": "audio_transcription",
|
||||
"output_cost_per_token": 2.1e-05,
|
||||
"source": "https://ai.google.dev/gemini-api/docs/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",
|
||||
|
|
|
|||
81
litellm/types/llms/gemini_audio_transcription.py
Normal file
81
litellm/types/llms/gemini_audio_transcription.py
Normal file
|
|
@ -0,0 +1,81 @@
|
|||
from typing import Literal, Required
|
||||
|
||||
from pydantic import BaseModel, ConfigDict
|
||||
from typing_extensions import ReadOnly, TypedDict
|
||||
|
||||
|
||||
class GeminiTranscriptionAudioInput(TypedDict):
|
||||
type: ReadOnly[Literal["audio"]]
|
||||
data: ReadOnly[str]
|
||||
mime_type: ReadOnly[str]
|
||||
|
||||
|
||||
class GeminiTranscriptionVerbatimMode(TypedDict, total=False):
|
||||
type: ReadOnly[Required[Literal["verbatim"]]]
|
||||
timestamp_granularities: ReadOnly[tuple[Literal["word"], ...]]
|
||||
diarization_mode: ReadOnly[Literal["speaker"]]
|
||||
|
||||
|
||||
class GeminiTranscriptionConfig(TypedDict, total=False):
|
||||
language_codes: ReadOnly[tuple[str, ...]]
|
||||
mode: ReadOnly[GeminiTranscriptionVerbatimMode]
|
||||
|
||||
|
||||
class GeminiTranscriptionGenerationConfig(TypedDict):
|
||||
transcription_config: ReadOnly[GeminiTranscriptionConfig]
|
||||
|
||||
|
||||
class GeminiTranscriptionInteractionRequest(TypedDict, total=False):
|
||||
model: ReadOnly[Required[str]]
|
||||
input: ReadOnly[Required[tuple[GeminiTranscriptionAudioInput, ...]]]
|
||||
generation_config: ReadOnly[GeminiTranscriptionGenerationConfig]
|
||||
|
||||
|
||||
class GeminiTranscriptionWordAnnotation(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
type: str | None = None
|
||||
text: str | None = None
|
||||
speaker: str | None = None
|
||||
start_offset: str | None = None
|
||||
end_offset: str | None = None
|
||||
|
||||
|
||||
class GeminiTranscriptionContent(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
type: str | None = None
|
||||
text: str | None = None
|
||||
annotations: tuple[GeminiTranscriptionWordAnnotation, ...] = ()
|
||||
|
||||
|
||||
class GeminiTranscriptionStep(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
type: str | None = None
|
||||
content: tuple[GeminiTranscriptionContent, ...] = ()
|
||||
|
||||
|
||||
class GeminiTranscriptionModalityTokens(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
modality: str | None = None
|
||||
tokens: int = 0
|
||||
|
||||
|
||||
class GeminiTranscriptionUsage(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
total_tokens: int = 0
|
||||
total_input_tokens: int = 0
|
||||
total_output_tokens: int = 0
|
||||
input_tokens_by_modality: tuple[GeminiTranscriptionModalityTokens, ...] = ()
|
||||
|
||||
|
||||
class GeminiTranscriptionInteractionResponse(BaseModel):
|
||||
model_config = ConfigDict(extra="ignore")
|
||||
|
||||
id: str | None = None
|
||||
status: str | None = None
|
||||
usage: GeminiTranscriptionUsage | None = None
|
||||
steps: tuple[GeminiTranscriptionStep, ...] = ()
|
||||
|
|
@ -8503,6 +8503,12 @@ class ProviderConfigManager:
|
|||
)
|
||||
|
||||
return VertexAIAudioTranscriptionConfig()
|
||||
elif litellm.LlmProviders.GEMINI == provider:
|
||||
from litellm.llms.gemini.audio_transcription.transformation import (
|
||||
GeminiAudioTranscriptionConfig,
|
||||
)
|
||||
|
||||
return GeminiAudioTranscriptionConfig()
|
||||
return None
|
||||
|
||||
@staticmethod
|
||||
|
|
|
|||
|
|
@ -51340,6 +51340,43 @@
|
|||
"supports_audio_output": true,
|
||||
"tpm": 250000
|
||||
},
|
||||
"gemini/gemini-3.5-transcribe": {
|
||||
"input_cost_per_audio_token": 2e-06,
|
||||
"input_cost_per_token": 2e-06,
|
||||
"litellm_provider": "gemini",
|
||||
"mode": "audio_transcription",
|
||||
"output_cost_per_token": 1.2e-05,
|
||||
"source": "https://ai.google.dev/gemini-api/docs/pricing",
|
||||
"supported_endpoints": [
|
||||
"/v1/audio/transcriptions"
|
||||
],
|
||||
"supported_modalities": [
|
||||
"text",
|
||||
"audio"
|
||||
],
|
||||
"supported_output_modalities": [
|
||||
"text"
|
||||
],
|
||||
"supports_audio_input": true
|
||||
},
|
||||
"gemini/gemini-3.5-transcribe-live": {
|
||||
"input_cost_per_audio_token": 3.5e-06,
|
||||
"input_cost_per_token": 3.5e-06,
|
||||
"litellm_provider": "gemini",
|
||||
"mode": "audio_transcription",
|
||||
"output_cost_per_token": 2.1e-05,
|
||||
"source": "https://ai.google.dev/gemini-api/docs/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,248 @@
|
|||
import base64
|
||||
import json
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
|
||||
import litellm
|
||||
from litellm.llms.gemini.audio_transcription.transformation import (
|
||||
GeminiAudioTranscriptionConfig,
|
||||
)
|
||||
from litellm.llms.gemini.common_utils import GeminiError
|
||||
from litellm.types.utils import LlmProviders
|
||||
from litellm.utils import ProviderConfigManager
|
||||
|
||||
AUDIO_BYTES = b"RIFF....WAVEfmt fake-wav-bytes"
|
||||
|
||||
COMPLETED_RESPONSE = {
|
||||
"id": "v1_abc123",
|
||||
"status": "completed",
|
||||
"usage": {
|
||||
"total_tokens": 200,
|
||||
"total_input_tokens": 200,
|
||||
"input_tokens_by_modality": [
|
||||
{"modality": "text", "tokens": 1},
|
||||
{"modality": "audio", "tokens": 199},
|
||||
],
|
||||
"total_output_tokens": 0,
|
||||
},
|
||||
"steps": [
|
||||
{
|
||||
"type": "model_generation",
|
||||
"content": [
|
||||
{
|
||||
"type": "text",
|
||||
"text": "Hello world.",
|
||||
"annotations": [
|
||||
{
|
||||
"type": "word_info",
|
||||
"text": "Hello",
|
||||
"speaker": "spk:0",
|
||||
"start_offset": "0.100s",
|
||||
"end_offset": "0.400s",
|
||||
},
|
||||
{
|
||||
"type": "word_info",
|
||||
"text": "world.",
|
||||
"speaker": "spk:1",
|
||||
"start_offset": "0.500s",
|
||||
"end_offset": "0.900s",
|
||||
},
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def make_response(payload):
|
||||
return httpx.Response(200, json=payload, request=httpx.Request("POST", "https://example.test"))
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def config():
|
||||
return GeminiAudioTranscriptionConfig()
|
||||
|
||||
|
||||
def test_provider_config_manager_returns_gemini_config():
|
||||
provider_config = ProviderConfigManager.get_provider_audio_transcription_config(
|
||||
model="gemini-3.5-transcribe", provider=LlmProviders.GEMINI
|
||||
)
|
||||
assert isinstance(provider_config, GeminiAudioTranscriptionConfig)
|
||||
|
||||
|
||||
class TestValidateEnvironment:
|
||||
def test_sets_api_key_and_revision_headers(self, config):
|
||||
headers = config.validate_environment(
|
||||
headers={},
|
||||
model="gemini-3.5-transcribe",
|
||||
messages=[],
|
||||
optional_params={},
|
||||
litellm_params={},
|
||||
api_key="test-key",
|
||||
)
|
||||
assert headers["x-goog-api-key"] == "test-key"
|
||||
assert headers["Api-Revision"] == "2026-05-20"
|
||||
assert headers["Content-Type"] == "application/json"
|
||||
|
||||
def test_missing_api_key_raises(self, config, monkeypatch):
|
||||
monkeypatch.delenv("GOOGLE_API_KEY", raising=False)
|
||||
monkeypatch.delenv("GEMINI_API_KEY", raising=False)
|
||||
with pytest.raises(GeminiError) as excinfo:
|
||||
config.validate_environment(
|
||||
headers={},
|
||||
model="gemini-3.5-transcribe",
|
||||
messages=[],
|
||||
optional_params={},
|
||||
litellm_params={},
|
||||
)
|
||||
assert excinfo.value.status_code == 401
|
||||
|
||||
|
||||
class TestGetCompleteUrl:
|
||||
def test_defaults_to_interactions_endpoint(self, config):
|
||||
url = config.get_complete_url(
|
||||
api_base=None,
|
||||
api_key=None,
|
||||
model="gemini-3.5-transcribe",
|
||||
optional_params={},
|
||||
litellm_params={},
|
||||
)
|
||||
assert url == "https://generativelanguage.googleapis.com/v1beta/interactions"
|
||||
|
||||
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",
|
||||
optional_params={},
|
||||
litellm_params={},
|
||||
)
|
||||
assert url == "http://localhost:8080/v1beta/interactions"
|
||||
|
||||
|
||||
class TestTransformRequest:
|
||||
def test_builds_json_interaction_request(self, config):
|
||||
request_data = config.transform_audio_transcription_request(
|
||||
model="gemini/gemini-3.5-transcribe",
|
||||
audio_file=("sample.wav", AUDIO_BYTES, "audio/wav"),
|
||||
optional_params={},
|
||||
litellm_params={},
|
||||
)
|
||||
assert request_data.files is None
|
||||
assert json.loads(json.dumps(request_data.data)) == {
|
||||
"model": "gemini-3.5-transcribe",
|
||||
"input": [
|
||||
{
|
||||
"type": "audio",
|
||||
"data": base64.b64encode(AUDIO_BYTES).decode("utf-8"),
|
||||
"mime_type": "audio/wav",
|
||||
}
|
||||
],
|
||||
}
|
||||
|
||||
def test_language_maps_to_bcp47_language_codes(self, config):
|
||||
request_data = config.transform_audio_transcription_request(
|
||||
model="gemini-3.5-transcribe",
|
||||
audio_file=("sample.wav", AUDIO_BYTES, "audio/wav"),
|
||||
optional_params={"language": "en"},
|
||||
litellm_params={},
|
||||
)
|
||||
transcription_config = request_data.data["generation_config"]["transcription_config"]
|
||||
assert json.loads(json.dumps(transcription_config)) == {"language_codes": ["en-US"]}
|
||||
|
||||
def test_word_timestamp_granularity_maps_to_verbatim_diarization_mode(self, config):
|
||||
request_data = config.transform_audio_transcription_request(
|
||||
model="gemini-3.5-transcribe",
|
||||
audio_file=("sample.wav", AUDIO_BYTES, "audio/wav"),
|
||||
optional_params={"timestamp_granularities": ["word"]},
|
||||
litellm_params={},
|
||||
)
|
||||
transcription_config = request_data.data["generation_config"]["transcription_config"]
|
||||
assert json.loads(json.dumps(transcription_config)) == {
|
||||
"mode": {
|
||||
"type": "verbatim",
|
||||
"timestamp_granularities": ["word"],
|
||||
"diarization_mode": "speaker",
|
||||
}
|
||||
}
|
||||
|
||||
def test_segment_granularity_sends_no_mode(self, config):
|
||||
request_data = config.transform_audio_transcription_request(
|
||||
model="gemini-3.5-transcribe",
|
||||
audio_file=("sample.wav", AUDIO_BYTES, "audio/wav"),
|
||||
optional_params={"timestamp_granularities": ["segment"]},
|
||||
litellm_params={},
|
||||
)
|
||||
assert "generation_config" not in request_data.data
|
||||
|
||||
|
||||
class TestTransformResponse:
|
||||
def test_completed_interaction_maps_to_transcription_response(self, config):
|
||||
response = config.transform_audio_transcription_response(make_response(COMPLETED_RESPONSE))
|
||||
assert response.text == "Hello world."
|
||||
assert response["task"] == "transcribe"
|
||||
assert response["words"] == [
|
||||
{"word": "Hello", "start": 0.1, "end": 0.4, "speaker": "spk:0"},
|
||||
{"word": "world.", "start": 0.5, "end": 0.9, "speaker": "spk:1"},
|
||||
]
|
||||
assert response["duration"] == 0.9
|
||||
assert response.usage.input_tokens == 200
|
||||
assert response.usage.output_tokens == 0
|
||||
assert response.usage.total_tokens == 200
|
||||
assert response.usage.input_token_details.audio_tokens == 199
|
||||
assert response.usage.input_token_details.text_tokens == 1
|
||||
|
||||
def test_non_completed_status_raises(self, config):
|
||||
with pytest.raises(GeminiError, match="did not complete"):
|
||||
config.transform_audio_transcription_response(
|
||||
make_response({**COMPLETED_RESPONSE, "status": "in_progress"})
|
||||
)
|
||||
|
||||
def test_non_json_response_raises(self, config):
|
||||
raw = httpx.Response(200, text="<html>oops</html>", request=httpx.Request("POST", "https://example.test"))
|
||||
with pytest.raises(GeminiError, match="non-JSON"):
|
||||
config.transform_audio_transcription_response(raw)
|
||||
|
||||
def test_word_without_offsets_survives(self, config):
|
||||
payload = json.loads(json.dumps(COMPLETED_RESPONSE))
|
||||
payload["steps"][0]["content"][0]["annotations"] = [{"type": "word_info", "text": "Hello"}]
|
||||
response = config.transform_audio_transcription_response(make_response(payload))
|
||||
assert response["words"] == [{"word": "Hello"}]
|
||||
assert response.get("duration") is None
|
||||
|
||||
|
||||
class TestCostRegression:
|
||||
@pytest.fixture
|
||||
def local_cost_map(self, monkeypatch):
|
||||
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
|
||||
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
|
||||
|
||||
def test_registry_entries(self, local_cost_map):
|
||||
batch_entry = litellm.model_cost["gemini/gemini-3.5-transcribe"]
|
||||
assert batch_entry["mode"] == "audio_transcription"
|
||||
assert batch_entry["input_cost_per_audio_token"] == 2e-06
|
||||
assert batch_entry["input_cost_per_token"] == 2e-06
|
||||
assert batch_entry["output_cost_per_token"] == 1.2e-05
|
||||
assert batch_entry["supported_endpoints"] == ["/v1/audio/transcriptions"]
|
||||
|
||||
live_entry = litellm.model_cost["gemini/gemini-3.5-transcribe-live"]
|
||||
assert live_entry["mode"] == "audio_transcription"
|
||||
assert live_entry["input_cost_per_audio_token"] == 3.5e-06
|
||||
assert live_entry["input_cost_per_token"] == 3.5e-06
|
||||
assert live_entry["output_cost_per_token"] == 2.1e-05
|
||||
assert live_entry["supported_endpoints"] == ["/v1/realtime"]
|
||||
|
||||
def test_completion_cost_bills_provider_reported_tokens(self, config, local_cost_map):
|
||||
payload = json.loads(json.dumps(COMPLETED_RESPONSE))
|
||||
payload["usage"]["total_output_tokens"] = 10
|
||||
payload["usage"]["total_tokens"] = 210
|
||||
response = config.transform_audio_transcription_response(make_response(payload))
|
||||
cost = litellm.completion_cost(
|
||||
completion_response=response,
|
||||
model="gemini/gemini-3.5-transcribe",
|
||||
call_type="transcription",
|
||||
)
|
||||
assert cost == pytest.approx(199 * 2e-06 + 1 * 2e-06 + 10 * 1.2e-05)
|
||||
|
|
@ -1864,3 +1864,80 @@ def test_map_openai_params_drops_stock_voice_case_insensitively():
|
|||
|
||||
passthrough = cfg.map_openai_params(optional_params={}, non_default_params={"voice": "Kore"})
|
||||
assert passthrough["generationConfig"]["speechConfig"]["voiceConfig"]["prebuiltVoiceConfig"]["voiceName"] == "Kore"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=False)
|
||||
def patch_gemini_transcribe_live_cost_map_entry(monkeypatch):
|
||||
"""Inject the gemini-3.5-transcribe-live registry entry locally.
|
||||
|
||||
litellm.model_cost is fetched from main branch at import time, so in CI
|
||||
the entry may not exist yet. Also stamp supported_output_modalities on a
|
||||
chat model to prove mode, not output modalities, drives the discriminator.
|
||||
"""
|
||||
for m in ["gemini-3.5-transcribe-live", "gemini/gemini-3.5-transcribe-live"]:
|
||||
entry = dict(litellm.model_cost.get(m, {}))
|
||||
entry["mode"] = "audio_transcription"
|
||||
monkeypatch.setitem(litellm.model_cost, m, entry)
|
||||
chat_entry = dict(litellm.model_cost.get("gemini-2.5-flash", {}))
|
||||
chat_entry["supported_output_modalities"] = ["text"]
|
||||
monkeypatch.setitem(litellm.model_cost, "gemini-2.5-flash", chat_entry)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("model", ["gemini-3.5-transcribe-live", "gemini/gemini-3.5-transcribe-live"])
|
||||
def test_gemini_transcribe_live_eager_setup_uses_text_modality(model, patch_gemini_transcribe_live_cost_map_entry):
|
||||
"""Regression: the hardcoded AUDIO eager setup closes transcribe-live sessions with 1007."""
|
||||
config = GeminiRealtimeConfig()
|
||||
|
||||
setup = json.loads(config.session_configuration_request(model))["setup"]
|
||||
|
||||
assert setup["generationConfig"]["responseModalities"] == ["TEXT"]
|
||||
|
||||
|
||||
def test_gemini_transcribe_live_session_update_defaults_to_text_modality(
|
||||
patch_gemini_transcribe_live_cost_map_entry,
|
||||
):
|
||||
config = GeminiRealtimeConfig()
|
||||
session_update = {
|
||||
"type": "session.update",
|
||||
"session": {"instructions": "Transcribe the audio."},
|
||||
}
|
||||
|
||||
messages = config.transform_realtime_request(
|
||||
json.dumps(session_update),
|
||||
"gemini-3.5-transcribe-live",
|
||||
session_configuration_request=None,
|
||||
)
|
||||
|
||||
setup = json.loads(messages[0])["setup"]
|
||||
assert setup["generationConfig"]["responseModalities"] == ["TEXT"]
|
||||
|
||||
|
||||
@pytest.mark.parametrize("modalities", [["audio"], ["audio", "text"]])
|
||||
def test_gemini_transcribe_live_coerces_audio_modality_to_text(
|
||||
modalities, patch_gemini_transcribe_live_cost_map_entry
|
||||
):
|
||||
config = GeminiRealtimeConfig()
|
||||
session_update = {
|
||||
"type": "session.update",
|
||||
"session": {"modalities": modalities},
|
||||
}
|
||||
|
||||
messages = config.transform_realtime_request(
|
||||
json.dumps(session_update),
|
||||
"gemini-3.5-transcribe-live",
|
||||
session_configuration_request=None,
|
||||
)
|
||||
|
||||
setup = json.loads(messages[0])["setup"]
|
||||
assert setup["generationConfig"]["responseModalities"] == ["TEXT"]
|
||||
|
||||
|
||||
def test_gemini_chat_model_with_text_output_modalities_keeps_audio_eager_setup(
|
||||
patch_gemini_transcribe_live_cost_map_entry,
|
||||
):
|
||||
"""Chat entries also declare supported_output_modalities ["text"]; they must keep AUDIO."""
|
||||
config = GeminiRealtimeConfig()
|
||||
|
||||
setup = json.loads(config.session_configuration_request("gemini-2.5-flash"))["setup"]
|
||||
|
||||
assert setup["generationConfig"]["responseModalities"] == ["AUDIO"]
|
||||
|
|
|
|||
|
|
@ -343,6 +343,31 @@ def test_transcription_cost_uses_token_pricing(_local_model_cost_map):
|
|||
assert pytest.approx(cost, rel=1e-6) == expected_cost
|
||||
|
||||
|
||||
def test_transcription_token_pricing_is_provider_aware(_local_model_cost_map):
|
||||
"""Regression: the token-priced transcription path hardcoded provider openai,
|
||||
so gemini transcription models raised "This model isn't mapped yet"."""
|
||||
from litellm import completion_cost
|
||||
|
||||
usage = Usage(
|
||||
prompt_tokens=200,
|
||||
completion_tokens=10,
|
||||
total_tokens=210,
|
||||
prompt_tokens_details=PromptTokensDetailsWrapper(text_tokens=1, audio_tokens=199),
|
||||
)
|
||||
response = TranscriptionResponse(text="demo text")
|
||||
response.usage = usage
|
||||
|
||||
cost = completion_cost(
|
||||
completion_response=response,
|
||||
model="gemini/gemini-3.5-transcribe",
|
||||
custom_llm_provider="gemini",
|
||||
call_type="atranscription",
|
||||
)
|
||||
|
||||
expected_cost = (199 * 2e-06) + (1 * 2e-06) + (10 * 1.2e-05)
|
||||
assert pytest.approx(cost, rel=1e-6) == expected_cost
|
||||
|
||||
|
||||
def test_transcription_cost_falls_back_to_duration(_local_model_cost_map):
|
||||
from litellm import completion_cost
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue