mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-17 04:34:25 +00:00
feat(soniox): decode proxy form fields for context, translation and flags (#40605)
* feat(soniox): decode proxy form fields for context, translation and flags Multipart form fields reach the proxy as strings, so Soniox received context and translation as JSON strings and booleans as "true". Decode them at the handler entry point and add typed shapes for the context and translation request fields. * fix(soniox): keep bracket-prefixed free-form context and move types under llms Free-form context such as "[inaudible] consultation" is valid Soniox input, so a string that fails JSON parsing is now forwarded unchanged instead of returning a 400. The Soniox request types live next to the provider code. * test(soniox): exercise SDK-typed context and translation shapes through the decoder * refactor(soniox): derive handler options and provider params without mutating a local dict * chore(soniox): drop explanatory comment on the non-forwarded param set * refactor(soniox): return invalid form bool as a value and raise once in the handler * fix(soniox): validate decoded form fields and keep language with hints Validate structured context before sending it as an object, answer 400 for a bad polling interval or flag unless drop_params is on, nest SDK bracketed extra_body fields, keep the OpenAI language at the head of language_hints, and return the translated text for one-way translation without tags. * fix(soniox): keep the source transcript, rebuild SDK list fields, and reject malformed JSON values * fix(soniox): keep cleanup strings whole and drop the unused translation type * fix(soniox): let explicit language_hints win over language and test the proxy form path --------- Co-authored-by: mateo-berri <277851410+mateo-berri@users.noreply.github.com>
This commit is contained in:
Dan Lemon
GitHub
parent
cda19d6a29
commit
c8b73cde84
5 changed files with 672 additions and 59 deletions
|
|
@ -19,11 +19,14 @@ import asyncio
|
|||
import math
|
||||
import time
|
||||
from collections.abc import Coroutine, Mapping, Sequence
|
||||
from types import MappingProxyType
|
||||
from typing import TYPE_CHECKING, Any, Final
|
||||
|
||||
import httpx
|
||||
from pydantic import TypeAdapter
|
||||
from typing_extensions import ReadOnly, TypedDict
|
||||
|
||||
import litellm
|
||||
from litellm.litellm_core_utils.audio_utils.utils import (
|
||||
get_audio_file_name,
|
||||
process_audio_file,
|
||||
|
|
@ -35,7 +38,12 @@ from litellm.llms.custom_httpx.http_handler import (
|
|||
get_httpx_client,
|
||||
)
|
||||
from litellm.llms.soniox.audio_transcription.transformation import (
|
||||
SONIOX_HANDLER_ONLY_PARAMS,
|
||||
SonioxAudioTranscriptionConfig,
|
||||
SonioxInvalidFormParam,
|
||||
decode_soniox_form_params,
|
||||
raise_soniox_form_error,
|
||||
soniox_language_hints,
|
||||
)
|
||||
from litellm.llms.soniox.common_utils import (
|
||||
SONIOX_DEFAULT_CLEANUP,
|
||||
|
|
@ -58,6 +66,39 @@ else:
|
|||
LiteLLMLoggingObj = Any
|
||||
|
||||
|
||||
_CLEANUP_ENTRIES: Final = TypeAdapter(tuple[object, ...])
|
||||
_NOT_FORWARDED_TO_SONIOX: Final[frozenset[str]] = frozenset(
|
||||
(*SONIOX_HANDLER_ONLY_PARAMS, "audio_url", "file_id", "response_format", "language")
|
||||
)
|
||||
|
||||
|
||||
def _optional_str(value: object) -> str | None:
|
||||
return None if value is None else str(value)
|
||||
|
||||
|
||||
def _float_or_default(value: object, default: float) -> float:
|
||||
try:
|
||||
parsed: Final = float(str(value))
|
||||
except ValueError:
|
||||
return default
|
||||
return parsed if math.isfinite(parsed) else default
|
||||
|
||||
|
||||
def _int_or_default(value: object, default: int) -> int:
|
||||
try:
|
||||
return int(value) if isinstance(value, (int, float)) else int(str(value))
|
||||
except (ValueError, OverflowError):
|
||||
return default
|
||||
|
||||
|
||||
def _cleanup_targets(raw: object) -> tuple[str, ...]:
|
||||
if raw is None:
|
||||
return ()
|
||||
if isinstance(raw, str):
|
||||
return (raw,)
|
||||
return tuple(target for target in _CLEANUP_ENTRIES.validate_python(raw) if isinstance(target, str))
|
||||
|
||||
|
||||
class _TranscriptionMeta(TypedDict, total=False):
|
||||
"""Fields the handler reads from a Soniox transcription object."""
|
||||
|
||||
|
|
@ -182,7 +223,7 @@ class SonioxAudioTranscriptionHandler:
|
|||
) -> tuple[
|
||||
dict[str, str], # auth headers
|
||||
str, # api_base (no trailing slash)
|
||||
dict[str, object], # body for POST /v1/transcriptions (without file_id/audio_url)
|
||||
Mapping[str, object], # body for POST /v1/transcriptions (without file_id/audio_url)
|
||||
_HandlerOptions, # handler-only options (poll interval, cleanup, ...)
|
||||
]:
|
||||
# Validate env -> auth headers.
|
||||
|
|
@ -198,52 +239,42 @@ class SonioxAudioTranscriptionHandler:
|
|||
|
||||
base_url: Final = get_soniox_api_base(api_base)
|
||||
|
||||
# Operate on a local copy so we don't mutate the caller's dict
|
||||
# (the caller may reuse `optional_params` for retries or logging).
|
||||
params: Final = dict(optional_params)
|
||||
|
||||
# Pull handler-only kwargs out of params so they aren't sent
|
||||
# to Soniox.
|
||||
poll_interval = float(params.pop("soniox_polling_interval", SONIOX_DEFAULT_POLL_INTERVAL))
|
||||
try:
|
||||
max_attempts = int(params.pop("soniox_max_polling_attempts", SONIOX_DEFAULT_MAX_POLL_ATTEMPTS))
|
||||
except (ValueError, OverflowError):
|
||||
max_attempts = SONIOX_DEFAULT_MAX_POLL_ATTEMPTS
|
||||
cleanup_raw: Final = params.pop("soniox_cleanup", SONIOX_DEFAULT_CLEANUP)
|
||||
if cleanup_raw is None:
|
||||
cleanup: list[str] = []
|
||||
elif isinstance(cleanup_raw, str):
|
||||
cleanup = [cleanup_raw]
|
||||
else:
|
||||
cleanup = list(cleanup_raw)
|
||||
filename_override: Final = params.pop("filename", None)
|
||||
decoded: Final = decode_soniox_form_params(
|
||||
optional_params, drop_params=litellm_params.get("drop_params") is True or litellm.drop_params is True
|
||||
)
|
||||
if isinstance(decoded, SonioxInvalidFormParam):
|
||||
raise_soniox_form_error(decoded)
|
||||
|
||||
# Server-side clamps. Caller-supplied poll settings (from request kwargs)
|
||||
# are bounded so an authenticated caller cannot force a worker into a
|
||||
# tight poll loop (zero interval) or pin it indefinitely (huge attempt
|
||||
# count). Total polling time is bounded by
|
||||
# SONIOX_MAX_POLL_ATTEMPTS * SONIOX_MAX_POLL_INTERVAL.
|
||||
if not math.isfinite(poll_interval):
|
||||
poll_interval = SONIOX_DEFAULT_POLL_INTERVAL
|
||||
clamped_poll_interval: Final = max(SONIOX_MIN_POLL_INTERVAL, min(poll_interval, SONIOX_MAX_POLL_INTERVAL))
|
||||
clamped_max_attempts: Final = max(1, min(max_attempts, SONIOX_MAX_POLL_ATTEMPTS))
|
||||
poll_interval: Final = _float_or_default(decoded.get("soniox_polling_interval"), SONIOX_DEFAULT_POLL_INTERVAL)
|
||||
max_attempts: Final = _int_or_default(
|
||||
decoded.get("soniox_max_polling_attempts"), SONIOX_DEFAULT_MAX_POLL_ATTEMPTS
|
||||
)
|
||||
cleanup: Final = _cleanup_targets(decoded.get("soniox_cleanup", SONIOX_DEFAULT_CLEANUP))
|
||||
language_hints: Final = soniox_language_hints(decoded.get("language"), decoded.get("language_hints"))
|
||||
|
||||
# response_format is handled by LiteLLM post-processing, not Soniox.
|
||||
handler_opts: Final[_HandlerOptions] = {
|
||||
"poll_interval": clamped_poll_interval,
|
||||
"max_attempts": clamped_max_attempts,
|
||||
"poll_interval": max(SONIOX_MIN_POLL_INTERVAL, min(poll_interval, SONIOX_MAX_POLL_INTERVAL)),
|
||||
"max_attempts": max(1, min(max_attempts, SONIOX_MAX_POLL_ATTEMPTS)),
|
||||
"cleanup": cleanup,
|
||||
"filename_override": filename_override,
|
||||
"audio_url": params.pop("audio_url", None),
|
||||
"file_id": params.pop("file_id", None),
|
||||
"response_format": params.pop("response_format", None),
|
||||
"filename_override": _optional_str(decoded.get("filename")),
|
||||
"audio_url": _optional_str(decoded.get("audio_url")),
|
||||
"file_id": _optional_str(decoded.get("file_id")),
|
||||
"response_format": _optional_str(decoded.get("response_format")),
|
||||
}
|
||||
|
||||
# Soniox does not accept `language` directly; map_openai_params should
|
||||
# already have translated it, but drop any leftover to be safe.
|
||||
params.pop("language", None)
|
||||
|
||||
return auth_headers, base_url, params, handler_opts
|
||||
provider_params: Final = MappingProxyType(
|
||||
{
|
||||
**{key: value for key, value in decoded.items() if key not in _NOT_FORWARDED_TO_SONIOX},
|
||||
**({} if language_hints is None else {"language_hints": language_hints}),
|
||||
}
|
||||
)
|
||||
return auth_headers, base_url, provider_params, handler_opts
|
||||
|
||||
def _build_create_body(
|
||||
self,
|
||||
|
|
|
|||
|
|
@ -9,9 +9,14 @@ async API requires multiple HTTP calls and does not fit the single-request
|
|||
contract of `base_llm_http_handler.audio_transcriptions`.
|
||||
"""
|
||||
|
||||
from typing import Any, Final
|
||||
import re
|
||||
from collections.abc import Mapping
|
||||
from dataclasses import dataclass
|
||||
from types import MappingProxyType
|
||||
from typing import Any, Final, NoReturn
|
||||
|
||||
from httpx import Headers, Response
|
||||
from pydantic import TypeAdapter, ValidationError
|
||||
|
||||
from litellm.llms.base_llm.audio_transcription.transformation import (
|
||||
AudioTranscriptionRequestData,
|
||||
|
|
@ -26,6 +31,7 @@ from litellm.llms.soniox.common_utils import (
|
|||
render_soniox_tokens_as_srt,
|
||||
render_soniox_tokens_as_vtt,
|
||||
)
|
||||
from litellm.llms.soniox.types import SonioxContext
|
||||
from litellm.types.llms.openai import (
|
||||
AllMessageValues,
|
||||
OpenAIAudioTranscriptionOptionalParams,
|
||||
|
|
@ -58,6 +64,152 @@ SONIOX_HANDLER_ONLY_PARAMS: Final[list[str]] = [
|
|||
]
|
||||
|
||||
|
||||
SONIOX_BOOL_PARAMS: Final[frozenset[str]] = frozenset(
|
||||
{"enable_speaker_diarization", "enable_language_identification", "language_hints_strict"}
|
||||
)
|
||||
_JSON_OBJECT: Final = TypeAdapter(dict[str, object])
|
||||
_STRUCTURED_CONTEXT: Final = TypeAdapter(SonioxContext)
|
||||
_FORM_BOOL: Final = TypeAdapter(bool)
|
||||
_ITEMS: Final = TypeAdapter(tuple[object, ...])
|
||||
_BRACKETED_FORM_FIELD: Final = re.compile(
|
||||
r"(?P<param>context|translation)\[(?P<field>\w+)\](?:\[\]\[(?P<entry_field>\w+)\])?"
|
||||
)
|
||||
_BracketedField = tuple[str, str, str | None, object]
|
||||
|
||||
|
||||
@dataclass(frozen=True, slots=True)
|
||||
class SonioxInvalidFormParam:
|
||||
key: str
|
||||
value: str
|
||||
expected: str
|
||||
|
||||
|
||||
def raise_soniox_form_error(error: SonioxInvalidFormParam) -> NoReturn:
|
||||
raise SonioxException(
|
||||
message=f"`{error.key}` must be {error.expected}, got {error.value!r}", status_code=400, headers=None
|
||||
)
|
||||
|
||||
|
||||
def soniox_language_hints(language: object, hints: object) -> object:
|
||||
"""Soniox has no `language` field, so the OpenAI `language` becomes `language_hints` when none are set."""
|
||||
if hints is not None or not isinstance(language, str) or not language:
|
||||
return hints
|
||||
return (language,)
|
||||
|
||||
|
||||
def _decode_bool(key: str, value: str) -> bool | SonioxInvalidFormParam:
|
||||
try:
|
||||
return _FORM_BOOL.validate_python(value.strip())
|
||||
except ValidationError:
|
||||
return SonioxInvalidFormParam(key=key, value=value, expected="a boolean")
|
||||
|
||||
|
||||
def _decode_polling_interval(value: str) -> str | SonioxInvalidFormParam:
|
||||
try:
|
||||
float(value)
|
||||
except ValueError:
|
||||
return SonioxInvalidFormParam(key="soniox_polling_interval", value=value, expected="a number")
|
||||
return value
|
||||
|
||||
|
||||
def _decode_context(value: str) -> SonioxContext | str:
|
||||
if not value.lstrip().startswith("{"):
|
||||
return value
|
||||
try:
|
||||
return _STRUCTURED_CONTEXT.validate_json(value)
|
||||
except ValidationError:
|
||||
return value
|
||||
|
||||
|
||||
def _decode_translation(value: str) -> Mapping[str, object] | SonioxInvalidFormParam:
|
||||
try:
|
||||
return _JSON_OBJECT.validate_json(value)
|
||||
except ValidationError:
|
||||
return SonioxInvalidFormParam(key="translation", value=value, expected="a JSON object")
|
||||
|
||||
|
||||
def _decode_language_hints(value: str) -> tuple[object, ...] | SonioxInvalidFormParam:
|
||||
if not value.lstrip().startswith("["):
|
||||
return tuple(hint.strip() for hint in value.split(",") if hint.strip())
|
||||
try:
|
||||
return _ITEMS.validate_json(value)
|
||||
except ValidationError:
|
||||
return SonioxInvalidFormParam(key="language_hints", value=value, expected="a JSON list")
|
||||
|
||||
|
||||
def _decode_form_value(key: str, value: object) -> object | SonioxInvalidFormParam:
|
||||
if not isinstance(value, str):
|
||||
return value
|
||||
if key in SONIOX_BOOL_PARAMS:
|
||||
return _decode_bool(key, value)
|
||||
if key == "soniox_polling_interval":
|
||||
return _decode_polling_interval(value)
|
||||
if key == "context":
|
||||
return _decode_context(value)
|
||||
if key == "translation":
|
||||
return _decode_translation(value)
|
||||
if key == "language_hints":
|
||||
return _decode_language_hints(value)
|
||||
return value
|
||||
|
||||
|
||||
def _nested_form_param(param: str, fields: tuple[_BracketedField, ...]) -> Mapping[str, object]:
|
||||
owned: Final = tuple(field for field in fields if field[0] == param)
|
||||
flat: Final = {name: value for _, name, entry_field, value in owned if entry_field is None}
|
||||
listed: Final = {name for _, name, entry_field, _ in owned if entry_field is not None}
|
||||
return {
|
||||
**flat,
|
||||
**{
|
||||
name: (
|
||||
{
|
||||
entry_field: value
|
||||
for _, owner, entry_field, value in owned
|
||||
if owner == name and entry_field is not None
|
||||
},
|
||||
)
|
||||
for name in listed
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def _is_soniox_context(value: object) -> bool:
|
||||
try:
|
||||
_STRUCTURED_CONTEXT.validate_python(value)
|
||||
except ValidationError:
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
def _nest_bracketed_form_fields(params: Mapping[str, object]) -> Mapping[str, object]:
|
||||
"""The OpenAI SDK sends a dict in `extra_body` as `context[terms]` and `context[general][][key]` form keys."""
|
||||
matches: Final = {key: _BRACKETED_FORM_FIELD.fullmatch(key) for key in params}
|
||||
fields: Final = tuple(
|
||||
(match["param"], match["field"], match["entry_field"], params[key]) for key, match in matches.items() if match
|
||||
)
|
||||
nested: Final = {param: _nested_form_param(param, fields) for param in {field[0] for field in fields}}
|
||||
kept: Final = {param: value for param, value in nested.items() if param != "context" or _is_soniox_context(value)}
|
||||
unbracketed: Final = {
|
||||
key: value for key, value in params.items() if (match := matches[key]) is None or match["param"] not in kept
|
||||
}
|
||||
return {**kept, **unbracketed}
|
||||
|
||||
|
||||
def decode_soniox_form_params(
|
||||
optional_params: Mapping[str, object],
|
||||
drop_params: bool = False,
|
||||
) -> Mapping[str, object] | SonioxInvalidFormParam:
|
||||
"""Multipart form fields reach the proxy as strings; restore the JSON types Soniox expects."""
|
||||
decoded: Final = {
|
||||
key: _decode_form_value(key, value) for key, value in _nest_bracketed_form_fields(optional_params).items()
|
||||
}
|
||||
invalid: Final = next((value for value in decoded.values() if isinstance(value, SonioxInvalidFormParam)), None)
|
||||
if invalid is not None and not drop_params:
|
||||
return invalid
|
||||
return MappingProxyType(
|
||||
{key: value for key, value in decoded.items() if not isinstance(value, SonioxInvalidFormParam)}
|
||||
)
|
||||
|
||||
|
||||
class SonioxAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
|
||||
"""Configuration for Soniox async speech-to-text transcription."""
|
||||
|
||||
|
|
@ -74,25 +226,26 @@ class SonioxAudioTranscriptionConfig(BaseAudioTranscriptionConfig):
|
|||
model: str,
|
||||
drop_params: bool,
|
||||
) -> dict:
|
||||
# Translate the OpenAI `language` param into Soniox `language_hints`.
|
||||
if "language" in non_default_params and non_default_params["language"]:
|
||||
language: Final = non_default_params["language"]
|
||||
existing_hints: Final = optional_params.get("language_hints")
|
||||
if not existing_hints:
|
||||
optional_params["language_hints"] = [language]
|
||||
elif language not in existing_hints:
|
||||
optional_params["language_hints"] = [language] + list(existing_hints)
|
||||
|
||||
# Capture response_format for post-processing (not sent to Soniox API).
|
||||
if "response_format" in non_default_params:
|
||||
optional_params["response_format"] = non_default_params["response_format"]
|
||||
|
||||
# Pass through Soniox-native kwargs unchanged.
|
||||
for key in SONIOX_PASSTHROUGH_PARAMS + SONIOX_HANDLER_ONLY_PARAMS:
|
||||
if key in non_default_params and non_default_params[key] is not None:
|
||||
optional_params[key] = non_default_params[key]
|
||||
|
||||
return optional_params
|
||||
language: Final[object] = non_default_params.get("language")
|
||||
language_params: Final[Mapping[str, object]] = (
|
||||
{
|
||||
"language": language,
|
||||
"language_hints": soniox_language_hints(language, optional_params.get("language_hints")),
|
||||
}
|
||||
if language
|
||||
else {}
|
||||
)
|
||||
response_format_params: Final[Mapping[str, object]] = (
|
||||
{"response_format": non_default_params["response_format"]}
|
||||
if "response_format" in non_default_params
|
||||
else {}
|
||||
)
|
||||
passthrough_params: Final[Mapping[str, object]] = {
|
||||
key: non_default_params[key]
|
||||
for key in SONIOX_PASSTHROUGH_PARAMS + SONIOX_HANDLER_ONLY_PARAMS
|
||||
if non_default_params.get(key) is not None
|
||||
}
|
||||
return {**optional_params, **language_params, **response_format_params, **passthrough_params}
|
||||
|
||||
def get_error_class(self, error_message: str, status_code: int, headers: dict | Headers) -> BaseLLMException:
|
||||
return SonioxException(message=error_message, status_code=status_code, headers=headers)
|
||||
|
|
|
|||
24
litellm/llms/soniox/types.py
Normal file
24
litellm/llms/soniox/types.py
Normal file
|
|
@ -0,0 +1,24 @@
|
|||
from collections.abc import Sequence
|
||||
|
||||
from pydantic import ConfigDict, with_config
|
||||
from typing_extensions import ReadOnly, TypedDict
|
||||
|
||||
|
||||
class SonioxContextGeneralEntry(TypedDict):
|
||||
key: ReadOnly[str]
|
||||
value: ReadOnly[str]
|
||||
|
||||
|
||||
class SonioxTranslationTerm(TypedDict):
|
||||
source: ReadOnly[str]
|
||||
target: ReadOnly[str]
|
||||
|
||||
|
||||
@with_config(ConfigDict(extra="forbid"))
|
||||
class SonioxContext(TypedDict, total=False):
|
||||
"""Soniox `context` request field: https://soniox.com/docs/stt/concepts/context"""
|
||||
|
||||
general: ReadOnly[Sequence[SonioxContextGeneralEntry]]
|
||||
text: ReadOnly[str]
|
||||
terms: ReadOnly[Sequence[str]]
|
||||
translation_terms: ReadOnly[Sequence[SonioxTranslationTerm]]
|
||||
|
|
@ -339,6 +339,65 @@ class TestPollLimitsClamping:
|
|||
a worker on tight poll loops.
|
||||
"""
|
||||
|
||||
def test_should_raise_400_when_form_bool_param_is_invalid(self):
|
||||
handler = SonioxAudioTranscriptionHandler()
|
||||
with pytest.raises(SonioxException) as exc_info:
|
||||
handler._prepare(
|
||||
audio_file=None,
|
||||
optional_params={
|
||||
"enable_speaker_diarization": "maybe",
|
||||
"audio_url": "https://example.com/a.wav",
|
||||
},
|
||||
litellm_params={},
|
||||
api_key="sk-test",
|
||||
api_base=None,
|
||||
provider_config=SonioxAudioTranscriptionConfig(),
|
||||
headers={},
|
||||
)
|
||||
assert exc_info.value.status_code == 400
|
||||
assert "enable_speaker_diarization" in str(exc_info.value)
|
||||
|
||||
def test_should_raise_400_when_poll_interval_is_not_a_number(self):
|
||||
with pytest.raises(SonioxException) as exc_info:
|
||||
SonioxAudioTranscriptionHandler()._prepare(
|
||||
audio_file=None,
|
||||
optional_params={"soniox_polling_interval": "abc", "audio_url": "https://example.com/a.wav"},
|
||||
litellm_params={},
|
||||
api_key="sk-test",
|
||||
api_base=None,
|
||||
provider_config=SonioxAudioTranscriptionConfig(),
|
||||
headers={},
|
||||
)
|
||||
assert exc_info.value.status_code == 400
|
||||
assert "soniox_polling_interval" in str(exc_info.value)
|
||||
|
||||
def test_should_use_default_poll_interval_for_bad_value_when_drop_params_is_set(self):
|
||||
from litellm.llms.soniox.common_utils import SONIOX_DEFAULT_POLL_INTERVAL
|
||||
|
||||
_, _, _, opts = SonioxAudioTranscriptionHandler()._prepare(
|
||||
audio_file=None,
|
||||
optional_params={"soniox_polling_interval": "abc", "audio_url": "https://example.com/a.wav"},
|
||||
litellm_params={"drop_params": True},
|
||||
api_key="sk-test",
|
||||
api_base=None,
|
||||
provider_config=SonioxAudioTranscriptionConfig(),
|
||||
headers={},
|
||||
)
|
||||
assert opts["poll_interval"] == SONIOX_DEFAULT_POLL_INTERVAL
|
||||
|
||||
@pytest.mark.parametrize("raw, expected", [(10.0, 10), (10, 10), ("10", 10)])
|
||||
def test_should_keep_numeric_max_polling_attempts(self, raw: object, expected: int):
|
||||
_, _, _, opts = SonioxAudioTranscriptionHandler()._prepare(
|
||||
audio_file=None,
|
||||
optional_params={"soniox_max_polling_attempts": raw, "audio_url": "https://example.com/a.wav"},
|
||||
litellm_params={},
|
||||
api_key="sk-test",
|
||||
api_base=None,
|
||||
provider_config=SonioxAudioTranscriptionConfig(),
|
||||
headers={},
|
||||
)
|
||||
assert opts["max_attempts"] == expected
|
||||
|
||||
def test_should_clamp_poll_interval_to_minimum(self):
|
||||
from litellm.llms.soniox.common_utils import SONIOX_MIN_POLL_INTERVAL
|
||||
|
||||
|
|
@ -707,6 +766,112 @@ class TestCleanupNormalization:
|
|||
assert "https://api.soniox.com/v1/transcriptions/tx_1" in deletes
|
||||
|
||||
|
||||
def test_should_skip_non_string_cleanup_entries(self):
|
||||
_, _, _, opts = SonioxAudioTranscriptionHandler()._prepare(
|
||||
audio_file=None,
|
||||
optional_params={"soniox_cleanup": ["transcription", 1, None], "audio_url": "https://example.com/a.wav"},
|
||||
litellm_params={},
|
||||
api_key="sk-test",
|
||||
api_base=None,
|
||||
provider_config=SonioxAudioTranscriptionConfig(),
|
||||
headers={},
|
||||
)
|
||||
assert opts["cleanup"] == ("transcription",)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"language_hints, expected",
|
||||
[("en,es", ["en", "es"]), (["es"], ["es"]), (None, ["de"])],
|
||||
)
|
||||
def test_should_use_language_only_without_language_hints(self, language_hints, expected, monkeypatch):
|
||||
import litellm
|
||||
|
||||
monkeypatch.setattr("time.sleep", lambda *_: None)
|
||||
client = _MockSyncClient(
|
||||
{
|
||||
"POST https://api.soniox.com/v1/transcriptions": [_make_response({"id": "tx_1"})],
|
||||
"GET https://api.soniox.com/v1/transcriptions/tx_1": [_make_response({"status": "completed"})],
|
||||
"GET https://api.soniox.com/v1/transcriptions/tx_1/transcript": [
|
||||
_make_response({"text": "hallo", "tokens": []})
|
||||
],
|
||||
"DELETE https://api.soniox.com/v1/transcriptions/tx_1": [_make_response({})],
|
||||
}
|
||||
)
|
||||
litellm.transcription(
|
||||
model="soniox/stt-async-v4",
|
||||
file=("a.wav", b"RIFF", "audio/wav"),
|
||||
language="de",
|
||||
api_key="sk-test",
|
||||
client=client,
|
||||
audio_url="https://example.com/a.wav",
|
||||
**({} if language_hints is None else {"language_hints": language_hints}),
|
||||
)
|
||||
body = next(c for c in client.calls if c["method"] == "POST")["json"]
|
||||
assert list(body["language_hints"]) == expected
|
||||
assert "language" not in body
|
||||
|
||||
|
||||
class TestProxyFormFields:
|
||||
def test_should_send_soniox_types_for_fields_the_proxy_parsed_from_an_sdk_form(self, monkeypatch):
|
||||
import litellm
|
||||
from starlette.requests import Request
|
||||
|
||||
from litellm.proxy.common_utils.http_parsing_utils import get_form_data
|
||||
|
||||
monkeypatch.setattr("time.sleep", lambda *_: None)
|
||||
multipart = httpx.Request(
|
||||
"POST",
|
||||
"http://proxy/v1/audio/transcriptions",
|
||||
data={
|
||||
"model": "soniox/stt-async-v4",
|
||||
"audio_url": "https://example.com/a.wav",
|
||||
"context[terms][]": ["Celebrex"],
|
||||
"context[general][][key]": "domain",
|
||||
"context[general][][value]": "Healthcare",
|
||||
"translation[type]": "one_way",
|
||||
"translation[target_language]": "de",
|
||||
"enable_speaker_diarization": "true",
|
||||
"language_hints[]": ["en", "es"],
|
||||
},
|
||||
files={"file": ("a.wav", b"RIFF", "audio/wav")},
|
||||
)
|
||||
multipart_body = multipart.read()
|
||||
|
||||
async def receive():
|
||||
return {"type": "http.request", "body": multipart_body, "more_body": False}
|
||||
|
||||
request = Request(
|
||||
{
|
||||
"type": "http",
|
||||
"method": "POST",
|
||||
"path": "/v1/audio/transcriptions",
|
||||
"headers": [(b"content-type", multipart.headers["content-type"].encode())],
|
||||
},
|
||||
receive,
|
||||
)
|
||||
form = asyncio.run(get_form_data(request))
|
||||
client = _MockSyncClient(
|
||||
{
|
||||
"POST https://api.soniox.com/v1/transcriptions": [_make_response({"id": "tx_1"})],
|
||||
"GET https://api.soniox.com/v1/transcriptions/tx_1": [_make_response({"status": "completed"})],
|
||||
"GET https://api.soniox.com/v1/transcriptions/tx_1/transcript": [
|
||||
_make_response({"text": "hello", "tokens": []})
|
||||
],
|
||||
"DELETE https://api.soniox.com/v1/transcriptions/tx_1": [_make_response({})],
|
||||
}
|
||||
)
|
||||
litellm.transcription(
|
||||
file=("a.wav", b"RIFF", "audio/wav"),
|
||||
api_key="sk-test",
|
||||
client=client,
|
||||
**{key: value for key, value in form.items() if key != "file"},
|
||||
)
|
||||
body = json.loads(json.dumps(next(c for c in client.calls if c["method"] == "POST")["json"]))
|
||||
assert body["context"] == {"terms": ["Celebrex"], "general": [{"key": "domain", "value": "Healthcare"}]}
|
||||
assert body["translation"] == {"type": "one_way", "target_language": "de"}
|
||||
assert body["enable_speaker_diarization"] is True
|
||||
assert body["language_hints"] == ["en", "es"]
|
||||
|
||||
|
||||
class TestErrorResponses:
|
||||
def test_should_raise_on_4xx_during_create_with_json_error(self, monkeypatch):
|
||||
monkeypatch.setattr("time.sleep", lambda *_: None)
|
||||
|
|
@ -780,6 +945,51 @@ class TestPassthroughBodyBuilding:
|
|||
post_call = next(c for c in client.calls if c["method"] == "POST")
|
||||
assert "context" not in post_call["json"]
|
||||
|
||||
def test_should_send_form_string_params_to_soniox_as_json_types(self, monkeypatch):
|
||||
"""Simulates the proxy path, where every multipart form field is a string."""
|
||||
monkeypatch.setattr("time.sleep", lambda *_: None)
|
||||
responses = {
|
||||
"POST https://api.soniox.com/v1/transcriptions": [
|
||||
_make_response({"id": "tx_1"}),
|
||||
],
|
||||
"GET https://api.soniox.com/v1/transcriptions/tx_1": [
|
||||
_make_response({"status": "completed"}),
|
||||
],
|
||||
"GET https://api.soniox.com/v1/transcriptions/tx_1/transcript": [
|
||||
_make_response({"text": "ok", "tokens": []}),
|
||||
],
|
||||
"DELETE https://api.soniox.com/v1/transcriptions/tx_1": [
|
||||
_make_response({}),
|
||||
],
|
||||
}
|
||||
client = _MockSyncClient(responses)
|
||||
form_params = {
|
||||
"audio_url": "https://example.com/a.wav",
|
||||
"context": '{"terms": ["Celebrex"], "general": [{"key": "domain", "value": "Healthcare"}]}',
|
||||
"translation": '{"type": "one_way", "target_language": "es"}',
|
||||
"language_hints": "en,es",
|
||||
"language_hints_strict": "true",
|
||||
"enable_speaker_diarization": "true",
|
||||
"enable_language_identification": "false",
|
||||
"soniox_polling_interval": "0.5",
|
||||
}
|
||||
SonioxAudioTranscriptionHandler().audio_transcriptions(
|
||||
audio_file=None,
|
||||
optional_params=form_params,
|
||||
litellm_params={},
|
||||
atranscription=False,
|
||||
**_common_call_kwargs(client),
|
||||
)
|
||||
body = next(c for c in client.calls if c["method"] == "POST")["json"]
|
||||
assert body["context"] == {"terms": ["Celebrex"], "general": [{"key": "domain", "value": "Healthcare"}]}
|
||||
assert body["translation"] == {"type": "one_way", "target_language": "es"}
|
||||
assert list(body["language_hints"]) == ["en", "es"]
|
||||
assert body["language_hints_strict"] is True
|
||||
assert body["enable_speaker_diarization"] is True
|
||||
assert body["enable_language_identification"] is False
|
||||
assert "soniox_polling_interval" not in body
|
||||
assert form_params["context"].startswith("{"), "caller's params must not be mutated"
|
||||
|
||||
|
||||
class TestSecretRedaction:
|
||||
"""Secret-bearing fields must be redacted before reaching logging callbacks.
|
||||
|
|
|
|||
|
|
@ -9,8 +9,16 @@ import pytest
|
|||
|
||||
from litellm.llms.soniox.audio_transcription.transformation import (
|
||||
SonioxAudioTranscriptionConfig,
|
||||
SonioxInvalidFormParam,
|
||||
decode_soniox_form_params,
|
||||
raise_soniox_form_error,
|
||||
)
|
||||
from litellm.llms.soniox.common_utils import SonioxException
|
||||
from litellm.llms.soniox.types import (
|
||||
SonioxContext,
|
||||
SonioxContextGeneralEntry,
|
||||
SonioxTranslationTerm,
|
||||
)
|
||||
from litellm.types.utils import TranscriptionResponse
|
||||
|
||||
|
||||
|
|
@ -40,9 +48,9 @@ class TestMapOpenAIParams:
|
|||
model="stt-async-v4",
|
||||
drop_params=False,
|
||||
)
|
||||
assert result["language_hints"] == ["en"]
|
||||
assert list(result["language_hints"]) == ["en"]
|
||||
|
||||
def test_should_prepend_language_to_existing_hints(self):
|
||||
def test_should_keep_explicit_hints_over_language(self):
|
||||
cfg = SonioxAudioTranscriptionConfig()
|
||||
result = cfg.map_openai_params(
|
||||
non_default_params={"language": "en"},
|
||||
|
|
@ -50,7 +58,7 @@ class TestMapOpenAIParams:
|
|||
model="stt-async-v4",
|
||||
drop_params=False,
|
||||
)
|
||||
assert result["language_hints"] == ["en", "fr"]
|
||||
assert list(result["language_hints"]) == ["fr"]
|
||||
|
||||
def test_should_not_duplicate_language_already_in_hints(self):
|
||||
cfg = SonioxAudioTranscriptionConfig()
|
||||
|
|
@ -60,7 +68,7 @@ class TestMapOpenAIParams:
|
|||
model="stt-async-v4",
|
||||
drop_params=False,
|
||||
)
|
||||
assert result["language_hints"] == ["en", "fr"]
|
||||
assert list(result["language_hints"]) == ["en", "fr"]
|
||||
|
||||
def test_should_passthrough_soniox_native_kwargs(self):
|
||||
cfg = SonioxAudioTranscriptionConfig()
|
||||
|
|
@ -243,6 +251,21 @@ class TestTransformAudioTranscriptionResponse:
|
|||
assert "Speaker 1:" in resp.text
|
||||
assert "Speaker 2:" in resp.text
|
||||
|
||||
def test_should_keep_source_transcript_for_translation_without_tags(self):
|
||||
cfg = SonioxAudioTranscriptionConfig()
|
||||
payload = {
|
||||
"transcript": {
|
||||
"text": "Hello there.",
|
||||
"tokens": [
|
||||
{"text": "Hello", "translation_status": "original"},
|
||||
{"text": " there.", "translation_status": "original"},
|
||||
{"text": "Hallo", "translation_status": "translation"},
|
||||
{"text": " zusammen.", "translation_status": "translation"},
|
||||
],
|
||||
}
|
||||
}
|
||||
assert cfg.build_response_from_payload(payload).text == "Hello there."
|
||||
|
||||
def test_should_set_language_when_all_tokens_share_one(self):
|
||||
cfg = SonioxAudioTranscriptionConfig()
|
||||
payload = {
|
||||
|
|
@ -678,3 +701,175 @@ class TestGetErrorClass:
|
|||
err = cfg.get_error_class(error_message="boom", status_code=500, headers={})
|
||||
assert isinstance(err, SonioxException)
|
||||
assert err.status_code == 500
|
||||
|
||||
|
||||
class TestDecodeSonioxFormParams:
|
||||
"""Proxy multipart form fields arrive as strings; Soniox needs real JSON types."""
|
||||
|
||||
def test_should_decode_json_string_context_with_all_four_sections(self):
|
||||
context = {
|
||||
"general": [{"key": "domain", "value": "Healthcare"}],
|
||||
"text": "Follow-up visit for a patient on blood thinners.",
|
||||
"terms": ["Celebrex", "Zyrtec"],
|
||||
"translation_terms": [{"source": "Mr. Smith", "target": "Sr. Smith"}],
|
||||
}
|
||||
result = decode_soniox_form_params({"context": json.dumps(context)})
|
||||
assert result["context"] == context
|
||||
|
||||
def test_should_keep_plain_text_context_as_string(self):
|
||||
result = decode_soniox_form_params({"context": "medical conversation"})
|
||||
assert result["context"] == "medical conversation"
|
||||
|
||||
def test_should_leave_sdk_typed_values_untouched(self):
|
||||
context = SonioxContext(
|
||||
general=[SonioxContextGeneralEntry(key="domain", value="Healthcare")],
|
||||
text="Follow-up visit notes.",
|
||||
terms=["Celebrex"],
|
||||
translation_terms=[SonioxTranslationTerm(source="Mr. Smith", target="Sr. Smith")],
|
||||
)
|
||||
translation = {"type": "one_way", "target_language": "es"}
|
||||
result = decode_soniox_form_params(
|
||||
{
|
||||
"context": context,
|
||||
"translation": translation,
|
||||
"enable_speaker_diarization": True,
|
||||
"language_hints": ["en", "es"],
|
||||
}
|
||||
)
|
||||
assert result["context"] is context
|
||||
assert result["translation"] is translation
|
||||
assert result["enable_speaker_diarization"] is True
|
||||
assert result["language_hints"] == ["en", "es"]
|
||||
|
||||
@pytest.mark.parametrize("raw", ["[inaudible] consultation", "{unbalanced", '{"terms": ["Celebrex"'])
|
||||
def test_should_keep_bracket_prefixed_free_form_context_as_string(self, raw: str):
|
||||
assert decode_soniox_form_params({"context": raw})["context"] == raw
|
||||
|
||||
def test_should_decode_json_string_translation(self):
|
||||
result = decode_soniox_form_params({"translation": '{"type": "one_way", "target_language": "es"}'})
|
||||
assert result["translation"] == {"type": "one_way", "target_language": "es"}
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"raw, expected",
|
||||
[("true", True), ("True", True), ("1", True), ("false", False), ("FALSE", False), ("0", False)],
|
||||
)
|
||||
def test_should_decode_boolean_strings(self, raw: str, expected: bool):
|
||||
result = decode_soniox_form_params(
|
||||
{
|
||||
"enable_speaker_diarization": raw,
|
||||
"enable_language_identification": raw,
|
||||
"language_hints_strict": raw,
|
||||
}
|
||||
)
|
||||
assert result == {
|
||||
"enable_speaker_diarization": expected,
|
||||
"enable_language_identification": expected,
|
||||
"language_hints_strict": expected,
|
||||
}
|
||||
|
||||
@pytest.mark.parametrize("raw, expected", [("on", True), ("yes", True), ("off", False), ("no", False)])
|
||||
def test_should_decode_html_style_boolean_strings(self, raw: str, expected: bool):
|
||||
assert decode_soniox_form_params({"enable_speaker_diarization": raw})["enable_speaker_diarization"] is expected
|
||||
|
||||
def test_should_return_error_value_on_non_boolean_string(self):
|
||||
result = decode_soniox_form_params({"context": '{"general": []}', "enable_speaker_diarization": "maybe"})
|
||||
assert result == SonioxInvalidFormParam(key="enable_speaker_diarization", value="maybe", expected="a boolean")
|
||||
|
||||
def test_should_drop_non_boolean_string_when_drop_params_is_set(self):
|
||||
result = decode_soniox_form_params(
|
||||
{"enable_speaker_diarization": "maybe", "language_hints_strict": "true"}, drop_params=True
|
||||
)
|
||||
assert result == {"language_hints_strict": True}
|
||||
|
||||
def test_should_return_error_value_on_non_numeric_polling_interval(self):
|
||||
result = decode_soniox_form_params({"soniox_polling_interval": "abc"})
|
||||
assert result == SonioxInvalidFormParam(key="soniox_polling_interval", value="abc", expected="a number")
|
||||
|
||||
def test_should_drop_non_numeric_polling_interval_when_drop_params_is_set(self):
|
||||
assert decode_soniox_form_params({"soniox_polling_interval": "abc"}, drop_params=True) == {}
|
||||
|
||||
def test_should_map_invalid_form_value_error_to_400(self):
|
||||
with pytest.raises(SonioxException) as exc_info:
|
||||
raise_soniox_form_error(
|
||||
SonioxInvalidFormParam(key="enable_speaker_diarization", value="maybe", expected="a boolean")
|
||||
)
|
||||
assert exc_info.value.status_code == 400
|
||||
assert "enable_speaker_diarization" in str(exc_info.value)
|
||||
assert "maybe" in str(exc_info.value)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"raw", ['["Celebrex", "Zyrtec"]', '{"note": "follow-up visit"}', '{"terms": "Celebrex"}', '"quoted"']
|
||||
)
|
||||
def test_should_keep_json_context_soniox_cannot_take_as_an_object_as_string(self, raw: str):
|
||||
assert decode_soniox_form_params({"context": raw})["context"] == raw
|
||||
|
||||
def test_should_nest_openai_sdk_bracketed_form_fields(self):
|
||||
result = decode_soniox_form_params(
|
||||
{
|
||||
"context[terms]": ["Celebrex", "Zyrtec"],
|
||||
"context[text]": "Follow-up visit.",
|
||||
"translation[type]": "one_way",
|
||||
"translation[target_language]": "de",
|
||||
"client_reference_id": "ref-1",
|
||||
}
|
||||
)
|
||||
assert result == {
|
||||
"context": {"terms": ["Celebrex", "Zyrtec"], "text": "Follow-up visit."},
|
||||
"translation": {"type": "one_way", "target_language": "de"},
|
||||
"client_reference_id": "ref-1",
|
||||
}
|
||||
|
||||
def test_should_prefer_explicit_field_over_bracketed_fields(self):
|
||||
result = decode_soniox_form_params({"context": "plain text", "context[terms]": ["Celebrex"]})
|
||||
assert result == {"context": "plain text"}
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"raw, expected",
|
||||
[('["en", "es"]', ["en", "es"]), ("en,es", ["en", "es"]), ("en, es ,", ["en", "es"]), ("en", ["en"])],
|
||||
)
|
||||
def test_should_decode_language_hints_strings(self, raw: str, expected: list):
|
||||
assert list(decode_soniox_form_params({"language_hints": raw})["language_hints"]) == expected
|
||||
|
||||
@pytest.mark.parametrize("raw", ['{"type": "one_way"', "de", '["one_way"]'])
|
||||
def test_should_return_error_value_on_translation_that_is_not_a_json_object(self, raw: str):
|
||||
assert decode_soniox_form_params({"translation": raw}) == SonioxInvalidFormParam(
|
||||
key="translation", value=raw, expected="a JSON object"
|
||||
)
|
||||
|
||||
def test_should_return_error_value_on_malformed_json_language_hints(self):
|
||||
assert decode_soniox_form_params({"language_hints": "[en, es]"}) == SonioxInvalidFormParam(
|
||||
key="language_hints", value="[en, es]", expected="a JSON list"
|
||||
)
|
||||
|
||||
def test_should_drop_malformed_json_params_when_drop_params_is_set(self):
|
||||
result = decode_soniox_form_params(
|
||||
{"translation": '{"type": "one_way"', "language_hints": "[en, es]", "language_hints_strict": "true"},
|
||||
drop_params=True,
|
||||
)
|
||||
assert result == {"language_hints_strict": True}
|
||||
|
||||
def test_should_rebuild_openai_sdk_list_of_object_context_fields(self):
|
||||
result = decode_soniox_form_params(
|
||||
{
|
||||
"context[terms]": ["Celebrex"],
|
||||
"context[general][][key]": "domain",
|
||||
"context[general][][value]": "Healthcare",
|
||||
"context[translation_terms][][source]": "Mr. Smith",
|
||||
"context[translation_terms][][target]": "Sr. Smith",
|
||||
}
|
||||
)
|
||||
assert result == {
|
||||
"context": {
|
||||
"terms": ["Celebrex"],
|
||||
"general": ({"key": "domain", "value": "Healthcare"},),
|
||||
"translation_terms": ({"source": "Mr. Smith", "target": "Sr. Smith"},),
|
||||
}
|
||||
}
|
||||
|
||||
def test_should_leave_bracketed_context_soniox_cannot_take_unnested(self):
|
||||
params = {"context[terms]": ["Celebrex"], "context[note]": "follow-up"}
|
||||
assert decode_soniox_form_params(params) == params
|
||||
|
||||
def test_should_not_json_decode_unrelated_string_params(self):
|
||||
result = decode_soniox_form_params({"client_reference_id": '["not", "json-decoded"]'})
|
||||
assert result["client_reference_id"] == '["not", "json-decoded"]'
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue