scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-05 07:37:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 360

Merge 25f8d676f4 into 0a1dbc0fb4

Browse source
This commit is contained in:
Ben Langfeld 2026-10-27 15:07:10 -03:00 • committed by GitHub
commit 4be634d4ba
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 453 additions and 59 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

254
litellm/llms/elevenlabs/audio_transcription/transformation.py
View file

@ -2,22 +2,31 @@
Translates from OpenAI's `/v1/audio/transcriptions` to ElevenLabs's `/v1/speech-to-text`
"""
from collections.abc import Iterable, Mapping
import io
from collections.abc import Iterable, Iterator, Mapping, Sequence
from itertools import chain, groupby
from types import MappingProxyType
from typing import Final, Protocol
from httpx import Headers, Response
from pydantic import ConfigDict, TypeAdapter
from pydantic import ConfigDict, TypeAdapter, ValidationError
import litellm
from litellm.litellm_core_utils.audio_utils.utils import process_audio_file
from litellm.litellm_core_utils.url_utils import join_api_base
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.secret_managers.main import get_secret_str
from litellm.types.llms.elevenlabs import (
ElevenLabsSTTChunk,
ElevenLabsSTTMultichannelResponse,
ElevenLabsSTTWord,
OpenAIDiarizedSegment,
)
from litellm.types.llms.openai import (
AllMessageValues,
OpenAIAudioTranscriptionOptionalParams,
)
from litellm.types.utils import FileTypes, TranscriptionResponse
from litellm.types.utils import FileTypes, TranscriptionResponse, TranscriptionUsageDurationObject
from ...base_llm.audio_transcription.transformation import (
AudioTranscriptionRequestData,
@ -28,6 +37,10 @@ from ..common_utils import ElevenLabsException
_JSON_OBJECT: Final = TypeAdapter(Mapping[str, object], config=ConfigDict(hide_input_in_errors=True))
_JSON_OBJECTS: Final = TypeAdapter(Iterable[Mapping[str, object]], config=ConfigDict(hide_input_in_errors=True))
_HIDDEN_PARAMS: Final = TypeAdapter(dict[str, object])
_CHANNELS: Final = TypeAdapter(int)
_DIARIZATION_KEYS: Final = frozenset(("diarize", "use_multi_channel", "multichannel_output_style"))
# OpenAI parameter names that ElevenLabs spells differently
_ELEVENLABS_PARAM_NAMES: Final = MappingProxyType({"language": "language_code"})
class _JsonResponse(Protocol):
@ -44,7 +57,7 @@ class ElevenLabsAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
return litellm.LlmProviders.ELEVENLABS.value
def get_supported_openai_params(self, model: str) -> list[OpenAIAudioTranscriptionOptionalParams]:
return ["language", "temperature"]
return ["language", "temperature", "response_format"]
def map_openai_params(
self,
@ -54,14 +67,17 @@ class ElevenLabsAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
drop_params: bool,
) -> dict:
supported_params: Final = self.get_supported_openai_params(model)
for k, v in non_default_params.items():
if k in supported_params:
if k == "language":
# Map OpenAI language format to ElevenLabs language_code
optional_params["language_code"] = v
else:
optional_params[k] = v
return optional_params
params: Final = _JSON_OBJECT.validate_python(non_default_params)
mapped_params: Final = {
_ELEVENLABS_PARAM_NAMES.get(k, k): v
for k, v in params.items()
if k in supported_params and k != "response_format"
}
# ElevenLabs always returns JSON; the only response_format that changes the
# request is diarized_json, which maps to diarization. Other formats are
# shaped from the same JSON at response time.
diarize_params: Final = {"diarize": True} if params.get("response_format") == "diarized_json" else {}
return {**optional_params, **mapped_params, **diarize_params}
def get_error_class(self, error_message: str, status_code: int, headers: dict | Headers) -> BaseLLMException:
return ElevenLabsException(message=error_message, status_code=status_code, headers=headers)
@ -84,31 +100,25 @@ class ElevenLabsAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
# Use common utility to process the audio file
processed_audio: Final = process_audio_file(audio_file)
supported_params: Final = self.get_supported_openai_params(model)
# Prepare form data
form_data: Final = {"model_id": model}
#########################################################
# Add OpenAI Compatible Parameters
#########################################################
for key, value in optional_params.items():
if key in self.get_supported_openai_params(model) and value is not None:
# Convert values to strings for form data, but skip None values
form_data[key] = str(value)
#########################################################
# Add Provider Specific Parameters
#########################################################
provider_specific_params: Final = self.get_provider_specific_params(
model=model,
optional_params=optional_params,
openai_params=self.get_supported_openai_params(model),
provider_specific_params: Final = _resolve_diarization_params(
self.get_provider_specific_params(
model=model,
optional_params=optional_params,
openai_params=supported_params,
),
processed_audio.file_content,
)
for key, value in provider_specific_params.items():
form_data[key] = str(value)
#########################################################
#########################################################
openai_params: Final = {
key: str(value) for key, value in optional_params.items() if key in supported_params and value is not None
}
form_data: Final = {
"model_id": model,
**openai_params,
**{key: str(value) for key, value in provider_specific_params.items()},
}
# Prepare files
files: Final = {
@ -126,33 +136,16 @@ class ElevenLabsAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
raw_response: Response,
) -> TranscriptionResponse:
"""
Transforms the raw response from ElevenLabs to the TranscriptionResponse format
Transforms the raw ElevenLabs speech-to-text response.
When the response carries speaker information (diarize=true) or per-channel
transcripts (use_multi_channel=true), it is shaped into OpenAI's
diarized_json form (segments with speaker labels + usage). Otherwise it
falls back to a flat transcript with word-level timestamps.
"""
try:
response_json: Final = _json_body(raw_response)
response_object: Final = _JSON_OBJECT.validate_python(response_json)
# Extract the main transcript text
text: Final = response_object.get("text", "")
# Create TranscriptionResponse object
response: Final = TranscriptionResponse(text=text)
# Add additional metadata matching OpenAI format
response["task"] = "transcribe"
response["language"] = response_object.get("language_code", "unknown")
# Map ElevenLabs words to OpenAI format
if "words" in response_object:
response["words"] = [
{
"word": word_data.get("text", ""),
"start": word_data.get("start", 0),
"end": word_data.get("end", 0),
}
for word_data in _JSON_OBJECTS.validate_python(response_object["words"])
if word_data.get("type") == "word"
]
response: Final = _transcription_response(_JSON_OBJECT.validate_python(response_json))
# Store full response in hidden params
response.hidden_params = _HIDDEN_PARAMS.validate_python(response_json)
@ -195,3 +188,148 @@ class ElevenLabsAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
headers.update(auth_header)
return headers
def _truthy(value: object) -> bool:
return str(value).strip().lower() == "true"
def _channel_count(audio_content: bytes) -> int:
"""Best-effort channel count from the audio header; 1 if it can't be read."""
try:
import soundfile
except ImportError:
return 1
try:
channels: Final = soundfile.info(io.BytesIO(audio_content)).channels # pyright: ignore[reportUnknownMemberType] # soundfile ships no type hints
return _CHANNELS.validate_python(channels)
except (RuntimeError, TypeError, ValidationError):
return 1
def _resolve_diarization_params(provider_params: Mapping[str, object], audio_content: bytes) -> Mapping[str, object]:
"""
ElevenLabs rejects diarize and use_multi_channel together. Diarization
requested on multi-channel audio uses the channel as the speaker; on mono it
falls back to acoustic diarization. An explicit use_multi_channel from the
caller wins over auto-detection, and an unreadable header falls back to mono.
"""
diarize_requested: Final = _truthy(provider_params.get("diarize"))
multichannel: Final = (
_truthy(provider_params["use_multi_channel"])
if "use_multi_channel" in provider_params
else diarize_requested and _channel_count(audio_content) > 1
)
others: Final = {key: value for key, value in provider_params.items() if key not in _DIARIZATION_KEYS}
if multichannel:
return {
**others,
"use_multi_channel": True,
"multichannel_output_style": provider_params.get("multichannel_output_style", "combined"),
}
if diarize_requested:
return {**others, "diarize": True}
return others
def _transcription_response(response_object: Mapping[str, object]) -> TranscriptionResponse:
"""
The audio-transcription response transformer only sees the provider body, so
the output shape is decided from it: per-channel transcripts or any word with
a speaker id produce diarized segments, anything else stays a flat transcript.
"""
if "transcripts" in response_object:
return _diarized_response(_combine_channels(ElevenLabsSTTMultichannelResponse.model_validate(response_object)))
words: Final = tuple(_JSON_OBJECTS.validate_python(response_object.get("words", ())))
if any(word.get("speaker_id") is not None for word in words):
response: Final = _diarized_response(ElevenLabsSTTChunk.model_validate(response_object))
# Callers already passing diarize=True read the flat word list, so it stays alongside the segments
response["words"] = _openai_words(words)
return response
return _flat_response(response_object, words)
def _flat_response(
response_object: Mapping[str, object], words: tuple[Mapping[str, object], ...]
) -> TranscriptionResponse:
response: Final = TranscriptionResponse(text=response_object.get("text", ""))
# Add additional metadata matching OpenAI format
response["task"] = "transcribe"
response["language"] = response_object.get("language_code", "unknown")
# Map ElevenLabs words to OpenAI format
if "words" in response_object:
response["words"] = _openai_words(words)
return response
def _openai_words(words: tuple[Mapping[str, object], ...]) -> Sequence[Mapping[str, object]]:
return [
{
"word": word_data.get("text", ""),
"start": word_data.get("start", 0),
"end": word_data.get("end", 0),
}
for word_data in words
if word_data.get("type") == "word"
]
def _diarized_response(transcript: ElevenLabsSTTChunk) -> TranscriptionResponse:
response: Final = TranscriptionResponse(text=transcript.text)
response["task"] = "transcribe"
response["language"] = transcript.language_code or "unknown"
response["segments"] = _build_segments(transcript.words)
if transcript.audio_duration_secs is not None:
response["duration"] = transcript.audio_duration_secs
response["usage"] = TranscriptionUsageDurationObject(type="duration", seconds=transcript.audio_duration_secs)
return response
def _combine_channels(parsed: ElevenLabsSTTMultichannelResponse) -> ElevenLabsSTTChunk:
"""
Flatten per-channel transcripts into a single time-ordered word stream, using
the channel as the speaker so the two sides of e.g. a stereo call don't collide
on the per-channel speaker ids ElevenLabs assigns independently.
"""
words: Final = chain.from_iterable(
_channel_words(position, transcript) for position, transcript in enumerate(parsed.transcripts)
)
return ElevenLabsSTTChunk(
text=" ".join(transcript.text for transcript in parsed.transcripts if transcript.text),
language_code=next(
(transcript.language_code for transcript in parsed.transcripts if transcript.language_code), None
),
words=tuple(sorted(words, key=lambda word: word.start if word.start is not None else 0.0)),
audio_duration_secs=parsed.audio_duration_secs,
)
def _channel_words(position: int, transcript: ElevenLabsSTTChunk) -> Iterator[ElevenLabsSTTWord]:
"""
The list position is the fallback channel when channel_index is absent, so
distinct channels never collapse onto the same speaker.
"""
channel: Final = transcript.channel_index if transcript.channel_index is not None else position
return (word.model_copy(update={"speaker_id": f"speaker_{channel}"}) for word in transcript.words)
def _build_segments(words: tuple[ElevenLabsSTTWord, ...]) -> tuple[OpenAIDiarizedSegment, ...]:
spoken: Final = (word for word in words if word.type == "word" and word.start is not None)
groups: Final = (tuple(group) for _, group in groupby(spoken, key=lambda word: word.speaker_id))
return tuple(
OpenAIDiarizedSegment(
id=f"segment_{index}",
start=group[0].start if group[0].start is not None else 0.0,
end=group[-1].end if group[-1].end is not None else 0.0,
speaker=group[0].speaker_id or "speaker_0",
text=" ".join(word.text for word in group),
type="transcript.text.segment",
)
for index, group in enumerate(groups)
)

