scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-04 02:31:27 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 70

feat(azure): add speech transcription config support

Browse source 
Co-authored-by: ishaan-berri <ishaan-berri@users.noreply.github.com>
This commit is contained in:
oss-agent-shin 2026-05-08 18:03:19 +00:00
parent 98cd057f38
commit af826f2368
No known key found for this signature in database
10 changed files with 470 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
litellm/__init__.py
View file

@ -1842,6 +1842,9 @@ if TYPE_CHECKING:
from .llms.azure.completion.transformation import (
AzureOpenAITextConfig as AzureOpenAITextConfig,
)
from .llms.azure.audio_transcription.transformation import (
AzureSpeechAudioTranscriptionConfig as AzureSpeechAudioTranscriptionConfig,
)
from .llms.hosted_vllm.chat.transformation import (
HostedVLLMChatConfig as HostedVLLMChatConfig,
)

5
litellm/_lazy_imports_registry.py
View file

@ -271,6 +271,7 @@ LLM_CONFIG_NAMES = (
"AzureOpenAIConfig",
"AzureOpenAIGPT5Config",
"AzureOpenAITextConfig",
"AzureSpeechAudioTranscriptionConfig",
"HostedVLLMChatConfig",
"HostedVLLMEmbeddingConfig",
# Alias for backwards compatibility
@ -1045,6 +1046,10 @@ _LLM_CONFIGS_IMPORT_MAP = {
".llms.azure.completion.transformation",
"AzureOpenAITextConfig",
),
"AzureSpeechAudioTranscriptionConfig": (
".llms.azure.audio_transcription.transformation",
"AzureSpeechAudioTranscriptionConfig",
),
"HostedVLLMChatConfig": (
".llms.hosted_vllm.chat.transformation",
"HostedVLLMChatConfig",

3
litellm/llms/azure/audio_transcription/__init__.py Normal file
View file

@ -0,0 +1,3 @@
from .transformation import AzureSpeechAudioTranscriptionConfig
__all__ = ["AzureSpeechAudioTranscriptionConfig"]

198
litellm/llms/azure/audio_transcription/transformation.py Normal file
View file

@ -0,0 +1,198 @@
"""
Azure AI Speech (Cognitive Services) speech-to-text transformation.
Maps OpenAI-compatible audio transcription calls to Azure Speech REST
recognition for short audio.
"""
from typing import Any, Dict, List, Optional, Union
from urllib.parse import urlencode, urlparse
import httpx
from litellm.litellm_core_utils.audio_utils.utils import 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.types.llms.openai import (
AllMessageValues,
OpenAIAudioTranscriptionOptionalParams,
)
from litellm.types.utils import FileTypes, TranscriptionResponse
class AzureSpeechAudioTranscriptionException(BaseLLMException):
pass
class AzureSpeechAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
"""
Configuration for Azure AI Speech (Cognitive Services) STT.
Reference:
https://learn.microsoft.com/en-us/azure/ai-services/speech-service/rest-speech-to-text-short
"""
COGNITIVE_SERVICES_DOMAIN = "api.cognitive.microsoft.com"
STT_SPEECH_DOMAIN = "stt.speech.microsoft.com"
STT_ENDPOINT_PATH = "/speech/recognition/conversation/cognitiveservices/v1"
DEFAULT_LANGUAGE = "en-US"
def get_supported_openai_params(
self, model: str
) -> List[OpenAIAudioTranscriptionOptionalParams]:
return ["language", "response_format"]
def map_openai_params(
self,
non_default_params: dict,
optional_params: dict,
model: str,
drop_params: bool,
) -> dict:
supported_params = self.get_supported_openai_params(model=model)
for key, value in non_default_params.items():
if key in supported_params:
optional_params[key] = value
return optional_params
def validate_environment(
self,
headers: dict,
model: str,
messages: List[AllMessageValues],
optional_params: dict,
litellm_params: dict,
api_key: Optional[str] = None,
api_base: Optional[str] = None,
) -> dict:
if not api_key:
raise AzureSpeechAudioTranscriptionException(
message="api_key is required for Azure AI Speech transcription.",
status_code=401,
)
validated_headers = headers.copy()
validated_headers["Ocp-Apim-Subscription-Key"] = api_key
validated_headers["Content-Type"] = validated_headers.get(
"Content-Type", "audio/wav"
)
validated_headers["Accept"] = "application/json"
return validated_headers
def get_complete_url(
self,
api_base: Optional[str],
api_key: Optional[str],
model: str,
optional_params: dict,
litellm_params: dict,
stream: Optional[bool] = None,
) -> str:
if api_base is None:
raise AzureSpeechAudioTranscriptionException(
message=(
"api_base is required for Azure AI Speech transcription. "
"Use a Cognitive Services endpoint like "
"https://{region}.api.cognitive.microsoft.com or an STT "
"endpoint like https://{region}.stt.speech.microsoft.com."
),
status_code=400,
)
base_url = self._resolve_stt_base_url(api_base=api_base)
query_params = {
"language": optional_params.get("language", self.DEFAULT_LANGUAGE),
"format": self._get_azure_response_format(
optional_params.get("response_format")
),
}
return f"{base_url}{self.STT_ENDPOINT_PATH}?{urlencode(query_params)}"
def transform_audio_transcription_request(
self,
model: str,
audio_file: FileTypes,
optional_params: dict,
litellm_params: dict,
) -> AudioTranscriptionRequestData:
processed_audio = process_audio_file(audio_file)
return AudioTranscriptionRequestData(
data=processed_audio.file_content,
files=None,
content_type=processed_audio.content_type,
)
def transform_audio_transcription_response(
self,
raw_response: httpx.Response,
) -> TranscriptionResponse:
response_json = raw_response.json()
text = self._extract_text(response_json)
response = TranscriptionResponse(text=text)
response._hidden_params = response_json
return response
def get_error_class(
self, error_message: str, status_code: int, headers: Union[dict, httpx.Headers]
) -> BaseLLMException:
return AzureSpeechAudioTranscriptionException(
message=error_message,
status_code=status_code,
headers=headers,
)
def _resolve_stt_base_url(self, api_base: str) -> str:
api_base = api_base.rstrip("/")
parsed_url = urlparse(api_base)
hostname = parsed_url.hostname or ""
if self._is_cognitive_services_endpoint(hostname=hostname):
region = self._extract_region_from_hostname(
hostname=hostname, domain=self.COGNITIVE_SERVICES_DOMAIN
)
return self._build_stt_base_url(region=region)
if self._is_stt_endpoint(hostname=hostname):
return f"{parsed_url.scheme}://{hostname}"
return api_base
def _is_cognitive_services_endpoint(self, hostname: str) -> bool:
return hostname == self.COGNITIVE_SERVICES_DOMAIN or hostname.endswith(
f".{self.COGNITIVE_SERVICES_DOMAIN}"
)
def _is_stt_endpoint(self, hostname: str) -> bool:
return hostname == self.STT_SPEECH_DOMAIN or hostname.endswith(
f".{self.STT_SPEECH_DOMAIN}"
)
def _extract_region_from_hostname(self, hostname: str, domain: str) -> str:
if hostname.endswith(f".{domain}"):
return hostname[: -len(f".{domain}")]
return ""
def _build_stt_base_url(self, region: str) -> str:
if region:
return f"https://{region}.{self.STT_SPEECH_DOMAIN}"
return f"https://{self.STT_SPEECH_DOMAIN}"
def _get_azure_response_format(self, response_format: Optional[str]) -> str:
if response_format == "verbose_json":
return "detailed"
return "simple"
def _extract_text(self, response_json: Dict[str, Any]) -> str:
if isinstance(response_json.get("DisplayText"), str):
return response_json["DisplayText"]
nbest = response_json.get("NBest")
if isinstance(nbest, list) and nbest:
best = nbest[0]
if isinstance(best, dict):
return best.get("Display") or best.get("Lexical") or ""
return ""

2
litellm/llms/custom_httpx/llm_http_handler.py
View file

@ -1177,6 +1177,8 @@ class BaseLLMHTTPHandler:
data = transformed_result.data
files = transformed_result.files
if transformed_result.content_type is not None:
headers["Content-Type"] = transformed_result.content_type
## LOGGING
logging_obj.pre_call(

92
litellm/main.py
View file

@ -94,7 +94,11 @@ from litellm.litellm_core_utils.mock_functions import (
from litellm.litellm_core_utils.prompt_templates.common_utils import (
get_content_from_model_response,
)
from litellm.llms.base_llm import BaseConfig, BaseImageGenerationConfig
from litellm.llms.base_llm import (
BaseAudioTranscriptionConfig,
BaseConfig,
BaseImageGenerationConfig,
)
from litellm.llms.base_llm.base_model_iterator import (
convert_model_response_to_streaming,
)
@ -6438,6 +6442,65 @@ async def atranscription(*args, **kwargs) -> TranscriptionResponse:
)
def _azure_speech_audio_transcriptions(
model: str,
file: FileTypes,
optional_params: dict,
litellm_params_dict: dict,
model_response: TranscriptionResponse,
atranscription: bool,
client: Optional[
Union[
openai.AsyncOpenAI,
openai.OpenAI,
openai.AzureOpenAI,
openai.AsyncAzureOpenAI,
]
],
timeout: float,
max_retries: int,
litellm_logging_obj: LiteLLMLoggingObj,
api_base: Optional[str],
api_key: Optional[str],
custom_llm_provider: str,
provider_config: BaseAudioTranscriptionConfig,
shared_session: Optional["ClientSession"],
) -> Union[TranscriptionResponse, Coroutine[Any, Any, TranscriptionResponse]]:
api_base = api_base or litellm.api_base or get_secret_str("AZURE_API_BASE")
api_key = (
api_key
or litellm.api_key
or litellm.azure_key
or get_secret_str("AZURE_API_KEY")
)
return base_llm_http_handler.audio_transcriptions(
model=model,
audio_file=file,
optional_params=optional_params,
litellm_params=litellm_params_dict,
model_response=model_response,
atranscription=atranscription,
client=(
client
if client is not None
and (
isinstance(client, HTTPHandler) or isinstance(client, AsyncHTTPHandler)
)
else None
),
timeout=timeout,
max_retries=max_retries,
logging_obj=litellm_logging_obj,
api_base=api_base,
api_key=api_key,
custom_llm_provider=custom_llm_provider,
headers={},
provider_config=provider_config,
shared_session=shared_session,
)
@client
def transcription(
model: str,
@ -6500,8 +6563,7 @@ def transcription(
api_key=api_key,
) # type: ignore
if dynamic_api_key is not None:
api_key = dynamic_api_key
api_key = dynamic_api_key if dynamic_api_key is not None else api_key
optional_params = get_optional_params_transcription(
model=model,
@ -6541,7 +6603,29 @@ def transcription(
provider=LlmProviders(custom_llm_provider),
)
if custom_llm_provider == "azure":
if (
custom_llm_provider == "azure"
and provider_config is not None
and model.startswith("speech/")
):
response = _azure_speech_audio_transcriptions(
model=model,
file=file,
optional_params=optional_params,
litellm_params_dict=litellm_params_dict,
model_response=model_response,
atranscription=atranscription,
client=client,
timeout=timeout,
max_retries=max_retries,
litellm_logging_obj=litellm_logging_obj,
api_base=api_base,
api_key=api_key,
custom_llm_provider=custom_llm_provider,
provider_config=provider_config,
shared_session=shared_session,
)
elif custom_llm_provider == "azure":
# azure configs
api_base = api_base or litellm.api_base or get_secret_str("AZURE_API_BASE")

10
litellm/model_prices_and_context_window_backup.json
View file

@ -5651,6 +5651,16 @@
"mode": "audio_speech",
"source": "https://azure.microsoft.com/en-us/pricing/calculator/"
},
"azure/speech/azure-stt": {
"input_cost_per_second": 0.0,
"litellm_provider": "azure",
"mode": "audio_transcription",
"output_cost_per_second": 0.0,
"source": "https://azure.microsoft.com/en-us/pricing/details/cognitive-services/speech-services/",
"supported_endpoints": [
"/v1/audio/transcriptions"
]
},
"azure/tts-1": {
"input_cost_per_character": 1.5e-05,
"litellm_provider": "azure",

6
litellm/utils.py
View file

@ -8500,6 +8500,12 @@ class ProviderConfigManager:
model: str,
provider: LlmProviders,
) -> Optional[BaseAudioTranscriptionConfig]:
if litellm.LlmProviders.AZURE == provider and model.startswith("speech/"):
from litellm.llms.azure.audio_transcription.transformation import (
AzureSpeechAudioTranscriptionConfig,
)
return AzureSpeechAudioTranscriptionConfig()
if litellm.LlmProviders.FIREWORKS_AI == provider:
return litellm.FireworksAIAudioTranscriptionConfig()
elif litellm.LlmProviders.DEEPGRAM == provider:

10
model_prices_and_context_window.json
View file

@ -5651,6 +5651,16 @@
"mode": "audio_speech",
"source": "https://azure.microsoft.com/en-us/pricing/calculator/"
},
"azure/speech/azure-stt": {
"input_cost_per_second": 0.0,
"litellm_provider": "azure",
"mode": "audio_transcription",
"output_cost_per_second": 0.0,
"source": "https://azure.microsoft.com/en-us/pricing/details/cognitive-services/speech-services/",
"supported_endpoints": [
"/v1/audio/transcriptions"
]
},
"azure/tts-1": {
"input_cost_per_character": 1.5e-05,
"litellm_provider": "azure",

145
tests/test_litellm/llms/azure/test_azure_speech_audio_transcription.py Normal file
View file

@ -0,0 +1,145 @@
import io
from unittest.mock import MagicMock
import httpx
import pytest
import litellm
from litellm.llms.base_llm.audio_transcription.transformation import (
AudioTranscriptionRequestData,
BaseAudioTranscriptionConfig,
)
from litellm.llms.azure.audio_transcription.transformation import (
AzureSpeechAudioTranscriptionConfig,
)
from litellm.types.utils import TranscriptionResponse
from litellm.utils import ProviderConfigManager
def test_azure_speech_audio_transcription_config_installed():
config = ProviderConfigManager.get_provider_audio_transcription_config(
model="speech/azure-stt",
provider=litellm.LlmProviders.AZURE,
)
assert isinstance(config, BaseAudioTranscriptionConfig)
assert isinstance(config, AzureSpeechAudioTranscriptionConfig)
def test_azure_speech_audio_transcription_builds_stt_url_from_cognitive_endpoint():
config = AzureSpeechAudioTranscriptionConfig()
url = config.get_complete_url(
api_base="https://eastus.api.cognitive.microsoft.com/",
api_key="test-key",
model="speech/azure-stt",
optional_params={"language": "fr-FR", "response_format": "verbose_json"},
litellm_params={},
)
assert (
url
== "https://eastus.stt.speech.microsoft.com/speech/recognition/conversation/cognitiveservices/v1?language=fr-FR&format=detailed"
)
def test_azure_speech_audio_transcription_accepts_stt_endpoint_base():
config = AzureSpeechAudioTranscriptionConfig()
url = config.get_complete_url(
api_base="https://westus.stt.speech.microsoft.com",
api_key="test-key",
model="speech/azure-stt",
optional_params={},
litellm_params={},
)
assert (
url
== "https://westus.stt.speech.microsoft.com/speech/recognition/conversation/cognitiveservices/v1?language=en-US&format=simple"
)
def test_azure_speech_audio_transcription_validate_environment():
config = AzureSpeechAudioTranscriptionConfig()
headers = config.validate_environment(
headers={},
model="speech/azure-stt",
messages=[],
optional_params={},
litellm_params={},
api_key="test-key",
)
assert headers["Ocp-Apim-Subscription-Key"] == "test-key"
assert headers["Content-Type"] == "audio/wav"
assert headers["Accept"] == "application/json"
def test_azure_speech_audio_transcription_request_transform():
config = AzureSpeechAudioTranscriptionConfig()
audio = io.BytesIO(b"RIFF....WAVE")
request_data = config.transform_audio_transcription_request(
model="speech/azure-stt",
audio_file=audio,
optional_params={},
litellm_params={},
)
assert isinstance(request_data, AudioTranscriptionRequestData)
assert request_data.data == b"RIFF....WAVE"
assert request_data.files is None
assert request_data.content_type == "audio/wav"
@pytest.mark.parametrize(
"payload,expected_text",
[
({"DisplayText": "hello world"}, "hello world"),
(
{
"RecognitionStatus": "Success",
"NBest": [{"Display": "best text", "Confidence": 0.91}],
},
"best text",
),
],
)
def test_azure_speech_audio_transcription_response_transform(payload, expected_text):
config = AzureSpeechAudioTranscriptionConfig()
response = httpx.Response(200, json=payload)
result = config.transform_audio_transcription_response(response)
assert isinstance(result, TranscriptionResponse)
assert result.text == expected_text
assert result._hidden_params == payload
def test_azure_speech_transcription_routes_through_provider_config(monkeypatch):
expected = TranscriptionResponse(text="hello")
audio_handler = MagicMock(return_value=expected)
monkeypatch.setattr(
litellm.main.base_llm_http_handler,
"audio_transcriptions",
audio_handler,
)
response = litellm.transcription(
model="azure/speech/azure-stt",
file=io.BytesIO(b"RIFF....WAVE"),
api_base="https://eastus.api.cognitive.microsoft.com",
api_key="test-key",
language="en-US",
)
assert response is expected
audio_handler.assert_called_once()
assert isinstance(
audio_handler.call_args.kwargs["provider_config"],
AzureSpeechAudioTranscriptionConfig,
)
assert audio_handler.call_args.kwargs["custom_llm_provider"] == "azure"