scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-25 05:54:53 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 125

Merge b81e63020a into 7ef92689da

Browse source
This commit is contained in:
Zahidur Rahman Faisal 2026-10-19 16:28:39 +02:00 • committed by GitHub
commit 6db25f52e7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
8 changed files with 340 additions and 13 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

42
litellm/litellm_core_utils/audio_utils/utils.py
View file

@ -5,7 +5,7 @@ Utils used for litellm.transcription() and litellm.atranscription()
import hashlib
import os
from dataclasses import dataclass
from typing import Final
from typing import Final, Literal
from litellm.types.files import (
AUDIO_FILE_TYPES,
@ -401,3 +401,43 @@ def speech_media_type_from_audio_bytes(audio: bytes) -> str | None:
if (audio[1] & _ADTS_SYNC_AND_LAYER_MASK) == _ADTS_SYNC_AND_LAYER:
return _adts_aac_frame_media_type(audio)
return _mpeg_audio_frame_media_type(audio)
GeminiTranscriptionMode = Literal["smart", "verbatim"]
GEMINI_TRANSCRIPTION_MODES: Final[frozenset[str]] = frozenset(("smart", "verbatim"))
def normalize_gemini_transcription_mode(mode: object) -> GeminiTranscriptionMode | None:
"""
Parse the provider-specific `mode` param Gemini 3.5 Transcribe accepts
(https://ai.google.dev/gemini-api/docs/transcribe): `smart` cleans up
disfluencies and self-corrections, `verbatim` (Google's default) keeps them.
Case-insensitive; None when the caller sent nothing.
Raises ValueError (unknown value) or TypeError (not a string) so the provider
config can turn it into a 400.
"""
if mode is None:
return None
if not isinstance(mode, str):
raise TypeError(f"Unsupported transcription mode {mode!r}. Supported values: smart, verbatim")
if mode.lower() not in GEMINI_TRANSCRIPTION_MODES:
raise ValueError(f"Unsupported transcription mode {mode!r}. Supported values: smart, verbatim")
return "smart" if mode.lower() == "smart" else "verbatim"
def normalize_custom_vocabulary(custom_vocabulary: object) -> tuple[str, ...]:
"""
Parse the provider-specific `custom_vocabulary` param: a sequence of terms, or
a comma-separated string (what a multipart form field arrives as). Blank
terms are dropped and order is kept.
"""
if custom_vocabulary is None:
return ()
if isinstance(custom_vocabulary, str):
raw_terms: tuple[object, ...] = tuple(custom_vocabulary.split(","))
elif isinstance(custom_vocabulary, (list, tuple)):
raw_terms = tuple(custom_vocabulary)
else:
raise TypeError("custom_vocabulary must be a list of terms or a comma-separated string")
return tuple(term.strip() for term in raw_terms if isinstance(term, str) and term.strip())

59
litellm/llms/gemini/audio_transcription/transformation.py
View file

@ -6,6 +6,8 @@ from httpx import Headers, Response
from litellm.litellm_core_utils.audio_utils.subtitle_utils import SUBTITLE_RESPONSE_FORMATS
from litellm.litellm_core_utils.audio_utils.utils import (
normalize_custom_vocabulary,
normalize_gemini_transcription_mode,
normalize_transcription_language_to_bcp47,
process_audio_file,
)
@ -42,6 +44,13 @@ 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
Provider-specific params (passed as extra kwargs / form fields):
- `mode`: "smart" (disfluency removal, self-corrections, formatting) or
"verbatim" (Google's default). Smart mode cannot be combined with word
timestamps or subtitle formats.
- `custom_vocabulary`: terms to bias recognition towards, as a list or a
comma-separated string.
"""
def get_supported_openai_params(
@ -190,6 +199,8 @@ _WORD_TIMESTAMP_CONFIG: Final[GeminiTranscriptionConfig] = {
"diarization_mode": "speaker",
},
}
_SMART_MODE_CONFIG: Final[GeminiTranscriptionConfig] = {"mode": {"type": "smart"}}
_VERBATIM_MODE_CONFIG: Final[GeminiTranscriptionConfig] = {"mode": {"type": "verbatim"}}
def _build_interaction_request(
@ -220,17 +231,53 @@ def _language_config(language: object) -> GeminiTranscriptionConfig:
return language_config
def _timestamp_config(timestamp_granularities: object, response_format: object) -> GeminiTranscriptionConfig:
wants_word_timestamps: Final = (
isinstance(timestamp_granularities, list) and "word" in timestamp_granularities
) or (isinstance(response_format, str) and response_format in SUBTITLE_RESPONSE_FORMATS)
return _WORD_TIMESTAMP_CONFIG if wants_word_timestamps else _EMPTY_TRANSCRIPTION_CONFIG
def _wants_word_timestamps(timestamp_granularities: object, response_format: object) -> bool:
return (isinstance(timestamp_granularities, list) and "word" in timestamp_granularities) or (
isinstance(response_format, str) and response_format in SUBTITLE_RESPONSE_FORMATS
)
def _mode_config(mode: object, wants_word_timestamps: bool) -> GeminiTranscriptionConfig:
try:
normalized_mode: Final = normalize_gemini_transcription_mode(mode)
except (TypeError, ValueError) as error:
raise GeminiError(status_code=400, message=str(error)) from error
if normalized_mode == "smart":
if wants_word_timestamps:
raise GeminiError(
status_code=400,
message=(
"Gemini smart transcription mode cannot be combined with word timestamps "
"(timestamp_granularities=['word'] or a subtitle response_format)."
),
)
return _SMART_MODE_CONFIG
if wants_word_timestamps:
return _WORD_TIMESTAMP_CONFIG
return _VERBATIM_MODE_CONFIG if normalized_mode == "verbatim" else _EMPTY_TRANSCRIPTION_CONFIG
def _vocabulary_config(custom_vocabulary: object) -> GeminiTranscriptionConfig:
try:
terms: Final = normalize_custom_vocabulary(custom_vocabulary)
except (TypeError, ValueError) as error:
raise GeminiError(status_code=400, message=str(error)) from error
if not terms:
return _EMPTY_TRANSCRIPTION_CONFIG
vocabulary_config: Final[GeminiTranscriptionConfig] = {"custom_vocabulary": terms}
return vocabulary_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"), optional_params.get("response_format")),
**_mode_config(
optional_params.get("mode"),
_wants_word_timestamps(
optional_params.get("timestamp_granularities"), optional_params.get("response_format")
),
),
**_vocabulary_config(optional_params.get("custom_vocabulary")),
}
return transcription_config

41
litellm/llms/vertex_ai/audio_transcription/gemini_transcribe_transformation.py
View file

@ -7,6 +7,8 @@ from httpx import Headers, Response
import litellm
from litellm.exceptions import UnsupportedParamsError
from litellm.litellm_core_utils.audio_utils.utils import (
normalize_custom_vocabulary,
normalize_gemini_transcription_mode,
normalize_transcription_language_to_bcp47,
process_audio_file,
)
@ -47,6 +49,17 @@ AUDIO_MODALITY: Final = "AUDIO"
class VertexGeminiAudioTranscriptionConfig(BaseAudioTranscriptionConfig, VertexBase):
"""
Maps OpenAI /v1/audio/transcriptions onto Vertex generateContent for
gemini-3.5-transcribe, using `generationConfig.audioTranscriptionConfig`.
Provider-specific params (passed as extra kwargs / form fields):
- `mode`: "smart" (disfluency removal, self-corrections, formatting) or
"verbatim" (Google's default), sent as `audioTranscriptionConfig.mode`.
- `custom_vocabulary`: terms to bias recognition towards, as a list or a
comma-separated string, sent as `audioTranscriptionConfig.customVocabulary`.
"""
def __init__(self) -> None:
BaseAudioTranscriptionConfig.__init__(self)
VertexBase.__init__(self)
@ -166,7 +179,11 @@ class VertexGeminiAudioTranscriptionConfig(BaseAudioTranscriptionConfig, VertexB
),
),
generationConfig=VertexGeminiTranscriptionGenerationConfig(
audioTranscriptionConfig=_audio_transcription_config(optional_params.get("language"))
audioTranscriptionConfig=_audio_transcription_config(
language=optional_params.get("language"),
mode=optional_params.get("mode"),
custom_vocabulary=optional_params.get("custom_vocabulary"),
)
),
)
return AudioTranscriptionRequestData(data=dict(request_body))
@ -210,7 +227,21 @@ class VertexGeminiAudioTranscriptionConfig(BaseAudioTranscriptionConfig, VertexB
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),))
def _audio_transcription_config(
language: object, mode: object, custom_vocabulary: object
) -> VertexGeminiTranscriptionAudioConfig:
try:
normalized_mode: Final = normalize_gemini_transcription_mode(mode)
terms: Final = normalize_custom_vocabulary(custom_vocabulary)
except (TypeError, ValueError) as error:
raise VertexAIError(status_code=400, message=str(error)) from error
language_config: Final[VertexGeminiTranscriptionAudioConfig] = (
{"languageCodes": (normalize_transcription_language_to_bcp47(language),)}
if isinstance(language, str) and language
else {}
)
mode_config: Final[VertexGeminiTranscriptionAudioConfig] = (
{"mode": "SMART" if normalized_mode == "smart" else "VERBATIM"} if normalized_mode is not None else {}
)
vocabulary_config: Final[VertexGeminiTranscriptionAudioConfig] = {"customVocabulary": terms} if terms else {}
return VertexGeminiTranscriptionAudioConfig(**language_config, **mode_config, **vocabulary_config)

7
litellm/types/llms/gemini_audio_transcription.py
View file

@ -18,9 +18,14 @@ class GeminiTranscriptionVerbatimMode(TypedDict, total=False):
diarization_mode: ReadOnly[Literal["speaker"]]
class GeminiTranscriptionSmartMode(TypedDict):
type: ReadOnly[Literal["smart"]]
class GeminiTranscriptionConfig(TypedDict, total=False):
language_codes: ReadOnly[tuple[str, ...]]
mode: ReadOnly[GeminiTranscriptionVerbatimMode]
mode: ReadOnly[GeminiTranscriptionVerbatimMode | GeminiTranscriptionSmartMode]
custom_vocabulary: ReadOnly[tuple[str, ...]]
class GeminiTranscriptionGenerationConfig(TypedDict):

2
litellm/types/llms/vertex_ai_gemini_transcription.py
View file

@ -22,6 +22,8 @@ class VertexGeminiTranscriptionContent(TypedDict):
class VertexGeminiTranscriptionAudioConfig(TypedDict, total=False):
languageCodes: ReadOnly[tuple[str, ...]]
mode: ReadOnly[Literal["SMART", "VERBATIM"]]
customVocabulary: ReadOnly[tuple[str, ...]]
class VertexGeminiTranscriptionGenerationConfig(TypedDict):

40
tests/unit/litellm_core_utils/test_audio_utils.py
View file

@ -15,6 +15,8 @@ from litellm.litellm_core_utils.audio_utils.utils import (
get_audio_file_content_hash,
get_audio_file_for_health_check,
get_audio_file_name,
normalize_custom_vocabulary,
normalize_gemini_transcription_mode,
process_audio_file,
)
@ -409,3 +411,41 @@ class TestSpeechMediaTypeFromAudioBytes:
from litellm.litellm_core_utils.audio_utils.utils import speech_media_type_from_audio_bytes
assert speech_media_type_from_audio_bytes(audio) == expected
class TestNormalizeGeminiTranscriptionMode:
@pytest.mark.parametrize(
"raw,expected",
[("smart", "smart"), ("SMART", "smart"), ("Verbatim", "verbatim"), ("verbatim", "verbatim")],
)
def test_case_insensitive_values_normalize(self, raw: str, expected: str) -> None:
assert normalize_gemini_transcription_mode(raw) == expected
def test_missing_mode_is_none(self) -> None:
assert normalize_gemini_transcription_mode(None) is None
@pytest.mark.parametrize("raw", ["loud", ""])
def test_unknown_mode_raises(self, raw: str) -> None:
with pytest.raises(ValueError, match="Unsupported transcription mode"):
normalize_gemini_transcription_mode(raw)
@pytest.mark.parametrize("raw", [1, ["smart"]])
def test_non_string_mode_raises(self, raw: object) -> None:
with pytest.raises(TypeError, match="Unsupported transcription mode"):
normalize_gemini_transcription_mode(raw)
class TestNormalizeCustomVocabulary:
def test_list_is_stripped_and_blanks_dropped(self) -> None:
assert normalize_custom_vocabulary([" LiteLLM ", "", "Kubernetes", " "]) == ("LiteLLM", "Kubernetes")
def test_comma_separated_string_is_split(self) -> None:
assert normalize_custom_vocabulary("LiteLLM, Kubernetes,,HV-12 ") == ("LiteLLM", "Kubernetes", "HV-12")
def test_missing_vocabulary_is_empty(self) -> None:
assert normalize_custom_vocabulary(None) == ()
@pytest.mark.parametrize("raw", [42, {"term": "x"}])
def test_unsupported_shape_raises(self, raw: object) -> None:
with pytest.raises(TypeError, match="custom_vocabulary"):
normalize_custom_vocabulary(raw)

100
tests/unit/llms/gemini/audio_transcription/test_gemini_audio_transcription_transformation.py
View file

@ -1,5 +1,6 @@
import base64
import json
from collections.abc import Mapping
import httpx
import pytest
@ -214,6 +215,105 @@ class TestTransformRequest:
assert "generation_config" not in request_data.data
class TestSmartModeAndVocabulary:
def test_smart_mode_sends_smart_transcription_config(self, config: GeminiAudioTranscriptionConfig) -> None:
request_data = config.transform_audio_transcription_request(
model="gemini-3.5-transcribe",
audio_file=("sample.wav", AUDIO_BYTES, "audio/wav"),
optional_params={"mode": "SMART", "language": "en"},
litellm_params={},
)
transcription_config = json.loads(json.dumps(request_data.data["generation_config"]["transcription_config"]))
assert transcription_config == {"language_codes": ["en-US"], "mode": {"type": "smart"}}
def test_verbatim_mode_is_sent_explicitly(self, config: GeminiAudioTranscriptionConfig) -> None:
request_data = config.transform_audio_transcription_request(
model="gemini-3.5-transcribe",
audio_file=("sample.wav", AUDIO_BYTES, "audio/wav"),
optional_params={"mode": "verbatim"},
litellm_params={},
)
transcription_config = json.loads(json.dumps(request_data.data["generation_config"]["transcription_config"]))
assert transcription_config == {"mode": {"type": "verbatim"}}
def test_verbatim_mode_keeps_word_timestamps(self, config: GeminiAudioTranscriptionConfig) -> None:
request_data = config.transform_audio_transcription_request(
model="gemini-3.5-transcribe",
audio_file=("sample.wav", AUDIO_BYTES, "audio/wav"),
optional_params={"mode": "verbatim", "timestamp_granularities": ["word"]},
litellm_params={},
)
mode = json.loads(json.dumps(request_data.data["generation_config"]["transcription_config"]["mode"]))
assert mode["type"] == "verbatim"
assert mode["timestamp_granularities"] == ["word"]
@pytest.mark.parametrize(
"optional_params",
[
{"mode": "smart", "timestamp_granularities": ["word"]},
{"mode": "smart", "response_format": "srt"},
],
)
def test_smart_mode_with_word_timestamps_is_rejected(
self, config: GeminiAudioTranscriptionConfig, optional_params: Mapping[str, object]
) -> None:
with pytest.raises(GeminiError, match="smart transcription mode cannot be combined") as excinfo:
config.transform_audio_transcription_request(
model="gemini-3.5-transcribe",
audio_file=("sample.wav", AUDIO_BYTES, "audio/wav"),
optional_params=optional_params,
litellm_params={},
)
assert excinfo.value.status_code == 400
def test_unknown_mode_is_rejected(self, config: GeminiAudioTranscriptionConfig) -> None:
with pytest.raises(GeminiError, match="Unsupported transcription mode") as excinfo:
config.transform_audio_transcription_request(
model="gemini-3.5-transcribe",
audio_file=("sample.wav", AUDIO_BYTES, "audio/wav"),
optional_params={"mode": "loud"},
litellm_params={},
)
assert excinfo.value.status_code == 400
@pytest.mark.parametrize(
"custom_vocabulary",
[["LiteLLM", "Kubernetes"], "LiteLLM, Kubernetes"],
)
def test_custom_vocabulary_accepts_list_or_comma_separated_string(
self, config: GeminiAudioTranscriptionConfig, custom_vocabulary: list[str] | str
) -> None:
request_data = config.transform_audio_transcription_request(
model="gemini-3.5-transcribe",
audio_file=("sample.wav", AUDIO_BYTES, "audio/wav"),
optional_params={"custom_vocabulary": custom_vocabulary},
litellm_params={},
)
transcription_config = json.loads(json.dumps(request_data.data["generation_config"]["transcription_config"]))
assert transcription_config == {"custom_vocabulary": ["LiteLLM", "Kubernetes"]}
def test_empty_vocabulary_sends_no_transcription_config(self, config: GeminiAudioTranscriptionConfig) -> None:
request_data = config.transform_audio_transcription_request(
model="gemini-3.5-transcribe",
audio_file=("sample.wav", AUDIO_BYTES, "audio/wav"),
optional_params={"custom_vocabulary": " , "},
litellm_params={},
)
assert "generation_config" not in request_data.data
def test_mode_and_vocabulary_reach_optional_params_as_provider_specific_params(self) -> None:
optional_params = litellm.utils.get_optional_params_transcription(
model="gemini-3.5-transcribe",
custom_llm_provider="gemini",
language="en",
mode="smart",
custom_vocabulary=["LiteLLM"],
)
assert optional_params["mode"] == "smart"
assert optional_params["custom_vocabulary"] == ["LiteLLM"]
assert optional_params["language"] == "en"
class TestTransformResponse:
def test_completed_interaction_maps_to_transcription_response(self, config):
response = config.transform_audio_transcription_response(make_response(COMPLETED_RESPONSE))

62
tests/unit/llms/vertex_ai/audio_transcription/test_vertex_ai_gemini_transcribe_transformation.py
View file

@ -218,6 +218,68 @@ class TestTransformRequest:
assert round_tripped["contents"][0]["role"] == "user"
class TestSmartModeAndVocabulary:
@pytest.mark.parametrize("mode,expected", [("smart", "SMART"), ("SMART", "SMART"), ("verbatim", "VERBATIM")])
def test_mode_maps_to_audio_transcription_config_mode(
self, config: VertexGeminiAudioTranscriptionConfig, mode: str, expected: str
) -> None:
request_data = config.transform_audio_transcription_request(
model="gemini-3.5-transcribe-preview",
audio_file=AUDIO_BYTES,
optional_params={"mode": mode, "language": "en"},
litellm_params={},
)
audio_config = json.loads(json.dumps(request_data.data["generationConfig"]["audioTranscriptionConfig"]))
assert audio_config == {"languageCodes": ["en-US"], "mode": expected}
@pytest.mark.parametrize(
"custom_vocabulary",
[["LiteLLM", "Kubernetes"], "LiteLLM, Kubernetes"],
)
def test_custom_vocabulary_accepts_list_or_comma_separated_string(
self, config: VertexGeminiAudioTranscriptionConfig, custom_vocabulary: list[str] | str
) -> None:
request_data = config.transform_audio_transcription_request(
model="gemini-3.5-transcribe-preview",
audio_file=AUDIO_BYTES,
optional_params={"custom_vocabulary": custom_vocabulary},
litellm_params={},
)
audio_config = json.loads(json.dumps(request_data.data["generationConfig"]["audioTranscriptionConfig"]))
assert audio_config == {"customVocabulary": ["LiteLLM", "Kubernetes"]}
def test_no_mode_leaves_vertex_default(self, config: VertexGeminiAudioTranscriptionConfig) -> None:
request_data = config.transform_audio_transcription_request(
model="gemini-3.5-transcribe-preview",
audio_file=AUDIO_BYTES,
optional_params={"language": "en"},
litellm_params={},
)
assert "mode" not in request_data.data["generationConfig"]["audioTranscriptionConfig"]
def test_unknown_mode_is_rejected(self, config: VertexGeminiAudioTranscriptionConfig) -> None:
with pytest.raises(VertexAIError, match="Unsupported transcription mode") as excinfo:
config.transform_audio_transcription_request(
model="gemini-3.5-transcribe-preview",
audio_file=AUDIO_BYTES,
optional_params={"mode": "loud"},
litellm_params={},
)
assert excinfo.value.status_code == 400
def test_mode_and_vocabulary_survive_drop_params(self) -> None:
optional_params = get_optional_params_transcription(
model="gemini-3.5-transcribe-preview",
custom_llm_provider="vertex_ai",
language="en",
mode="smart",
custom_vocabulary="LiteLLM, Kubernetes",
drop_params=True,
)
assert optional_params["mode"] == "smart"
assert optional_params["custom_vocabulary"] == "LiteLLM, Kubernetes"
class TestTransformResponse:
def test_generate_content_response(self, config):
raw_response = httpx.Response(status_code=200, json=GENERATE_CONTENT_RESPONSE)