42
litellm/types/llms/elevenlabs.py Normal file
View file

@ -0,0 +1,42 @@
from typing import Literal
from pydantic import BaseModel, ConfigDict
from typing_extensions import ReadOnly, TypedDict
class ElevenLabsSTTWord(BaseModel):
model_config = ConfigDict(frozen=True)
text: str
type: Literal["word", "spacing", "audio_event"]
start: float | None = None
end: float | None = None
speaker_id: str | None = None
channel_index: int | None = None
class ElevenLabsSTTChunk(BaseModel):
model_config = ConfigDict(frozen=True)
text: str = ""
language_code: str | None = None
language_probability: float | None = None
words: tuple[ElevenLabsSTTWord, ...] = ()
channel_index: int | None = None
audio_duration_secs: float | None = None
class ElevenLabsSTTMultichannelResponse(BaseModel):
model_config = ConfigDict(frozen=True)
transcripts: tuple[ElevenLabsSTTChunk, ...]
audio_duration_secs: float | None = None
class OpenAIDiarizedSegment(TypedDict):
id: ReadOnly[str]
start: ReadOnly[float]
end: ReadOnly[float]
speaker: ReadOnly[str]
text: ReadOnly[str]
type: ReadOnly[Literal["transcript.text.segment"]]

216
tests/unit/llms/elevenlabs/audio_transcription/test_transformation.py
View file

