This commit is contained in:
AaronHowell 2026-11-06 07:23:13 +00:00 • committed by GitHub
commit 52ef5e8984
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
15 changed files with 545 additions and 62 deletions

View file

@ -157,6 +157,7 @@ from litellm.utils import (
CostPerToken,
EmbeddingResponse,
ImageResponse,
ModelInfoLookupMiss,
ModelResponse,
ModelResponseStream,
ProviderConfigManager,
@ -164,6 +165,7 @@ from litellm.utils import (
TranscriptionResponse,
cached_cost_map_key,
cached_get_model_info_helper,
lookup_model_info_for_registration,
mapped_model_fill_missing,
model_info_with_pricing_overrides,
runtime_model_cost_registrations,
@ -519,6 +521,7 @@ def cost_per_token(
request_model: str | None = None, # original request model for router detection
custom_model_info: OCRPricing | None = None,
bedrock_model_info: ModelInfo | None = None,
deployment_model_info: ModelInfo | None = None,
) -> tuple[float, float]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@ -604,7 +607,7 @@ def cost_per_token(
cached_tokens=_cache_read_tokens,
cache_creation_tokens=_cache_creation_tokens,
custom_cost_per_second=custom_cost_per_second,
custom_cost_per_token=custom_cost_per_token,
custom_cost_per_token=None if deployment_model_info is not None else custom_cost_per_token,
)
if response_cost is not None:
@ -813,19 +816,27 @@ def cost_per_token(
optional_params=(getattr(response, "_hidden_params", None) if response else None),
)
elif (
per_second_cost := _per_second_pricing_cost(
model=model,
custom_llm_provider=custom_llm_provider,
response_time_ms=response_time_ms,
audio_seconds=audio_transcription_file_duration,
deployment_model_info is None
and (
per_second_cost := _per_second_pricing_cost(
model=model,
custom_llm_provider=custom_llm_provider,
response_time_ms=response_time_ms,
audio_seconds=audio_transcription_file_duration,
)
)
) is not None:
is not None
):
return per_second_cost
elif custom_llm_provider == "vertex_ai":
cost_router: Final = google_cost_router(
model=model_without_prefix,
custom_llm_provider=custom_llm_provider,
call_type=call_type,
cost_router: Final = (
"cost_per_token"
if deployment_model_info is not None
else google_cost_router(
model=model_without_prefix,
custom_llm_provider=custom_llm_provider,
call_type=call_type,
)
)
if cost_router == "cost_per_character":
return google_cost_per_character(
@ -844,11 +855,16 @@ def cost_per_token(
usage=usage_block,
service_tier=service_tier,
vertex_location=vertex_location,
deployment_model_info=deployment_model_info,
)
elif custom_llm_provider == "anthropic":
return anthropic_cost_per_token(model=model, usage=usage_block, service_tier=service_tier)
return anthropic_cost_per_token(
model=model, usage=usage_block, service_tier=service_tier, model_info=deployment_model_info
)
elif custom_llm_provider == "bedrock":
resolved_bedrock_info: Final = bedrock_model_info or _bedrock_pricing_info(model, None, "bedrock", False)
resolved_bedrock_info: Final = (
deployment_model_info or bedrock_model_info or _bedrock_pricing_info(model, None, "bedrock", False)
)
if resolved_bedrock_info is not None:
return generic_cost_per_token(
model=model,
@ -864,36 +880,47 @@ def cost_per_token(
usage=usage_block,
service_tier=service_tier,
data_residency=data_residency,
model_info=deployment_model_info,
)
elif custom_llm_provider == "databricks":
return databricks_cost_per_token(model=model, usage=usage_block, service_tier=service_tier)
return databricks_cost_per_token(
model=model, usage=usage_block, service_tier=service_tier, model_info=deployment_model_info
)
elif custom_llm_provider == "fireworks_ai":
return fireworks_ai_cost_per_token(model=model, usage=usage_block)
return fireworks_ai_cost_per_token(model=model, usage=usage_block, deployment_model_info=deployment_model_info)
elif custom_llm_provider == "azure":
return azure_openai_cost_per_token(
model=model,
usage=usage_block,
response_time_ms=response_time_ms,
service_tier=service_tier,
model_info=deployment_model_info,
)
elif custom_llm_provider == "gemini":
return gemini_cost_per_token(model=model, usage=usage_block, service_tier=service_tier)
return gemini_cost_per_token(
model=model, usage=usage_block, service_tier=service_tier, model_info=deployment_model_info
)
elif custom_llm_provider == "deepseek":
return deepseek_cost_per_token(model=model, usage=usage_block)
return deepseek_cost_per_token(model=model, usage=usage_block, model_info=deployment_model_info)
elif custom_llm_provider == "tencent":
return tencent_cost_per_token(model=model, usage=usage_block)
return tencent_cost_per_token(model=model, usage=usage_block, model_info=deployment_model_info)
elif custom_llm_provider == "perplexity":
return perplexity_cost_per_token(model=model, usage=usage_block)
return perplexity_cost_per_token(model=model, usage=usage_block, deployment_model_info=deployment_model_info)
elif custom_llm_provider == "xai":
return xai_cost_per_token(model=model, usage=usage_block)
return xai_cost_per_token(model=model, usage=usage_block, model_info=deployment_model_info)
elif custom_llm_provider == "lemonade":
return lemonade_cost_per_token(model=model, usage=usage_block)
return lemonade_cost_per_token(model=model, usage=usage_block, model_info=deployment_model_info)
elif custom_llm_provider in ("dashscope", "qwencloud", "qwen_ai_platform"):
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, custom_llm_provider=custom_llm_provider)
return dashscope_cost_per_token(
model=model,
usage=usage_block,
custom_llm_provider=custom_llm_provider,
deployment_model_info=deployment_model_info,
)
elif custom_llm_provider == "azure_ai":
return azure_ai_cost_per_token(
model=model,
@ -901,16 +928,22 @@ def cost_per_token(
response_time_ms=response_time_ms,
request_model=request_model,
service_tier=service_tier,
model_info=deployment_model_info,
)
else:
model_info: Final = cached_get_model_info_helper(model=model, custom_llm_provider=custom_llm_provider)
if _has_token_or_tiered_pricing(model_info):
model_info: Final = (
deployment_model_info
if deployment_model_info is not None
else cached_get_model_info_helper(model=model, custom_llm_provider=custom_llm_provider)
)
if deployment_model_info is not None or _has_token_or_tiered_pricing(model_info):
return generic_cost_per_token(
model=model,
usage=usage_block,
custom_llm_provider=custom_llm_provider,
service_tier=service_tier,
data_residency=data_residency,
model_info=deployment_model_info,
)
verbose_logger.debug("No per-token, tiered, or per-second pricing for model=%s; cost is 0", model)
return 0.0, 0.0
@ -1858,9 +1891,20 @@ def completion_cost(
cache_creation_input_tokens: int | None = None
cache_read_input_tokens: int | None = None
audio_transcription_file_duration: float = 0.0
deployment_token_info: Final = _deployment_token_model_info(
model,
custom_llm_provider,
litellm_logging_obj,
custom_pricing,
router_model_id,
pricing_provider,
completion_response,
)
provider_usage_object: Final = get_usage_object(completion_response=completion_response)
cost_per_token_usage_object: Final[Usage | None] = (
_without_provider_stated_cost(provider_usage_object) if custom_pricing else provider_usage_object
_without_provider_stated_cost(provider_usage_object)
if custom_pricing or deployment_token_info is not None
else provider_usage_object
)
rerank_billed_units: RerankBilledUnits | None = None
@ -1873,7 +1917,7 @@ def completion_cost(
served=_served_service_tier(completion_response, cost_per_token_usage_object),
)
explicit_pricing: Final = custom_pricing is True or base_model is not None
explicit_pricing: Final = custom_pricing is True or base_model is not None or deployment_token_info is not None
caller_set_prices: Final = (
explicit_pricing or custom_cost_per_token is not None or custom_cost_per_second is not None
)
@ -2323,6 +2367,7 @@ def completion_cost(
response=completion_response,
custom_model_info=_ocr_model_info(litellm_logging_obj, custom_pricing, router_model_id),
bedrock_model_info=bedrock_model_info,
deployment_model_info=deployment_token_info,
)
# Get additional costs from provider (e.g., routing fees, infrastructure costs)
@ -2413,8 +2458,8 @@ def completion_cost(
service_tier=service_tier,
data_residency=data_residency,
vertex_location=vertex_location,
custom_cost_per_token=custom_cost_per_token,
model_info=bedrock_model_info,
custom_cost_per_token=None if deployment_token_info is not None else custom_cost_per_token,
model_info=deployment_token_info or bedrock_model_info,
)
_reasoning_cost = _token_type_breakdown.reasoning_cost
_cache_read_cost = _token_type_breakdown.cache_read_cost
@ -2561,7 +2606,20 @@ def response_cost_calculator(
getattr(response_object, HIDDEN_PARAMS_ATTR),
)
provider_response_cost: Final = get_response_cost_from_hidden_params(hidden_params)
if provider_response_cost is not None and not custom_pricing:
if (
provider_response_cost is not None
and not custom_pricing
and _deployment_token_model_info(
model,
custom_llm_provider,
litellm_logging_obj,
custom_pricing,
router_model_id,
pricing_provider,
response_object,
)
is None
):
if _breakdown_total(litellm_logging_obj) != provider_response_cost:
header_priced_model: Final = select_model_name_for_cost_calc(
model=model,
@ -2680,6 +2738,65 @@ def _deployment_model_info(
)
def _deployment_token_model_info(
model: str | None,
custom_llm_provider: str | None,
litellm_logging_obj: LitellmLoggingObject | None,
custom_pricing: bool | None,
router_model_id: str | None,
pricing_provider: str | None = None,
response: object | None = None,
) -> ModelInfo | None:
if router_model_id is None:
return None
registration: Final = next(
(
entry
for key, entry in runtime_model_cost_registrations(router_model_id, custom_llm_provider or "")
if key == router_model_id
),
None,
)
if not custom_pricing and (registration is None or not _has_deployment_token_pricing(registration)):
return None
deployment_info: Final = _deployment_model_info(litellm_logging_obj, True, router_model_id)
if deployment_info is None:
return None
if not _has_deployment_token_pricing(deployment_info):
return None
if litellm_logging_obj is not None and litellm_logging_obj.spilled_over_ptu_for(response):
return None
if model is None:
return deployment_info
prefix: Final = model.partition("/")[0]
public_provider: Final = pricing_provider or (prefix if prefix in LlmProvidersSet else custom_llm_provider)
public_info: Final = lookup_model_info_for_registration(model, public_provider)
if isinstance(public_info, ModelInfoLookupMiss):
return deployment_info
public_entry: Final = _raw_cost_map_entry(public_info["key"])
if public_entry is None or not (
_cost_map_entry_prices_anything(public_entry) or _has_deployment_token_pricing(public_entry)
):
return deployment_info
effective_info: Final = model_info_with_pricing_overrides(
{**public_info, "mode": public_info.get("mode") or "chat"}, deployment_info
)
off_peak: Final = deployment_info.get("off_peak_pricing")
return {
**effective_info,
"off_peak_pricing": off_peak if off_peak is not None else effective_info.get("off_peak_pricing"),
}
def _has_deployment_token_pricing(info: Mapping[str, object]) -> bool:
return any(
value is not None
and ("token" in field or field in ("tiered_pricing", "off_peak_pricing"))
and (is_server_derived_pricing_key(field) or field == "off_peak_pricing")
for field, value in info.items()
)
def _ocr_model_info(
litellm_logging_obj: LitellmLoggingObject | None,
custom_pricing: bool | None,

View file

@ -4,7 +4,7 @@ Helper util for handling azure openai-specific cost calculation
"""
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
from litellm.types.utils import Usage
from litellm.types.utils import ModelInfo, Usage
def cost_per_token(
@ -12,6 +12,7 @@ def cost_per_token(
usage: Usage,
response_time_ms: float | None = 0.0,
service_tier: str | None = None,
model_info: ModelInfo | None = None,
) -> tuple[float, float]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@ -30,4 +31,5 @@ def cost_per_token(
usage=usage,
custom_llm_provider="azure",
service_tier=service_tier,
model_info=model_info,
)

View file

@ -7,7 +7,7 @@ from typing import Final
from litellm._logging import verbose_logger
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
from litellm.types.utils import Usage
from litellm.types.utils import ModelInfo, Usage
from litellm.utils import get_model_info
@ -63,10 +63,12 @@ def calculate_azure_model_router_flat_cost(model: str, prompt_tokens: int) -> fl
return 0.0
def _response_model_cost(model: str, usage: Usage, service_tier: str | None) -> tuple[float, float]:
def _response_model_cost(
model: str, usage: Usage, service_tier: str | None, model_info: ModelInfo | None
) -> tuple[float, float]:
try:
return generic_cost_per_token(
model=model, usage=usage, custom_llm_provider="azure_ai", service_tier=service_tier
model=model, usage=usage, custom_llm_provider="azure_ai", service_tier=service_tier, model_info=model_info
)
except Exception as e:
if not is_azure_model_router(model):
@ -93,6 +95,7 @@ def cost_per_token(
response_time_ms: float | None = 0.0,
request_model: str | None = None,
service_tier: str | None = None,
model_info: ModelInfo | None = None,
) -> tuple[float, float]:
"""
Price the response model's own tokens for Azure AI, plus the Model Router fee exactly once when either the
@ -111,6 +114,7 @@ def cost_per_token(
response_time_ms: Optional response time in milliseconds
request_model: Optional[str], the original request model name; a Model Router name adds the routing fee
service_tier: Optional service tier the request was priced on
model_info: Optional resolved deployment prices
Returns:
Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
@ -118,7 +122,9 @@ def cost_per_token(
Raises:
ValueError: If a model that is not a Model Router name is missing from the cost map
"""
prompt_cost, completion_cost = _response_model_cost(model=model, usage=usage, service_tier=service_tier)
prompt_cost, completion_cost = _response_model_cost(
model=model, usage=usage, service_tier=service_tier, model_info=model_info
)
fee_name: Final = _router_fee_name(model=model, request_model=request_model)
if fee_name is None:
return prompt_cost, completion_cost

View file

@ -115,6 +115,7 @@ def cost_per_token(
usage: Usage,
custom_llm_provider: str = "dashscope",
current_time: datetime | None = None,
deployment_model_info: ModelInfo | None = None,
) -> tuple[float, float]:
"""
Calculate cost per token for Dashscope models.
@ -127,11 +128,16 @@ def cost_per_token(
usage: LiteLLM Usage block
custom_llm_provider: The provider id the request resolved to; dashscope or one of its brand aliases
current_time: The moment the request is billed at; defaults to now, UTC
deployment_model_info: Optional resolved deployment prices
Returns:
Tuple[float, float] - (prompt_cost_in_usd, completion_cost_in_usd)
"""
model_info: Final = get_model_info(model=model, custom_llm_provider=custom_llm_provider)
model_info: Final = (
deployment_model_info
if deployment_model_info is not None
else get_model_info(model=model, custom_llm_provider=custom_llm_provider)
)
breakdown: Final = _extract_token_breakdown(usage)
raw_tiers: Final = model_info.get("tiered_pricing")
tiered_pricing: Final = raw_tiers if isinstance(raw_tiers, list) else None

View file

@ -7,7 +7,7 @@ from types import MappingProxyType
from typing import Final
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
from litellm.types.utils import Usage
from litellm.types.utils import ModelInfo, Usage
_LEGACY_ENDPOINT_NAMES: Final = MappingProxyType(
{
@ -30,7 +30,9 @@ def _registry_key(model: str) -> str:
)
def cost_per_token(model: str, usage: Usage, service_tier: str | None = None) -> tuple[float, float]:
def cost_per_token(
model: str, usage: Usage, service_tier: str | None = None, model_info: ModelInfo | None = None
) -> tuple[float, float]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@ -46,4 +48,5 @@ def cost_per_token(model: str, usage: Usage, service_tier: str | None = None) ->
usage=usage,
custom_llm_provider="databricks",
service_tier=service_tier,
model_info=model_info,
)

View file

@ -5,13 +5,13 @@ Handles prompt caching scenario.
"""
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
from litellm.types.utils import Usage
from litellm.types.utils import ModelInfo, Usage
def cost_per_token(model: str, usage: Usage) -> tuple[float, float]:
def cost_per_token(model: str, usage: Usage, model_info: ModelInfo | None = None) -> tuple[float, float]:
"""
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.
"""
return generic_cost_per_token(model=model, usage=usage, custom_llm_provider="deepseek")
return generic_cost_per_token(model=model, usage=usage, custom_llm_provider="deepseek", model_info=model_info)

View file

@ -71,7 +71,9 @@ def _resolve_routed_model_info(model: str) -> ModelInfo:
return get_model_info(model=base_model, custom_llm_provider="fireworks_ai")
def cost_per_token(model: str, usage: Usage, current_time: datetime | None = None) -> tuple[float, float]:
def cost_per_token(
model: str, usage: Usage, current_time: datetime | None = None, deployment_model_info: ModelInfo | None = None
) -> tuple[float, float]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens,
swapping in the model's off_peak_pricing rates while one of its windows is open.
@ -84,11 +86,11 @@ def cost_per_token(model: str, usage: Usage, current_time: datetime | None = Non
Returns:
Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
"""
model_info: Final = _resolve_model_info(model)
model_info: Final = deployment_model_info if deployment_model_info is not None else _resolve_model_info(model)
return generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=model_info["litellm_provider"],
custom_llm_provider="fireworks_ai" if deployment_model_info is not None else model_info["litellm_provider"],
model_info=model_info,
current_time=current_time,
)

View file

@ -10,7 +10,9 @@ if TYPE_CHECKING:
from litellm.types.utils import ModelInfo, Usage
def cost_per_token(model: str, usage: "Usage", service_tier: str | None = None) -> tuple[float, float]:
def cost_per_token(
model: str, usage: "Usage", service_tier: str | None = None, model_info: "ModelInfo | None" = None
) -> tuple[float, float]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@ -23,6 +25,7 @@ def cost_per_token(model: str, usage: "Usage", service_tier: str | None = None)
usage=usage,
custom_llm_provider="gemini",
service_tier=service_tier,
model_info=model_info,
)

View file

@ -7,28 +7,33 @@ This prevents cost calculation errors when using models not in model_prices_and_
from typing import Final
from litellm.types.utils import Usage
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
from litellm.types.utils import ModelInfo, Usage
def cost_per_token(
model: str,
usage: Usage,
model_info: ModelInfo | None = None,
) -> tuple[float, float]:
"""
Calculate cost per token for Lemonade models.
Calculate configured token costs, or return zero for unpriced 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
without errors for any Lemonade model, regardless of whether it's in the
Since Lemonade is a local/self-hosted deployment, unpriced models have no per-token costs.
This function returns (0.0, 0.0) for those models, regardless of whether they are in the
model_prices_and_context_window.json file.
Args:
model: The model name (with or without "lemonade/" prefix)
usage: Usage object containing token counts
model_info: Optional resolved deployment prices
Returns:
Tuple of (prompt_cost, completion_cost) - always (0.0, 0.0) for Lemonade
Tuple of (prompt_cost, completion_cost)
"""
if model_info is not None:
return generic_cost_per_token(model=model, usage=usage, custom_llm_provider="lemonade", model_info=model_info)
# Lemonade is self-hosted/local, so cost is always 0
prompt_cost: Final = 0.0
completion_cost: Final = 0.0

View file

@ -24,6 +24,7 @@ def cost_per_token(
usage: Usage,
service_tier: str | None = None,
data_residency: str | None = None,
model_info: ModelInfo | None = None,
) -> tuple[float, float]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@ -44,6 +45,7 @@ def cost_per_token(
custom_llm_provider="openai",
service_tier=service_tier,
data_residency=data_residency,
model_info=model_info,
)

View file

@ -8,11 +8,13 @@ from typing import Final
from litellm.litellm_core_utils.core_helpers import finite_reported_cost
from litellm.litellm_core_utils.llm_cost_calc.utils import TokenRates, apply_off_peak_pricing
from litellm.types.utils import Usage
from litellm.types.utils import ModelInfo, Usage
from litellm.utils import get_model_info
def cost_per_token(model: str, usage: Usage, current_time: datetime | None = None) -> tuple[float, float]:
def cost_per_token(
model: str, usage: Usage, current_time: datetime | None = None, deployment_model_info: ModelInfo | None = None
) -> tuple[float, float]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
The manual fallback swaps in the model's off_peak_pricing rates while one of its windows is open.
@ -35,13 +37,19 @@ def cost_per_token(model: str, usage: Usage, current_time: datetime | None = Non
total_cost = cost_info.get("total_cost")
elif isinstance(cost_info, (int, float)) and not isinstance(cost_info, bool):
total_cost = float(cost_info)
reported_cost: Final = None if total_cost is None else finite_reported_cost(float(total_cost))
reported_cost: Final = (
None if total_cost is None or deployment_model_info is not None else finite_reported_cost(float(total_cost))
)
if reported_cost is not None:
return (0.0, reported_cost)
## FALLBACK: Calculate cost manually if Perplexity doesn't provide it
## GET MODEL INFO
model_info: Final = get_model_info(model=model, custom_llm_provider="perplexity")
model_info: Final = (
deployment_model_info
if deployment_model_info is not None
else get_model_info(model=model, custom_llm_provider="perplexity")
)
def _safe_float_cast(value: str | float | None | object, default: float = 0.0) -> float:
"""Safely cast a value to float with proper type handling for mypy."""

View file

@ -9,8 +9,8 @@ from litellm.litellm_core_utils.llm_cost_calc.utils import (
from litellm.types.utils import ImageResponse, ModelInfo, Usage
def cost_per_token(model: str, usage: Usage) -> tuple[float, float]:
return generic_cost_per_token(model=model, usage=usage, custom_llm_provider="tencent")
def cost_per_token(model: str, usage: Usage, model_info: ModelInfo | None = None) -> tuple[float, float]:
return generic_cost_per_token(model=model, usage=usage, custom_llm_provider="tencent", model_info=model_info)
def _deployment_image_token_pricing(model_info: ModelInfo | None) -> ModelInfo | None:

View file

@ -207,6 +207,7 @@ def cost_per_token(
usage: Usage,
service_tier: str | None = None,
vertex_location: str | None = None,
deployment_model_info: ModelInfo | None = None,
) -> tuple[float, float]:
"""
Calculates the cost per token for a given model, prompt tokens, and completion tokens.
@ -229,7 +230,11 @@ def cost_per_token(
"""
## GET MODEL INFO
model_info: Final = litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider)
model_info: Final = (
deployment_model_info
if deployment_model_info is not None
else litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider)
)
## HANDLE 128k+ PRICING
input_cost_per_token_above_128k_tokens: Final = model_info.get("input_cost_per_token_above_128k_tokens")
@ -248,4 +253,5 @@ def cost_per_token(
usage=usage,
service_tier=service_tier,
vertex_location=vertex_location,
model_info=model_info,
)

View file

@ -62,7 +62,7 @@ def _split_reported_cost(model: str, usage: Usage, reported_cost: float) -> tupl
return input_cost, reported_cost - input_cost
def cost_per_token(model: str, usage: Usage) -> tuple[float, float]:
def cost_per_token(model: str, usage: Usage, model_info: "ModelInfo | None" = None) -> tuple[float, float]:
"""
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.
@ -74,13 +74,13 @@ def cost_per_token(model: str, usage: Usage) -> tuple[float, float]:
Returns:
Tuple[float, float] - prompt_cost_in_usd, completion_cost_in_usd
"""
reported_cost: Final = _cost_reported_by_xai(usage)
reported_cost: Final = _cost_reported_by_xai(usage) if model_info is None else None
if reported_cost is not None:
return _split_reported_cost(model, usage, reported_cost)
return _token_math_cost_per_token(model, usage)
return _token_math_cost_per_token(model, usage, model_info)
def _token_math_cost_per_token(model: str, usage: Usage) -> tuple[float, float]:
def _token_math_cost_per_token(model: str, usage: Usage, model_info: "ModelInfo | None" = None) -> tuple[float, float]:
# XAI-specific completion cost: completion is billed as visible + reasoning
# tokens. Detect when the transformation layer already folded them so we
# don't double-count; fall back to raw xAI shape for callers that bypass
@ -105,7 +105,9 @@ def _token_math_cost_per_token(model: str, usage: Usage) -> tuple[float, float]:
completion_tokens_details=None,
)
prompt_cost, completion_cost = generic_cost_per_token(model=model, usage=modified_usage, custom_llm_provider="xai")
prompt_cost, completion_cost = generic_cost_per_token(
model=model, usage=modified_usage, custom_llm_provider="xai", model_info=model_info
)
return prompt_cost, completion_cost

View file

@ -3,6 +3,7 @@ import asyncio
import datetime
import time
from collections.abc import Iterator, Mapping
from itertools import product
from types import MappingProxyType, SimpleNamespace
from typing import Final, Literal, cast
@ -9650,6 +9651,294 @@ def isolated_pricing_registry(local_model_cost_map: None, monkeypatch: pytest.Mo
monkeypatch.setattr(litellm, "models_by_provider", dict(litellm.models_by_provider))
@pytest.mark.parametrize(
("adapter", "pricing"),
tuple(product(
("anthropic", "openai", "azure", "azure_ai", "gemini", "databricks", "bedrock", "vertex_ai", "bigmodel"),
("flat", "zero", "off_peak", "priority", "tiered"),
)) + tuple(product(
("deepseek", "tencent", "fireworks_ai", "xai", "dashscope", "lemonade"),
("flat", "zero", "off_peak", "tiered"),
)),
)
@pytest.mark.parametrize("legacy_flat_price", (False, True))
@pytest.mark.parametrize("custom_pricing", (False, True))
@pytest.mark.usefixtures("isolated_pricing_registry")
def test_selected_deployment_prices_survive_transport_provider_changes(
adapter: str, pricing: str, legacy_flat_price: bool, custom_pricing: bool
) -> None:
from litellm._internal_context import pinned_billing_time
deployment_id: Final = "transport-pricing-deployment"
rates: Final = {
"input_cost_per_token": 0.001,
"output_cost_per_token": 0.002,
"cache_read_input_token_cost": 0.0001,
"cache_creation_input_token_cost": 0.00125,
}
effective_rates: Final = (
{field: 0.0 for field in rates}
if pricing == "zero"
else {field: rate / 2 for field, rate in rates.items()}
if pricing == "off_peak"
else {field: rate * 2 for field, rate in rates.items()}
if pricing in ("priority", "tiered")
else rates
)
entry: Final = {
**(effective_rates if pricing == "zero" else rates),
**({"off_peak_pricing": {"hours_utc": "00:00-00:00", **effective_rates}} if pricing == "off_peak" else {}),
**({f"{field}_priority": rate for field, rate in effective_rates.items()} if pricing == "priority" else {}),
**({"tiered_pricing": [{"range": [0, 100], **effective_rates}]} if pricing == "tiered" else {}),
"mode": "chat",
}
response: Final = ModelResponse(
model="unmapped-upstream-model",
usage=Usage(
prompt_tokens=40,
completion_tokens=1,
total_tokens=41,
prompt_tokens_details=PromptTokensDetailsWrapper(text_tokens=24, cached_tokens=10, cache_creation_tokens=6),
),
)
expected_input: Final = (
24 * effective_rates["input_cost_per_token"]
+ 10 * effective_rates["cache_read_input_token_cost"]
+ 6 * effective_rates["cache_creation_input_token_cost"]
)
expected_output: Final = effective_rates["output_cost_per_token"]
for provider in ("deepseek", adapter, "deepseek"):
litellm.register_model({deployment_id: {**entry, "litellm_provider": provider}})
logging_obj: Final = Logging(
model="deepseek/unmapped-upstream-model",
messages=[],
stream=False,
call_type="completion",
start_time=datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc),
litellm_call_id="transport-pricing",
function_id="transport-pricing",
)
with pinned_billing_time(datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)):
cost: Final = response_cost_calculator(
response_object=response,
model="deepseek/unmapped-upstream-model",
custom_llm_provider=adapter,
call_type="completion",
optional_params={},
custom_pricing=custom_pricing,
custom_cost_per_token=(
{"input_cost_per_token": rates["input_cost_per_token"], "output_cost_per_token": rates["output_cost_per_token"]}
if legacy_flat_price
else None
),
router_model_id=deployment_id,
service_tier="priority" if pricing == "priority" else None,
litellm_logging_obj=logging_obj,
)
assert cost == pytest.approx(expected_input + expected_output)
assert logging_obj.cost_breakdown["input_cost"] == pytest.approx(expected_input)
assert logging_obj.cost_breakdown["output_cost"] == pytest.approx(expected_output)
assert logging_obj.cost_breakdown.get("cache_read_cost", 0.0) == pytest.approx(10 * effective_rates["cache_read_input_token_cost"])
assert logging_obj.cost_breakdown.get("cache_creation_cost", 0.0) == pytest.approx(6 * effective_rates["cache_creation_input_token_cost"])
assert litellm.model_cost[deployment_id]["litellm_provider"] == provider
if provider != adapter and adapter in ("openai", "anthropic"):
with pytest.raises(litellm.ModelNotMappedError, match="isn't mapped yet"):
litellm.get_model_info(model=deployment_id, custom_llm_provider=adapter)
@pytest.mark.parametrize("field", ("input_cost_per_token", "output_cost_per_token", "cache_read_input_token_cost"))
@pytest.mark.parametrize("rate", (0.0, 0.001))
@pytest.mark.usefixtures("isolated_pricing_registry")
def test_deployment_partial_prices_keep_public_rates_and_sibling_isolation(field: str, rate: float) -> None:
public_model: Final = "openai/owned-public-pricing-model"
public_rates: Final = {
"input_cost_per_token": 0.005,
"output_cost_per_token": 0.008,
"cache_read_input_token_cost": 0.0005,
"cache_creation_input_token_cost": 0.00625,
}
litellm.register_model({public_model: {**public_rates, "litellm_provider": "openai", "mode": "chat"}})
router: Final = litellm.Router(
model_list=[
{
"model_name": "partial-priced",
"litellm_params": {"model": public_model, "api_key": "fake-key", field: rate},
"model_info": {"id": "partial-price-deployment"},
},
{
"model_name": "public-priced",
"litellm_params": {"model": public_model, "api_key": "fake-key"},
"model_info": {"id": "public-price-deployment"},
},
]
)
assert len(router.model_list) == 2
response: Final = ModelResponse(
model=public_model,
usage=Usage(
prompt_tokens=10,
completion_tokens=5,
total_tokens=15,
prompt_tokens_details=PromptTokensDetailsWrapper(text_tokens=4, cached_tokens=4, cache_creation_tokens=2),
),
)
effective_rates: Final = {**public_rates, field: rate}
assert completion_cost(
completion_response=response,
model=public_model,
custom_llm_provider="openai",
custom_pricing=True,
router_model_id="partial-price-deployment",
) == pytest.approx(
4 * effective_rates["input_cost_per_token"]
+ 4 * effective_rates["cache_read_input_token_cost"]
+ 2 * effective_rates["cache_creation_input_token_cost"]
+ 5 * effective_rates["output_cost_per_token"]
)
for deployment_id, selected_rates in (
("partial-price-deployment", effective_rates),
("public-price-deployment", public_rates),
):
assert completion_cost(
completion_response=response,
model=public_model,
custom_llm_provider="openai",
custom_pricing=False,
router_model_id=deployment_id,
) == pytest.approx(
4 * selected_rates["input_cost_per_token"]
+ 4 * selected_rates["cache_read_input_token_cost"]
+ 2 * selected_rates["cache_creation_input_token_cost"]
+ 5 * selected_rates["output_cost_per_token"]
)
@pytest.mark.parametrize("pricing", ("input", "zero", "cache", "complete", "off_peak"))
@pytest.mark.parametrize("prefixed_model", (False, True))
@pytest.mark.parametrize("custom_pricing", (False, True))
@pytest.mark.usefixtures("isolated_pricing_registry")
def test_deployment_prices_fill_only_missing_components_from_the_pricing_provider(
pricing: str, prefixed_model: bool, custom_pricing: bool, monkeypatch: pytest.MonkeyPatch
) -> None:
from litellm._internal_context import pinned_billing_time
public_model: Final = "owned-partial-pricing-model"
public_rates: Final = {
"input_cost_per_token": 0.005,
"output_cost_per_token": 0.008,
"cache_read_input_token_cost": 0.0005,
"cache_creation_input_token_cost": 0.00625,
}
monkeypatch.setitem(litellm.model_cost, f"deepseek/{public_model}", {**public_rates, "litellm_provider": "deepseek", "mode": "chat"})
monkeypatch.setitem(litellm.model_cost, f"anthropic/{public_model}", {**{field: value * 10 for field, value in public_rates.items()}, "litellm_provider": "anthropic", "mode": "chat"})
overrides: Final = (
{"input_cost_per_token": 0.001, "output_cost_per_token": 0.002}
if pricing == "complete"
else {"cache_read_input_token_cost": 0.0001}
if pricing == "cache"
else {"input_cost_per_token": 0.0 if pricing == "zero" else 0.001}
if pricing != "off_peak"
else {}
)
off_peak_rates: Final = {field: value / 2 for field, value in public_rates.items()}
effective_rates: Final = off_peak_rates if pricing == "off_peak" else {**public_rates, **overrides}
deployment_id: Final = "owned-partial-price-deployment"
litellm.register_model({deployment_id: {
**overrides,
**({"off_peak_pricing": {"hours_utc": "00:00-00:00", **off_peak_rates}} if pricing == "off_peak" else {}),
"litellm_provider": "deepseek",
"mode": "chat",
}})
response: Final = ModelResponse(model=public_model, usage=Usage(
prompt_tokens=10,
completion_tokens=5,
total_tokens=15,
prompt_tokens_details=PromptTokensDetailsWrapper(text_tokens=4, cached_tokens=4, cache_creation_tokens=2),
cost=99.0,
))
response._hidden_params["response_cost"] = 99.0
with pinned_billing_time(datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)):
cost: Final = response_cost_calculator(
response_object=response,
model=f"deepseek/{public_model}" if prefixed_model else public_model,
custom_llm_provider="anthropic",
pricing_provider=None if prefixed_model else "deepseek",
router_model_id=deployment_id,
custom_pricing=custom_pricing,
call_type="completion",
optional_params={},
)
assert cost == pytest.approx(
4 * effective_rates["input_cost_per_token"]
+ 4 * effective_rates["cache_read_input_token_cost"]
+ 2 * effective_rates["cache_creation_input_token_cost"]
+ 5 * effective_rates["output_cost_per_token"]
)
assert litellm.model_cost[f"deepseek/{public_model}"]["input_cost_per_token"] == public_rates["input_cost_per_token"]
assert litellm.model_cost[deployment_id]["litellm_provider"] == "deepseek"
@pytest.mark.parametrize("custom_pricing", (False, True))
@pytest.mark.usefixtures("isolated_pricing_registry")
def test_perplexity_deployment_prices_keep_citation_reasoning_and_search_charges(custom_pricing: bool) -> None:
deployment_id: Final = "perplexity-deployment-pricing"
litellm.register_model({deployment_id: {
"input_cost_per_token": 0.001,
"output_cost_per_token": 0.002,
"output_cost_per_reasoning_token": 0.003,
"citation_cost_per_token": 0.004,
"search_queries_cost_per_query": 0.005,
"litellm_provider": "deepseek",
"mode": "chat",
}})
response: Final = ModelResponse(model="owned-perplexity-model", usage=Usage(
prompt_tokens=10, completion_tokens=5, total_tokens=15,
citation_tokens=3, reasoning_tokens=2, cost={"total_cost": 99.0},
prompt_tokens_details=PromptTokensDetailsWrapper(web_search_requests=2),
))
assert completion_cost(
completion_response=response,
model="owned-perplexity-model",
custom_llm_provider="perplexity",
router_model_id=deployment_id,
custom_pricing=custom_pricing,
) == pytest.approx(10 * 0.001 + 3 * 0.004 + 3 * 0.002 + 2 * 0.003 + 2 * 0.005)
@pytest.mark.parametrize("pricing", ("off_peak", "priority", "tiered", "reasoning", "long_context"))
@pytest.mark.usefixtures("isolated_pricing_registry")
def test_deployment_flat_prices_override_public_rate_variants(pricing: str, monkeypatch: pytest.MonkeyPatch) -> None:
from litellm._internal_context import pinned_billing_time
public_model: Final = "deepseek/owned-public-rate-variants"
public_rates: Final = {"input_cost_per_token": 0.005, "output_cost_per_token": 0.008}
variant: Final = (
{"off_peak_pricing": {"hours_utc": "00:00-00:00", **public_rates}}
if pricing == "off_peak"
else {"tiered_pricing": [{"range": [0, 1000], **public_rates}]}
if pricing == "tiered"
else {"output_cost_per_reasoning_token": 0.01}
if pricing == "reasoning"
else {f"{field}_priority": rate for field, rate in public_rates.items()}
if pricing == "priority"
else {f"{field}_above_100_tokens": rate for field, rate in public_rates.items()}
)
monkeypatch.setitem(litellm.model_cost, public_model, {**public_rates, **variant, "litellm_provider": "deepseek", "mode": "chat"})
deployment_id: Final = "owned-flat-price-deployment"
litellm.register_model({deployment_id: {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002, "litellm_provider": "deepseek", "mode": "chat"}})
response: Final = ModelResponse(model=public_model, usage=Usage(prompt_tokens=200, completion_tokens=20, total_tokens=220, completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=5)))
with pinned_billing_time(datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)):
cost: Final = completion_cost(
completion_response=response,
model=public_model,
custom_llm_provider="openai",
router_model_id=deployment_id,
custom_pricing=True,
service_tier="priority" if pricing == "priority" else None,
)
assert cost == pytest.approx(200 * 0.001 + 20 * 0.002)
_CUSTOM_PRICING_MODEL: Final = "migration-custom-pricing-chat"
_CUSTOM_PRICING_RATES: Final = MappingProxyType(
{"input_cost_per_token": 0.0000008, "output_cost_per_token": 0.0000032}
@ -9747,6 +10036,38 @@ def test_router_cache_only_pricing_preserves_uncached_token_charges(rate: float,
assert response._hidden_params["response_cost"] == pytest.approx(_CUSTOM_PRICING_EXPECTED_COST)
@pytest.mark.parametrize("async_mode", (False, True))
@pytest.mark.asyncio
@pytest.mark.usefixtures("isolated_pricing_registry")
@respx.mock
async def test_router_transport_calls_keep_deployment_off_peak_prices(async_mode: bool) -> None:
from litellm._internal_context import pinned_billing_time
respx.post("https://api.openai.com/v1/chat/completions").mock(side_effect=_custom_pricing_upstream)
discounted_rates: Final = {field: rate / 2 for field, rate in _CUSTOM_PRICING_RATES.items()}
router: Final = litellm.Router(
model_list=[
{
"model_name": "off-peak-deployment",
"litellm_params": {
"model": f"openai/{_CUSTOM_PRICING_MODEL}",
"api_key": "sk-test",
**_CUSTOM_PRICING_RATES,
},
"model_info": {
"id": "off-peak-deployment-id",
"litellm_provider": "deepseek",
"off_peak_pricing": {"hours_utc": "00:00-00:00", **discounted_rates},
},
}
]
)
request: Final = {"model": "off-peak-deployment", "messages": [{"role": "user", "content": "How are you?"}]}
with pinned_billing_time(datetime.datetime(2026, 1, 1, tzinfo=datetime.timezone.utc)):
response: Final = await router.acompletion(**request) if async_mode else router.completion(**request)
assert response._hidden_params["response_cost"] == pytest.approx(_CUSTOM_PRICING_EXPECTED_COST / 2)
@pytest.mark.usefixtures("isolated_pricing_registry")
@respx.mock
def test_streamed_custom_pricing_cost_matches_non_streamed_cost() -> None: