fix(transcription): tolerate non-conforming usage in json response_format

litellm/litellm_core_utils/llm_response_utils/convert_dict_to_response.py

@@ -5,6 +5,8 @@
 import traceback
 from typing import Dict, Iterable, List, Literal, Optional, Tuple, Union, cast
 
+from pydantic import ValidationError
+
 import litellm
 from litellm._logging import verbose_logger
 from litellm.constants import RESPONSE_FORMAT_TOOL_NAME
@@ -52,6 +54,26 @@
 _MODEL_RESPONSE_FIELDS: frozenset = frozenset(ModelResponse.model_fields.keys()) | {"usage"}
 
 
+def _parse_transcription_usage(
+    usage: dict,
+) -> Optional[Union[TranscriptionUsageDurationObject, TranscriptionUsageTokensObject]]:
+    usage_type = usage.get("type")
+    model = (
+        TranscriptionUsageDurationObject
+        if usage_type == "duration"
+        else TranscriptionUsageTokensObject
+        if usage_type == "tokens"
+        else None
+    )
+    if model is None:
+        return None
+    try:
+        return model.model_validate(usage)
+    except ValidationError as e:
+        verbose_logger.debug(f"Dropping unparseable transcription usage {usage}: {e}")
+        return None
+
+
 def _normalize_images_for_message(
     images: Optional[List[dict]],
 ) -> Optional[List[ImageURLListItem]]:
@@ -827,14 +849,7 @@ def convert_to_model_response_object(
                     setattr(model_response_object, key, response_object[key])
 
             if "usage" in response_object and response_object["usage"] is not None:
-                tr_usage_object: Optional[Union[TranscriptionUsageDurationObject, TranscriptionUsageTokensObject]] = (
-                    None
-                )
-
-                if response_object["usage"].get("type", None) == "duration":
-                    tr_usage_object = TranscriptionUsageDurationObject(**response_object["usage"])
-                elif response_object["usage"].get("type", None) == "tokens":
-                    tr_usage_object = TranscriptionUsageTokensObject(**response_object["usage"])
+                tr_usage_object = _parse_transcription_usage(response_object["usage"])
                 if tr_usage_object is not None:
                     setattr(model_response_object, "usage", tr_usage_object)
 

litellm/types/utils.py

@@ -2384,7 +2384,7 @@ class TranscriptionUsageTokensObject(BaseModel):
     input_tokens: int
     output_tokens: int
     total_tokens: int
-    input_token_details: TranscriptionUsageInputTokenDetailsObject
+    input_token_details: Optional[TranscriptionUsageInputTokenDetailsObject] = None
 
 
 class TranscriptionResponse(OpenAIObject):

tests/test_litellm/llms/openai/transcriptions/test_transcription_duration_hidden.py

@@ -16,6 +16,7 @@
 from litellm.types.utils import (
     TranscriptionResponse,
     TranscriptionUsageDurationObject,
+    TranscriptionUsageTokensObject,
 )
 
 
@@ -61,6 +62,75 @@ def test_usage_duration_object_accepts_float_seconds(self):
         )
 
 
+class TestTokensUsageParsingIsResilient:
+    """
+    Non-OpenAI OpenAI-compatible servers (e.g. llama.cpp) can return a
+    `usage.type == "tokens"` object that omits or nulls fields OpenAI always
+    sends. A non-conforming usage must never sink a successful transcription
+    (regression for the input_token_details=None ValidationError in #33764).
+    """
+
+    def _convert(self, usage):
+        return convert_to_model_response_object(
+            response_object={"text": "hello world", "usage": usage},
+            model_response_object=TranscriptionResponse(),
+            response_type="audio_transcription",
+        )
+
+    def test_null_input_token_details_does_not_raise(self):
+        result = self._convert(
+            {
+                "type": "tokens",
+                "input_tokens": 10,
+                "output_tokens": 5,
+                "total_tokens": 15,
+                "input_token_details": None,
+            }
+        )
+        assert result.text == "hello world"
+        assert isinstance(result.usage, TranscriptionUsageTokensObject)
+        assert result.usage.input_tokens == 10
+        assert result.usage.input_token_details is None
+
+    def test_missing_input_token_details_does_not_raise(self):
+        result = self._convert(
+            {
+                "type": "tokens",
+                "input_tokens": 10,
+                "output_tokens": 5,
+                "total_tokens": 15,
+            }
+        )
+        assert isinstance(result.usage, TranscriptionUsageTokensObject)
+        assert result.usage.total_tokens == 15
+        assert result.usage.input_token_details is None
+
+    def test_full_token_usage_still_parses(self):
+        result = self._convert(
+            {
+                "type": "tokens",
+                "input_tokens": 10,
+                "output_tokens": 5,
+                "total_tokens": 15,
+                "input_token_details": {"audio_tokens": 3, "text_tokens": 7},
+            }
+        )
+        assert isinstance(result.usage, TranscriptionUsageTokensObject)
+        assert result.usage.input_token_details is not None
+        assert result.usage.input_token_details.audio_tokens == 3
+
+    def test_unparseable_usage_is_dropped_not_raised(self):
+        """A tokens usage missing required counts should drop usage, not crash."""
+        result = self._convert({"type": "tokens", "foo": "bar"})
+        assert result.text == "hello world"
+        assert getattr(result, "usage", None) is None
+
+    def test_unknown_usage_type_is_dropped(self):
+        result = self._convert({"type": "something_new", "value": 1})
+        assert result.text == "hello world"
+        assert getattr(result, "usage", None) is None
+
+
 class TestTranscriptionDurationNotInResponseBody:
     """Duration calculated internally should be in _hidden_params, not in the response body."""