diff --git a/litellm/llms/soniox/audio_transcription/handler.py b/litellm/llms/soniox/audio_transcription/handler.py
index b53955ae75d..f86bc142898 100644
--- a/litellm/llms/soniox/audio_transcription/handler.py
+++ b/litellm/llms/soniox/audio_transcription/handler.py
@@ -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,
diff --git a/litellm/llms/soniox/audio_transcription/transformation.py b/litellm/llms/soniox/audio_transcription/transformation.py
index aa1ff239320..30d872f248f 100644
--- a/litellm/llms/soniox/audio_transcription/transformation.py
+++ b/litellm/llms/soniox/audio_transcription/transformation.py
@@ -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"(?Pcontext|translation)\[(?P\w+)\](?:\[\]\[(?P\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)
diff --git a/litellm/llms/soniox/types.py b/litellm/llms/soniox/types.py
new file mode 100644
index 00000000000..806d2a8fb0e
--- /dev/null
+++ b/litellm/llms/soniox/types.py
@@ -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]]
diff --git a/tests/unit/llms/soniox/audio_transcription/test_soniox_audio_transcription_handler.py b/tests/unit/llms/soniox/audio_transcription/test_soniox_audio_transcription_handler.py
index d2d7d2247f1..9c70ebe7ece 100644
--- a/tests/unit/llms/soniox/audio_transcription/test_soniox_audio_transcription_handler.py
+++ b/tests/unit/llms/soniox/audio_transcription/test_soniox_audio_transcription_handler.py
@@ -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.
diff --git a/tests/unit/llms/soniox/audio_transcription/test_soniox_audio_transcription_transformation.py b/tests/unit/llms/soniox/audio_transcription/test_soniox_audio_transcription_transformation.py
index c99b28adc49..e3e1c9ab493 100644
--- a/tests/unit/llms/soniox/audio_transcription/test_soniox_audio_transcription_transformation.py
+++ b/tests/unit/llms/soniox/audio_transcription/test_soniox_audio_transcription_transformation.py
@@ -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"]'