@ -1,15 +1,19 @@
import io
import sys
from collections.abc import Iterator
from typing import Final
import httpx
import numpy as np
import pytest
import respx
import soundfile
import litellm
from litellm.caching.llm_caching_handler import LLMClientCache
from litellm.llms.elevenlabs.audio_transcription.transformation import ElevenLabsAudioTranscriptionConfig
from litellm.llms.elevenlabs.text_to_speech.transformation import ElevenLabsTextToSpeechConfig
from litellm.types.utils import TranscriptionResponse
from litellm.types.utils import TranscriptionResponse, TranscriptionUsageDurationObject
from typing import Any, Dict
from unittest.mock import patch, MagicMock
@ -129,6 +133,216 @@ def test_transform_audio_transcription_response_wraps_malformed_payloads_with_th
assert str(exc_info.value).endswith(f"\nResponse: {raw_response.text}")
def _wav_bytes(channels: int) -> bytes:
buffer: Final = io.BytesIO()
samples: Final = np.zeros((1600, channels) if channels > 1 else 1600, dtype="float32")
soundfile.write(buffer, samples, 16000, format="WAV")
return buffer.getvalue()
def _map_params(non_default_params: dict[str, object]) -> dict:
return ElevenLabsAudioTranscriptionConfig().map_openai_params(
non_default_params=non_default_params,
optional_params={},
model="scribe_v2",
drop_params=False,
)
def _form_data(audio: bytes, optional_params: dict[str, object]) -> dict | bytes:
return (
ElevenLabsAudioTranscriptionConfig()
.transform_audio_transcription_request(
model="scribe_v2",
audio_file=audio,
optional_params=optional_params,
litellm_params={},
)
.data
)
def test_response_format_is_a_supported_openai_param():
assert "response_format" in ElevenLabsAudioTranscriptionConfig().get_supported_openai_params("scribe_v2")
def test_diarized_json_response_format_maps_to_diarize():
assert _map_params({"response_format": "diarized_json"}) == {"diarize": True}
def test_other_response_formats_do_not_enable_diarize():
assert _map_params({"response_format": "json"}) == {}
def test_language_maps_to_language_code():
assert _map_params({"language": "pt"}) == {"language_code": "pt"}
@pytest.mark.parametrize(
("channels", "optional_params", "expected"),
[
pytest.param(1, {"diarize": True}, {"diarize": "True"}, id="mono-uses-acoustic-diarize"),
pytest.param(
2,
{"diarize": True},
{"use_multi_channel": "True", "multichannel_output_style": "combined"},
id="stereo-switches-to-multichannel",
),
pytest.param(
1,
{"diarize": True, "use_multi_channel": True},
{"use_multi_channel": "True", "multichannel_output_style": "combined"},
id="explicit-multichannel-wins-on-mono",
),
pytest.param(
2,
{"diarize": True, "use_multi_channel": False},
{"diarize": "True"},
id="explicit-no-multichannel-keeps-diarize-on-stereo",
),
pytest.param(2, {}, {}, id="no-diarization-requested"),
],
)
def test_diarization_mechanism_follows_audio_channel_count(
channels: int, optional_params: dict[str, object], expected: dict[str, str]
):
assert _form_data(_wav_bytes(channels), optional_params) == {"model_id": "scribe_v2", **expected}
def test_unreadable_audio_falls_back_to_acoustic_diarize():
assert _form_data(b"not audio at all", {"diarize": True}) == {"model_id": "scribe_v2", "diarize": "True"}
def test_stereo_audio_without_soundfile_installed_falls_back_to_acoustic_diarize(monkeypatch: pytest.MonkeyPatch):
stereo: Final = _wav_bytes(2)
monkeypatch.setitem(sys.modules, "soundfile", None)
assert _form_data(stereo, {"diarize": True}) == {"model_id": "scribe_v2", "diarize": "True"}
def test_diarized_response_builds_speaker_segments():
response: Final = _transform(
{
"language_code": "por",
"text": "Bom dia. Oi?",
"audio_duration_secs": 70.5,
"words": [
{"text": "ev", "start": 1.0, "end": 2.0, "type": "audio_event", "speaker_id": "speaker_0"},
{"text": "Bom", "start": 3.0, "end": 3.4, "type": "word", "speaker_id": "speaker_1"},
{"text": "dia", "start": 3.4, "end": 3.8, "type": "word", "speaker_id": "speaker_1"},
{"text": "Oi", "start": 5.0, "end": 5.3, "type": "word", "speaker_id": "speaker_0"},
],
}
)
assert response.text == "Bom dia. Oi?"
assert response["language"] == "por"
assert response["segments"] == (
{
"id": "segment_0",
"start": 3.0,
"end": 3.8,
"speaker": "speaker_1",
"text": "Bom dia",
"type": "transcript.text.segment",
},
{
"id": "segment_1",
"start": 5.0,
"end": 5.3,
"speaker": "speaker_0",
"text": "Oi",
"type": "transcript.text.segment",
},
)
assert response["words"] == [
{"word": "Bom", "start": 3.0, "end": 3.4},
{"word": "dia", "start": 3.4, "end": 3.8},
{"word": "Oi", "start": 5.0, "end": 5.3},
]
def test_diarized_response_reports_fractional_duration_as_typed_usage():
response: Final = _transform(
{
"text": "oi",
"audio_duration_secs": 70.5,
"words": [{"text": "oi", "start": 1.0, "end": 1.5, "type": "word", "speaker_id": "speaker_0"}],
}
)
assert response["usage"] == TranscriptionUsageDurationObject(type="duration", seconds=70.5)
assert response["duration"] == 70.5
assert response.model_dump()["usage"] == {"type": "duration", "seconds": 70.5}
def test_multichannel_response_uses_the_channel_as_speaker():
response: Final = _transform(
{
"audio_duration_secs": 40.0,
"transcripts": [
{
"channel_index": 1,
"language_code": "por",
"text": "lado B",
"words": [
{"text": "lado", "start": 10.0, "end": 10.5, "type": "word", "speaker_id": "speaker_0"},
{"text": "B", "start": 10.5, "end": 10.8, "type": "word", "speaker_id": "speaker_0"},
],
},
{
"channel_index": 0,
"language_code": "por",
"text": "lado A",
"words": [
{"text": "lado", "start": 1.0, "end": 1.5, "type": "word", "speaker_id": "speaker_0"},
{"text": "A", "start": 1.5, "end": 1.8, "type": "word", "speaker_id": "speaker_0"},
],
},
],
}
)
assert [(s["speaker"], s["text"], s["start"]) for s in response["segments"]] == [
("speaker_0", "lado A", 1.0),
("speaker_1", "lado B", 10.0),
]
assert response.text == "lado B lado A"
assert response["language"] == "por"
assert response["duration"] == 40.0
def test_multichannel_response_without_channel_index_keeps_speakers_distinct():
response: Final = _transform(
{
"transcripts": [
{"text": "a", "words": [{"text": "a", "start": 1.0, "end": 1.2, "type": "word"}]},
{"text": "b", "words": [{"text": "b", "start": 2.0, "end": 2.2, "type": "word"}]},
],
}
)
assert [s["speaker"] for s in response["segments"]] == ["speaker_0", "speaker_1"]
assert response["usage"] is None
@pytest.mark.parametrize(
"payload",
[
{"text": "t", "words": [{"type": "word", "text": None, "speaker_id": "speaker_0"}]},
{"transcripts": "not a list"},
{"transcripts": [{"words": [{"type": "unknown", "text": "x"}]}]},
],
)
def test_malformed_diarized_payloads_raise_with_the_raw_body(payload: object):
raw_response: Final = httpx.Response(200, json=payload)
with pytest.raises(ValueError, match="Error transforming ElevenLabs response: ") as exc_info:
ElevenLabsAudioTranscriptionConfig().transform_audio_transcription_response(raw_response=raw_response)
assert str(exc_info.value).endswith(f"\nResponse: {raw_response.text}")
class TestElevenLabsAudioTranscription:
@pytest.mark.usefixtures("elevenlabs_api_key_env")
def test_elevenlabs_diarize_parameter_passthrough(self):