fix: refactor cost tracking

This commit is contained in:
Krrish Dholakia 2026-03-23 21:19:20 -07:00
parent 51ef8a96f0
commit 01b416efb7
66 changed files with 917 additions and 378 deletions

View file

@ -23,6 +23,8 @@ from litellm.litellm_core_utils.llm_cost_calc.usage_object_transformation import
)
from litellm.litellm_core_utils.llm_cost_calc.utils import (
CostCalculatorUtils,
InputCostBreakdown,
OutputCostBreakdown,
_generic_cost_per_character,
_get_service_tier_cost_key,
_parse_prompt_tokens_details,
@ -286,7 +288,7 @@ def cost_per_token( # noqa: PLR0915
response: Optional[Any] = None,
### REQUEST MODEL ###
request_model: Optional[str] = None, # original request model for router detection
) -> Tuple[float, float]: # type: ignore
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@ -331,7 +333,7 @@ def cost_per_token( # noqa: PLR0915
)
if response_cost is not None:
return response_cost[0], response_cost[1]
return InputCostBreakdown(total=response_cost[0]), OutputCostBreakdown(total=response_cost[1])
# given
prompt_tokens_cost_usd_dollar: float = 0
@ -413,41 +415,47 @@ def cost_per_token( # noqa: PLR0915
prompt_cost = _prompt_cost
completion_cost = _completion_cost
elif cost_metric == "cost_per_token":
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=model_without_prefix,
usage=usage_block,
custom_llm_provider=custom_llm_provider,
service_tier=service_tier,
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
return prompt_cost, completion_cost
return InputCostBreakdown(total=prompt_cost), OutputCostBreakdown(total=completion_cost)
elif call_type == "arerank" or call_type == "rerank":
return rerank_cost(
_rerank_result = rerank_cost(
model=model,
custom_llm_provider=custom_llm_provider,
billed_units=rerank_billed_units,
)
return InputCostBreakdown(total=_rerank_result[0]), OutputCostBreakdown(total=_rerank_result[1])
elif call_type == "avector_store_search" or call_type == "vector_store_search":
return vector_store_search_cost(
_vs_result = vector_store_search_cost(
model=model,
custom_llm_provider=custom_llm_provider,
response=cast(VectorStoreSearchResponse, response),
)
return InputCostBreakdown(total=_vs_result[0]), OutputCostBreakdown(total=_vs_result[1])
elif call_type == "ocr" or call_type == "aocr":
return ocr_cost(
_ocr_result = ocr_cost(
model=model,
custom_llm_provider=custom_llm_provider,
response=response,
)
return InputCostBreakdown(total=_ocr_result[0]), OutputCostBreakdown(total=_ocr_result[1])
elif (
call_type == "aretrieve_batch"
or call_type == "retrieve_batch"
or call_type == CallTypes.aretrieve_batch
or call_type == CallTypes.retrieve_batch
):
return batch_cost_calculator(
_batch_result = batch_cost_calculator(
usage=usage_block, model=model, custom_llm_provider=custom_llm_provider
)
return InputCostBreakdown(total=_batch_result[0]), OutputCostBreakdown(total=_batch_result[1])
elif call_type == "atranscription" or call_type == "transcription":
if _transcription_usage_has_token_details(usage_block):
return openai_cost_per_token(
@ -456,16 +464,17 @@ def cost_per_token( # noqa: PLR0915
service_tier=service_tier,
)
return openai_cost_per_second(
_transcription_result = openai_cost_per_second(
model=model_without_prefix,
custom_llm_provider=custom_llm_provider,
duration=audio_transcription_file_duration,
)
return InputCostBreakdown(total=_transcription_result[0]), OutputCostBreakdown(total=_transcription_result[1])
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(
_search_result = search_provider_cost_per_query(
model=model,
custom_llm_provider=custom_llm_provider,
number_of_queries=number_of_queries or 1,
@ -475,6 +484,7 @@ def cost_per_token( # noqa: PLR0915
else None
),
)
return InputCostBreakdown(total=_search_result[0]), OutputCostBreakdown(total=_search_result[1])
elif custom_llm_provider == "vertex_ai":
cost_router = google_cost_router(
model=model_without_prefix,
@ -482,63 +492,94 @@ def cost_per_token( # noqa: PLR0915
call_type=call_type,
)
if cost_router == "cost_per_character":
return google_cost_per_character(
_char_result = google_cost_per_character(
model=model_without_prefix,
custom_llm_provider=custom_llm_provider,
prompt_characters=prompt_characters,
completion_characters=completion_characters,
usage=usage_block,
)
return InputCostBreakdown(total=_char_result[0]), OutputCostBreakdown(total=_char_result[1])
elif cost_router == "cost_per_token":
return google_cost_per_token(
_input_bd, _output_bd = google_cost_per_token(
model=model_without_prefix,
custom_llm_provider=custom_llm_provider,
usage=usage_block,
service_tier=service_tier,
)
return _input_bd, _output_bd
else:
raise ValueError(
f"Unknown google cost_router: {cost_router} for model={model}"
)
elif custom_llm_provider == "anthropic":
return anthropic_cost_per_token(model=model, usage=usage_block)
_input_bd, _output_bd = anthropic_cost_per_token(model=model, usage=usage_block)
return _input_bd, _output_bd
elif custom_llm_provider == "bedrock":
return bedrock_cost_per_token(
_input_bd, _output_bd = bedrock_cost_per_token(
model=model, usage=usage_block, service_tier=service_tier
)
return _input_bd, _output_bd
elif custom_llm_provider == "openai":
return openai_cost_per_token(
_input_bd, _output_bd = openai_cost_per_token(
model=model, usage=usage_block, service_tier=service_tier
)
return _input_bd, _output_bd
elif custom_llm_provider == "databricks":
return databricks_cost_per_token(model=model, usage=usage_block)
_input_bd, _output_bd = databricks_cost_per_token(
model=model, usage=usage_block
)
return _input_bd, _output_bd
elif custom_llm_provider == "fireworks_ai":
return fireworks_ai_cost_per_token(model=model, usage=usage_block)
_input_bd, _output_bd = fireworks_ai_cost_per_token(
model=model, usage=usage_block
)
return _input_bd, _output_bd
elif custom_llm_provider == "azure":
return azure_openai_cost_per_token(
_input_bd, _output_bd = azure_openai_cost_per_token(
model=model, usage=usage_block, response_time_ms=response_time_ms
)
return _input_bd, _output_bd
elif custom_llm_provider == "gemini":
return gemini_cost_per_token(
_input_bd, _output_bd = gemini_cost_per_token(
model=model, usage=usage_block, service_tier=service_tier
)
return _input_bd, _output_bd
elif custom_llm_provider == "deepseek":
return deepseek_cost_per_token(model=model, usage=usage_block)
_input_bd, _output_bd = deepseek_cost_per_token(
model=model, usage=usage_block
)
return _input_bd, _output_bd
elif custom_llm_provider == "perplexity":
return perplexity_cost_per_token(model=model, usage=usage_block)
_input_bd, _output_bd = perplexity_cost_per_token(
model=model, usage=usage_block
)
return _input_bd, _output_bd
elif custom_llm_provider == "xai":
return xai_cost_per_token(model=model, usage=usage_block)
_input_bd, _output_bd = xai_cost_per_token(model=model, usage=usage_block)
return _input_bd, _output_bd
elif custom_llm_provider == "lemonade":
return lemonade_cost_per_token(model=model, usage=usage_block)
_input_bd, _output_bd = lemonade_cost_per_token(
model=model, usage=usage_block
)
return _input_bd, _output_bd
elif custom_llm_provider == "dashscope":
from litellm.llms.dashscope.cost_calculator import (
cost_per_token as dashscope_cost_per_token,
)
return dashscope_cost_per_token(model=model, usage=usage_block)
_input_bd, _output_bd = dashscope_cost_per_token(
model=model, usage=usage_block
)
return _input_bd, _output_bd
elif custom_llm_provider == "azure_ai":
return azure_ai_cost_per_token(
_input_bd, _output_bd = azure_ai_cost_per_token(
model=model,
usage=usage_block,
response_time_ms=response_time_ms,
request_model=request_model,
)
return _input_bd, _output_bd
else:
model_info = _cached_get_model_info_helper(
model=model, custom_llm_provider=custom_llm_provider
@ -548,12 +589,13 @@ def cost_per_token( # noqa: PLR0915
model_info.get("input_cost_per_token", 0) > 0
or model_info.get("output_cost_per_token", 0) > 0
):
return generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=model,
usage=usage_block,
custom_llm_provider=custom_llm_provider,
service_tier=service_tier,
)
return _input_bd, _output_bd
if (
model_info.get("input_cost_per_second", None) is not None
@ -591,7 +633,7 @@ def cost_per_token( # noqa: PLR0915
prompt_tokens_cost_usd_dollar,
completion_tokens_cost_usd_dollar,
)
return prompt_tokens_cost_usd_dollar, completion_tokens_cost_usd_dollar
return InputCostBreakdown(total=prompt_tokens_cost_usd_dollar), OutputCostBreakdown(total=completion_tokens_cost_usd_dollar)
def get_replicate_completion_pricing(completion_response: dict, total_time=0.0):
@ -965,6 +1007,8 @@ def _store_cost_breakdown_in_logging_obj(
margin_percent: Optional[float] = None,
margin_fixed_amount: Optional[float] = None,
margin_total_amount: Optional[float] = None,
input_cost_breakdown: Optional[InputCostBreakdown] = None,
output_cost_breakdown: Optional[OutputCostBreakdown] = None,
) -> None:
"""
Helper function to store cost breakdown in the logging object.
@ -1000,6 +1044,8 @@ def _store_cost_breakdown_in_logging_obj(
margin_percent=margin_percent,
margin_fixed_amount=margin_fixed_amount,
margin_total_amount=margin_total_amount,
input_cost_breakdown=input_cost_breakdown,
output_cost_breakdown=output_cost_breakdown,
)
except Exception as breakdown_error:
@ -1489,8 +1535,8 @@ def completion_cost( # noqa: PLR0915
request_model_for_cost = litellm_logging_obj.model
(
prompt_tokens_cost_usd_dollar,
completion_tokens_cost_usd_dollar,
input_cost_breakdown,
output_cost_breakdown,
) = cost_per_token(
model=model,
prompt_tokens=prompt_tokens or 0,
@ -1512,6 +1558,8 @@ def completion_cost( # noqa: PLR0915
response=completion_response,
request_model=request_model_for_cost,
)
prompt_tokens_cost_usd_dollar = input_cost_breakdown["total"]
completion_tokens_cost_usd_dollar = output_cost_breakdown["total"]
# Get additional costs from provider (e.g., routing fees, infrastructure costs)
if custom_llm_provider == "azure_ai":
@ -1602,6 +1650,8 @@ def completion_cost( # noqa: PLR0915
margin_percent=margin_percent,
margin_fixed_amount=margin_fixed_amount,
margin_total_amount=margin_total_amount,
input_cost_breakdown=input_cost_breakdown,
output_cost_breakdown=output_cost_breakdown,
)
return _final_cost
@ -2279,15 +2329,15 @@ def handle_realtime_stream_cost_calculation(
try:
if model_name is None:
continue
_input_cost_per_token, _output_cost_per_token = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=model_name,
usage=combined_usage_object,
custom_llm_provider=custom_llm_provider,
)
except Exception:
continue
input_cost_per_token += _input_cost_per_token
output_cost_per_token += _output_cost_per_token
input_cost_per_token += _input_bd["total"]
output_cost_per_token += _output_bd["total"]
break # exit if we find a valid model
total_cost = input_cost_per_token + output_cost_per_token

View file

@ -91,6 +91,10 @@ from litellm.types.llms.openai import (
from litellm.types.mcp import MCPPostCallResponseObject
from litellm.types.prompts.init_prompts import PromptSpec
from litellm.types.rerank import RerankResponse
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
)
from litellm.types.utils import (
CachingDetails,
CallTypes,
@ -1371,6 +1375,8 @@ class Logging(LiteLLMLoggingBaseClass):
margin_percent: Optional[float] = None,
margin_fixed_amount: Optional[float] = None,
margin_total_amount: Optional[float] = None,
input_cost_breakdown: Optional[InputCostBreakdown] = None,
output_cost_breakdown: Optional[OutputCostBreakdown] = None,
) -> None:
"""
Helper method to store cost breakdown in the logging object.
@ -1387,6 +1393,8 @@ class Logging(LiteLLMLoggingBaseClass):
margin_percent: Margin percentage applied (0.10 = 10%)
margin_fixed_amount: Fixed margin amount in USD
margin_total_amount: Total margin added in USD
input_cost_breakdown: Granular input cost components
output_cost_breakdown: Granular output cost components
"""
self.cost_breakdown = CostBreakdown(
@ -1420,6 +1428,11 @@ class Logging(LiteLLMLoggingBaseClass):
if margin_total_amount is not None:
self.cost_breakdown["margin_total_amount"] = margin_total_amount
if input_cost_breakdown is not None:
self.cost_breakdown["input_cost_breakdown"] = input_cost_breakdown # type: ignore
if output_cost_breakdown is not None:
self.cost_breakdown["output_cost_breakdown"] = output_cost_breakdown # type: ignore
def _response_cost_calculator(
self,
result: Union[

View file

@ -422,6 +422,33 @@ def calculate_cache_writing_cost(
return total_cost
class _InputCostBreakdownRequired(TypedDict):
total: float
class InputCostBreakdown(_InputCostBreakdownRequired, total=False):
"""Granular breakdown of input/prompt token costs."""
text_cost: float
cache_read_cost: float
cache_creation_cost: float
audio_cost: float
image_cost: float
class _OutputCostBreakdownRequired(TypedDict):
total: float
class OutputCostBreakdown(_OutputCostBreakdownRequired, total=False):
"""Granular breakdown of output/completion token costs."""
text_cost: float
reasoning_cost: float
audio_cost: float
image_cost: float
class PromptTokensDetailsResult(TypedDict):
cache_hit_tokens: int
cache_creation_tokens: int
@ -549,41 +576,46 @@ def _calculate_input_cost(
cache_creation_cost: float,
cache_creation_cost_above_1hr: float,
service_tier: Optional[str] = None,
) -> float:
) -> InputCostBreakdown:
"""
Calculates the input cost for a given model, prompt tokens, and completion tokens.
Returns an InputCostBreakdown with per-component costs and a total.
"""
prompt_cost = float(prompt_tokens_details["text_tokens"]) * prompt_base_cost
breakdown = InputCostBreakdown(total=0.0)
### CACHE READ COST - Now uses tiered pricing
prompt_cost += float(prompt_tokens_details["cache_hit_tokens"]) * cache_read_cost
text_cost = float(prompt_tokens_details["text_tokens"]) * prompt_base_cost
breakdown["text_cost"] = text_cost
### AUDIO COST
cache_rd_cost = float(prompt_tokens_details["cache_hit_tokens"]) * cache_read_cost
breakdown["cache_read_cost"] = cache_rd_cost
audio_cost_val = 0.0
if prompt_tokens_details["audio_tokens"]:
audio_cost_key = _get_service_tier_cost_key(
"input_cost_per_audio_token", service_tier
)
prompt_cost += calculate_cost_component(
audio_cost_val = calculate_cost_component(
model_info, audio_cost_key, prompt_tokens_details["audio_tokens"]
)
breakdown["audio_cost"] = audio_cost_val
### IMAGE TOKEN COST
image_cost_val = 0.0
if prompt_tokens_details["image_tokens"]:
# For image token costs:
# First check if input_cost_per_image_token is available. If not, default to generic input_cost_per_token.
image_token_cost_key = "input_cost_per_image_token"
if model_info.get(image_token_cost_key) is None:
image_token_cost_key = "input_cost_per_token"
prompt_cost += calculate_cost_component(
image_cost_val = calculate_cost_component(
model_info, image_token_cost_key, prompt_tokens_details["image_tokens"]
)
breakdown["image_cost"] = image_cost_val
### CACHE WRITING COST - Now uses tiered pricing
cache_cr_cost = 0.0
if (
prompt_tokens_details["cache_creation_tokens"]
or prompt_tokens_details["cache_creation_token_details"] is not None
):
prompt_cost += calculate_cache_writing_cost(
cache_cr_cost = calculate_cache_writing_cost(
cache_creation_tokens=prompt_tokens_details["cache_creation_tokens"],
cache_creation_token_details=prompt_tokens_details[
"cache_creation_token_details"
@ -591,30 +623,37 @@ def _calculate_input_cost(
cache_creation_cost_above_1hr=cache_creation_cost_above_1hr,
cache_creation_cost=cache_creation_cost,
)
breakdown["cache_creation_cost"] = cache_cr_cost
### CHARACTER COST
extra_cost = 0.0
if prompt_tokens_details["character_count"]:
prompt_cost += calculate_cost_component(
extra_cost += calculate_cost_component(
model_info,
"input_cost_per_character",
prompt_tokens_details["character_count"],
)
### IMAGE COUNT COST
if prompt_tokens_details["image_count"]:
prompt_cost += calculate_cost_component(
extra_cost += calculate_cost_component(
model_info, "input_cost_per_image", prompt_tokens_details["image_count"]
)
### VIDEO LENGTH COST
if prompt_tokens_details["video_length_seconds"]:
prompt_cost += calculate_cost_component(
extra_cost += calculate_cost_component(
model_info,
"input_cost_per_video_per_second",
prompt_tokens_details["video_length_seconds"],
)
# Fold character/image-count/video costs into text_cost (they are non-token input costs)
breakdown["text_cost"] = breakdown.get("text_cost", 0.0) + extra_cost
return prompt_cost
breakdown["total"] = (
breakdown.get("text_cost", 0.0)
+ breakdown.get("cache_read_cost", 0.0)
+ breakdown.get("cache_creation_cost", 0.0)
+ breakdown.get("audio_cost", 0.0)
+ breakdown.get("image_cost", 0.0)
)
return breakdown
def generic_cost_per_token( # noqa: PLR0915
@ -622,7 +661,7 @@ def generic_cost_per_token( # noqa: PLR0915
usage: Usage,
custom_llm_provider: str,
service_tier: Optional[str] = None,
) -> Tuple[float, float]:
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@ -633,7 +672,7 @@ def generic_cost_per_token( # noqa: PLR0915
- usage: LiteLLM Usage block, containing anthropic caching information
Returns:
Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
Tuple[InputCostBreakdown, OutputCostBreakdown] - granular input and output cost breakdowns
"""
## GET MODEL INFO
@ -641,7 +680,6 @@ def generic_cost_per_token( # noqa: PLR0915
## CALCULATE INPUT COST
### Cost of processing (non-cache hit + cache hit) + Cost of cache-writing (cache writing)
prompt_cost = 0.0
### PROCESSING COST
prompt_tokens_details = PromptTokensDetailsResult(
cache_hit_tokens=0,
@ -696,7 +734,7 @@ def generic_cost_per_token( # noqa: PLR0915
model_info=model_info, usage=usage, service_tier=service_tier
)
prompt_cost = _calculate_input_cost(
input_breakdown = _calculate_input_cost(
prompt_tokens_details=prompt_tokens_details,
model_info=model_info,
prompt_base_cost=prompt_base_cost,
@ -719,15 +757,9 @@ def generic_cost_per_token( # noqa: PLR0915
reasoning_tokens = completion_tokens_details["reasoning_tokens"]
image_tokens = completion_tokens_details["image_tokens"]
# Handle text_tokens calculation:
# 1. If text_tokens is explicitly provided and > 0, use it
# 2. If there's a breakdown (reasoning/audio/image tokens), calculate text_tokens as the remainder
# 3. If no breakdown at all, assume all completion_tokens are text_tokens
has_token_breakdown = image_tokens > 0 or audio_tokens > 0 or reasoning_tokens > 0
if text_tokens == 0:
if has_token_breakdown:
# Calculate text tokens as remainder when we have a breakdown
# This handles cases like OpenAI's reasoning models where text_tokens isn't provided
text_tokens = max(
0,
usage.completion_tokens
@ -736,13 +768,17 @@ def generic_cost_per_token( # noqa: PLR0915
- image_tokens,
)
else:
# No breakdown at all, all tokens are text tokens
text_tokens = usage.completion_tokens
is_text_tokens_total = True
output_breakdown = OutputCostBreakdown(total=0.0)
## TEXT COST
completion_cost = float(text_tokens) * completion_base_cost
output_text_cost = float(text_tokens) * completion_base_cost
output_breakdown["text_cost"] = output_text_cost
## AUDIO COST
output_audio_cost = 0.0
if not is_text_tokens_total and audio_tokens is not None and audio_tokens > 0:
_output_cost_per_audio_token = _get_cost_per_unit(
model_info, "output_cost_per_audio_token", None
@ -752,9 +788,11 @@ def generic_cost_per_token( # noqa: PLR0915
if _output_cost_per_audio_token is not None
else completion_base_cost
)
completion_cost += float(audio_tokens) * _output_cost_per_audio_token
output_audio_cost = float(audio_tokens) * _output_cost_per_audio_token
output_breakdown["audio_cost"] = output_audio_cost
## REASONING COST
output_reasoning_cost = 0.0
if not is_text_tokens_total and reasoning_tokens and reasoning_tokens > 0:
_output_cost_per_reasoning_token = _get_cost_per_unit(
model_info, "output_cost_per_reasoning_token", None
@ -764,9 +802,13 @@ def generic_cost_per_token( # noqa: PLR0915
if _output_cost_per_reasoning_token is not None
else completion_base_cost
)
completion_cost += float(reasoning_tokens) * _output_cost_per_reasoning_token
output_reasoning_cost = (
float(reasoning_tokens) * _output_cost_per_reasoning_token
)
output_breakdown["reasoning_cost"] = output_reasoning_cost
## IMAGE COST
output_image_cost = 0.0
if not is_text_tokens_total and image_tokens and image_tokens > 0:
_output_cost_per_image_token = _get_cost_per_unit(
model_info, "output_cost_per_image_token", None
@ -776,9 +818,14 @@ def generic_cost_per_token( # noqa: PLR0915
if _output_cost_per_image_token is not None
else completion_base_cost
)
completion_cost += float(image_tokens) * _output_cost_per_image_token
output_image_cost = float(image_tokens) * _output_cost_per_image_token
output_breakdown["image_cost"] = output_image_cost
return prompt_cost, completion_cost
output_breakdown["total"] = (
output_text_cost + output_audio_cost + output_reasoning_cost + output_image_cost
)
return input_breakdown, output_breakdown
def calculate_image_response_cost_from_usage(
@ -831,12 +878,12 @@ def calculate_image_response_cost_from_usage(
),
)
prompt_cost, completion_cost = generic_cost_per_token(
input_breakdown, output_breakdown = generic_cost_per_token(
model=model,
usage=normalized_usage,
custom_llm_provider=custom_llm_provider,
)
return prompt_cost + completion_cost
return input_breakdown["total"] + output_breakdown["total"]
class CostCalculatorUtils:

View file

@ -5,13 +5,19 @@ Helper util for handling amazon nova cost calculation
from typing import TYPE_CHECKING, Tuple
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
generic_cost_per_token,
)
if TYPE_CHECKING:
from litellm.types.utils import Usage
def cost_per_token(model: str, usage: "Usage") -> Tuple[float, float]:
def cost_per_token(
model: str, usage: "Usage"
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
Follows the same logic as Anthropic's cost per token calculation.

View file

@ -6,9 +6,8 @@ Helper util for handling anthropic-specific cost calculation
from typing import TYPE_CHECKING, Optional, Tuple
from litellm.litellm_core_utils.llm_cost_calc.utils import (
_get_token_base_cost,
_parse_prompt_tokens_details,
calculate_cache_writing_cost,
InputCostBreakdown,
OutputCostBreakdown,
generic_cost_per_token,
)
@ -17,45 +16,9 @@ if TYPE_CHECKING:
import litellm
def _compute_cache_only_cost(model_info: "ModelInfo", usage: "Usage") -> float:
"""
Return only the cache-related portion of the prompt cost (cache read + cache write).
These costs must NOT be scaled by geo/speed multipliers because the old
explicit ``fast/`` model entries carried unchanged cache rates while
multiplying only the regular input/output token costs.
"""
if usage.prompt_tokens_details is None:
return 0.0
prompt_tokens_details = _parse_prompt_tokens_details(usage)
(
_,
_,
cache_creation_cost,
cache_creation_cost_above_1hr,
cache_read_cost,
) = _get_token_base_cost(model_info=model_info, usage=usage)
cache_cost = float(prompt_tokens_details["cache_hit_tokens"]) * cache_read_cost
if (
prompt_tokens_details["cache_creation_tokens"]
or prompt_tokens_details["cache_creation_token_details"] is not None
):
cache_cost += calculate_cache_writing_cost(
cache_creation_tokens=prompt_tokens_details["cache_creation_tokens"],
cache_creation_token_details=prompt_tokens_details[
"cache_creation_token_details"
],
cache_creation_cost_above_1hr=cache_creation_cost_above_1hr,
cache_creation_cost=cache_creation_cost,
)
return cache_cost
def cost_per_token(model: str, usage: "Usage") -> Tuple[float, float]:
def cost_per_token(
model: str, usage: "Usage"
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@ -64,13 +27,12 @@ def cost_per_token(model: str, usage: "Usage") -> Tuple[float, float]:
- usage: LiteLLM Usage block, containing anthropic caching information
Returns:
Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
Tuple[InputCostBreakdown, OutputCostBreakdown] - granular input and output cost breakdowns
"""
prompt_cost, completion_cost = generic_cost_per_token(
input_bd, output_bd = generic_cost_per_token(
model=model, usage=usage, custom_llm_provider="anthropic"
)
# Apply provider_specific_entry multipliers for geo/speed routing
try:
model_info = litellm.get_model_info(
model=model, custom_llm_provider="anthropic"
@ -78,23 +40,46 @@ def cost_per_token(model: str, usage: "Usage") -> Tuple[float, float]:
provider_specific_entry: dict = model_info.get("provider_specific_entry") or {}
multiplier = 1.0
inference_geo = getattr(usage, "inference_geo", None)
if (
hasattr(usage, "inference_geo")
and usage.inference_geo
and usage.inference_geo.lower() not in ["global", "not_available"]
inference_geo
and inference_geo.lower() not in ["global", "not_available"]
):
multiplier *= provider_specific_entry.get(usage.inference_geo.lower(), 1.0)
if hasattr(usage, "speed") and usage.speed == "fast":
multiplier *= provider_specific_entry.get(inference_geo.lower(), 1.0)
speed = getattr(usage, "speed", None)
if speed == "fast":
multiplier *= provider_specific_entry.get("fast", 1.0)
if multiplier != 1.0:
cache_cost = _compute_cache_only_cost(model_info=model_info, usage=usage)
prompt_cost = (prompt_cost - cache_cost) * multiplier + cache_cost
completion_cost *= multiplier
# Multiply text/audio/image input costs; leave cache costs unchanged
input_bd["text_cost"] = input_bd.get("text_cost", 0.0) * multiplier
input_bd["audio_cost"] = input_bd.get("audio_cost", 0.0) * multiplier
input_bd["image_cost"] = input_bd.get("image_cost", 0.0) * multiplier
input_bd["total"] = (
input_bd.get("text_cost", 0.0)
+ input_bd.get("cache_read_cost", 0.0)
+ input_bd.get("cache_creation_cost", 0.0)
+ input_bd.get("audio_cost", 0.0)
+ input_bd.get("image_cost", 0.0)
)
# Multiply all output cost components
output_bd["text_cost"] = output_bd.get("text_cost", 0.0) * multiplier
output_bd["reasoning_cost"] = (
output_bd.get("reasoning_cost", 0.0) * multiplier
)
output_bd["audio_cost"] = output_bd.get("audio_cost", 0.0) * multiplier
output_bd["image_cost"] = output_bd.get("image_cost", 0.0) * multiplier
output_bd["total"] = (
output_bd.get("text_cost", 0.0)
+ output_bd.get("reasoning_cost", 0.0)
+ output_bd.get("audio_cost", 0.0)
+ output_bd.get("image_cost", 0.0)
)
except Exception:
pass
return prompt_cost, completion_cost
return input_bd, output_bd
def get_cost_for_anthropic_web_search(

View file

@ -6,14 +6,18 @@ Helper util for handling azure openai-specific cost calculation
from typing import Optional, Tuple
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
generic_cost_per_token,
)
from litellm.types.utils import Usage
from litellm.utils import get_model_info
def cost_per_token(
model: str, usage: Usage, response_time_ms: Optional[float] = 0.0
) -> Tuple[float, float]:
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@ -22,7 +26,7 @@ def cost_per_token(
- usage: LiteLLM Usage block, containing caching and audio token information
Returns:
Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
Tuple[InputCostBreakdown, OutputCostBreakdown] - granular input and output cost breakdowns
"""
## GET MODEL INFO
model_info = get_model_info(model=model, custom_llm_provider="azure")
@ -36,13 +40,12 @@ def cost_per_token(
verbose_logger.debug(
f"For model={model} - output_cost_per_second: {model_info.get('output_cost_per_second')}; response time: {response_time_ms}"
)
## COST PER SECOND ##
prompt_cost = 0.0
completion_cost = model_info["output_cost_per_second"] * response_time_ms / 1000
return prompt_cost, completion_cost
return (
InputCostBreakdown(total=0.0),
OutputCostBreakdown(total=completion_cost, text_cost=completion_cost),
)
## Use generic cost calculator for all other cases
## This properly handles: text tokens, audio tokens, cached tokens, reasoning tokens, etc.
return generic_cost_per_token(
model=model,
usage=usage,

View file

@ -6,7 +6,11 @@ Handles Azure AI Foundry Model Router flat cost and other Azure AI specific pric
from typing import Optional, Tuple
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
generic_cost_per_token,
)
from litellm.types.utils import Usage
from litellm.utils import get_model_info
@ -65,7 +69,7 @@ def cost_per_token(
usage: Usage,
response_time_ms: Optional[float] = 0.0,
request_model: Optional[str] = None,
) -> Tuple[float, float]:
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculate the cost per token for Azure AI models.
@ -80,44 +84,33 @@ def cost_per_token(
request_model: Optional[str], the original request model name (to detect router usage)
Returns:
Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
Tuple[InputCostBreakdown, OutputCostBreakdown] - granular input and output cost breakdowns
Raises:
ValueError: If the model is not found in the cost map and cost cannot be calculated
(except for Model Router models where we return just the routing flat cost)
"""
prompt_cost = 0.0
completion_cost = 0.0
input_bd = InputCostBreakdown(total=0.0)
output_bd = OutputCostBreakdown(total=0.0)
# Determine if this was a model router request
# Check both the response model and the request model
is_router_request = _is_azure_model_router(model) or (
request_model is not None and _is_azure_model_router(request_model)
)
# Calculate base cost using generic cost calculator
# This may raise an exception if the model is not in the cost map
try:
prompt_cost, completion_cost = generic_cost_per_token(
input_bd, output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider="azure_ai",
)
except Exception as e:
# For Model Router, the model name (e.g., "azure-model-router") may not be in the cost map
# because it's a routing service, not an actual model. In this case, we continue
# to calculate just the routing flat cost.
if not _is_azure_model_router(model):
# Re-raise for non-router models - they should have pricing defined
raise
verbose_logger.debug(
f"Azure AI Model Router: model '{model}' not in cost map, calculating routing flat cost only. Error: {e}"
)
# Add flat cost for Azure Model Router
# The flat cost is defined in model_prices_and_context_window.json for azure_ai/model_router
if is_router_request:
# Use the request model for flat cost calculation if available, otherwise use response model
router_model_for_calc = request_model if request_model else model
router_flat_cost = calculate_azure_model_router_flat_cost(
router_model_for_calc, usage.prompt_tokens
@ -129,7 +122,7 @@ def cost_per_token(
f"({usage.prompt_tokens} tokens × ${router_flat_cost / usage.prompt_tokens:.9f}/token)"
)
# Add flat cost to prompt cost
prompt_cost += router_flat_cost
input_bd["text_cost"] = input_bd.get("text_cost", 0.0) + router_flat_cost
input_bd["total"] = input_bd.get("total", 0.0) + router_flat_cost
return prompt_cost, completion_cost
return input_bd, output_bd

View file

@ -5,7 +5,11 @@ Helper util for handling bedrock-specific cost calculation
from typing import TYPE_CHECKING, Optional, Tuple
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
generic_cost_per_token,
)
if TYPE_CHECKING:
from litellm.types.utils import Usage
@ -13,7 +17,7 @@ if TYPE_CHECKING:
def cost_per_token(
model: str, usage: "Usage", service_tier: Optional[str] = None
) -> Tuple[float, float]:
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.

View file

@ -7,6 +7,10 @@ Handles tiered pricing and prompt caching scenarios.
from dataclasses import dataclass
from typing import List, Optional, Tuple
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
)
from litellm.types.utils import ModelInfo, Usage
from litellm.utils import get_model_info
@ -124,8 +128,8 @@ def _calculate_prompt_cost(
breakdown: TokenBreakdown,
model_info: ModelInfo,
tiered_pricing: Optional[List[dict]],
) -> float:
"""Calculate total prompt cost including cached tokens."""
) -> Tuple[float, float]:
"""Calculate prompt text cost and cache read cost separately."""
if tiered_pricing:
text_cost = _calculate_tiered_cost(
tokens=breakdown.text_tokens,
@ -138,28 +142,27 @@ def _calculate_prompt_cost(
cost_key="cache_read_input_token_cost",
fallback_cost_key="input_cost_per_token",
)
return text_cost + cache_cost
return text_cost, cache_cost
input_cost = float(model_info.get("input_cost_per_token") or 0.0)
# For cache_cost, first try the specific key, then fall back to input_cost.
cache_cost_val = model_info.get("cache_read_input_token_cost")
if cache_cost_val is None:
cache_cost = input_cost
cache_rate = input_cost
else:
cache_cost = float(cache_cost_val)
cache_rate = float(cache_cost_val)
return (breakdown.text_tokens * input_cost) + (breakdown.cached_tokens * cache_cost)
return (breakdown.text_tokens * input_cost), (breakdown.cached_tokens * cache_rate)
def _calculate_completion_cost(
breakdown: TokenBreakdown,
model_info: ModelInfo,
tiered_pricing: Optional[List[dict]],
) -> float:
"""Calculate total completion cost including reasoning tokens."""
) -> Tuple[float, float]:
"""Calculate completion text cost and reasoning cost separately."""
if tiered_pricing:
completion_cost = _calculate_tiered_cost(
text_cost = _calculate_tiered_cost(
tokens=breakdown.completion_tokens,
tiered_pricing=tiered_pricing,
cost_key="output_cost_per_token",
@ -170,23 +173,24 @@ def _calculate_completion_cost(
cost_key="output_cost_per_reasoning_token",
fallback_cost_key="output_cost_per_token",
)
return completion_cost + reasoning_cost
return text_cost, reasoning_cost
output_cost = float(model_info.get("output_cost_per_token") or 0.0)
# For reasoning_cost, first try the specific key, then fall back to output_cost.
reasoning_cost_val = model_info.get("output_cost_per_reasoning_token")
if reasoning_cost_val is None:
reasoning_cost = output_cost
reasoning_rate = output_cost
else:
reasoning_cost = float(reasoning_cost_val)
reasoning_rate = float(reasoning_cost_val)
return (breakdown.completion_tokens * output_cost) + (
breakdown.reasoning_tokens * reasoning_cost
return (breakdown.completion_tokens * output_cost), (
breakdown.reasoning_tokens * reasoning_rate
)
def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
def cost_per_token(
model: str, usage: Usage
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculate cost per token for Dashscope models.
@ -197,7 +201,7 @@ def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
usage: LiteLLM Usage block
Returns:
Tuple[float, float] - (prompt_cost_in_usd, completion_cost_in_usd)
Tuple[InputCostBreakdown, OutputCostBreakdown] - granular input and output cost breakdowns
"""
model_info = get_model_info(model=model, custom_llm_provider="dashscope")
breakdown = _extract_token_breakdown(usage)
@ -207,11 +211,22 @@ def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
else None
)
prompt_cost = _calculate_prompt_cost(
text_cost, cache_read_cost = _calculate_prompt_cost(
breakdown=breakdown, model_info=model_info, tiered_pricing=tiered_pricing
)
completion_cost = _calculate_completion_cost(
output_text_cost, reasoning_cost = _calculate_completion_cost(
breakdown=breakdown, model_info=model_info, tiered_pricing=tiered_pricing
)
return prompt_cost, completion_cost
return (
InputCostBreakdown(
total=text_cost + cache_read_cost,
text_cost=text_cost,
cache_read_cost=cache_read_cost,
),
OutputCostBreakdown(
total=output_text_cost + reasoning_cost,
text_cost=output_text_cost,
reasoning_cost=reasoning_cost,
),
)

View file

@ -5,20 +5,26 @@ Helper util for handling databricks-specific cost calculation
from typing import Tuple
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
)
from litellm.types.utils import Usage
from litellm.utils import get_model_info
def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
def cost_per_token(
model: str, usage: Usage
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
Input:
- model: str, the model name without provider prefix
- usage: LiteLLM Usage block, containing anthropic caching information
- usage: LiteLLM Usage block
Returns:
Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
Tuple[InputCostBreakdown, OutputCostBreakdown] - granular input and output cost breakdowns
"""
base_model = model
if model.startswith("databricks/dbrx-instruct") or model.startswith(
@ -53,14 +59,15 @@ def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
"llama-2-70b-chat"
):
base_model = "databricks-llama-2-70b-chat"
## GET MODEL INFO
model_info = get_model_info(model=base_model, custom_llm_provider="databricks")
## CALCULATE INPUT COST
prompt_cost: float = usage["prompt_tokens"] * model_info["input_cost_per_token"]
completion_cost: float = (
usage["completion_tokens"] * model_info["output_cost_per_token"]
)
## CALCULATE OUTPUT COST
completion_cost = usage["completion_tokens"] * model_info["output_cost_per_token"]
return prompt_cost, completion_cost
return (
InputCostBreakdown(total=prompt_cost, text_cost=prompt_cost),
OutputCostBreakdown(total=completion_cost, text_cost=completion_cost),
)

View file

@ -6,11 +6,17 @@ Handles prompt caching scenario.
from typing import Tuple
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
generic_cost_per_token,
)
from litellm.types.utils import Usage
def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
def cost_per_token(
model: str, usage: Usage
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.

View file

@ -10,6 +10,10 @@ from litellm.constants import (
FIREWORKS_AI_56_B_MOE,
FIREWORKS_AI_176_B_MOE,
)
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
)
from litellm.types.utils import Usage
from litellm.utils import get_model_info
@ -54,33 +58,33 @@ def get_base_model_for_pricing(model_name: str) -> str:
return "fireworks-ai-default"
def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
def cost_per_token(
model: str, usage: Usage
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
Input:
- model: str, the model name without provider prefix
- usage: LiteLLM Usage block, containing anthropic caching information
- usage: LiteLLM Usage block
Returns:
Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
Tuple[InputCostBreakdown, OutputCostBreakdown] - granular input and output cost breakdowns
"""
## check if model mapped, else use default pricing
try:
model_info = get_model_info(model=model, custom_llm_provider="fireworks_ai")
except Exception:
base_model = get_base_model_for_pricing(model_name=model)
## GET MODEL INFO
model_info = get_model_info(
model=base_model, custom_llm_provider="fireworks_ai"
)
## CALCULATE INPUT COST
prompt_cost: float = usage["prompt_tokens"] * model_info["input_cost_per_token"]
completion_cost: float = (
usage["completion_tokens"] * model_info["output_cost_per_token"]
)
## CALCULATE OUTPUT COST
completion_cost = usage["completion_tokens"] * model_info["output_cost_per_token"]
return prompt_cost, completion_cost
return (
InputCostBreakdown(total=prompt_cost, text_cost=prompt_cost),
OutputCostBreakdown(total=completion_cost, text_cost=completion_cost),
)

View file

@ -7,12 +7,16 @@ Handles the context caching for Gemini API.
from typing import TYPE_CHECKING, Optional, Tuple
if TYPE_CHECKING:
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
)
from litellm.types.utils import ModelInfo, Usage
def cost_per_token(
model: str, usage: "Usage", service_tier: Optional[str] = None
) -> Tuple[float, float]:
) -> Tuple["InputCostBreakdown", "OutputCostBreakdown"]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.

View file

@ -6,18 +6,22 @@ This prevents cost calculation errors when using models not in model_prices_and_
"""
from typing import Tuple
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
)
from litellm.types.utils import Usage
def cost_per_token(
model: str,
usage: Usage,
) -> Tuple[float, float]:
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculate cost per token for Lemonade models.
Since Lemonade is a local/self-hosted deployment, there are no per-token costs.
This function returns (0.0, 0.0) for all models to allow cost tracking to work
Returns zero-cost breakdowns for all models to allow cost tracking to work
without errors for any Lemonade model, regardless of whether it's in the
model_prices_and_context_window.json file.
@ -26,10 +30,6 @@ def cost_per_token(
usage: Usage object containing token counts
Returns:
Tuple of (prompt_cost, completion_cost) - always (0.0, 0.0) for Lemonade
Tuple of (InputCostBreakdown, OutputCostBreakdown) - always zero for Lemonade
"""
# Lemonade is self-hosted/local, so cost is always 0
prompt_cost = 0.0
completion_cost = 0.0
return prompt_cost, completion_cost
return InputCostBreakdown(total=0.0), OutputCostBreakdown(total=0.0)

View file

@ -6,7 +6,11 @@ Helper util for handling openai-specific cost calculation
from typing import Literal, Optional, Tuple
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
generic_cost_per_token,
)
from litellm.types.utils import CallTypes, ModelInfo, Usage
from litellm.utils import get_model_info
@ -20,7 +24,7 @@ def cost_router(call_type: CallTypes) -> Literal["cost_per_token", "cost_per_sec
def cost_per_token(
model: str, usage: Usage, service_tier: Optional[str] = None
) -> Tuple[float, float]:
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@ -29,9 +33,8 @@ def cost_per_token(
- usage: LiteLLM Usage block, containing anthropic caching information
Returns:
Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
Tuple[InputCostBreakdown, OutputCostBreakdown] - granular input and output cost breakdowns
"""
## CALCULATE INPUT COST
return generic_cost_per_token(
model=model,
usage=usage,

View file

@ -4,7 +4,7 @@ Cost calculator for OpenAI image generation models (gpt-image-1, gpt-image-1-min
These models use token-based pricing instead of pixel-based pricing like DALL-E.
"""
from typing import Optional
from typing import Any, Optional, cast
from litellm import verbose_logger
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
@ -48,21 +48,21 @@ def cost_calculator(
from litellm.responses.utils import ResponseAPILoggingUtils
chat_usage = (
ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(usage)
ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(cast(Any, usage))
)
# Use generic_cost_per_token for cost calculation
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=model,
usage=chat_usage,
custom_llm_provider=custom_llm_provider or "openai",
)
total_cost = prompt_cost + completion_cost
total_cost = _input_bd["total"] + _output_bd["total"]
verbose_logger.debug(
f"OpenAI gpt-image cost calculation for {model}: "
f"prompt_cost=${prompt_cost:.6f}, completion_cost=${completion_cost:.6f}, "
f"prompt_cost=${_input_bd['total']:.6f}, completion_cost=${_output_bd['total']:.6f}, "
f"total=${total_cost:.6f}"
)

View file

@ -5,11 +5,29 @@ Helper util for handling perplexity-specific cost calculation
from typing import Tuple, Union
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
)
from litellm.types.utils import Usage
from litellm.utils import get_model_info
def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
def _safe_float_cast(
value: Union[str, int, float, None, object], default: float = 0.0
) -> float:
"""Safely cast a value to float with proper type handling for mypy."""
if value is None:
return default
try:
return float(value) # type: ignore
except (ValueError, TypeError):
return default
def cost_per_token(
model: str, usage: Usage
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@ -18,51 +36,40 @@ def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
- usage: LiteLLM Usage block, containing perplexity-specific usage information
Returns:
Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
Tuple[InputCostBreakdown, OutputCostBreakdown] - granular input and output cost breakdowns
"""
## USE PRE-CALCULATED COST FROM PERPLEXITY IF AVAILABLE
## Perplexity returns accurate cost in usage.cost.total_cost including request fees
cost_info = getattr(usage, "cost", None)
if cost_info is not None and isinstance(cost_info, dict):
total_cost = cost_info.get("total_cost")
if total_cost is not None:
# Return total cost as completion_cost (prompt_cost=0) since Perplexity
# doesn't break down by input/output in their cost object
return (0.0, float(total_cost))
cost_val = float(total_cost)
return (
InputCostBreakdown(total=0.0),
OutputCostBreakdown(total=cost_val, text_cost=cost_val),
)
## FALLBACK: Calculate cost manually if Perplexity doesn't provide it
## GET MODEL INFO
model_info = get_model_info(model=model, custom_llm_provider="perplexity")
def _safe_float_cast(
value: Union[str, int, float, None, object], default: float = 0.0
) -> float:
"""Safely cast a value to float with proper type handling for mypy."""
if value is None:
return default
try:
return float(value) # type: ignore
except (ValueError, TypeError):
return default
## CALCULATE INPUT COST
input_cost_per_token = _safe_float_cast(model_info.get("input_cost_per_token"))
prompt_cost: float = (usage.prompt_tokens or 0) * input_cost_per_token
text_cost: float = (usage.prompt_tokens or 0) * input_cost_per_token
## ADD CITATION TOKENS COST (if present)
## CITATION TOKENS COST
citation_tokens = getattr(usage, "citation_tokens", 0) or 0
citation_cost_value = model_info.get("citation_cost_per_token")
citation_cost = 0.0
if citation_tokens > 0 and citation_cost_value is not None:
citation_cost_per_token = _safe_float_cast(citation_cost_value)
prompt_cost += citation_tokens * citation_cost_per_token
citation_cost = citation_tokens * _safe_float_cast(citation_cost_value)
text_cost += citation_cost
prompt_cost = text_cost
## CALCULATE OUTPUT COST
output_cost_per_token = _safe_float_cast(model_info.get("output_cost_per_token"))
completion_cost: float = (usage.completion_tokens or 0) * output_cost_per_token
output_text_cost: float = (usage.completion_tokens or 0) * output_cost_per_token
## ADD REASONING TOKENS COST (if present)
## REASONING TOKENS COST
reasoning_tokens = getattr(usage, "reasoning_tokens", 0) or 0
# Also check completion_tokens_details if reasoning_tokens is not directly available
if (
reasoning_tokens == 0
and hasattr(usage, "completion_tokens_details")
@ -72,26 +79,24 @@ def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
getattr(usage.completion_tokens_details, "reasoning_tokens", 0) or 0
)
reasoning_cost = 0.0
reasoning_cost_value = model_info.get("output_cost_per_reasoning_token")
if reasoning_tokens > 0 and reasoning_cost_value is not None:
reasoning_cost_per_token = _safe_float_cast(reasoning_cost_value)
completion_cost += reasoning_tokens * reasoning_cost_per_token
reasoning_cost = reasoning_tokens * _safe_float_cast(reasoning_cost_value)
## ADD SEARCH QUERIES COST (if present)
## SEARCH QUERIES COST
num_search_queries = 0
if hasattr(usage, "prompt_tokens_details") and usage.prompt_tokens_details:
num_search_queries = (
getattr(usage.prompt_tokens_details, "web_search_requests", 0) or 0
)
# Check both possible keys for search cost (legacy and current)
search_cost = 0.0
search_cost_value = model_info.get(
"search_queries_cost_per_query"
) or model_info.get("search_context_cost_per_query")
if num_search_queries > 0 and search_cost_value is not None:
# Handle both dict and float formats
if isinstance(search_cost_value, dict):
# Use the "low" size as default - tests expect 0.005 / 1000
search_cost_per_query = (
_safe_float_cast(search_cost_value.get("search_context_size_low", 0))
/ 1000
@ -99,7 +104,14 @@ def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
else:
search_cost_per_query = _safe_float_cast(search_cost_value)
search_cost = num_search_queries * search_cost_per_query
# Add search cost to completion cost (similar to how other providers handle it)
completion_cost += search_cost
return prompt_cost, completion_cost
completion_cost = output_text_cost + reasoning_cost + search_cost
return (
InputCostBreakdown(total=prompt_cost, text_cost=text_cost),
OutputCostBreakdown(
total=completion_cost,
text_cost=output_text_cost + search_cost,
reasoning_cost=reasoning_cost,
),
)

View file

@ -5,6 +5,8 @@ from typing import Literal, Optional, Tuple, Union
import litellm
from litellm import verbose_logger
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
_is_above_128k,
generic_cost_per_token,
)
@ -89,11 +91,12 @@ def cost_per_character(
## CALCULATE INPUT COST
if prompt_characters is None:
prompt_cost, _ = cost_per_token(
_input_bd, _ = cost_per_token(
model=model,
custom_llm_provider=custom_llm_provider,
usage=usage,
)
prompt_cost = _input_bd["total"]
else:
try:
if (
@ -126,19 +129,21 @@ def cost_per_character(
str(e)
)
)
prompt_cost, _ = cost_per_token(
_input_bd, _ = cost_per_token(
model=model,
custom_llm_provider=custom_llm_provider,
usage=usage,
)
prompt_cost = _input_bd["total"]
## CALCULATE OUTPUT COST
if completion_characters is None:
_, completion_cost = cost_per_token(
_, _output_bd = cost_per_token(
model=model,
custom_llm_provider=custom_llm_provider,
usage=usage,
)
completion_cost = _output_bd["total"]
else:
completion_tokens = usage.completion_tokens
try:
@ -173,11 +178,12 @@ def cost_per_character(
str(e)
)
)
_, completion_cost = cost_per_token(
_, _output_bd = cost_per_token(
model=model,
custom_llm_provider=custom_llm_provider,
usage=usage,
)
completion_cost = _output_bd["total"]
return prompt_cost, completion_cost
@ -225,7 +231,7 @@ def cost_per_token(
custom_llm_provider: str,
usage: Usage,
service_tier: Optional[str] = None,
) -> Tuple[float, float]:
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@ -238,7 +244,7 @@ def cost_per_token(
("priority" for ON_DEMAND_PRIORITY, "flex" for FLEX/batch).
Returns:
Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
Tuple[InputCostBreakdown, OutputCostBreakdown] - granular input and output cost breakdowns
Raises:
Exception if model requires >128k pricing, but model cost not mapped
@ -260,10 +266,14 @@ def cost_per_token(
input_cost_per_token_above_128k_tokens is not None
or output_cost_per_token_above_128k_tokens is not None
):
return _handle_128k_pricing(
prompt_cost, completion_cost = _handle_128k_pricing(
model_info=model_info,
usage=usage,
)
return (
InputCostBreakdown(total=prompt_cost, text_cost=prompt_cost),
OutputCostBreakdown(total=completion_cost, text_cost=completion_cost),
)
return generic_cost_per_token(
model=model,

View file

@ -6,14 +6,20 @@ Helper util for handling XAI-specific cost calculation
from typing import TYPE_CHECKING, Tuple
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
generic_cost_per_token,
)
from litellm.types.utils import Usage
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
if TYPE_CHECKING:
from litellm.types.utils import ModelInfo
def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
def cost_per_token(
model: str, usage: Usage
) -> Tuple[InputCostBreakdown, OutputCostBreakdown]:
"""
Calculates the cost per token for a given XAI model, prompt tokens, and completion tokens.
Uses the generic cost calculator for all pricing logic, with XAI-specific reasoning token handling.
@ -23,10 +29,8 @@ def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
- usage: LiteLLM Usage block, containing XAI-specific usage information
Returns:
Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
Tuple[InputCostBreakdown, OutputCostBreakdown] - granular input and output cost breakdowns
"""
# XAI-specific completion cost calculation
# For XAI models, completion is billed as (visible completion tokens + reasoning tokens)
completion_tokens = int(getattr(usage, "completion_tokens", 0) or 0)
reasoning_tokens = 0
if hasattr(usage, "completion_tokens_details") and usage.completion_tokens_details:
@ -44,12 +48,10 @@ def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
completion_tokens_details=None,
)
prompt_cost, completion_cost = generic_cost_per_token(
return generic_cost_per_token(
model=model, usage=modified_usage, custom_llm_provider="xai"
)
return prompt_cost, completion_cost
def cost_per_web_search_request(usage: "Usage", model_info: "ModelInfo") -> float:
"""
@ -75,8 +77,8 @@ def cost_per_web_search_request(usage: "Usage", model_info: "ModelInfo") -> floa
num_sources_used = int(usage.prompt_tokens_details.web_search_requests)
# Fallback: try to get from num_sources_used if set directly
elif hasattr(usage, "num_sources_used") and usage.num_sources_used is not None:
num_sources_used = int(usage.num_sources_used)
elif hasattr(usage, "num_sources_used") and getattr(usage, "num_sources_used", None) is not None:
num_sources_used = int(getattr(usage, "num_sources_used"))
total_cost = cost_per_source * num_sources_used

View file

@ -2767,6 +2767,33 @@ class CachingDetails(TypedDict):
"""
class _InputCostBreakdownPublicRequired(TypedDict):
total: float
class InputCostBreakdownPublic(_InputCostBreakdownPublicRequired, total=False):
"""Granular breakdown of input/prompt token costs (public-facing, stored in spend logs)."""
text_cost: float
cache_read_cost: float
cache_creation_cost: float
audio_cost: float
image_cost: float
class _OutputCostBreakdownPublicRequired(TypedDict):
total: float
class OutputCostBreakdownPublic(_OutputCostBreakdownPublicRequired, total=False):
"""Granular breakdown of output/completion token costs (public-facing, stored in spend logs)."""
text_cost: float
reasoning_cost: float
audio_cost: float
image_cost: float
class CostBreakdown(TypedDict, total=False):
"""
Detailed cost breakdown for a request
@ -2787,6 +2814,8 @@ class CostBreakdown(TypedDict, total=False):
margin_percent: float # Margin percentage applied (e.g., 0.10 = 10%) (optional)
margin_fixed_amount: float # Fixed margin amount in USD (optional)
margin_total_amount: float # Total margin added in USD (optional)
input_cost_breakdown: InputCostBreakdownPublic # Granular input cost components
output_cost_breakdown: OutputCostBreakdownPublic # Granular output cost components
class StandardLoggingPayloadStatusFields(TypedDict, total=False):

View file

@ -453,30 +453,36 @@ class TestBedrockGovCloudSupport:
usage = Usage(prompt_tokens=20, completion_tokens=10, total_tokens=30)
# Test base model with standard region
base_prompt_cost, base_completion_cost = cost_per_token(
base_input_bd, base_output_bd = cost_per_token(
model="anthropic.claude-3-5-sonnet-20240620-v1:0",
prompt_tokens=20,
completion_tokens=10,
custom_llm_provider="bedrock",
region_name="us-east-1",
)
base_prompt_cost = base_input_bd["total"]
base_completion_cost = base_output_bd["total"]
# Test gov models with gov regions
gov_east_prompt_cost, gov_east_completion_cost = cost_per_token(
gov_east_input_bd, gov_east_output_bd = cost_per_token(
model="anthropic.claude-3-5-sonnet-20240620-v1:0",
prompt_tokens=20,
completion_tokens=10,
custom_llm_provider="bedrock",
region_name="us-gov-east-1",
)
gov_east_prompt_cost = gov_east_input_bd["total"]
gov_east_completion_cost = gov_east_output_bd["total"]
gov_west_prompt_cost, gov_west_completion_cost = cost_per_token(
gov_west_input_bd, gov_west_output_bd = cost_per_token(
model="anthropic.claude-3-5-sonnet-20240620-v1:0",
prompt_tokens=20,
completion_tokens=10,
custom_llm_provider="bedrock",
region_name="us-gov-west-1",
)
gov_west_prompt_cost = gov_west_input_bd["total"]
gov_west_completion_cost = gov_west_output_bd["total"]
# Expected costs:
# Base model: 20 * 3e-06 + 10 * 1.5e-05 = 0.00006 + 0.00015 = 0.00021

View file

@ -583,12 +583,14 @@ def test_gemini_completion_cost(provider):
output_cost = output_tokens * model_info["output_cost_per_token"]
## CALCULATED COST
calculated_input_cost, calculated_output_cost = cost_per_token(
input_bd, output_bd = cost_per_token(
model=model_name,
prompt_tokens=prompt_tokens,
completion_tokens=output_tokens,
custom_llm_provider=provider,
)
calculated_input_cost = input_bd["total"]
calculated_output_cost = output_bd["total"]
assert calculated_input_cost == input_cost
assert calculated_output_cost == output_cost
@ -613,12 +615,14 @@ def test_vertex_ai_completion_cost():
expected_input_cost = prompt_tokens * model_info["input_cost_per_token"]
## CALCULATED COST
calculated_input_cost, calculated_output_cost = cost_per_token(
input_bd, output_bd = cost_per_token(
model="gemini-2.0-flash",
custom_llm_provider="vertex_ai",
prompt_tokens=prompt_tokens,
completion_tokens=0,
)
calculated_input_cost = input_bd["total"]
calculated_output_cost = output_bd["total"]
assert round(expected_input_cost, 6) == round(calculated_input_cost, 6)
print("expected_input_cost: {}".format(expected_input_cost))
@ -721,12 +725,14 @@ def test_vertex_ai_embedding_completion_cost(caplog):
expected_input_cost = input_tokens * model_info["input_cost_per_token"]
## CALCULATED COST
calculated_input_cost, calculated_output_cost = cost_per_token(
input_bd, output_bd = cost_per_token(
model="text-embedding-004",
custom_llm_provider="vertex_ai",
prompt_tokens=input_tokens,
call_type="aembedding",
)
calculated_input_cost = input_bd["total"]
calculated_output_cost = output_bd["total"]
assert round(expected_input_cost, 6) == round(calculated_input_cost, 6)
print("expected_input_cost: {}".format(expected_input_cost))
@ -908,7 +914,9 @@ def test_completion_cost_anthropic():
"call_type": "acompletion",
}
input_cost, output_cost = cost_per_token(**data)
input_bd, output_bd = cost_per_token(**data)
input_cost = input_bd["total"]
output_cost = output_bd["total"]
assert input_cost > 0
assert output_cost > 0
@ -2261,25 +2269,31 @@ def test_completion_cost_params():
Relevant Issue: https://github.com/BerriAI/litellm/issues/6133
"""
litellm.set_verbose = True
resp1_prompt_cost, resp1_completion_cost = cost_per_token(
resp1_input_bd, resp1_output_bd = cost_per_token(
model="gemini-2.0-flash",
prompt_tokens=1000,
completion_tokens=1000,
custom_llm_provider="vertex_ai_beta",
)
resp1_prompt_cost = resp1_input_bd["total"]
resp1_completion_cost = resp1_output_bd["total"]
resp2_prompt_cost, resp2_completion_cost = cost_per_token(
resp2_input_bd, resp2_output_bd = cost_per_token(
model="gemini-2.0-flash", prompt_tokens=1000, completion_tokens=1000
)
resp2_prompt_cost = resp2_input_bd["total"]
resp2_completion_cost = resp2_output_bd["total"]
assert resp2_prompt_cost > 0
assert resp1_prompt_cost == resp2_prompt_cost
assert resp1_completion_cost == resp2_completion_cost
resp3_prompt_cost, resp3_completion_cost = cost_per_token(
resp3_input_bd, resp3_output_bd = cost_per_token(
model="vertex_ai/gemini-2.0-flash", prompt_tokens=1000, completion_tokens=1000
)
resp3_prompt_cost = resp3_input_bd["total"]
resp3_completion_cost = resp3_output_bd["total"]
assert resp3_prompt_cost > 0
@ -2295,11 +2309,13 @@ def test_completion_cost_params_2():
prompt_tokens = 1000
completion_tokens = 1000
resp1_prompt_cost, resp1_completion_cost = cost_per_token(
resp1_input_bd, resp1_output_bd = cost_per_token(
model="gemini-2.0-flash",
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
)
resp1_prompt_cost = resp1_input_bd["total"]
resp1_completion_cost = resp1_output_bd["total"]
print(resp1_prompt_cost, resp1_completion_cost)

View file

@ -33,6 +33,8 @@ sys.path.insert(
) # Adds the parent directory to the system path
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
PromptTokensDetailsResult,
_calculate_input_cost,
calculate_cache_writing_cost,
@ -64,11 +66,13 @@ def test_reasoning_tokens_no_price_set():
audio_tokens=None, cached_tokens=None, text_tokens=17, image_tokens=None
),
)
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider="openai",
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
assert round(prompt_cost, 10) == round(
model_cost_map["input_cost_per_token"] * usage.prompt_tokens,
10,
@ -106,11 +110,13 @@ def test_reasoning_tokens_gemini():
),
)
model_cost_map = litellm.model_cost[model]
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
assert round(prompt_cost, 10) == round(
model_cost_map["input_cost_per_token"] * usage.prompt_tokens,
@ -152,11 +158,13 @@ def test_reasoning_tokens_gemini_3_1_flash_lite():
),
)
model_cost_map = litellm.model_cost[model]
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
assert round(prompt_cost, 10) == round(
model_cost_map["input_cost_per_token"] * usage.prompt_tokens,
@ -207,9 +215,11 @@ def test_image_tokens_with_custom_pricing():
"litellm.litellm_core_utils.llm_cost_calc.utils.get_model_info",
return_value=mock_model_info,
):
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model="test-model", usage=usage, custom_llm_provider="gemini"
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
# Expected costs:
# Prompt: 14 * 1e-6
@ -253,9 +263,11 @@ def test_image_tokens_fallback_to_base_cost():
"litellm.litellm_core_utils.llm_cost_calc.utils.get_model_info",
return_value=mock_model_info,
):
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model="test-model", usage=usage, custom_llm_provider="gemini"
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
# Expected costs:
# Prompt: 14 * 1e-6
@ -282,11 +294,13 @@ def test_generic_cost_per_token_above_200k_tokens():
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
)
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
assert round(prompt_cost, 10) == round(
model_cost_map["input_cost_per_token_above_200k_tokens"] * usage.prompt_tokens,
10,
@ -313,11 +327,13 @@ def test_generic_cost_per_token_gpt54_above_272k_tokens():
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
)
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
expected_prompt = model_cost_map["input_cost_per_token_above_272k_tokens"] * prompt_tokens
expected_completion = model_cost_map["output_cost_per_token_above_272k_tokens"] * completion_tokens
assert round(prompt_cost, 10) == round(expected_prompt, 10)
@ -346,11 +362,12 @@ def test_generic_cost_per_token_anthropic_prompt_caching():
custom_llm_provider = "vertex_ai"
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
)
prompt_cost = _input_bd["total"]
print(f"prompt_cost: {prompt_cost}")
assert prompt_cost < 0.085
@ -375,11 +392,12 @@ def test_generic_cost_per_token_anthropic_prompt_caching_with_cache_creation():
custom_llm_provider = "anthropic"
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
)
prompt_cost = _input_bd["total"]
print(f"prompt_cost: {prompt_cost}")
assert round(prompt_cost, 3) == 0.029
@ -423,9 +441,11 @@ def test_string_cost_values():
"litellm.litellm_core_utils.llm_cost_calc.utils.get_model_info",
return_value=mock_model_info,
):
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model="test-model", usage=usage, custom_llm_provider="test-provider"
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
# Calculate expected costs manually
# Prompt cost = text_tokens * input_cost + audio_tokens * audio_cost + cached_tokens * cache_read_cost + cache_creation_tokens * cache_creation_cost
@ -521,9 +541,11 @@ def test_string_cost_values_edge_cases():
"litellm.litellm_core_utils.llm_cost_calc.utils.get_model_info",
return_value=mock_model_info,
):
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model="test-model", usage=usage, custom_llm_provider="test-provider"
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
# Expected costs:
# Prompt: 1000 * 1e-6 + 100 * 0 (invalid string becomes 0)
@ -558,9 +580,11 @@ def test_string_cost_values_with_threshold():
"litellm.litellm_core_utils.llm_cost_calc.utils.get_model_info",
return_value=mock_model_info,
):
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model="test-model", usage=usage, custom_llm_provider="test-provider"
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
# Expected costs using threshold pricing (string values converted to float)
expected_prompt_cost = 250000 * 5e-7 # threshold cost
@ -684,8 +708,8 @@ def test_cache_writing_cost_with_zero_creation_tokens_and_ephemeral_details():
# Expected: (100 * 3.75e-06) + (200 * 6e-06) = 0.000375 + 0.0012 = 0.001575
expected = (100 * cache_creation_cost) + (200 * cache_creation_cost_above_1hr)
assert result > 0, "Cost should not be zero when ephemeral token details are present"
assert round(result, 6) == round(expected, 6)
assert result["total"] > 0, "Cost should not be zero when ephemeral token details are present"
assert round(result["total"], 6) == round(expected, 6)
def test_service_tier_flex_pricing():
@ -706,22 +730,22 @@ def test_service_tier_flex_pricing():
)
# Test standard pricing
std_cost = generic_cost_per_token(
std_input_bd, std_output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
service_tier=None
)
std_total = std_cost[0] + std_cost[1]
std_total = std_input_bd["total"] + std_output_bd["total"]
# Test flex pricing
flex_cost = generic_cost_per_token(
flex_input_bd, flex_output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
service_tier="flex"
)
flex_total = flex_cost[0] + flex_cost[1]
flex_total = flex_input_bd["total"] + flex_output_bd["total"]
# Verify flex is approximately 50% of standard
assert std_total > 0, "Standard cost should be greater than 0"
@ -736,8 +760,8 @@ def test_service_tier_flex_pricing():
expected_flex_completion = 500 * 2e-07 # 0.0001
expected_flex_total = expected_flex_prompt + expected_flex_completion
assert abs(flex_cost[0] - expected_flex_prompt) < 1e-10, f"Flex prompt cost mismatch: {flex_cost[0]} vs {expected_flex_prompt}"
assert abs(flex_cost[1] - expected_flex_completion) < 1e-10, f"Flex completion cost mismatch: {flex_cost[1]} vs {expected_flex_completion}"
assert abs(flex_input_bd["total"] - expected_flex_prompt) < 1e-10, f"Flex prompt cost mismatch: {flex_input_bd['total']} vs {expected_flex_prompt}"
assert abs(flex_output_bd["total"] - expected_flex_completion) < 1e-10, f"Flex completion cost mismatch: {flex_output_bd['total']} vs {expected_flex_completion}"
assert abs(flex_total - expected_flex_total) < 1e-10, f"Flex total cost mismatch: {flex_total} vs {expected_flex_total}"
@ -759,7 +783,7 @@ def test_service_tier_default_pricing():
)
# Test with no service tier (should use standard)
default_cost = generic_cost_per_token(
default_input_bd, default_output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
@ -767,7 +791,7 @@ def test_service_tier_default_pricing():
)
# Test with explicit standard service tier
standard_cost = generic_cost_per_token(
standard_input_bd, standard_output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
@ -775,8 +799,8 @@ def test_service_tier_default_pricing():
)
# Both should be identical
assert abs(default_cost[0] - standard_cost[0]) < 1e-10, "Default and standard prompt costs should be identical"
assert abs(default_cost[1] - standard_cost[1]) < 1e-10, "Default and standard completion costs should be identical"
assert abs(default_input_bd["total"] - standard_input_bd["total"]) < 1e-10, "Default and standard prompt costs should be identical"
assert abs(default_output_bd["total"] - standard_output_bd["total"]) < 1e-10, "Default and standard completion costs should be identical"
# Verify specific costs match expected standard values
# gpt-5-nano standard: input=5e-08, output=4e-07
@ -784,8 +808,8 @@ def test_service_tier_default_pricing():
expected_standard_completion = 500 * 4e-07 # 0.0002
expected_standard_total = expected_standard_prompt + expected_standard_completion
assert abs(default_cost[0] - expected_standard_prompt) < 1e-10, f"Standard prompt cost mismatch: {default_cost[0]} vs {expected_standard_prompt}"
assert abs(default_cost[1] - expected_standard_completion) < 1e-10, f"Standard completion cost mismatch: {default_cost[1]} vs {expected_standard_completion}"
assert abs(default_input_bd["total"] - expected_standard_prompt) < 1e-10, f"Standard prompt cost mismatch: {default_input_bd['total']} vs {expected_standard_prompt}"
assert abs(default_output_bd["total"] - expected_standard_completion) < 1e-10, f"Standard completion cost mismatch: {default_output_bd['total']} vs {expected_standard_completion}"
def test_service_tier_fallback_pricing():
@ -806,31 +830,31 @@ def test_service_tier_fallback_pricing():
)
# Test standard pricing
std_cost = generic_cost_per_token(
std_input_bd, std_output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
service_tier=None
)
std_total = std_cost[0] + std_cost[1]
std_total = std_input_bd["total"] + std_output_bd["total"]
# Test flex pricing (should fall back to standard since gpt-4 doesn't have flex keys)
flex_cost = generic_cost_per_token(
flex_input_bd, flex_output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
service_tier="flex"
)
flex_total = flex_cost[0] + flex_cost[1]
flex_total = flex_input_bd["total"] + flex_output_bd["total"]
# Test priority pricing (should fall back to standard since gpt-4 doesn't have priority keys)
priority_cost = generic_cost_per_token(
priority_input_bd, priority_output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
service_tier="priority"
)
priority_total = priority_cost[0] + priority_cost[1]
priority_total = priority_input_bd["total"] + priority_output_bd["total"]
# All should be identical (fallback to standard)
assert abs(std_total - flex_total) < 1e-10, f"Standard and flex costs should be identical (fallback): {std_total} vs {flex_total}"
@ -847,8 +871,8 @@ def test_service_tier_fallback_pricing():
expected_standard_completion = 500 * 6e-05 # 0.03
expected_standard_total = expected_standard_prompt + expected_standard_completion
assert abs(std_cost[0] - expected_standard_prompt) < 1e-10, f"Standard prompt cost mismatch: {std_cost[0]} vs {expected_standard_prompt}"
assert abs(std_cost[1] - expected_standard_completion) < 1e-10, f"Standard completion cost mismatch: {std_cost[1]} vs {expected_standard_completion}"
assert abs(std_input_bd["total"] - expected_standard_prompt) < 1e-10, f"Standard prompt cost mismatch: {std_input_bd['total']} vs {expected_standard_prompt}"
assert abs(std_output_bd["total"] - expected_standard_completion) < 1e-10, f"Standard completion cost mismatch: {std_output_bd['total']} vs {expected_standard_completion}"
@pytest.mark.parametrize(
@ -892,11 +916,13 @@ def test_gemini_image_generation_cost_with_zero_text_tokens(model: str):
)
model_cost_map = litellm.model_cost[model]
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
# Expected costs:
# - text_tokens: 0 * output_cost_per_token = 0
@ -1069,11 +1095,13 @@ def test_bedrock_anthropic_prompt_caching():
custom_llm_provider = "bedrock"
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
assert prompt_cost >= 0
assert completion_cost >= 0
@ -1110,11 +1138,13 @@ def test_reasoning_tokens_without_text_tokens_gpt5_nano():
),
)
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
# gpt-5-nano pricing: $0.05/1M input, $0.40/1M output
expected_prompt_cost = 17 * 0.05 / 1_000_000
@ -1155,11 +1185,13 @@ def test_image_count_prevents_text_tokens_fallback():
),
)
prompt_cost, completion_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model="amazon.nova-2-multimodal-embeddings-v1:0",
usage=usage,
custom_llm_provider="bedrock",
)
prompt_cost = _input_bd["total"]
completion_cost = _output_bd["total"]
# Cost should be 1 * input_cost_per_image ($6e-05) = $0.00006
# NOT 768 * input_cost_per_token ($1.35e-07) + $0.00006 = $0.000164
@ -1169,3 +1201,123 @@ def test_image_count_prevents_text_tokens_fallback():
f"got {prompt_cost}. text_tokens fallback may be double-charging."
)
assert completion_cost == 0.0
def test_generic_cost_per_token_returns_input_cost_breakdown():
"""Test that generic_cost_per_token returns InputCostBreakdown with granular input costs."""
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
model = "gpt-4o"
usage = Usage(
prompt_tokens=300,
completion_tokens=100,
total_tokens=400,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=100, text_tokens=200, audio_tokens=None, image_tokens=None
),
)
input_bd, output_bd = generic_cost_per_token(
model=model, usage=usage, custom_llm_provider="openai"
)
assert "total" in input_bd
assert input_bd["total"] > 0
assert "text_cost" in input_bd
assert input_bd["text_cost"] > 0
assert "cache_read_cost" in input_bd
assert input_bd["cache_read_cost"] > 0
component_sum = input_bd.get("text_cost", 0) + input_bd.get("cache_read_cost", 0) + input_bd.get("cache_creation_cost", 0) + input_bd.get("audio_cost", 0) + input_bd.get("image_cost", 0)
assert abs(input_bd["total"] - component_sum) < 1e-12
def test_generic_cost_per_token_returns_output_cost_breakdown_with_reasoning():
"""Test that generic_cost_per_token returns OutputCostBreakdown with reasoning costs."""
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
model = "gemini-2.5-flash"
usage = Usage(
prompt_tokens=17,
completion_tokens=1578,
total_tokens=1595,
completion_tokens_details=CompletionTokensDetailsWrapper(
reasoning_tokens=952, text_tokens=626,
),
prompt_tokens_details=PromptTokensDetailsWrapper(
text_tokens=17,
),
)
input_bd, output_bd = generic_cost_per_token(
model=model, usage=usage, custom_llm_provider="gemini"
)
assert "total" in output_bd
assert output_bd["total"] > 0
assert "text_cost" in output_bd
assert output_bd["text_cost"] > 0
assert "reasoning_cost" in output_bd
assert output_bd["reasoning_cost"] > 0
component_sum = output_bd.get("text_cost", 0) + output_bd.get("reasoning_cost", 0) + output_bd.get("audio_cost", 0) + output_bd.get("image_cost", 0)
assert abs(output_bd["total"] - component_sum) < 1e-12
def test_generic_cost_per_token_breakdown_cache_creation():
"""Test that cache creation costs appear in InputCostBreakdown."""
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
model = "claude-haiku-4-5-20251001"
usage = Usage(
completion_tokens=90,
prompt_tokens=28436,
total_tokens=28526,
prompt_tokens_details=None,
cache_creation_input_tokens=2000,
)
input_bd, output_bd = generic_cost_per_token(
model=model, usage=usage, custom_llm_provider="anthropic"
)
assert "cache_creation_cost" in input_bd
assert input_bd["cache_creation_cost"] > 0
assert input_bd["total"] > 0
def test_generic_cost_per_token_breakdown_components_sum_to_total():
"""Test that all breakdown components sum to their respective totals."""
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
model = "gpt-4o"
usage = Usage(
prompt_tokens=1000,
completion_tokens=500,
total_tokens=1500,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=200, text_tokens=800,
),
completion_tokens_details=CompletionTokensDetailsWrapper(
text_tokens=500,
),
)
input_bd, output_bd = generic_cost_per_token(
model=model, usage=usage, custom_llm_provider="openai"
)
input_sum = sum(v for k, v in input_bd.items() if k != "total")
assert abs(input_bd["total"] - input_sum) < 1e-12
output_sum = sum(v for k, v in output_bd.items() if k != "total")
assert abs(output_bd["total"] - output_sum) < 1e-12
def test_generic_cost_per_token_backward_compat_total_matches_old_behavior():
"""Test that input_bd['total'] and output_bd['total'] match the old float return values."""
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
model = "gpt-4o"
model_cost_map = litellm.model_cost[model]
usage = Usage(
prompt_tokens=100,
completion_tokens=200,
total_tokens=300,
)
input_bd, output_bd = generic_cost_per_token(
model=model, usage=usage, custom_llm_provider="openai"
)
expected_prompt = model_cost_map["input_cost_per_token"] * 100
expected_completion = model_cost_map["output_cost_per_token"] * 200
assert abs(input_bd["total"] - expected_prompt) < 1e-12
assert abs(output_bd["total"] - expected_completion) < 1e-12

View file

@ -12,6 +12,7 @@ from litellm.llms.anthropic.chat.transformation import AnthropicConfig
from litellm.llms.anthropic.experimental_pass_through.messages.transformation import (
AnthropicMessagesConfig,
)
from litellm.litellm_core_utils.llm_cost_calc.utils import InputCostBreakdown, OutputCostBreakdown
from litellm.types.utils import ServerToolUse
@ -3103,13 +3104,21 @@ def test_fast_mode_cost_calculation():
from litellm.llms.anthropic.cost_calculation import cost_per_token
from litellm.types.utils import Usage
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
)
base_prompt = 0.005
base_completion = 0.025
with patch(
"litellm.llms.anthropic.cost_calculation.generic_cost_per_token"
) as mock_cost, patch("litellm.get_model_info") as mock_info:
mock_cost.return_value = (base_prompt, base_completion)
mock_cost.return_value = (
InputCostBreakdown(total=base_prompt, text_cost=base_prompt),
OutputCostBreakdown(total=base_completion, text_cost=base_completion),
)
mock_info.return_value = {"provider_specific_entry": {"fast": 1.1, "us": 1.1}}
usage_fast = Usage(
@ -3118,7 +3127,7 @@ def test_fast_mode_cost_calculation():
speed="fast",
)
prompt_cost, completion_cost = cost_per_token(
input_bd, output_bd = cost_per_token(
model="claude-opus-4-6",
usage=usage_fast,
)
@ -3129,8 +3138,8 @@ def test_fast_mode_cost_calculation():
assert mock_cost.call_args[1]["custom_llm_provider"] == "anthropic"
# 1.1x multiplier applied
assert abs(prompt_cost - base_prompt * 1.1) < 1e-10
assert abs(completion_cost - base_completion * 1.1) < 1e-10
assert abs(input_bd["total"] - base_prompt * 1.1) < 1e-10
assert abs(output_bd["total"] - base_completion * 1.1) < 1e-10
def test_fast_mode_with_inference_geo():
@ -3139,6 +3148,10 @@ def test_fast_mode_with_inference_geo():
provider_specific_entry (1.1 * 1.1 = 1.21x for claude-opus-4-6).
"""
from litellm.litellm_core_utils.llm_cost_calc.utils import (
InputCostBreakdown,
OutputCostBreakdown,
)
from litellm.llms.anthropic.cost_calculation import cost_per_token
from litellm.types.utils import Usage
@ -3148,7 +3161,10 @@ def test_fast_mode_with_inference_geo():
with patch(
"litellm.llms.anthropic.cost_calculation.generic_cost_per_token"
) as mock_cost, patch("litellm.get_model_info") as mock_info:
mock_cost.return_value = (base_prompt, base_completion)
mock_cost.return_value = (
InputCostBreakdown(total=base_prompt, text_cost=base_prompt),
OutputCostBreakdown(total=base_completion, text_cost=base_completion),
)
mock_info.return_value = {"provider_specific_entry": {"fast": 1.1, "us": 1.1}}
usage = Usage(
@ -3158,7 +3174,7 @@ def test_fast_mode_with_inference_geo():
inference_geo="us",
)
prompt_cost, completion_cost = cost_per_token(
input_bd, output_bd = cost_per_token(
model="claude-opus-4-6",
usage=usage,
)
@ -3170,8 +3186,8 @@ def test_fast_mode_with_inference_geo():
# 1.1 (fast) * 1.1 (us) = 1.21x multiplier applied
expected_multiplier = 1.1 * 1.1
assert abs(prompt_cost - base_prompt * expected_multiplier) < 1e-10
assert abs(completion_cost - base_completion * expected_multiplier) < 1e-10
assert abs(input_bd["total"] - base_prompt * expected_multiplier) < 1e-10
assert abs(output_bd["total"] - base_completion * expected_multiplier) < 1e-10
def test_fast_mode_parameter_in_supported_params():

View file

@ -90,7 +90,9 @@ class TestAzureModelRouterFlatCost:
total_tokens=1500,
)
prompt_cost, completion_cost = cost_per_token(model=model, usage=usage)
input_bd, output_bd = cost_per_token(model=model, usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Calculate expected flat cost
expected_flat_cost = (
@ -117,7 +119,9 @@ class TestAzureModelRouterFlatCost:
total_tokens=150_000,
)
prompt_cost, completion_cost = cost_per_token(model=model, usage=usage)
input_bd, output_bd = cost_per_token(model=model, usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Calculate expected flat cost
expected_flat_cost = (
@ -142,7 +146,9 @@ class TestAzureModelRouterFlatCost:
total_tokens=1_100_000,
)
prompt_cost, completion_cost = cost_per_token(model=model, usage=usage)
input_bd, output_bd = cost_per_token(model=model, usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Calculate expected flat cost
expected_flat_cost = AZURE_MODEL_ROUTER_FLAT_COST_PER_M_INPUT_TOKENS
@ -162,7 +168,9 @@ class TestAzureModelRouterFlatCost:
total_tokens=1500,
)
prompt_cost, completion_cost = cost_per_token(model=model, usage=usage)
input_bd, output_bd = cost_per_token(model=model, usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# No flat cost should be added for non-Model Router models
# The cost might be 0 or based on the model's pricing
@ -182,7 +190,9 @@ class TestAzureModelRouterFlatCost:
cache_creation_input_tokens=200,
)
prompt_cost, completion_cost = cost_per_token(model=model, usage=usage)
input_bd, output_bd = cost_per_token(model=model, usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Flat cost is based on ALL prompt tokens (including cached)
expected_flat_cost = (
@ -215,11 +225,13 @@ class TestAzureModelRouterFlatCost:
# Request model is the router - user called azure_ai/model_router/model-router
request_model = "azure_ai/model_router/model-router"
prompt_cost, completion_cost = cost_per_token(
input_bd, output_bd = cost_per_token(
model=response_model,
usage=usage,
request_model=request_model,
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected: model cost (from gpt-5-nano) + router flat cost
expected_flat_cost = (

View file

@ -42,9 +42,11 @@ class TestDashscopeCostCalculator:
usage = Usage(prompt_tokens=1000, completion_tokens=500)
# We call the specific calculator for dashscope
prompt_cost, completion_cost = dashscope_cost_per_token(
input_bd, output_bd = dashscope_cost_per_token(
model="qwen-max", usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
model_info = litellm.get_model_info("dashscope/qwen-max")
expected_prompt_cost = 1000 * model_info["input_cost_per_token"]
@ -60,9 +62,11 @@ class TestDashscopeCostCalculator:
"""
# Tier 1 for qwen-flash is [0, 256,000] tokens
usage = Usage(prompt_tokens=100000, completion_tokens=50000)
prompt_cost, completion_cost = dashscope_cost_per_token(
input_bd, output_bd = dashscope_cost_per_token(
model="qwen-flash", usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
model_info = litellm.get_model_info("dashscope/qwen-flash")
tier_1_pricing = model_info["tiered_pricing"][0]
@ -80,9 +84,11 @@ class TestDashscopeCostCalculator:
"""
# Tiering for qwen-flash: Tier 1: [0, 256k], Tier 2: [256k, 1M]
usage = Usage(prompt_tokens=300000, completion_tokens=300000)
prompt_cost, completion_cost = dashscope_cost_per_token(
input_bd, output_bd = dashscope_cost_per_token(
model="qwen-flash", usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
model_info = litellm.get_model_info("dashscope/qwen-flash")
tier_1 = model_info["tiered_pricing"][0]
@ -113,7 +119,8 @@ class TestDashscopeCostCalculator:
prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=10000),
)
prompt_cost, _ = dashscope_cost_per_token(model="qwen3-coder-plus", usage=usage)
input_bd, _ = dashscope_cost_per_token(model="qwen3-coder-plus", usage=usage)
prompt_cost = input_bd["total"]
model_info = litellm.get_model_info("dashscope/qwen3-coder-plus")
tier_1 = model_info["tiered_pricing"][0]
@ -140,7 +147,8 @@ class TestDashscopeCostCalculator:
prompt_tokens=1200000, completion_tokens=1000
) # Max defined range for qwen-flash is 1M
prompt_cost, _ = dashscope_cost_per_token(model="qwen-flash", usage=usage)
input_bd, _ = dashscope_cost_per_token(model="qwen-flash", usage=usage)
prompt_cost = input_bd["total"]
model_info = litellm.get_model_info("dashscope/qwen-flash")
tier_1 = model_info["tiered_pricing"][0]

View file

@ -67,10 +67,12 @@ class TestPerplexityCostCalculator:
total_tokens=150
)
prompt_cost, completion_cost = perplexity_cost_per_token(
input_bd, output_bd = perplexity_cost_per_token(
model="sonar-deep-research",
usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs:
# Input: 100 tokens * $2e-6 = $0.0002
@ -92,10 +94,12 @@ class TestPerplexityCostCalculator:
# Add citation tokens
usage.citation_tokens = 25
prompt_cost, completion_cost = perplexity_cost_per_token(
input_bd, output_bd = perplexity_cost_per_token(
model="sonar-deep-research",
usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs:
# Input: 100 tokens * $2e-6 = $0.0002
@ -117,10 +121,12 @@ class TestPerplexityCostCalculator:
prompt_tokens_details=PromptTokensDetailsWrapper(web_search_requests=3)
)
prompt_cost, completion_cost = perplexity_cost_per_token(
input_bd, output_bd = perplexity_cost_per_token(
model="sonar-deep-research",
usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs:
# Input: 100 tokens * $2e-6 = $0.0002
@ -144,10 +150,12 @@ class TestPerplexityCostCalculator:
# Set reasoning tokens directly
usage.reasoning_tokens = 20
prompt_cost, completion_cost = perplexity_cost_per_token(
input_bd, output_bd = perplexity_cost_per_token(
model="sonar-deep-research",
usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs:
# Input: 100 tokens * $2e-6 = $0.0002
@ -169,10 +177,12 @@ class TestPerplexityCostCalculator:
reasoning_tokens=20 # This should be stored in completion_tokens_details
)
prompt_cost, completion_cost = perplexity_cost_per_token(
input_bd, output_bd = perplexity_cost_per_token(
model="sonar-deep-research",
usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs:
# Input: 100 tokens * $2e-6 = $0.0002
@ -198,10 +208,12 @@ class TestPerplexityCostCalculator:
# Add custom fields
usage.citation_tokens = 30
prompt_cost, completion_cost = perplexity_cost_per_token(
input_bd, output_bd = perplexity_cost_per_token(
model="sonar-deep-research",
usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs:
# Input: 100 tokens * $2e-6 = $0.0002
@ -229,10 +241,12 @@ class TestPerplexityCostCalculator:
# These should not raise errors and should not affect cost
usage.citation_tokens = 0
prompt_cost, completion_cost = perplexity_cost_per_token(
input_bd, output_bd = perplexity_cost_per_token(
model="sonar-deep-research",
usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Should be same as basic calculation
expected_prompt_cost = 100 * 2e-6
@ -260,10 +274,12 @@ class TestPerplexityCostCalculator:
# Missing search_queries_cost_per_query
}
prompt_cost, completion_cost = perplexity_cost_per_token(
input_bd, output_bd = perplexity_cost_per_token(
model="sonar-deep-research",
usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Should only calculate basic costs when fields are missing
expected_prompt_cost = 100 * 2e-6
@ -285,17 +301,21 @@ class TestPerplexityCostCalculator:
usage.citation_tokens = 20
# Test main cost calculator
prompt_cost, completion_cost_val = cost_per_token(
input_bd, output_bd = cost_per_token(
model="sonar-deep-research",
custom_llm_provider="perplexity",
usage_object=usage
)
prompt_cost = input_bd["total"]
completion_cost_val = output_bd["total"]
# Should match direct call to perplexity cost calculator
expected_prompt, expected_completion = perplexity_cost_per_token(
expected_input_bd, expected_output_bd = perplexity_cost_per_token(
model="sonar-deep-research",
usage=usage
)
expected_prompt = expected_input_bd["total"]
expected_completion = expected_output_bd["total"]
assert math.isclose(prompt_cost, expected_prompt, rel_tol=1e-6)
assert math.isclose(completion_cost_val, expected_completion, rel_tol=1e-6)
@ -356,10 +376,12 @@ class TestPerplexityCostCalculator:
usage.citation_tokens = citation_tokens
prompt_cost, completion_cost = perplexity_cost_per_token(
input_bd, output_bd = perplexity_cost_per_token(
model="sonar-deep-research",
usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Calculate expected costs
expected_prompt_cost = (100 * 2e-6) + (citation_tokens * 2e-6)
@ -395,10 +417,12 @@ class TestPerplexityCostCalculator:
"total_cost": 0.008
}
prompt_cost, completion_cost = perplexity_cost_per_token(
input_bd, output_bd = perplexity_cost_per_token(
model="sonar-pro",
usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# When Perplexity provides total_cost, we use it directly
# prompt_cost should be 0, completion_cost should be total_cost
@ -418,10 +442,12 @@ class TestPerplexityCostCalculator:
)
# No cost object - should use manual calculation
prompt_cost, completion_cost = perplexity_cost_per_token(
input_bd, output_bd = perplexity_cost_per_token(
model="sonar-deep-research",
usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Should calculate manually: 100 * 2e-6 + 50 * 8e-6
expected_prompt = 100 * 2e-6

View file

@ -138,11 +138,13 @@ class TestPerplexityIntegration:
usage.citation_tokens = 40
# Test main cost calculator
prompt_cost, completion_cost_val = cost_per_token(
input_bd, output_bd = cost_per_token(
model="sonar-deep-research",
custom_llm_provider="perplexity",
usage_object=usage
)
prompt_cost = input_bd["total"]
completion_cost_val = output_bd["total"]
# Calculate expected costs
expected_prompt_cost = (200 * 2e-6) + (40 * 2e-6) # Input + citation
@ -217,11 +219,13 @@ class TestPerplexityIntegration:
usage.prompt_tokens_details = PromptTokensDetailsWrapper(web_search_requests=0)
# Should not add any extra cost
prompt_cost, completion_cost_val = cost_per_token(
input_bd, output_bd = cost_per_token(
model="sonar-deep-research",
custom_llm_provider="perplexity",
usage_object=usage
)
prompt_cost = input_bd["total"]
completion_cost_val = output_bd["total"]
expected_prompt_cost = 100 * 2e-6
expected_completion_cost = 50 * 8e-6
@ -305,11 +309,13 @@ class TestPerplexityIntegration:
usage.prompt_tokens_details = PromptTokensDetailsWrapper(web_search_requests=1)
# Should work regardless of case
prompt_cost, completion_cost_val = cost_per_token(
input_bd, output_bd = cost_per_token(
model="sonar-deep-research",
custom_llm_provider=provider_name.lower(), # Normalize to lowercase
usage_object=usage
)
prompt_cost = input_bd["total"]
completion_cost_val = output_bd["total"]
# Should calculate costs correctly
expected_prompt_cost = (100 * 2e-6) + (10 * 2e-6)

View file

@ -237,11 +237,13 @@ def test_vercel_ai_gateway_glm46_cost_math():
key = "vercel_ai_gateway/zai/glm-4.6"
info = litellm.model_cost[key]
prompt_cost, completion_cost = cost_per_token(
input_bd, output_bd = cost_per_token(
model="vercel_ai_gateway/zai/glm-4.6",
prompt_tokens=1000,
completion_tokens=500,
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
assert math.isclose(prompt_cost, 1000 * info["input_cost_per_token"], rel_tol=1e-12)
assert math.isclose(completion_cost, 500 * info["output_cost_per_token"], rel_tol=1e-12)

View file

@ -40,7 +40,9 @@ class TestXAICostCalculator:
"""Test basic cost calculation without reasoning tokens."""
usage = Usage(prompt_tokens=12, completion_tokens=125, total_tokens=137)
prompt_cost, completion_cost = cost_per_token(model="grok-3-mini", usage=usage)
input_bd, output_bd = cost_per_token(model="grok-3-mini", usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs for grok-3-mini:
# Input: 12 tokens * $3e-7 = $0.0000036
@ -66,7 +68,9 @@ class TestXAICostCalculator:
),
)
prompt_cost, completion_cost = cost_per_token(model="grok-3-mini", usage=usage)
input_bd, output_bd = cost_per_token(model="grok-3-mini", usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs for grok-3-mini:
# Input: 12 tokens * $3e-7 = $0.0000036
@ -92,7 +96,9 @@ class TestXAICostCalculator:
),
)
prompt_cost, completion_cost = cost_per_token(model="grok-3-mini", usage=usage)
input_bd, output_bd = cost_per_token(model="grok-3-mini", usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs for grok-3-mini:
# Input: 12 tokens * $3e-7 = $0.0000036
@ -119,7 +125,9 @@ class TestXAICostCalculator:
),
)
prompt_cost, completion_cost = cost_per_token(model="grok-4", usage=usage)
input_bd, output_bd = cost_per_token(model="grok-4", usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs for grok-4:
# Input: 10 tokens * $3e-6 = $0.00003
@ -145,9 +153,11 @@ class TestXAICostCalculator:
),
)
prompt_cost, completion_cost = cost_per_token(
input_bd, output_bd = cost_per_token(
model="grok-3-fast-beta", usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs for grok-3-fast-beta:
# Input: 20 tokens * $5e-6 = $0.0001
@ -162,7 +172,9 @@ class TestXAICostCalculator:
"""Test cost calculation when completion_tokens_details is not present."""
usage = Usage(prompt_tokens=12, completion_tokens=125, total_tokens=137)
prompt_cost, completion_cost = cost_per_token(model="grok-3-mini", usage=usage)
input_bd, output_bd = cost_per_token(model="grok-3-mini", usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Should fall back to basic calculation
expected_prompt_cost = 12 * 3e-7
@ -186,7 +198,9 @@ class TestXAICostCalculator:
),
)
prompt_cost, completion_cost = cost_per_token(model="grok-3-mini", usage=usage)
input_bd, output_bd = cost_per_token(model="grok-3-mini", usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs:
# Input: 12 tokens * $3e-7 = $0.0000036
@ -213,7 +227,9 @@ class TestXAICostCalculator:
),
)
prompt_cost, completion_cost = cost_per_token(model="xai/grok-4-fast-reasoning", usage=usage)
input_bd, output_bd = cost_per_token(model="xai/grok-4-fast-reasoning", usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs for grok-4-fast-reasoning with tiered pricing:
# Input: 150000 tokens * $0.4e-6 (ALL tokens at tiered rate since input > 128k) = $0.06
@ -240,7 +256,9 @@ class TestXAICostCalculator:
),
)
prompt_cost, completion_cost = cost_per_token(model="xai/grok-4-fast-reasoning", usage=usage)
input_bd, output_bd = cost_per_token(model="xai/grok-4-fast-reasoning", usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs for grok-4-fast-reasoning with regular pricing:
# Input: 100000 tokens * $0.2e-6 (regular rate) = $0.02
@ -266,7 +284,9 @@ class TestXAICostCalculator:
),
)
prompt_cost, completion_cost = cost_per_token(model="xai/grok-4-latest", usage=usage)
input_bd, output_bd = cost_per_token(model="xai/grok-4-latest", usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs for grok-4-latest with tiered pricing:
# Input: 200000 tokens * $6e-6 (ALL tokens at tiered rate since input > 128k) = $1.2
@ -292,7 +312,9 @@ class TestXAICostCalculator:
),
)
prompt_cost, completion_cost = cost_per_token(model="xai/grok-4-fast-reasoning", usage=usage)
input_bd, output_bd = cost_per_token(model="xai/grok-4-fast-reasoning", usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Expected costs for grok-4-fast-reasoning:
# Input: 150000 tokens * $0.4e-6 (ALL tokens at tiered rate since input > 128k) = $0.06
@ -311,7 +333,9 @@ class TestXAICostCalculator:
total_tokens=200000,
)
prompt_cost, completion_cost = cost_per_token(model="grok-3-mini", usage=usage)
input_bd, output_bd = cost_per_token(model="grok-3-mini", usage=usage)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# grok-3-mini doesn't have tiered pricing, so should use regular rates:
# Input: 150000 tokens * $3e-7 (regular rate) = $0.045
@ -396,9 +420,11 @@ class TestXAICostCalculator:
"""Test cost calculation for grok-4.20-beta-0309-reasoning model."""
usage = Usage(prompt_tokens=100, completion_tokens=200, total_tokens=300)
prompt_cost, completion_cost = cost_per_token(
input_bd, output_bd = cost_per_token(
model="grok-4.20-beta-0309-reasoning", usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Input: 100 tokens * $2e-6 = $0.0002
# Output: 200 tokens * $6e-6 = $0.0012
@ -412,9 +438,11 @@ class TestXAICostCalculator:
"""Test cost calculation for grok-4.20-beta-0309-non-reasoning model."""
usage = Usage(prompt_tokens=50, completion_tokens=100, total_tokens=150)
prompt_cost, completion_cost = cost_per_token(
input_bd, output_bd = cost_per_token(
model="grok-4.20-beta-0309-non-reasoning", usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Input: 50 tokens * $2e-6 = $0.0001
# Output: 100 tokens * $6e-6 = $0.0006
@ -428,9 +456,11 @@ class TestXAICostCalculator:
"""Test cost calculation for grok-4.20-multi-agent-beta-0309 model."""
usage = Usage(prompt_tokens=200, completion_tokens=300, total_tokens=500)
prompt_cost, completion_cost = cost_per_token(
input_bd, output_bd = cost_per_token(
model="grok-4.20-multi-agent-beta-0309", usage=usage
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# Input: 200 tokens * $2e-6 = $0.0004
# Output: 300 tokens * $6e-6 = $0.0018

View file

@ -82,11 +82,13 @@ def test_zai_glm46_cost_calculation():
key = "zai/glm-4.6"
info = litellm.model_cost[key]
prompt_cost, completion_cost = cost_per_token(
input_bd, output_bd = cost_per_token(
model="zai/glm-4.6",
prompt_tokens=1000000, # 1M tokens
completion_tokens=1000000,
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# GLM-4.6: $0.6/M input, $2.2/M output
assert math.isclose(prompt_cost, 0.6, rel_tol=1e-6)
@ -128,11 +130,13 @@ def test_glm47_cost_calculation():
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
prompt_cost, completion_cost = cost_per_token(
input_bd, output_bd = cost_per_token(
model="zai/glm-4.7",
prompt_tokens=1000000, # 1M tokens
completion_tokens=1000000,
)
prompt_cost = input_bd["total"]
completion_cost = output_bd["total"]
# GLM-4.7: $0.6/M input, $2.2/M output (same as GLM-4.6)
assert math.isclose(prompt_cost, 0.6, rel_tol=1e-6)

View file

@ -17,7 +17,7 @@ from litellm.cost_calculator import (
response_cost_calculator,
)
from litellm.types.llms.openai import OpenAIRealtimeStreamList
from litellm.types.utils import ModelResponse, PromptTokensDetailsWrapper, Usage
from litellm.types.utils import CompletionTokensDetailsWrapper, ModelResponse, PromptTokensDetailsWrapper, Usage
from litellm.utils import TranscriptionResponse
@ -961,11 +961,13 @@ def test_gemini_25_explicit_caching_cost_direct_usage():
),
)
input_cost, output_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model="gemini/gemini-2.5-pro",
usage=usage,
custom_llm_provider="gemini",
)
input_cost = _input_bd["total"]
output_cost = _output_bd["total"]
total_cost = input_cost + output_cost
@ -1035,11 +1037,13 @@ def test_azure_ai_cache_cost_calculation():
cache_creation_input_tokens=100, # 100 cache creation tokens
)
input_cost, output_cost = generic_cost_per_token(
_input_bd, _output_bd = generic_cost_per_token(
model=test_model_id,
usage=usage,
custom_llm_provider="azure_ai",
)
input_cost = _input_bd["total"]
output_cost = _output_bd["total"]
total_cost = input_cost + output_cost
@ -1970,3 +1974,67 @@ def test_additional_costs_only_for_azure_ai():
completion_tokens=50,
)
assert result is None, "Vertex AI should have no additional costs"
def test_completion_cost_stores_granular_breakdown_on_logging_obj():
"""Test that completion_cost populates input/output cost breakdowns on the logging obj."""
from unittest.mock import MagicMock
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
response = ModelResponse(
id="test-id",
model="gpt-4o",
choices=[],
usage=Usage(
prompt_tokens=500,
completion_tokens=200,
total_tokens=700,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=100, text_tokens=400,
),
),
)
mock_logging_obj = MagicMock()
mock_logging_obj.model = "gpt-4o"
mock_logging_obj.cost_breakdown = None
cost = completion_cost(
completion_response=response,
model="gpt-4o",
custom_llm_provider="openai",
litellm_logging_obj=mock_logging_obj,
)
assert cost > 0
mock_logging_obj.set_cost_breakdown.assert_called_once()
call_kwargs = mock_logging_obj.set_cost_breakdown.call_args
if call_kwargs.kwargs:
assert "input_cost_breakdown" in call_kwargs.kwargs
assert "output_cost_breakdown" in call_kwargs.kwargs
assert call_kwargs.kwargs["input_cost_breakdown"]["total"] > 0
else:
pass
def test_completion_cost_works_without_logging_obj():
"""Test that completion_cost works fine when no logging obj is passed."""
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
response = ModelResponse(
id="test-id",
model="gpt-4o",
choices=[],
usage=Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150),
)
cost = completion_cost(
completion_response=response,
model="gpt-4o",
custom_llm_provider="openai",
)
assert isinstance(cost, float)
assert cost > 0