Add service_tier based pricing support for openai

Author: Sameerlite

litellm/cost_calculator.py

@@ -148,6 +148,8 @@ def cost_per_token(  # noqa: PLR0915
     ### CALL TYPE ###
     call_type: CallTypesLiteral = "completion",
     audio_transcription_file_duration: float = 0.0,  # for audio transcription calls - the file time in seconds
+    ### SERVICE TIER ###
+    service_tier: Optional[str] = None,  # for OpenAI service tier pricing
 ) -> Tuple[float, float]:  # type: ignore
     """
     Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@@ -278,6 +280,7 @@ def cost_per_token(  # noqa: PLR0915
                 model=model_without_prefix,
                 usage=usage_block,
                 custom_llm_provider=custom_llm_provider,
+                service_tier=service_tier,
             )
 
         return prompt_cost, completion_cost
@@ -327,7 +330,7 @@ def cost_per_token(  # noqa: PLR0915
     elif custom_llm_provider == "bedrock":
         return bedrock_cost_per_token(model=model, usage=usage_block)
     elif custom_llm_provider == "openai":
-        return openai_cost_per_token(model=model, usage=usage_block)
+        return openai_cost_per_token(model=model, usage=usage_block, service_tier=service_tier)
     elif custom_llm_provider == "databricks":
         return databricks_cost_per_token(model=model, usage=usage_block)
     elif custom_llm_provider == "fireworks_ai":
@@ -606,6 +609,8 @@ def completion_cost(  # noqa: PLR0915
     litellm_model_name: Optional[str] = None,
     router_model_id: Optional[str] = None,
     litellm_logging_obj: Optional[LitellmLoggingObject] = None,
+    ### SERVICE TIER ###
+    service_tier: Optional[str] = None,  # for OpenAI service tier pricing
 ) -> float:
     """
     Calculate the cost of a given completion call fot GPT-3.5-turbo, llama2, any litellm supported llm.
@@ -658,6 +663,10 @@ def completion_cost(  # noqa: PLR0915
             completion_response=completion_response
         )
         rerank_billed_units: Optional[RerankBilledUnits] = None
+        
+        # Extract service_tier from optional_params if not provided directly
+        if service_tier is None and optional_params is not None:
+            service_tier = optional_params.get("service_tier")
 
         selected_model = _select_model_name_for_cost_calc(
             model=model,
@@ -909,6 +918,7 @@ def completion_cost(  # noqa: PLR0915
                     call_type=cast(CallTypesLiteral, call_type),
                     audio_transcription_file_duration=audio_transcription_file_duration,
                     rerank_billed_units=rerank_billed_units,
+                    service_tier=service_tier,
                 )
                 _final_cost = (
                     prompt_tokens_cost_usd_dollar + completion_tokens_cost_usd_dollar
@@ -1003,6 +1013,8 @@ def response_cost_calculator(
     litellm_model_name: Optional[str] = None,
     router_model_id: Optional[str] = None,
     litellm_logging_obj: Optional[LitellmLoggingObject] = None,
+    ### SERVICE TIER ###
+    service_tier: Optional[str] = None,  # for OpenAI service tier pricing
 ) -> float:
     """
     Returns
@@ -1036,6 +1048,7 @@ def response_cost_calculator(
                 litellm_model_name=litellm_model_name,
                 router_model_id=router_model_id,
                 litellm_logging_obj=litellm_logging_obj,
+                service_tier=service_tier,
             )
         return response_cost
     except Exception as e:

litellm/litellm_core_utils/litellm_logging.py

@@ -1228,6 +1228,7 @@ def _response_cost_calculator(
                 "standard_built_in_tools_params": self.standard_built_in_tools_params,
                 "router_model_id": router_model_id,
                 "litellm_logging_obj": self,
+                "service_tier": self.optional_params.get("service_tier") if self.optional_params else None,
             }
         except Exception as e:  # error creating kwargs for cost calculation
             debug_info = StandardLoggingModelCostFailureDebugInformation(

litellm/litellm_core_utils/llm_cost_calc/utils.py

@@ -4,14 +4,15 @@
 from typing import Any, Literal, Optional, Tuple, TypedDict, cast
 
 import litellm
-from litellm._logging import verbose_logger
+from litellm._logging import verbose_logger, verbose_proxy_logger
 from litellm.types.utils import (
     CacheCreationTokenDetails,
     CallTypes,
     ImageResponse,
     ModelInfo,
     PassthroughCallTypes,
     Usage,
+    ServiceTier,
 )
 from litellm.utils import get_model_info
 
@@ -114,8 +115,30 @@ def _generic_cost_per_character(
     return prompt_cost, completion_cost
 
 
+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
+
+
 def _get_token_base_cost(
-    model_info: ModelInfo, usage: Usage
+    model_info: ModelInfo, usage: Usage, service_tier: Optional[str] = None
 ) -> Tuple[float, float, float, float, float]:
     """
     Return prompt cost, completion cost, and cache costs for a given model and usage.
@@ -126,21 +149,27 @@ def _get_token_base_cost(
     Returns:
         Tuple[float, float, float, float] - (prompt_cost, completion_cost, cache_creation_cost, cache_read_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_per_token")
+        float, _get_cost_per_unit(model_info, input_cost_key)
     )
     completion_base_cost = cast(
-        float, _get_cost_per_unit(model_info, "output_cost_per_token")
+        float, _get_cost_per_unit(model_info, output_cost_key)
     )
     cache_creation_cost = cast(
-        float, _get_cost_per_unit(model_info, "cache_creation_input_token_cost")
+        float, _get_cost_per_unit(model_info, cache_creation_cost_key)
     )
     cache_creation_cost_above_1hr = cast(
         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_input_token_cost")
+        float, _get_cost_per_unit(model_info, cache_read_cost_key)
     )
 
     ## CHECK IF ABOVE THRESHOLD
@@ -249,6 +278,29 @@ 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
+        for service_tier in ServiceTier:
+            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, '')
+                fallback_cost = model_info.get(base_key)
+                if isinstance(fallback_cost, float):
+                    return fallback_cost
+                if isinstance(fallback_cost, int):
+                    return float(fallback_cost)
+                if isinstance(fallback_cost, str):
+                    try:
+                        return float(fallback_cost)
+                    except ValueError:
+                        verbose_logger.exception(
+                            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
 
 
@@ -443,7 +495,7 @@ def _calculate_input_cost(
 
 
 def generic_cost_per_token(
-    model: str, usage: Usage, custom_llm_provider: str
+    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.
@@ -495,7 +547,7 @@ 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)
+    ) = _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,

litellm/llms/openai/cost_calculation.py

@@ -18,7 +18,7 @@ def cost_router(call_type: CallTypes) -> Literal["cost_per_token", "cost_per_sec
         return "cost_per_token"
 
 
-def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
+def cost_per_token(model: str, usage: Usage, service_tier: Optional[str] = None) -> Tuple[float, float]:
     """
     Calculates the cost per token for a given model, prompt tokens, and completion tokens.
 
@@ -31,7 +31,7 @@ def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
     """
     ## CALCULATE INPUT COST
     return generic_cost_per_token(
-        model=model, usage=usage, custom_llm_provider="openai"
+        model=model, usage=usage, custom_llm_provider="openai", service_tier=service_tier
     )
     # ### Non-cached text tokens
     # non_cached_text_tokens = usage.prompt_tokens