scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

Merge pull request #38540 from BerriAI/litellm_gemini_35_transcribe

Browse source 
feat(gemini): day-0 support for gemini-3.5-transcribe and transcribe-live
This commit is contained in:
Mateo Wang 2026-08-27 12:09:40 -07:00 • committed by GitHub
commit 493bca667b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 1091 additions and 17 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
litellm/cost_calculator.py
View file

@ -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,
)

1
litellm/litellm_core_utils/realtime_streaming.py
View file

@ -955,6 +955,7 @@ class RealTimeStreaming:
transcript = event.get("transcript", "")
self._collect_user_input_from_backend_event(cast(dict, event))
self.store_message(event_str)
self._capture_transcription_usage(event)
await self._send_event_to_client(event, event_str)
blocked = await self.run_realtime_guardrails(
cast(str, transcript),

0
litellm/llms/gemini/audio_transcription/__init__.py Normal file
View file

250
litellm/llms/gemini/audio_transcription/transformation.py Normal file
View 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

82
litellm/llms/gemini/realtime/transformation.py
View file

@ -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
@ -53,6 +53,7 @@ from litellm.types.llms.vertex_ai import (
)
from litellm.types.realtime import (
ALL_DELTA_TYPES,
RealtimeInputAudioTranscriptionUsage,
RealtimeModalityResponseTransformOutput,
RealtimeResponseTransformInput,
RealtimeResponseTypedDict,
@ -95,6 +96,18 @@ def _gemini_live_speech_config(voice: object) -> Mapping[str, object] | None:
return VertexGeminiConfig()._map_audio_params({"voice": voice})
# Google bills Live transcription at an estimated 25 audio tokens/sec of input and
# 175 text tokens/min of output (ai.google.dev/gemini-api/docs/pricing).
GEMINI_LIVE_TRANSCRIBE_AUDIO_TOKENS_PER_SECOND: Final = 25
GEMINI_LIVE_TRANSCRIBE_OUTPUT_TEXT_TOKENS_PER_MINUTE: Final = 175
PCM16_INPUT_AUDIO_BYTES_PER_SECOND: Final = 48000
def _base64_decoded_byte_count(data: str) -> int:
padding: Final = 2 if data.endswith("==") else 1 if data.endswith("=") else 0
return max(len(data) * 3 // 4 - padding, 0)
class GeminiRealtimeConfig(BaseRealtimeConfig):
_TOOL_CALL_ID_TO_NAME_MAX = 256 # LRU cap for call_id→name mapping
@ -104,6 +117,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
# Gemini Live sometimes emits usageMetadata in a standalone frame between
# turns; buffer it here so the next response.done carries the token counts.
self._pending_usage_metadata: dict | None = None
self._unbilled_input_audio_bytes: int = 0
def is_setup_message(self, msg_obj: dict) -> bool:
return "setup" in msg_obj
@ -384,17 +398,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 +458,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")
@ -558,9 +580,10 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
return self._handle_conversation_item(json_message)
if msg_type == "input_audio_buffer.append":
realtime_input_dict["audio"] = HttpxBlobType(
mimeType=self.get_audio_mime_type(), data=json_message["audio"]
)
audio_b64: Final = json_message["audio"]
if isinstance(audio_b64, str):
self._unbilled_input_audio_bytes += _base64_decoded_byte_count(audio_b64)
realtime_input_dict["audio"] = HttpxBlobType(mimeType=self.get_audio_mime_type(), data=audio_b64)
realtime_input_dict = cast(
BidiGenerateContentRealtimeInput,
@ -1151,6 +1174,23 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
raise ValueError(f"Unknown openai event: {key}, value: {value}")
return openai_event
def _consume_input_transcription_usage_estimate(self, model: str) -> RealtimeInputAudioTranscriptionUsage | None:
"""Gemini Live sends no usageMetadata for transcribe sessions; estimate billing from streamed audio duration."""
if self._unbilled_input_audio_bytes <= 0 or not self._is_text_only_live_model(model):
return None
audio_seconds: Final = self._unbilled_input_audio_bytes / PCM16_INPUT_AUDIO_BYTES_PER_SECOND
self._unbilled_input_audio_bytes = 0
audio_tokens: Final = round(audio_seconds * GEMINI_LIVE_TRANSCRIBE_AUDIO_TOKENS_PER_SECOND)
output_tokens: Final = round(audio_seconds * GEMINI_LIVE_TRANSCRIBE_OUTPUT_TEXT_TOKENS_PER_MINUTE / 60)
usage: Final[RealtimeInputAudioTranscriptionUsage] = {
"type": "tokens",
"input_tokens": audio_tokens,
"output_tokens": output_tokens,
"total_tokens": audio_tokens + output_tokens,
"input_token_details": {"text_tokens": 0, "audio_tokens": audio_tokens},
}
return usage
def transform_realtime_response(
self,
message: str | bytes,
@ -1190,6 +1230,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
if isinstance(server_content, dict):
input_tx: Final = server_content.get("inputTranscription")
if isinstance(input_tx, dict) and input_tx.get("text"):
transcription_usage: Final = self._consume_input_transcription_usage_estimate(model)
returned_message.append(
cast(
OpenAIRealtimeEvents,
@ -1199,6 +1240,7 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
"transcript": input_tx["text"],
"item_id": f"item_{uuid.uuid4()}",
"content_index": 0,
**({} if transcription_usage is None else {"usage": transcription_usage}),
},
)
)
@ -1235,6 +1277,12 @@ class GeminiRealtimeConfig(BaseRealtimeConfig):
)
)
# Transcription-only models emit generationComplete with no prior
# modelTurn delta; there is no started OpenAI response to close, so
# drop it and let siblings (turnComplete, usageMetadata) process.
if current_delta_type is None and "modelTurn" not in server_content:
server_content.pop("generationComplete", None)
# Mark transcription-only serverContent as handled so the main loop
# skips it; sibling keys like toolCall are still processed below.
_model_content_keys: Final = {
@ -1583,7 +1631,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

41
litellm/model_prices_and_context_window_backup.json
View file

@ -51340,6 +51340,47 @@
"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,
"tpm": 800000,
"rpm": 2000
},
"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,
"tpm": 250000,
"rpm": 10
},
"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
View 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, ...] = ()

