tiennm99/litellm
1
0
Fork 0
mirror of https://github.com/tiennm99/litellm.git synced 2026-07-13 13:10:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

(feat) audio transcriptions cost tracking (for azure/other non-openai models) + (fix) image generations - accurate cost tracking for dalle3/gpt-image-1 - uses the correct max image quality (#16076)

Browse Source 
* feat(audio_transcriptions/): calculate duration of audio file for cost calculation

Fixes https://github.com/BerriAI/litellm/issues/11846

Closes https://github.com/BerriAI/litellm/issues/14605

* fix(cost_calculator.py): correctly use base model, when set

Fixes issue where azure base model was being ignored

* feat(cost_calculator.py): fix default cost tracking quality param for image generation

* feat(image_generations/): return output_format, quality, size

aligns response to openai spec and improves cost tracking accuracy

* fix(cost_calculator.py): refactor cost calculation for image generation to use image response instead of hidden params

* build: update build

* fix: fix cost calculation

* build: update poetry lock

* fix: fix ruff checks
This commit is contained in:
Krish Dholakia
2025-11-08 13:54:37 -08:00
committed by GitHub
parent 3f8f2d2185
commit 9a88fe0861
17 changed files with 830 additions and 998 deletions
Download Patch File Download Diff File Expand all files Collapse all files
litellm/cost_calculator.py
+34 -30
View File
@@ -327,12 +327,16 @@ def cost_per_token( # noqa: PLR0915
elif call_type == "search" or call_type == "asearch":
# Search providers use per-query pricing
from litellm.search import search_provider_cost_per_query
return search_provider_cost_per_query(
model=model,
custom_llm_provider=custom_llm_provider,
number_of_queries=number_of_queries or 1,
optional_params=response._hidden_params if response and hasattr(response, "_hidden_params") else None
optional_params=(
response._hidden_params
if response and hasattr(response, "_hidden_params")
else None
),
)
elif custom_llm_provider == "vertex_ai":
cost_router = google_cost_router(
@@ -509,16 +513,18 @@ def _select_model_name_for_cost_calc(
else:
return_model = model
if base_model is not None:
elif base_model is not None:
return_model = base_model
if completion_response_model is None and hidden_params is not None:
elif completion_response_model is None and hidden_params is not None:
if (
hidden_params.get("model", None) is not None
and len(hidden_params["model"]) > 0
):
return_model = hidden_params.get("model", model)
if hidden_params is not None and hidden_params.get("region_name", None) is not None:
elif (
hidden_params is not None and hidden_params.get("region_name", None) is not None
):
region_name = hidden_params.get("region_name", None)
if return_model is None and completion_response_model is not None:
@@ -853,15 +859,6 @@ def completion_cost( # noqa: PLR0915
"custom_llm_provider", custom_llm_provider or None
)
region_name = hidden_params.get("region_name", region_name)
size = hidden_params.get("optional_params", {}).get(
"size", "1024-x-1024"
) # openai default
quality = hidden_params.get("optional_params", {}).get(
"quality", "standard"
) # openai default
n = hidden_params.get("optional_params", {}).get(
"n", 1
) # openai default
else:
if model is None:
raise ValueError(
@@ -888,7 +885,9 @@ def completion_cost( # noqa: PLR0915
str(e)
)
)
if CostCalculatorUtils._call_type_has_image_response(call_type):
if CostCalculatorUtils._call_type_has_image_response(
call_type
) and isinstance(completion_response, ImageResponse):
### IMAGE GENERATION COST CALCULATION ###
return CostCalculatorUtils.route_image_generation_cost_calculator(
model=model,
@@ -906,27 +905,32 @@ def completion_cost( # noqa: PLR0915
or call_type == CallTypes.avideo_remix.value
):
### VIDEO GENERATION COST CALCULATION ###
if completion_response is not None and hasattr(completion_response, 'usage'):
usage_obj = completion_response.usage
usage_obj = getattr(completion_response, "usage", None)
if completion_response is not None and usage_obj:
# Handle both dict and Pydantic Usage object
if isinstance(usage_obj, dict):
duration_seconds = usage_obj.get('duration_seconds', None)
duration_seconds = usage_obj.get("duration_seconds", None)
else:
duration_seconds = getattr(usage_obj, 'duration_seconds', None)
duration_seconds = getattr(
usage_obj, "duration_seconds", None
)
if duration_seconds is not None:
# Calculate cost based on video duration using video-specific cost calculation
from litellm.llms.openai.cost_calculation import video_generation_cost
from litellm.llms.openai.cost_calculation import (
video_generation_cost,
)
return video_generation_cost(
model=model,
duration_seconds=duration_seconds,
custom_llm_provider=custom_llm_provider
custom_llm_provider=custom_llm_provider,
)
# Fallback to default video cost calculation if no duration available
return default_video_cost_calculator(
model=model,
duration_seconds=0.0, # Default to 0 if no duration available
custom_llm_provider=custom_llm_provider
custom_llm_provider=custom_llm_provider,
)
elif (
call_type == CallTypes.speech.value
@@ -1460,13 +1464,13 @@ def default_video_cost_calculator(
model_name_without_custom_llm_provider = model.replace(
f"{custom_llm_provider}/", ""
)
base_model_name = f"{custom_llm_provider}/{model_name_without_custom_llm_provider}"
base_model_name = (
f"{custom_llm_provider}/{model_name_without_custom_llm_provider}"
)
verbose_logger.debug(
f"Looking up cost for video model: {base_model_name}"
)
verbose_logger.debug(f"Looking up cost for video model: {base_model_name}")
model_without_provider = model.split('/')[-1]
model_without_provider = model.split("/")[-1]
# Try model with provider first, fall back to base model name
cost_info: Optional[dict] = None
@@ -1480,7 +1484,7 @@ def default_video_cost_calculator(
if _model is not None and _model in litellm.model_cost:
cost_info = litellm.model_cost[_model]
break
# If still not found, try with custom_llm_provider prefix
if cost_info is None and custom_llm_provider:
prefixed_model = f"{custom_llm_provider}/{model}"
@@ -1495,12 +1499,12 @@ def default_video_cost_calculator(
video_cost_per_second = cost_info.get("output_cost_per_video_per_second")
if video_cost_per_second is not None:
return video_cost_per_second * duration_seconds
# Fallback to general output cost per second
output_cost_per_second = cost_info.get("output_cost_per_second")
if output_cost_per_second is not None:
return output_cost_per_second * duration_seconds
# If no cost information found, return 0
verbose_logger.info(
f"No cost information found for video model {model}. Please add pricing to model_prices_and_context_window.json"
litellm/integrations/custom_logger.py
+43 -45
View File
@@ -8,7 +8,6 @@ from typing import (
AsyncGenerator,
Dict,
List,
Literal,
Optional,
Tuple,
Union,
@@ -24,6 +23,7 @@ from litellm.types.llms.openai import AllMessageValues, ChatCompletionRequest
from litellm.types.utils import (
AdapterCompletionStreamWrapper,
CallTypes,
CallTypesLiteral,
LLMResponseTypes,
ModelResponse,
ModelResponseStream,
@@ -65,12 +65,11 @@ _BASE64_INLINE_PATTERN = re.compile(
class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callback#callback-class
# Class variables or attributes
def __init__(
self,
self,
turn_off_message_logging: bool = False,
# deprecated param, use `turn_off_message_logging` instead
message_logging: bool = True,
**kwargs
**kwargs,
) -> None:
"""
Args:
@@ -221,7 +220,7 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
) -> Optional[Any]:
"""
Allow modifying streaming chunks just before they're returned to the user.
This is called for each streaming chunk in the response.
"""
pass
@@ -292,18 +291,7 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
user_api_key_dict: UserAPIKeyAuth,
cache: DualCache,
data: dict,
call_type: Literal[
"completion",
"text_completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"pass_through_endpoint",
"rerank",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
) -> Optional[
Union[Exception, str, dict]
]: # raise exception if invalid, return a str for the user to receive - if rejected, or return a modified dictionary for passing into litellm
@@ -342,16 +330,7 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
self,
data: dict,
user_api_key_dict: UserAPIKeyAuth,
call_type: Literal[
"completion",
"embeddings",
"image_generation",
"moderation",
"audio_transcription",
"responses",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
) -> Any:
pass
@@ -435,7 +414,6 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
# MCP TOOL CALL HOOKS
#########################################################
async def async_post_mcp_tool_call_hook(
self, kwargs, response_obj: MCPPostCallResponseObject, start_time, end_time
) -> Optional[MCPPostCallResponseObject]:
@@ -519,18 +497,18 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
if LITELLM_METADATA_FIELD in request_kwargs:
return LITELLM_METADATA_FIELD
return OLD_LITELLM_METADATA_FIELD
def redact_standard_logging_payload_from_model_call_details(
self, model_call_details: Dict
) -> Dict:
"""
Only redacts messages and responses when self.turn_off_message_logging is True
By default, self.turn_off_message_logging is False and this does nothing.
Return a redacted deepcopy of the provided logging payload.
This is useful for logging payloads that contain sensitive information.
"""
from copy import copy
@@ -540,7 +518,7 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
if turn_off_message_logging is False:
return model_call_details
# Only make a shallow copy of the top-level dict to avoid deepcopy issues
# with complex objects like AuthenticationError that may be present
model_call_details_copy = copy(model_call_details)
@@ -553,7 +531,9 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
standard_logging_object_copy = copy(standard_logging_object)
if standard_logging_object_copy.get("messages") is not None:
standard_logging_object_copy["messages"] = [Message(content=redacted_str).model_dump()]
standard_logging_object_copy["messages"] = [
Message(content=redacted_str).model_dump()
]
if standard_logging_object_copy.get("response") is not None:
response = standard_logging_object_copy["response"]
@@ -580,11 +560,11 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
model_response_dict = model_response.model_dump()
standard_logging_object_copy["response"] = model_response_dict
model_call_details_copy["standard_logging_object"] = standard_logging_object_copy
model_call_details_copy["standard_logging_object"] = (
standard_logging_object_copy
)
return model_call_details_copy
async def get_proxy_server_request_from_cold_storage_with_object_key(
self,
object_key: str,
@@ -622,7 +602,9 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
verbose_logger.debug(f"Error in handle_callback_failure for {callback_name}: {str(e)}")
async def _strip_base64_from_messages(
self, payload: "StandardLoggingPayload", max_depth: int = DEFAULT_MAX_RECURSE_DEPTH_SENSITIVE_DATA_MASKER
self,
payload: "StandardLoggingPayload",
max_depth: int = DEFAULT_MAX_RECURSE_DEPTH_SENSITIVE_DATA_MASKER,
) -> "StandardLoggingPayload":
"""
Removes or redacts base64-encoded file data (e.g., PDFs, images, audio)
@@ -671,7 +653,9 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
verbose_logger.debug(f"[CustomLogger] Stripping base64 from {len(messages)} messages")
if messages:
payload["messages"] = self._process_messages(messages=messages, max_depth=max_depth)
payload["messages"] = self._process_messages(
messages=messages, max_depth=max_depth
)
total_items = 0
for m in payload.get("messages", []) or []:
@@ -684,9 +668,13 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
f"[CustomLogger] Completed base64 strip; retained {total_items} content items"
)
return payload
def _redact_base64(self, value: Any, depth: int = 0, max_depth: int = DEFAULT_MAX_RECURSE_DEPTH_SENSITIVE_DATA_MASKER) -> Any:
def _redact_base64(
self,
value: Any,
depth: int = 0,
max_depth: int = DEFAULT_MAX_RECURSE_DEPTH_SENSITIVE_DATA_MASKER,
) -> Any:
"""Recursively redact inline base64 from any nested structure with a max recursion depth limit."""
if depth > max_depth:
verbose_logger.warning(
@@ -703,10 +691,16 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
return value
if isinstance(value, list):
return [self._redact_base64(value=v, depth=depth + 1, max_depth=max_depth) for v in value]
return [
self._redact_base64(value=v, depth=depth + 1, max_depth=max_depth)
for v in value
]
if isinstance(value, dict):
return {k: self._redact_base64(value=v, depth=depth + 1, max_depth=max_depth) for k, v in value.items()}
return {
k: self._redact_base64(value=v, depth=depth + 1, max_depth=max_depth)
for k, v in value.items()
}
return value
@@ -729,10 +723,14 @@ class CustomLogger: # https://docs.litellm.ai/docs/observability/custom_callbac
cleaned: List[Any] = []
for c in contents:
if self._should_keep_content(content=c):
cleaned.append(self._redact_base64(value=c, max_depth=max_depth))
cleaned.append(
self._redact_base64(value=c, max_depth=max_depth)
)
msg["content"] = cleaned
else:
msg["content"] = self._redact_base64(value=contents, max_depth=max_depth)
msg["content"] = self._redact_base64(
value=contents, max_depth=max_depth
)
for key, val in list(msg.items()):
if key != "content":
litellm/litellm_core_utils/audio_utils/utils.py
+96 -23
View File
@@ -4,6 +4,7 @@ Utils used for litellm.transcription() and litellm.atranscription()
import os
from dataclasses import dataclass
from typing import Optional
from litellm.types.files import get_file_mime_type_from_extension
from litellm.types.utils import FileTypes
@@ -13,12 +14,13 @@ from litellm.types.utils import FileTypes
class ProcessedAudioFile:
"""
Processed audio file data.
Attributes:
file_content: The binary content of the audio file
filename: The filename (extracted or generated)
content_type: The MIME type of the audio file
"""
file_content: bytes
filename: str
content_type: str
@@ -27,61 +29,63 @@ class ProcessedAudioFile:
def process_audio_file(audio_file: FileTypes) -> ProcessedAudioFile:
"""
Common utility function to process audio files for audio transcription APIs.
Handles various input types:
- File paths (str, os.PathLike)
- Raw bytes/bytearray
- Tuples (filename, content, optional content_type)
- File-like objects with read() method
Args:
audio_file: The audio file input in various formats
Returns:
ProcessedAudioFile: Structured data with file content, filename, and content type
Raises:
ValueError: If audio_file type is unsupported or content cannot be extracted
"""
file_content = None
filename = None
if isinstance(audio_file, (bytes, bytearray)):
# Raw bytes
filename = 'audio.wav'
filename = "audio.wav"
file_content = bytes(audio_file)
elif isinstance(audio_file, (str, os.PathLike)):
# File path or PathLike
file_path = str(audio_file)
with open(file_path, 'rb') as f:
with open(file_path, "rb") as f:
file_content = f.read()
filename = file_path.split('/')[-1]
filename = file_path.split("/")[-1]
elif isinstance(audio_file, tuple):
# Tuple format: (filename, content, content_type) or (filename, content)
if len(audio_file) >= 2:
filename = audio_file[0] or 'audio.wav'
filename = audio_file[0] or "audio.wav"
content = audio_file[1]
if isinstance(content, (bytes, bytearray)):
file_content = bytes(content)
elif isinstance(content, (str, os.PathLike)):
# File path or PathLike
with open(str(content), 'rb') as f:
with open(str(content), "rb") as f:
file_content = f.read()
elif hasattr(content, 'read'):
elif hasattr(content, "read"):
# File-like object
file_content = content.read()
if hasattr(content, 'seek'):
if hasattr(content, "seek"):
content.seek(0)
else:
raise ValueError(f"Unsupported content type in tuple: {type(content)}")
else:
raise ValueError("Tuple must have at least 2 elements: (filename, content)")
elif hasattr(audio_file, 'read') and not isinstance(audio_file, (str, bytes, bytearray, tuple, os.PathLike)):
elif hasattr(audio_file, "read") and not isinstance(
audio_file, (str, bytes, bytearray, tuple, os.PathLike)
):
# File-like object (IO) - check this after all other types
filename = getattr(audio_file, 'name', 'audio.wav')
filename = getattr(audio_file, "name", "audio.wav")
file_content = audio_file.read() # type: ignore
# Reset file pointer if possible
if hasattr(audio_file, 'seek'):
if hasattr(audio_file, "seek"):
audio_file.seek(0) # type: ignore
else:
raise ValueError(f"Unsupported audio_file type: {type(audio_file)}")
@@ -90,20 +94,18 @@ def process_audio_file(audio_file: FileTypes) -> ProcessedAudioFile:
raise ValueError("Could not extract file content from audio_file")
# Determine content type using LiteLLM's file type utilities
content_type = 'audio/wav' # Default fallback
content_type = "audio/wav" # Default fallback
if filename:
try:
# Extract extension from filename
extension = filename.split('.')[-1].lower() if '.' in filename else 'wav'
extension = filename.split(".")[-1].lower() if "." in filename else "wav"
content_type = get_file_mime_type_from_extension(extension)
except ValueError:
# If extension is not recognized, fallback to audio/wav
content_type = 'audio/wav'
content_type = "audio/wav"
return ProcessedAudioFile(
file_content=file_content,
filename=filename,
content_type=content_type
file_content=file_content, filename=filename, content_type=content_type
)
@@ -134,3 +136,74 @@ def get_audio_file_for_health_check() -> FileTypes:
pwd = os.path.dirname(os.path.realpath(__file__))
file_path = os.path.join(pwd, "audio_health_check.wav")
return open(file_path, "rb")
def calculate_request_duration(file: FileTypes) -> Optional[float]:
"""
Calculate audio duration from file content.
Args:
file: The audio file (can be file path, bytes, or file-like object)
Returns:
Duration in seconds, or None if extraction fails or soundfile is not available
"""
try:
import soundfile as sf
except ImportError:
# soundfile not available, cannot extract duration
return None
try:
import io
# Handle different file input types
file_content: Optional[bytes] = None
if isinstance(file, (bytes, bytearray)):
# Raw bytes
file_content = bytes(file)
elif isinstance(file, (str, os.PathLike)):
# File path
with open(str(file), "rb") as f:
file_content = f.read()
elif isinstance(file, tuple):
# Tuple format: (filename, content, optional content_type)
if len(file) >= 2:
content = file[1]
if isinstance(content, bytes):
file_content = content
elif hasattr(content, "read") and not isinstance(
content, (str, os.PathLike)
):
# File-like object in tuple
current_pos = getattr(content, "tell", lambda: None)()
# Seek to start to ensure we read the entire content
if hasattr(content, "seek"):
content.seek(0)
file_content = content.read()
if current_pos is not None and hasattr(content, "seek"):
content.seek(current_pos)
elif hasattr(file, "read") and not isinstance(file, tuple):
# File-like object (including BytesIO)
current_position = file.tell() if hasattr(file, "tell") else None
# Seek to start to ensure we read the entire content
if hasattr(file, "seek"):
file.seek(0)
file_content = file.read()
# Reset file position if possible
if current_position is not None and hasattr(file, "seek"):
file.seek(current_position)
if file_content is None or not isinstance(file_content, bytes):
return None
# Extract duration using soundfile
file_object = io.BytesIO(file_content)
with sf.SoundFile(file_object) as audio:
duration = len(audio) / audio.samplerate
return duration
except Exception:
# Silently fail if duration extraction fails
return None
litellm/litellm_core_utils/llm_cost_calc/utils.py
+31 -21
View File
@@ -1,7 +1,7 @@
# What is this?
## Helper utilities for cost_per_token()
from typing import Any, Literal, Optional, Tuple, TypedDict, cast
from typing import Literal, Optional, Tuple, TypedDict, cast
import litellm
from litellm._logging import verbose_logger
@@ -118,21 +118,21 @@ def _generic_cost_per_character(
def _get_service_tier_cost_key(base_key: str, service_tier: Optional[str]) -> str:
"""
Get the appropriate cost key based on service tier.
Args:
base_key: The base cost key (e.g., "input_cost_per_token")
service_tier: The service tier ("flex", "priority", or None for standard)
Returns:
str: The cost key to use (e.g., "input_cost_per_token_flex" or "input_cost_per_token")
"""
if service_tier is None:
return base_key
# Only use service tier specific keys for "flex" and "priority"
if service_tier.lower() in [ServiceTier.FLEX.value, ServiceTier.PRIORITY.value]:
return f"{base_key}_{service_tier.lower()}"
# For any other service tier, use standard pricing
return base_key
@@ -152,15 +152,15 @@ def _get_token_base_cost(
# Get service tier aware cost keys
input_cost_key = _get_service_tier_cost_key("input_cost_per_token", service_tier)
output_cost_key = _get_service_tier_cost_key("output_cost_per_token", service_tier)
cache_creation_cost_key = _get_service_tier_cost_key("cache_creation_input_token_cost", service_tier)
cache_read_cost_key = _get_service_tier_cost_key("cache_read_input_token_cost", service_tier)
prompt_base_cost = cast(
float, _get_cost_per_unit(model_info, input_cost_key)
cache_creation_cost_key = _get_service_tier_cost_key(
"cache_creation_input_token_cost", service_tier
)
completion_base_cost = cast(
float, _get_cost_per_unit(model_info, output_cost_key)
cache_read_cost_key = _get_service_tier_cost_key(
"cache_read_input_token_cost", service_tier
)
prompt_base_cost = cast(float, _get_cost_per_unit(model_info, input_cost_key))
completion_base_cost = cast(float, _get_cost_per_unit(model_info, output_cost_key))
cache_creation_cost = cast(
float, _get_cost_per_unit(model_info, cache_creation_cost_key)
)
@@ -168,9 +168,7 @@ def _get_token_base_cost(
float,
_get_cost_per_unit(model_info, "cache_creation_input_token_cost_above_1hr"),
)
cache_read_cost = cast(
float, _get_cost_per_unit(model_info, cache_read_cost_key)
)
cache_read_cost = cast(float, _get_cost_per_unit(model_info, cache_read_cost_key))
## CHECK IF ABOVE THRESHOLD
threshold: Optional[float] = None
@@ -278,7 +276,7 @@ def _get_cost_per_unit(
verbose_logger.exception(
f"litellm.litellm_core_utils.llm_cost_calc.utils.py::calculate_cost_per_component(): Exception occured - {cost_per_unit}\nDefaulting to 0.0"
)
# If the service tier key doesn't exist or is None, try to fall back to the standard key
if cost_per_unit is None:
# Check if any service tier suffix exists in the cost key using ServiceTier enum
@@ -286,7 +284,7 @@ def _get_cost_per_unit(
suffix = f"_{service_tier.value}"
if suffix in cost_key:
# Extract the base key by removing the matched suffix
base_key = cost_key.replace(suffix, '')
base_key = cost_key.replace(suffix, "")
fallback_cost = model_info.get(base_key)
if isinstance(fallback_cost, float):
return fallback_cost
@@ -300,7 +298,7 @@ def _get_cost_per_unit(
f"litellm.litellm_core_utils.llm_cost_calc.utils.py::_get_cost_per_unit(): Exception occured - {fallback_cost}\nDefaulting to 0.0"
)
break # Only try the first matching suffix
return default_value
@@ -495,7 +493,10 @@ def _calculate_input_cost(
def generic_cost_per_token(
model: str, usage: Usage, custom_llm_provider: str, service_tier: Optional[str] = None
model: str,
usage: Usage,
custom_llm_provider: str,
service_tier: Optional[str] = None,
) -> Tuple[float, float]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@@ -547,7 +548,9 @@ def generic_cost_per_token(
cache_creation_cost,
cache_creation_cost_above_1hr,
cache_read_cost,
) = _get_token_base_cost(model_info=model_info, usage=usage, service_tier=service_tier)
) = _get_token_base_cost(
model_info=model_info, usage=usage, service_tier=service_tier
)
prompt_cost = _calculate_input_cost(
prompt_tokens_details=prompt_tokens_details,
@@ -631,7 +634,7 @@ class CostCalculatorUtils:
@staticmethod
def route_image_generation_cost_calculator(
model: str,
completion_response: Any,
completion_response: ImageResponse,
custom_llm_provider: Optional[str] = None,
quality: Optional[str] = None,
n: Optional[int] = None,
@@ -658,6 +661,13 @@ class CostCalculatorUtils:
cost_calculator as vertex_ai_image_cost_calculator,
)
if size is None:
size = completion_response.size or "1024-x-1024"
if quality is None:
quality = completion_response.quality or "standard"
if n is None:
n = len(completion_response.data) if completion_response.data else 0
if custom_llm_provider == litellm.LlmProviders.VERTEX_AI.value:
if isinstance(completion_response, ImageResponse):
return vertex_ai_image_cost_calculator(
litellm/llms/azure/azure.py
+34 -17
View File
@@ -36,6 +36,7 @@ from .common_utils import (
process_azure_headers,
select_azure_base_url_or_endpoint,
)
from .image_generation import get_azure_image_generation_config
class AzureOpenAIAssistantsAPIConfig:
@@ -1011,7 +1012,7 @@ class AzureChatCompletion(BaseAzureLLM, BaseLLM):
async def aimage_generation(
self,
data: dict,
model_response: ModelResponse,
model_response: Optional[ImageResponse],
azure_client_params: dict,
api_key: str,
input: list,
@@ -1020,6 +1021,7 @@ class AzureChatCompletion(BaseAzureLLM, BaseLLM):
client=None,
timeout=None,
) -> litellm.ImageResponse:
response: Optional[dict] = None
try:
# response = await azure_client.images.generate(**data, timeout=timeout)
@@ -1052,21 +1054,38 @@ class AzureChatCompletion(BaseAzureLLM, BaseLLM):
data=data,
headers=headers,
)
response = httpx_response.json()
stringified_response = response
## LOGGING
logging_obj.post_call(
input=input,
api_key=api_key,
additional_args={"complete_input_dict": data},
original_response=stringified_response,
)
return convert_to_model_response_object( # type: ignore
response_object=stringified_response,
model_response_object=model_response,
response_type="image_generation",
provider_config = get_azure_image_generation_config(
data.get("model", "dall-e-2")
)
if provider_config is not None:
return provider_config.transform_image_generation_response(
model=data.get("model", "dall-e-2"),
raw_response=httpx_response,
model_response=model_response or ImageResponse(),
logging_obj=logging_obj,
request_data=data,
optional_params=data,
litellm_params=data,
encoding=litellm.encoding,
)
else:
response = httpx_response.json()
stringified_response = response
## LOGGING
logging_obj.post_call(
input=input,
api_key=api_key,
additional_args={"complete_input_dict": data},
original_response=stringified_response,
)
return convert_to_model_response_object( # type: ignore
response_object=stringified_response,
model_response_object=model_response,
response_type="image_generation",
)
except Exception as e:
## LOGGING
logging_obj.post_call(
@@ -1124,9 +1143,7 @@ class AzureChatCompletion(BaseAzureLLM, BaseLLM):
if api_key is None and azure_ad_token_provider is not None:
azure_ad_token = azure_ad_token_provider()
if azure_ad_token:
headers.pop(
"api-key", None
)
headers.pop("api-key", None)
headers["Authorization"] = f"Bearer {azure_ad_token}"
# init AzureOpenAI Client
litellm/llms/openai/image_generation/dall_e_2_transformation.py
+50 -1
View File
@@ -1,9 +1,16 @@
from typing import List
from typing import TYPE_CHECKING, Any, List, Optional
import httpx
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
)
from litellm.types.llms.openai import OpenAIImageGenerationOptionalParams
from litellm.types.utils import ImageResponse
from litellm.utils import convert_to_model_response_object
if TYPE_CHECKING:
from litellm.litellm_core_utils.logging import Logging as LiteLLMLoggingObj
class DallE2ImageGenerationConfig(BaseImageGenerationConfig):
@@ -36,3 +43,45 @@ class DallE2ImageGenerationConfig(BaseImageGenerationConfig):
)
return optional_params
def transform_image_generation_response(
self,
model: str,
raw_response: httpx.Response,
model_response: ImageResponse,
logging_obj: "LiteLLMLoggingObj",
request_data: dict,
optional_params: dict,
litellm_params: dict,
encoding: Any,
api_key: Optional[str] = None,
json_mode: Optional[bool] = None,
) -> ImageResponse:
response = raw_response.json()
stringified_response = response
## LOGGING
logging_obj.post_call(
input=request_data.get("prompt", ""),
api_key=api_key,
additional_args={"complete_input_dict": request_data},
original_response=stringified_response,
)
image_response: ImageResponse = convert_to_model_response_object( # type: ignore
response_object=stringified_response,
model_response_object=model_response,
response_type="image_generation",
)
# set optional params
image_response.size = optional_params.get(
"size", "1024x1024"
) # default is always 1024x1024
image_response.quality = optional_params.get(
"quality", "standard"
) # always standard for dall-e-2
image_response.output_format = optional_params.get(
"output_format", "png"
) # always png for dall-e-2
return image_response
litellm/llms/openai/image_generation/dall_e_3_transformation.py
+50 -1
View File
@@ -1,9 +1,16 @@
from typing import List
from typing import TYPE_CHECKING, Any, List, Optional
import httpx
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
)
from litellm.types.llms.openai import OpenAIImageGenerationOptionalParams
from litellm.types.utils import ImageResponse
from litellm.utils import convert_to_model_response_object
if TYPE_CHECKING:
from litellm.litellm_core_utils.logging import Logging as LiteLLMLoggingObj
class DallE3ImageGenerationConfig(BaseImageGenerationConfig):
@@ -36,3 +43,45 @@ class DallE3ImageGenerationConfig(BaseImageGenerationConfig):
)
return optional_params
def transform_image_generation_response(
self,
model: str,
raw_response: httpx.Response,
model_response: ImageResponse,
logging_obj: "LiteLLMLoggingObj",
request_data: dict,
optional_params: dict,
litellm_params: dict,
encoding: Any,
api_key: Optional[str] = None,
json_mode: Optional[bool] = None,
) -> ImageResponse:
response = raw_response.json()
stringified_response = response
## LOGGING
logging_obj.post_call(
input=request_data.get("prompt", ""),
api_key=api_key,
additional_args={"complete_input_dict": request_data},
original_response=stringified_response,
)
image_response: ImageResponse = convert_to_model_response_object( # type: ignore
response_object=stringified_response,
model_response_object=model_response,
response_type="image_generation",
)
# set optional params
image_response.size = optional_params.get(
"size", "1024x1024"
) # default is always 1024x1024
image_response.quality = optional_params.get(
"quality", "hd"
) # always hd for dall-e-3
image_response.output_format = optional_params.get(
"output_format", "png"
) # always png for dall-e-3
return image_response
litellm/llms/openai/image_generation/gpt_transformation.py
+50 -1
View File
@@ -1,9 +1,16 @@
from typing import List
from typing import TYPE_CHECKING, Any, List, Optional
import httpx
from litellm.llms.base_llm.image_generation.transformation import (
BaseImageGenerationConfig,
)
from litellm.types.llms.openai import OpenAIImageGenerationOptionalParams
from litellm.types.utils import ImageResponse
from litellm.utils import convert_to_model_response_object
if TYPE_CHECKING:
from litellm.litellm_core_utils.logging import Logging as LiteLLMLoggingObj
class GPTImageGenerationConfig(BaseImageGenerationConfig):
@@ -45,3 +52,45 @@ class GPTImageGenerationConfig(BaseImageGenerationConfig):
)
return optional_params
def transform_image_generation_response(
self,
model: str,
raw_response: httpx.Response,
model_response: ImageResponse,
logging_obj: "LiteLLMLoggingObj",
request_data: dict,
optional_params: dict,
litellm_params: dict,
encoding: Any,
api_key: Optional[str] = None,
json_mode: Optional[bool] = None,
) -> ImageResponse:
response = raw_response.json()
stringified_response = response
## LOGGING
logging_obj.post_call(
input=request_data.get("prompt", ""),
api_key=api_key,
additional_args={"complete_input_dict": request_data},
original_response=stringified_response,
)
image_response: ImageResponse = convert_to_model_response_object( # type: ignore
response_object=stringified_response,
model_response_object=model_response,
response_type="image_generation",
)
# set optional params
image_response.size = optional_params.get(
"size", "1024x1024"
) # default is always 1024x1024
image_response.quality = optional_params.get(
"quality", "high"
) # always hd for dall-e-3
image_response.output_format = optional_params.get(
"response_format", "png"
) # always png for dall-e-3
return image_response
litellm/main.py
+29 -1
View File
@@ -65,7 +65,10 @@ from litellm.constants import (
)
from litellm.exceptions import LiteLLMUnknownProvider
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.audio_utils.utils import get_audio_file_for_health_check
from litellm.litellm_core_utils.audio_utils.utils import (
calculate_request_duration,
get_audio_file_for_health_check,
)
from litellm.litellm_core_utils.dd_tracing import tracer
from litellm.litellm_core_utils.get_provider_specific_headers import (
ProviderSpecificHeaderUtils,
@@ -5406,6 +5409,7 @@ async def atranscription(*args, **kwargs) -> TranscriptionResponse:
model = args[0] if len(args) > 0 else kwargs["model"]
### PASS ARGS TO Image Generation ###
kwargs["atranscription"] = True
file = kwargs.get("file", None)
custom_llm_provider = None
try:
# Use a partial function to pass your keyword arguments
@@ -5434,6 +5438,20 @@ async def atranscription(*args, **kwargs) -> TranscriptionResponse:
raise ValueError(
f"Invalid response from transcription provider, expected TranscriptionResponse, but got {type(response)}"
)
# Calculate and add duration if response is missing it
if (
response is not None
and not isinstance(response, Coroutine)
and file is not None
):
# Check if response is missing duration
existing_duration = getattr(response, "duration", None)
if existing_duration is None:
calculated_duration = calculate_request_duration(file)
if calculated_duration is not None:
setattr(response, "duration", calculated_duration)
return response
except Exception as e:
custom_llm_provider = custom_llm_provider or "openai"
@@ -5644,6 +5662,16 @@ def transcription(
headers={},
provider_config=provider_config,
)
# Calculate and add duration if response is missing it
if response is not None and not isinstance(response, Coroutine):
# Check if response is missing duration
existing_duration = getattr(response, "duration", None)
if existing_duration is None:
calculated_duration = calculate_request_duration(file)
if calculated_duration is not None:
setattr(response, "duration", calculated_duration)
if response is None:
raise ValueError("Unmapped provider passed in. Unable to get the response.")
return response
litellm/proxy/proxy_server.py
+7 -11
View File
@@ -270,9 +270,7 @@ from litellm.proxy.management_endpoints.customer_endpoints import (
from litellm.proxy.management_endpoints.internal_user_endpoints import (
router as internal_user_router,
)
from litellm.proxy.management_endpoints.internal_user_endpoints import (
user_update,
)
from litellm.proxy.management_endpoints.internal_user_endpoints import user_update
from litellm.proxy.management_endpoints.key_management_endpoints import (
delete_verification_tokens,
duration_in_seconds,
@@ -323,9 +321,7 @@ from litellm.proxy.ocr_endpoints.endpoints import router as ocr_router
from litellm.proxy.openai_files_endpoints.files_endpoints import (
router as openai_files_router,
)
from litellm.proxy.openai_files_endpoints.files_endpoints import (
set_files_config,
)
from litellm.proxy.openai_files_endpoints.files_endpoints import set_files_config
from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
passthrough_endpoint_router,
)
@@ -412,9 +408,7 @@ from litellm.types.proxy.management_endpoints.ui_sso import (
LiteLLM_UpperboundKeyGenerateParams,
)
from litellm.types.realtime import RealtimeQueryParams
from litellm.types.router import (
DeploymentTypedDict,
)
from litellm.types.router import DeploymentTypedDict
from litellm.types.router import ModelInfo as RouterModelInfo
from litellm.types.router import (
RouterGeneralSettings,
@@ -5081,7 +5075,9 @@ async def embeddings( # noqa: PLR0915
### CALL HOOKS ### - modify incoming data / reject request before calling the model
data = await proxy_logging_obj.pre_call_hook(
user_api_key_dict=user_api_key_dict, data=data, call_type="aembedding"
user_api_key_dict=user_api_key_dict,
data=data,
call_type=CallTypes.embedding.value,
)
tasks = []
@@ -5494,7 +5490,7 @@ async def audio_transcriptions(
data = await proxy_logging_obj.pre_call_hook(
user_api_key_dict=user_api_key_dict,
data=data,
call_type="audio_transcription",
call_type="transcription",
)
## ROUTE TO CORRECT ENDPOINT ##
litellm/proxy/utils.py
+6 -63
View File
@@ -33,7 +33,7 @@ from litellm.proxy._types import (
SpendLogsPayload,
)
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.utils import CallTypes
from litellm.types.utils import CallTypes, CallTypesLiteral
try:
import backoff
@@ -796,19 +796,7 @@ class ProxyLogging:
callback: CustomGuardrail,
data: dict,
user_api_key_dict: Optional[UserAPIKeyAuth],
call_type: Literal[
"completion",
"text_completion",
"embeddings",
"aembedding",
"image_generation",
"moderation",
"audio_transcription",
"pass_through_endpoint",
"rerank",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
) -> Optional[dict]:
"""
Process a guardrail callback during pre-call hook.
@@ -866,19 +854,7 @@ class ProxyLogging:
self,
user_api_key_dict: UserAPIKeyAuth,
data: None,
call_type: Literal[
"completion",
"text_completion",
"embeddings",
"aembedding",
"image_generation",
"moderation",
"audio_transcription",
"pass_through_endpoint",
"rerank",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
) -> None:
pass
@@ -887,19 +863,7 @@ class ProxyLogging:
self,
user_api_key_dict: UserAPIKeyAuth,
data: dict,
call_type: Literal[
"completion",
"text_completion",
"embeddings",
"aembedding",
"image_generation",
"moderation",
"audio_transcription",
"pass_through_endpoint",
"rerank",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
) -> dict:
pass
@@ -907,19 +871,7 @@ class ProxyLogging:
self,
user_api_key_dict: UserAPIKeyAuth,
data: Optional[dict],
call_type: Literal[
"completion",
"text_completion",
"embeddings",
"aembedding",
"image_generation",
"moderation",
"audio_transcription",
"pass_through_endpoint",
"rerank",
"mcp_call",
"anthropic_messages",
],
call_type: CallTypesLiteral,
) -> Optional[dict]:
"""
Allows users to modify/reject the incoming request to the proxy, without having to deal with parsing Request body.
@@ -1046,16 +998,7 @@ class ProxyLogging:
self,
data: dict,
user_api_key_dict: Optional[UserAPIKeyAuth],
call_type: Literal[
"completion",
"responses",
"embeddings",
"aembedding",
"image_generation",
"moderation",
"audio_transcription",
"mcp_call",
],
call_type: CallTypesLiteral,
):
"""
Runs the CustomGuardrail's async_moderation_hook() in parallel
litellm/types/utils.py
+8 -12
View File
@@ -1,16 +1,7 @@
import json
import time
from enum import Enum
from typing import (
TYPE_CHECKING,
Any,
Dict,
List,
Literal,
Mapping,
Optional,
Union,
)
from typing import TYPE_CHECKING, Any, Dict, List, Literal, Mapping, Optional, Union
from aiohttp import FormData
from openai._models import BaseModel as OpenAIObject
@@ -299,7 +290,7 @@ class CallTypes(str, Enum):
avideo_retrieve_job = "avideo_retrieve_job"
video_delete = "video_delete"
avideo_delete = "avideo_delete"
#########################################################
# Container Call Types
#########################################################
@@ -311,7 +302,7 @@ class CallTypes(str, Enum):
aretrieve_container = "aretrieve_container"
delete_container = "delete_container"
adelete_container = "adelete_container"
acancel_fine_tuning_job = "acancel_fine_tuning_job"
cancel_fine_tuning_job = "cancel_fine_tuning_job"
alist_fine_tuning_jobs = "alist_fine_tuning_jobs"
@@ -374,6 +365,7 @@ CallTypesLiteral = Literal[
"aocr",
"avector_store_search",
"vector_store_search",
"call_mcp_tool",
]
@@ -1775,6 +1767,10 @@ class ImageResponse(OpenAIImageResponse, BaseLiteLLMOpenAIResponseObject):
total_tokens=0,
)
super().__init__(created=created, data=_data, usage=_usage) # type: ignore
self.quality = kwargs.get("quality", None)
self.output_format = kwargs.get("output_format", None)
self.size = kwargs.get("size", None)
self._hidden_params = hidden_params or {}
def __contains__(self, key):
poetry.lock
Generated
+235 -721
View File
File diff suppressed because it is too large Load Diff
pyproject.toml
+2
View File
@@ -66,6 +66,7 @@ diskcache = {version = "^5.6.1", optional = true}
polars = {version = "^1.31.0", optional = true, python = ">=3.10"}
semantic-router = {version = "*", optional = true, python = ">=3.9"}
mlflow = {version = ">3.1.4", optional = true, python = ">=3.10"}
soundfile = {version = "^0.12.1", optional = true}
[tool.poetry.extras]
proxy = [
@@ -92,6 +93,7 @@ proxy = [
"litellm-enterprise",
"rich",
"polars",
"soundfile",
]
extra_proxy = [
requirements.txt
+1
View File
@@ -58,6 +58,7 @@ tenacity==8.5.0 # for retrying requests, when litellm.num_retries set
pydantic==2.10.2 # proxy + openai req.
jsonschema==4.22.0 # validating json schema
websockets==13.1.0 # for realtime API
soundfile==0.12.1 # for audio file processing
########################
# LITELLM ENTERPRISE DEPENDENCIES
tests/test_litellm/litellm_core_utils/test_audio_utils.py
+84 -27
View File
@@ -11,6 +11,7 @@ import pytest
from litellm.litellm_core_utils.audio_utils.utils import (
ProcessedAudioFile,
calculate_request_duration,
get_audio_file_for_health_check,
get_audio_file_name,
process_audio_file,
@@ -24,7 +25,7 @@ class TestProcessAudioFile:
"""Test processing raw bytes input"""
audio_data = b"fake audio data"
result = process_audio_file(audio_data)
assert isinstance(result, ProcessedAudioFile)
assert result.file_content == audio_data
assert result.filename == "audio.wav"
@@ -34,7 +35,7 @@ class TestProcessAudioFile:
"""Test processing bytearray input"""
audio_data = bytearray(b"fake audio data")
result = process_audio_file(audio_data)
assert isinstance(result, ProcessedAudioFile)
assert result.file_content == bytes(audio_data)
assert result.filename == "audio.wav"
@@ -43,14 +44,14 @@ class TestProcessAudioFile:
def test_process_file_path_input(self):
"""Test processing file path input"""
test_content = b"test audio content"
with tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) as temp_file:
temp_file.write(test_content)
temp_file_path = temp_file.name
try:
result = process_audio_file(temp_file_path)
assert isinstance(result, ProcessedAudioFile)
assert result.file_content == test_content
assert result.filename == os.path.basename(temp_file_path)
@@ -63,9 +64,9 @@ class TestProcessAudioFile:
filename = "test.wav"
audio_data = b"fake audio data"
audio_tuple = (filename, audio_data)
result = process_audio_file(audio_tuple)
assert isinstance(result, ProcessedAudioFile)
assert result.file_content == audio_data
assert result.filename == filename
@@ -74,17 +75,17 @@ class TestProcessAudioFile:
def test_process_tuple_input_with_file_path(self):
"""Test processing tuple input with file path content"""
test_content = b"test audio content"
with tempfile.NamedTemporaryFile(suffix=".flac", delete=False) as temp_file:
temp_file.write(test_content)
temp_file_path = temp_file.name
try:
filename = "custom_name.flac"
audio_tuple = (filename, temp_file_path)
result = process_audio_file(audio_tuple)
assert isinstance(result, ProcessedAudioFile)
assert result.file_content == test_content
assert result.filename == filename
@@ -97,14 +98,14 @@ class TestProcessAudioFile:
test_content = b"test audio content"
file_obj = io.BytesIO(test_content)
file_obj.name = "test_audio.ogg"
result = process_audio_file(file_obj)
assert isinstance(result, ProcessedAudioFile)
assert result.file_content == test_content
assert result.filename == "test_audio.ogg"
assert result.content_type == "audio/ogg"
# Verify file pointer was reset
assert file_obj.tell() == 0
@@ -112,9 +113,9 @@ class TestProcessAudioFile:
"""Test processing file-like object without name attribute"""
test_content = b"test audio content"
file_obj = io.BytesIO(test_content)
result = process_audio_file(file_obj)
assert isinstance(result, ProcessedAudioFile)
assert result.file_content == test_content
assert result.filename == "audio.wav"
@@ -124,17 +125,17 @@ class TestProcessAudioFile:
"""Test processing tuple with file-like object as content"""
test_content = b"test audio content"
file_obj = io.BytesIO(test_content)
filename = "custom.mp3"
audio_tuple = (filename, file_obj)
result = process_audio_file(audio_tuple)
assert isinstance(result, ProcessedAudioFile)
assert result.file_content == test_content
assert result.filename == filename
assert result.content_type == "audio/mpeg"
# Verify file pointer was reset
assert file_obj.tell() == 0
@@ -148,7 +149,7 @@ class TestProcessAudioFile:
("test.aac", "audio/aac"),
("test.m4a", "audio/x-m4a"),
]
for filename, expected_mime_type in test_cases:
audio_tuple = (filename, b"fake content")
result = process_audio_file(audio_tuple)
@@ -158,24 +159,25 @@ class TestProcessAudioFile:
"""Test MIME type fallback for unknown file extensions"""
audio_tuple = ("test.unknown", b"fake content")
result = process_audio_file(audio_tuple)
assert result.content_type == "audio/wav" # Should fallback to default
def test_process_pathlike_object(self):
"""Test processing os.PathLike object"""
test_content = b"test audio content"
with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as temp_file:
temp_file.write(test_content)
temp_file_path = temp_file.name
try:
# Convert to pathlib.Path
from pathlib import Path
path_obj = Path(temp_file_path)
result = process_audio_file(path_obj)
assert isinstance(result, ProcessedAudioFile)
assert result.file_content == test_content
assert result.filename == os.path.basename(temp_file_path)
@@ -202,7 +204,62 @@ class TestProcessAudioFile:
"""Test tuple with None filename gets default name"""
audio_tuple = (None, b"fake content")
result = process_audio_file(audio_tuple)
assert result.filename == "audio.wav"
assert result.content_type == "audio/wav"
class TestCalculateRequestDuration:
"""Test the calculate_request_duration function"""
@pytest.mark.skipif(
os.environ.get("SKIP_AUDIO_TESTS") == "true",
reason="Skipping audio tests - soundfile may not be available",
)
def test_bytesio_at_end_position(self):
"""
Test that calculate_request_duration handles BytesIO with file pointer at end.
This reproduces and verifies the fix for the OGG file bug where BytesIO
position was at the end after a previous read(), causing "Format not recognised" error.
"""
# Create a simple WAV file in memory (44 bytes header + some data)
# This is a minimal valid WAV file
wav_header = (
b"RIFF"
+ (36 + 8).to_bytes(4, "little") # ChunkSize
+ b"WAVE"
+ b"fmt "
+ (16).to_bytes(4, "little") # Subchunk1Size
+ (1).to_bytes(2, "little") # AudioFormat (PCM)
+ (1).to_bytes(2, "little") # NumChannels
+ (16000).to_bytes(4, "little") # SampleRate
+ (32000).to_bytes(4, "little") # ByteRate
+ (2).to_bytes(2, "little") # BlockAlign
+ (16).to_bytes(2, "little") # BitsPerSample
+ b"data"
+ (8).to_bytes(4, "little") # Subchunk2Size
+ b"\x00\x00\x00\x00\x00\x00\x00\x00" # Sample data
)
# Create BytesIO object
file_obj = io.BytesIO(wav_header)
file_obj.name = "test_audio.wav"
# Simulate the bug: something reads from the file first, moving position to end
_ = file_obj.read()
assert file_obj.tell() == len(wav_header), "File position should be at end"
# Call calculate_request_duration - this would fail before the fix
duration = calculate_request_duration(file_obj)
# Verify it succeeded (returns a duration, not None)
assert (
duration is not None
), "Duration should be calculated even when BytesIO is at end"
assert isinstance(duration, float), "Duration should be a float"
assert duration > 0, "Duration should be positive"
# Verify the file position was restored
assert file_obj.tell() == len(
wav_header
), "File position should be restored to original position"
tests/test_litellm/test_cost_calculator.py
+70 -24
View File
@@ -697,7 +697,7 @@ def test_cost_discount_vertex_ai():
# Save original config
original_discount_config = litellm.cost_discount_config.copy()
# Create mock response
response = ModelResponse(
id="test-id",
@@ -705,13 +705,9 @@ def test_cost_discount_vertex_ai():
created=1234567890,
model="gemini-pro",
object="chat.completion",
usage=Usage(
prompt_tokens=100,
completion_tokens=50,
total_tokens=150
)
usage=Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150),
)
# Calculate cost without discount
litellm.cost_discount_config = {}
cost_without_discount = completion_cost(
@@ -719,24 +715,24 @@ def test_cost_discount_vertex_ai():
model="vertex_ai/gemini-pro",
custom_llm_provider="vertex_ai",
)
# Set 5% discount for vertex_ai
litellm.cost_discount_config = {"vertex_ai": 0.05}
# Calculate cost with discount
cost_with_discount = completion_cost(
completion_response=response,
model="vertex_ai/gemini-pro",
custom_llm_provider="vertex_ai",
)
# Restore original config
litellm.cost_discount_config = original_discount_config
# Verify discount is applied (5% off means 95% of original cost)
expected_cost = cost_without_discount * 0.95
assert cost_with_discount == pytest.approx(expected_cost, rel=1e-9)
print(f"✓ Cost discount test passed:")
print(f" - Original cost: ${cost_without_discount:.6f}")
print(f" - Discounted cost (5% off): ${cost_with_discount:.6f}")
@@ -752,7 +748,7 @@ def test_cost_discount_not_applied_to_other_providers():
# Save original config
original_discount_config = litellm.cost_discount_config.copy()
# Create mock response for OpenAI
response = ModelResponse(
id="test-id",
@@ -760,23 +756,19 @@ def test_cost_discount_not_applied_to_other_providers():
created=1234567890,
model="gpt-4",
object="chat.completion",
usage=Usage(
prompt_tokens=100,
completion_tokens=50,
total_tokens=150
)
usage=Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150),
)
# Set discount only for vertex_ai (not openai)
litellm.cost_discount_config = {"vertex_ai": 0.05}
# Calculate cost for OpenAI - should NOT have discount applied
cost_with_selective_discount = completion_cost(
completion_response=response,
model="gpt-4",
custom_llm_provider="openai",
)
# Clear discount config
litellm.cost_discount_config = {}
cost_without_discount = completion_cost(
@@ -784,13 +776,67 @@ def test_cost_discount_not_applied_to_other_providers():
model="gpt-4",
custom_llm_provider="openai",
)
# Restore original config
litellm.cost_discount_config = original_discount_config
# Costs should be the same (no discount applied to OpenAI)
assert cost_with_selective_discount == cost_without_discount
print(f"✓ Selective discount test passed:")
print(f" - OpenAI cost (no discount configured): ${cost_without_discount:.6f}")
print(f" - Cost remains unchanged: ${cost_with_selective_discount:.6f}")
def test_azure_image_generation_cost_calculator():
from unittest.mock import MagicMock
from litellm.types.utils import (
ImageObject,
ImageResponse,
ImageUsage,
ImageUsageInputTokensDetails,
)
response_cost_calculator_kwargs = {
"response_object": ImageResponse(
created=1761785270,
background=None,
data=[
ImageObject(
b64_json=None,
revised_prompt="A futuristic, techno-inspired green duck wearing cool modern sunglasses. The duck has a sleek, metallic appearance with glowing neon green accents, standing on a high-tech urban background with holographic billboards and illuminated city lights in the distance. The duck's feathers have a glossy, high-tech sheen, resembling a robotic design but still maintaining its avian features. The scene has a vibrant, cyberpunk aesthetic with a neon color palette.",
url="https://dalleprodsec.blob.core.windows.net/private/images/caa17dc4-357d-4257-8938-eeea9baa8d0a/generated_00.png?se=2025-10-31T00%3A47%3A59Z&sig=KHRjLz3vMahbw94JtxL02S6t2AueeRMaiqj4z35HKDM%3D&ske=2025-11-05T00%3A26%3A20Z&skoid=e52d5ed7-0657-4f62-bc12-7e5dbb260a96&sks=b&skt=2025-10-29T00%3A26%3A20Z&sktid=33e01921-4d64-4f8c-a055-5bdaffd5e33d&skv=2020-10-02&sp=r&spr=https&sr=b&sv=2020-10-02",
)
],
output_format=None,
quality="hd",
size=None,
usage=ImageUsage(
input_tokens=0,
input_tokens_details=ImageUsageInputTokensDetails(
image_tokens=0, text_tokens=0
),
output_tokens=0,
total_tokens=0,
),
),
"model": "azure/dall-e-3",
"cache_hit": False,
"custom_llm_provider": "azure",
"base_model": "azure/dall-e-3",
"call_type": "aimage_generation",
"optional_params": {},
"custom_pricing": False,
"prompt": "",
"standard_built_in_tools_params": {
"web_search_options": None,
"file_search": None,
},
"router_model_id": "6738c432ffc9b733597c6b86613ca20dc5f49bde591fd3d03e7cd6aa25bb241e",
"litellm_logging_obj": MagicMock(),
"service_tier": None,
}
cost = response_cost_calculator(**response_cost_calculator_kwargs)
assert cost > 0.079