13
litellm/types/realtime.py
View file

@ -162,3 +162,16 @@ class RealtimeErrorDetail(TypedDict):
class RealtimeErrorEvent(TypedDict):
type: ReadOnly[Literal["error"]]
error: ReadOnly[RealtimeErrorDetail]
class RealtimeInputAudioTranscriptionUsageInputTokenDetails(TypedDict):
text_tokens: ReadOnly[int]
audio_tokens: ReadOnly[int]
class RealtimeInputAudioTranscriptionUsage(TypedDict):
type: ReadOnly[Literal["tokens"]]
input_tokens: ReadOnly[int]
output_tokens: ReadOnly[int]
total_tokens: ReadOnly[int]
input_token_details: ReadOnly[RealtimeInputAudioTranscriptionUsageInputTokenDetails]

6
litellm/utils.py
View file

@ -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

41
model_prices_and_context_window.json
View file

@ -51340,6 +51340,47 @@
"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,
"tpm": 800000,
"rpm": 2000
},
"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,
"tpm": 250000,
"rpm": 10
},
"perplexity/pplx-embed-context-v1-0.6b": {
"input_cost_per_token": 8e-09,
"litellm_provider": "perplexity",

62
tests/test_litellm/litellm_core_utils/test_realtime_streaming.py
View file

@ -2957,3 +2957,65 @@ async def test_log_messages_routes_async_logging_through_bounded_worker():
logging_obj.success_handler.assert_not_called()
# the bare create_task path must no longer be used for success logging
mock_create_task.assert_not_called()
@pytest.mark.asyncio
async def test_provider_config_path_captures_transcription_usage():
"""A transcription.completed event with usage from the provider transform must
land in the logged messages so realtime cost calculation can bill it."""
from typing import Final
from litellm.types.realtime import RealtimeInputAudioTranscriptionUsage, RealtimeResponseTypedDict
client_ws: Final = MagicMock()
client_ws.send_text = AsyncMock()
backend_ws: Final = MagicMock()
backend_ws.send = AsyncMock()
logging_obj: Final = MagicMock()
usage: Final[RealtimeInputAudioTranscriptionUsage] = {
"type": "tokens",
"input_tokens": 50,
"output_tokens": 6,
"total_tokens": 56,
"input_token_details": {"text_tokens": 0, "audio_tokens": 50},
}
transform_output: Final[RealtimeResponseTypedDict] = {
"response": {
"type": "conversation.item.input_audio_transcription.completed",
"event_id": "event_1",
"transcript": "ahoy",
"item_id": "item_1",
"content_index": 0,
"usage": usage,
},
"current_output_item_id": None,
"current_response_id": None,
"current_delta_chunks": None,
"current_conversation_id": None,
"current_item_chunks": None,
"current_delta_type": None,
"session_configuration_request": None,
}
provider_config: Final = MagicMock()
provider_config.transform_realtime_request = MagicMock(return_value=())
provider_config.transform_realtime_response = MagicMock(return_value=transform_output)
streaming: Final = RealTimeStreaming(
client_ws,
backend_ws,
logging_obj,
provider_config=provider_config,
model="gemini-3.5-transcribe-live",
)
await streaming._handle_provider_config_message("{}")
usage_events: Final = tuple(
message
for message in streaming.messages
if isinstance(message, dict)
and message.get("type") == "conversation.item.input_audio_transcription.completed"
and message.get("usage") == usage
)
assert len(usage_events) == 1

0
tests/test_litellm/llms/gemini/audio_transcription/__init__.py Normal file
View file

248
tests/test_litellm/llms/gemini/audio_transcription/test_gemini_audio_transcription_transformation.py Normal file
View file

@ -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)

255
tests/test_litellm/llms/gemini/realtime/test_gemini_realtime_transformation.py
View file

@ -1864,3 +1864,258 @@ 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"]
def test_generation_complete_without_prior_delta_keeps_turn_usage(patch_gemini_audio_cost_map_entries):
from typing import Final
from litellm.types.llms.gemini import BidiGenerateContentServerMessage
from litellm.types.realtime import RealtimeResponseTransformInput
config: Final = GeminiRealtimeConfig()
turn_end_frame: Final[BidiGenerateContentServerMessage] = {
"serverContent": {"generationComplete": True, "turnComplete": True},
"usageMetadata": {
"promptTokenCount": 200,
"totalTokenCount": 200,
"promptTokensDetails": [
{"modality": "AUDIO", "tokenCount": 199},
{"modality": "TEXT", "tokenCount": 1},
],
},
}
transform_input: Final[RealtimeResponseTransformInput] = {
"session_configuration_request": None,
"current_output_item_id": None,
"current_response_id": None,
"current_conversation_id": None,
"current_delta_chunks": None,
"current_item_chunks": None,
"current_delta_type": None,
}
result: Final = config.transform_realtime_response(
json.dumps(turn_end_frame),
"gemini-3.5-transcribe-live",
MagicMock(),
realtime_response_transform_input=transform_input,
)
done_events: Final = tuple(event for event in result["response"] if event["type"] == "response.done")
assert len(done_events) == 1
assert done_events[0]["response"]["usage"]["input_tokens"] == 200
def test_bare_generation_complete_without_prior_delta_is_dropped(patch_gemini_audio_cost_map_entries):
from typing import Final
from litellm.types.llms.gemini import BidiGenerateContentServerMessage
from litellm.types.realtime import RealtimeResponseTransformInput
config: Final = GeminiRealtimeConfig()
bare_frame: Final[BidiGenerateContentServerMessage] = {"serverContent": {"generationComplete": True}}
transform_input: Final[RealtimeResponseTransformInput] = {
"session_configuration_request": None,
"current_output_item_id": None,
"current_response_id": None,
"current_conversation_id": None,
"current_delta_chunks": None,
"current_item_chunks": None,
"current_delta_type": None,
}
result: Final = config.transform_realtime_response(
json.dumps(bare_frame),
"gemini-3.5-transcribe-live",
MagicMock(),
realtime_response_transform_input=transform_input,
)
assert result["response"] == []
def _input_audio_append_message(raw_byte_count: int) -> str:
import base64
return json.dumps(
{"type": "input_audio_buffer.append", "audio": base64.b64encode(b"\x00" * raw_byte_count).decode()}
)
def test_transcribe_live_completed_event_carries_estimated_usage(patch_gemini_transcribe_live_cost_map_entry):
"""Gemini Live sends no usageMetadata for transcribe sessions, so LiteLLM bills
from streamed audio duration at Google's published estimate (25 audio tok/sec in,
175 text tok/min out): 96000 pcm16 bytes = 2s at 24kHz -> 50 in / 6 out."""
from typing import Final
from litellm.types.llms.gemini import BidiGenerateContentServerMessage
from litellm.types.realtime import RealtimeInputAudioTranscriptionUsage, RealtimeResponseTransformInput
config: Final = GeminiRealtimeConfig()
config.transform_realtime_request(_input_audio_append_message(96000), "gemini-3.5-transcribe-live")
transcript_frame: Final[BidiGenerateContentServerMessage] = {
"serverContent": {"inputTranscription": {"text": "ahoy there"}}
}
transform_input: Final[RealtimeResponseTransformInput] = {
"session_configuration_request": None,
"current_output_item_id": None,
"current_response_id": None,
"current_conversation_id": None,
"current_delta_chunks": None,
"current_item_chunks": None,
"current_delta_type": None,
}
result: Final = config.transform_realtime_response(
json.dumps(transcript_frame),
"gemini-3.5-transcribe-live",
MagicMock(),
realtime_response_transform_input=transform_input,
)
completed: Final = tuple(
event
for event in result["response"]
if event["type"] == "conversation.item.input_audio_transcription.completed"
)
assert len(completed) == 1
assert completed[0]["transcript"] == "ahoy there"
expected_usage: Final[RealtimeInputAudioTranscriptionUsage] = {
"type": "tokens",
"input_tokens": 50,
"output_tokens": 6,
"total_tokens": 56,
"input_token_details": {"text_tokens": 0, "audio_tokens": 50},
}
assert completed[0]["usage"] == expected_usage
second: Final = config.transform_realtime_response(
json.dumps(transcript_frame),
"gemini-3.5-transcribe-live",
MagicMock(),
realtime_response_transform_input=transform_input,
)
second_completed: Final = tuple(
event
for event in second["response"]
if event["type"] == "conversation.item.input_audio_transcription.completed"
)
assert len(second_completed) == 1
assert "usage" not in second_completed[0]
def test_non_transcription_live_model_completed_event_has_no_usage(patch_gemini_audio_cost_map_entries):
"""Conversational Live models get their audio tokens from usageMetadata via
response.done; attaching estimated usage to their transcription events would
double-bill, so the estimate is gated to audio_transcription-mode models."""
from typing import Final
from litellm.types.llms.gemini import BidiGenerateContentServerMessage
from litellm.types.realtime import RealtimeResponseTransformInput
config: Final = GeminiRealtimeConfig()
config.transform_realtime_request(_input_audio_append_message(96000), "gemini-3.1-flash-live-preview")
transcript_frame: Final[BidiGenerateContentServerMessage] = {
"serverContent": {"inputTranscription": {"text": "ahoy there"}}
}
transform_input: Final[RealtimeResponseTransformInput] = {
"session_configuration_request": None,
"current_output_item_id": None,
"current_response_id": None,
"current_conversation_id": None,
"current_delta_chunks": None,
"current_item_chunks": None,
"current_delta_type": None,
}
result: Final = config.transform_realtime_response(
json.dumps(transcript_frame),
"gemini-3.1-flash-live-preview",
MagicMock(),
realtime_response_transform_input=transform_input,
)
completed: Final = tuple(
event
for event in result["response"]
if event["type"] == "conversation.item.input_audio_transcription.completed"
)
assert len(completed) == 1
assert "usage" not in completed[0]

25
tests/test_litellm/test_cost_calculator.py
View file

@ -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