# What is this? ## File for 'response_cost' calculation in Logging import logging import time from collections.abc import Mapping, Sequence from functools import lru_cache from typing import TYPE_CHECKING, Any, Final, Literal, cast from httpx import Response from pydantic import BaseModel import litellm import litellm._logging from litellm import verbose_logger from litellm.constants import ( DEFAULT_MAX_LRU_CACHE_SIZE, DEFAULT_REPLICATE_GPU_PRICE_PER_SECOND, ) from litellm.litellm_core_utils.llm_cost_calc.tool_call_cost_tracking import ( StandardBuiltInToolCostTracking, ) from litellm.litellm_core_utils.llm_cost_calc.usage_object_transformation import ( InteractionsUsageObjectTransformation, TranscriptionUsageObjectTransformation, ) from litellm.litellm_core_utils.llm_cost_calc.utils import ( CostCalculatorUtils, _generic_cost_per_character, _get_regional_uplift_multiplier, _get_service_tier_cost_key, calculate_cost_component, generic_cost_per_token, get_billable_input_tokens, get_token_type_cost_breakdown, parse_prompt_tokens_details, select_cost_metric_for_model, ) from litellm.llms.anthropic.cost_calculation import ( cost_per_token as anthropic_cost_per_token, ) from litellm.llms.azure.cost_calculation import ( cost_per_token as azure_openai_cost_per_token, ) from litellm.llms.azure_ai.cost_calculator import ( cost_per_token as azure_ai_cost_per_token, ) from litellm.llms.base_llm.search.transformation import SearchResponse from litellm.llms.bedrock.cost_calculation import ( cost_per_token as bedrock_cost_per_token, ) from litellm.llms.databricks.cost_calculator import ( cost_per_token as databricks_cost_per_token, ) from litellm.llms.deepseek.cost_calculator import ( cost_per_token as deepseek_cost_per_token, ) from litellm.llms.fireworks_ai.cost_calculator import ( cost_per_token as fireworks_ai_cost_per_token, ) from litellm.llms.gemini.cost_calculator import cost_per_token as gemini_cost_per_token from litellm.llms.lemonade.cost_calculator import ( cost_per_token as lemonade_cost_per_token, ) from litellm.llms.openai.cost_calculation import ( _video_output_cost_per_second, ) from litellm.llms.openai.cost_calculation import ( cost_per_second as openai_cost_per_second, ) from litellm.llms.openai.cost_calculation import ( cost_per_token as openai_cost_per_token, ) from litellm.llms.perplexity.cost_calculator import ( cost_per_token as perplexity_cost_per_token, ) from litellm.llms.tencent.cost_calculator import ( cost_per_token as tencent_cost_per_token, ) from litellm.llms.together_ai.cost_calculator import ( get_model_params_and_category, has_together_registry_pricing, ) from litellm.llms.vertex_ai.cost_calculator import ( cost_per_character as google_cost_per_character, ) from litellm.llms.vertex_ai.cost_calculator import ( cost_per_token as google_cost_per_token, ) from litellm.llms.vertex_ai.cost_calculator import cost_router as google_cost_router from litellm.llms.xai.cost_calculator import cost_per_token as xai_cost_per_token from litellm.responses.utils import ResponseAPILoggingUtils from litellm.types.agents import LiteLLMSendMessageResponse from litellm.types.llms.openai import ( HttpxBinaryResponseContent, ImageGenerationRequestQuality, OpenAIModerationResponse, OpenAIRealtimeStreamList, OpenAIRealtimeStreamResponseBaseObject, OpenAIRealtimeStreamSessionEvents, ResponseAPIUsage, ResponsesAPIResponse, ) from litellm.types.rerank import RerankBilledUnits, RerankResponse from litellm.types.utils import ( CallTypesLiteral, LiteLLMRealtimeStreamLoggingObject, LlmProviders, LlmProvidersSet, ModelInfo, PromptTokensDetailsWrapper, ServiceTier, StandardBuiltInToolsParams, TranscriptionUsageDurationObject, TranscriptionUsageTokensObject, Usage, VectorStoreSearchResponse, ) from litellm.utils import ( CallTypes, CostPerToken, EmbeddingResponse, ImageResponse, ModelResponse, ModelResponseStream, ProviderConfigManager, TextCompletionResponse, TranscriptionResponse, _cached_get_model_info_helper, token_counter, ) if TYPE_CHECKING: from litellm.litellm_core_utils.litellm_logging import ( Logging as LitellmLoggingObject, ) else: LitellmLoggingObject = Any # Pre-resolved CallTypes enum values for fast membership checks _A2A_CALL_TYPES: Final = frozenset( { CallTypes.asend_message.value, CallTypes.send_message.value, } ) _VIDEO_CALL_TYPES: Final = frozenset( { CallTypes.create_video.value, CallTypes.acreate_video.value, CallTypes.video_edit.value, CallTypes.avideo_edit.value, CallTypes.video_remix.value, CallTypes.avideo_remix.value, } ) _SPEECH_CALL_TYPES: Final = frozenset( { CallTypes.speech.value, CallTypes.aspeech.value, } ) _TRANSCRIPTION_CALL_TYPES: Final = frozenset( { CallTypes.atranscription.value, CallTypes.transcription.value, } ) _RERANK_CALL_TYPES: Final = frozenset( { CallTypes.rerank.value, CallTypes.arerank.value, } ) _SEARCH_CALL_TYPES: Final = frozenset( { CallTypes.search.value, CallTypes.asearch.value, } ) _AREALTIME_CALL_TYPE: Final = CallTypes.arealtime.value _MCP_CALL_TYPE: Final = CallTypes.call_mcp_tool.value def _cost_per_token_custom_pricing_helper( prompt_tokens: float = 0, completion_tokens: float = 0, response_time_ms: float | None = 0.0, cached_tokens: float = 0, cache_creation_tokens: float = 0, ### CUSTOM PRICING ### custom_cost_per_token: CostPerToken | None = None, custom_cost_per_second: float | None = None, ) -> tuple[float, float] | None: """Internal helper function for calculating cost, if custom pricing given. prompt_tokens is assumed to include both cached_tokens and cache_creation_tokens (OpenAI-compatible convention). Anthropic-style usage where prompt_tokens excludes cache tokens is handled at the caller (cost_per_token) before invoking this helper. """ if custom_cost_per_token is None and custom_cost_per_second is None: return None if custom_cost_per_token is not None: input_cost_per_token: Final = custom_cost_per_token["input_cost_per_token"] output_cost_per_token: Final = custom_cost_per_token["output_cost_per_token"] cache_read_input_token_cost: Final = custom_cost_per_token.get( "cache_read_input_token_cost", input_cost_per_token, ) cache_creation_input_token_cost: Final = custom_cost_per_token.get( "cache_creation_input_token_cost", input_cost_per_token, ) regular_prompt_tokens: Final = max( prompt_tokens - cached_tokens - cache_creation_tokens, 0, ) input_cost: Final = ( regular_prompt_tokens * input_cost_per_token + cached_tokens * cache_read_input_token_cost + cache_creation_tokens * cache_creation_input_token_cost ) output_cost = completion_tokens * output_cost_per_token return input_cost, output_cost elif custom_cost_per_second is not None: output_cost = custom_cost_per_second * (response_time_ms or 0.0) / 1000 return 0, output_cost return None def _get_additional_costs( model: str, custom_llm_provider: str | None, prompt_tokens: int, completion_tokens: int, ) -> dict | None: """ Calculate additional costs beyond standard token costs. This function delegates to provider-specific config classes to calculate any additional costs like routing fees, infrastructure costs, etc. Args: model: The model name custom_llm_provider: The provider name (optional) prompt_tokens: Number of prompt tokens completion_tokens: Number of completion tokens Returns: Optional dictionary with cost names and amounts, or None if no additional costs """ if not custom_llm_provider: return None try: config_class = None if custom_llm_provider == "azure_ai": from litellm.llms.azure_ai.common_utils import AzureFoundryModelInfo config_class = AzureFoundryModelInfo.get_azure_ai_config_for_model(model) # Add more providers here as needed # elif custom_llm_provider == "other_provider": # config_class = get_other_provider_config(model) if config_class and hasattr(config_class, "calculate_additional_costs"): return config_class.calculate_additional_costs( model=model, prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, ) except Exception as e: verbose_logger.debug("Error calculating additional costs: %s", e) return None def _transcription_usage_has_token_details( usage_block: Usage | None, ) -> bool: if usage_block is None: return False prompt_tokens_val: Final = getattr(usage_block, "prompt_tokens", 0) or 0 completion_tokens_val: Final = getattr(usage_block, "completion_tokens", 0) or 0 prompt_details: Final[PromptTokensDetailsWrapper | None] = getattr(usage_block, "prompt_tokens_details", None) if prompt_details is not None: audio_token_count: Final = getattr(prompt_details, "audio_tokens", 0) or 0 text_token_count: Final = getattr(prompt_details, "text_tokens", 0) or 0 if audio_token_count > 0 or text_token_count > 0: return True return (prompt_tokens_val > 0) or (completion_tokens_val > 0) def cost_per_token( model: str = "", prompt_tokens: int = 0, completion_tokens: int = 0, response_time_ms: float | None = 0.0, custom_llm_provider: str | None = None, region_name=None, ### CHARACTER PRICING ### prompt_characters: int | None = None, completion_characters: int | None = None, ### PROMPT CACHING PRICING ### - used for anthropic cache_creation_input_tokens: int | None = 0, cache_read_input_tokens: int | None = 0, ### CUSTOM PRICING ### custom_cost_per_token: CostPerToken | None = None, custom_cost_per_second: float | None = None, ### NUMBER OF QUERIES ### number_of_queries: int | None = None, ### USAGE OBJECT ### usage_object: Usage | None = None, # just read the usage object if provided ### BILLED UNITS ### rerank_billed_units: RerankBilledUnits | None = None, ### CALL TYPE ### call_type: CallTypesLiteral = "completion", audio_transcription_file_duration: float = 0.0, # for audio transcription calls - the file time in seconds ### SERVICE TIER ### service_tier: str | None = None, # for OpenAI service tier pricing ### DATA RESIDENCY ### data_residency: str | None = None, # for OpenAI regional-processing uplift (e.g. "eu", "us") ### VERTEX LOCATION ### vertex_location: str | None = None, # for Vertex AI regional-endpoint uplift (e.g. "us-east5", "global") response: Any | None = None, ### REQUEST MODEL ### request_model: str | None = None, # original request model for router detection ) -> tuple[float, float]: """ Calculates the cost per token for a given model, prompt tokens, and completion tokens. Parameters: model (str): The name of the model to use. Default is "" prompt_tokens (int): The number of tokens in the prompt. completion_tokens (int): The number of tokens in the completion. response_time (float): The amount of time, in milliseconds, it took the call to complete. prompt_characters (float): The number of characters in the prompt. Used for vertex ai cost calculation. completion_characters (float): The number of characters in the completion response. Used for vertex ai cost calculation. custom_llm_provider (str): The llm provider to whom the call was made (see init.py for full list) custom_cost_per_token: Optional[CostPerToken]: the cost per input + output token for the llm api call. custom_cost_per_second: Optional[float]: the cost per second for the llm api call. call_type: Optional[str]: the call type Returns: tuple: A tuple containing the cost in USD dollars for prompt tokens and completion tokens, respectively. """ if model is None: raise Exception("Invalid arg. Model cannot be none.") ## RECONSTRUCT USAGE BLOCK ## if usage_object is not None: usage_block = usage_object else: usage_block = Usage( prompt_tokens=prompt_tokens, completion_tokens=completion_tokens, total_tokens=prompt_tokens + completion_tokens, cache_creation_input_tokens=cache_creation_input_tokens, cache_read_input_tokens=cache_read_input_tokens, ) ## CUSTOM PRICING ## # Normalize cache token counts across providers: # - OpenAI-compatible: usage.prompt_tokens_details.cached_tokens # (prompt_tokens already INCLUDES cached_tokens) # - Anthropic: usage.cache_read_input_tokens / cache_creation_input_tokens # (prompt_tokens does NOT include these — adjust before calling helper) _cache_read_tokens: float = 0 _cache_creation_tokens: float = 0 _is_anthropic_style = False if usage_object is not None: _pt_details: Final[PromptTokensDetailsWrapper | None] = getattr(usage_object, "prompt_tokens_details", None) if _pt_details is not None: _cache_read_tokens = float(getattr(_pt_details, "cached_tokens", 0) or 0) # OpenAI-compatible providers report cache-write tokens under # either `cache_write_tokens` (kimi-k2) or `cache_creation_tokens`. # Mirror db_spend_update_writer to stay symmetric. _cache_creation_tokens = float( getattr(_pt_details, "cache_write_tokens", 0) or getattr(_pt_details, "cache_creation_tokens", 0) or 0 ) _anthropic_read: Final[int | None] = getattr(usage_object, "cache_read_input_tokens", None) _anthropic_create: Final[int | None] = getattr(usage_object, "cache_creation_input_tokens", None) if _anthropic_read is not None or _anthropic_create is not None: _is_anthropic_style = True if _anthropic_read is not None: _cache_read_tokens = float(_anthropic_read) if _anthropic_create is not None: _cache_creation_tokens = float(_anthropic_create) if not _cache_read_tokens and cache_read_input_tokens: _cache_read_tokens = float(cache_read_input_tokens) _is_anthropic_style = True if not _cache_creation_tokens and cache_creation_input_tokens: _cache_creation_tokens = float(cache_creation_input_tokens) _is_anthropic_style = True # Anthropic reports prompt_tokens as input_tokens (excluding cache tokens). # Adjust so the helper's "prompt_tokens includes cache tokens" invariant holds. _normalized_prompt_tokens = float(prompt_tokens) if _is_anthropic_style: _normalized_prompt_tokens += _cache_read_tokens + _cache_creation_tokens response_cost: Final = _cost_per_token_custom_pricing_helper( prompt_tokens=_normalized_prompt_tokens, completion_tokens=completion_tokens, response_time_ms=response_time_ms, 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, ) if response_cost is not None: return response_cost[0], response_cost[1] # given prompt_tokens_cost_usd_dollar: float = 0 completion_tokens_cost_usd_dollar: float = 0 model_cost_ref: Final = litellm.model_cost # Only callers that explicitly pass `custom_llm_provider` get the # dedup/prefix-join treatment. When provider is omitted, preserve legacy # behavior: `model_with_provider` stays equal to the raw `model` string # (provider is detected below for downstream use only). caller_supplied_provider: Final = custom_llm_provider is not None # `model` is normally a string, but callers that mock the transport can pass # non-string objects. Only run the string-based dedup/prefix-join when it is # actually a string — e.g. a MagicMock's `.startswith()` is always truthy and # its slices return new mocks, which would spin the dedup loop forever. model_is_str: Final = isinstance(model, str) # Router/proxy deployments may repeat the provider segment (e.g. model_name # "openai/openai/gpt-5.5"). Strip duplicated `{provider}/` chains before joining. if caller_supplied_provider and model_is_str: _dup_prefix: Final = f"{custom_llm_provider}/" while model.startswith(_dup_prefix): _remainder = model[len(_dup_prefix) :] if _remainder.startswith(_dup_prefix): model = _remainder else: break model_with_provider = model if caller_supplied_provider: _prov_prefix: Final = f"{custom_llm_provider}/" if model_is_str and model.startswith(_prov_prefix): model_with_provider = model else: model_with_provider = f"{custom_llm_provider}/{model}" if region_name is not None: model_with_provider_and_region: Final = f"{custom_llm_provider}/{region_name}/{model}" if model_with_provider_and_region in model_cost_ref: # use region based pricing, if it's available model_with_provider = model_with_provider_and_region else: _, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model) assert custom_llm_provider is not None # caller-supplied or get_llm_provider model_without_prefix = model model_parts: Final = model.split("/", 1) if len(model_parts) > 1: model_without_prefix = model_parts[1] else: model_without_prefix = model """ Code block that formats model to lookup in litellm.model_cost Option1. model = "bedrock/ap-northeast-1/anthropic.claude-instant-v1". This is the most accurate since it is region based. Should always be option 1 Option2. model = "openai/gpt-4" - model = provider/model Option3. model = "anthropic.claude-3" - model = model """ if model_with_provider in model_cost_ref: # Option 2. use model with provider, model = "openai/gpt-4" model = model_with_provider elif model in model_cost_ref: # Option 1. use model passed, model="gpt-4" model = model elif ( model_without_prefix in model_cost_ref ): # Option 3. if user passed model="bedrock/anthropic.claude-3", use model="anthropic.claude-3" model = model_without_prefix # see this https://learn.microsoft.com/en-us/azure/ai-services/openai/concepts/models if call_type == "speech" or call_type == "aspeech": speech_model_info = litellm.get_model_info(model=model_without_prefix, custom_llm_provider=custom_llm_provider) cost_metric: Final = select_cost_metric_for_model(speech_model_info) prompt_cost: float = 0.0 completion_cost: float = 0.0 if cost_metric == "cost_per_character": if prompt_characters is None: raise ValueError( f"prompt_characters must be provided for tts calls. prompt_characters={prompt_characters}, model={model}, custom_llm_provider={custom_llm_provider}, call_type={call_type}" ) _prompt_cost, _completion_cost = _generic_cost_per_character( model=model_without_prefix, custom_llm_provider=custom_llm_provider, prompt_characters=prompt_characters, completion_characters=0, custom_prompt_cost=None, custom_completion_cost=0, ) if _prompt_cost is None or _completion_cost is None: raise ValueError( f"cost for tts call is None. prompt_cost={_prompt_cost}, completion_cost={_completion_cost}, model={model_without_prefix}, custom_llm_provider={custom_llm_provider}, prompt_characters={prompt_characters}, completion_characters={completion_characters}" ) prompt_cost = _prompt_cost completion_cost = _completion_cost elif cost_metric == "cost_per_token": prompt_cost, completion_cost = generic_cost_per_token( model=model_without_prefix, usage=usage_block, custom_llm_provider=custom_llm_provider, service_tier=service_tier, data_residency=data_residency, ) return prompt_cost, completion_cost elif call_type == "arerank" or call_type == "rerank": return rerank_cost( model=model, custom_llm_provider=custom_llm_provider, billed_units=rerank_billed_units, ) elif call_type == "avector_store_search" or call_type == "vector_store_search": return vector_store_search_cost( model=model, custom_llm_provider=custom_llm_provider, response=cast(VectorStoreSearchResponse, response), ) elif call_type == "ocr" or call_type == "aocr": return ocr_cost( model=model, custom_llm_provider=custom_llm_provider, response=response, ) 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( usage=usage_block, model=model, custom_llm_provider=custom_llm_provider, data_residency=data_residency, ) elif call_type == "atranscription" or call_type == "transcription": if _transcription_usage_has_token_details(usage_block): return generic_cost_per_token( model=model_without_prefix, usage=usage_block, custom_llm_provider=custom_llm_provider, service_tier=service_tier, data_residency=data_residency, ) return openai_cost_per_second( model=model_without_prefix, custom_llm_provider=custom_llm_provider, duration=audio_transcription_file_duration, ) elif call_type == "search" or call_type == "asearch": # Search providers use per-query pricing from litellm.search import search_provider_cost_per_query return search_provider_cost_per_query( model=model, custom_llm_provider=custom_llm_provider, number_of_queries=number_of_queries or 1, optional_params=(response._hidden_params if response and hasattr(response, "_hidden_params") else None), ) 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, ) if cost_router == "cost_per_character": return 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, service_tier=service_tier, vertex_location=vertex_location, ) elif cost_router == "cost_per_token": return google_cost_per_token( model=model_without_prefix, custom_llm_provider=custom_llm_provider, usage=usage_block, service_tier=service_tier, vertex_location=vertex_location, ) elif custom_llm_provider == "anthropic": return anthropic_cost_per_token(model=model, usage=usage_block, service_tier=service_tier) elif custom_llm_provider == "bedrock": return bedrock_cost_per_token(model=model, usage=usage_block, service_tier=service_tier) elif custom_llm_provider == "openai": return openai_cost_per_token( model=model, usage=usage_block, service_tier=service_tier, data_residency=data_residency, ) elif custom_llm_provider == "databricks": return databricks_cost_per_token(model=model, usage=usage_block) elif custom_llm_provider == "fireworks_ai": return fireworks_ai_cost_per_token(model=model, usage=usage_block) 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, ) elif custom_llm_provider == "gemini": return gemini_cost_per_token(model=model, usage=usage_block, service_tier=service_tier) elif custom_llm_provider == "deepseek": return deepseek_cost_per_token(model=model, usage=usage_block) elif custom_llm_provider == "tencent": return tencent_cost_per_token(model=model, usage=usage_block) elif custom_llm_provider == "perplexity": return perplexity_cost_per_token(model=model, usage=usage_block) elif custom_llm_provider == "xai": return xai_cost_per_token(model=model, usage=usage_block) elif custom_llm_provider == "lemonade": return lemonade_cost_per_token(model=model, usage=usage_block) 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) elif custom_llm_provider == "azure_ai": return azure_ai_cost_per_token( model=model, usage=usage_block, response_time_ms=response_time_ms, request_model=request_model, service_tier=service_tier, ) else: model_info: Final = _cached_get_model_info_helper(model=model, custom_llm_provider=custom_llm_provider) if ( (model_info.get("input_cost_per_token") or 0.0) > 0 or (model_info.get("output_cost_per_token") or 0.0) > 0 or model_info.get("tiered_pricing") is not None ): return generic_cost_per_token( model=model, usage=usage_block, custom_llm_provider=custom_llm_provider, service_tier=service_tier, data_residency=data_residency, ) input_cost_per_second: Final = model_info.get("input_cost_per_second") if input_cost_per_second is not None and response_time_ms is not None: verbose_logger.debug( "For model=%s - input_cost_per_second: %s; response time: %s", model, input_cost_per_second, response_time_ms, ) ## COST PER SECOND ## prompt_tokens_cost_usd_dollar = input_cost_per_second * response_time_ms / 1000 output_cost_per_second: Final = model_info.get("output_cost_per_second") if output_cost_per_second is not None and response_time_ms is not None: verbose_logger.debug( "For model=%s - output_cost_per_second: %s; response time: %s", model, output_cost_per_second, response_time_ms, ) ## COST PER SECOND ## completion_tokens_cost_usd_dollar = output_cost_per_second * response_time_ms / 1000 verbose_logger.debug( "Returned custom cost for model=%s - prompt_tokens_cost_usd_dollar: %s, completion_tokens_cost_usd_dollar: %s", model, prompt_tokens_cost_usd_dollar, completion_tokens_cost_usd_dollar, ) return prompt_tokens_cost_usd_dollar, completion_tokens_cost_usd_dollar def get_replicate_completion_pricing(completion_response: dict, total_time=0.0): # see https://replicate.com/pricing # for all litellm currently supported LLMs, almost all requests go to a100_80gb a100_80gb_price_per_second_public: Final = ( DEFAULT_REPLICATE_GPU_PRICE_PER_SECOND # assume all calls sent to A100 80GB for now ) if total_time == 0.0: # total time is in ms start_time: Final = completion_response.get("created", time.time()) end_time: Final = getattr(completion_response, "ended", time.time()) total_time = end_time - start_time return a100_80gb_price_per_second_public * total_time / 1000 def has_hidden_params(obj: object) -> bool: return hasattr(obj, "_hidden_params") def _get_provider_for_cost_calc( model: str | None, custom_llm_provider: str | None = None, ) -> str | None: if custom_llm_provider is not None: return custom_llm_provider if model is None: return None try: _, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model) except Exception as e: verbose_logger.debug( "litellm.cost_calculator.py::_get_provider_for_cost_calc() - Error inferring custom_llm_provider - %s", e ) return None return custom_llm_provider def _get_hidden_str_for_cost_calc(hidden_params: object, key: str) -> str | None: if not isinstance(hidden_params, Mapping): return None value: Final[object] = hidden_params.get(key) return value if isinstance(value, str) and value else None def _select_model_name_for_cost_calc( model: str | None, completion_response: object | None, base_model: str | None = None, custom_pricing: bool | None = None, custom_llm_provider: str | None = None, router_model_id: str | None = None, ) -> str | None: """ 1. If custom pricing is true, return received model name 2. If base_model is set (e.g. for azure models), return that 3. If completion response has model set return that 4. Check if model is passed in return that """ return_model: str | None = None custom_llm_provider = _get_provider_for_cost_calc(model=model, custom_llm_provider=custom_llm_provider) completion_response_model: str | None = None if completion_response is not None: if isinstance(completion_response, BaseModel): completion_response_model = getattr(completion_response, "model", None) elif isinstance(completion_response, dict): completion_response_model = completion_response.get("model", None) hidden_params: Final[dict | None] = getattr(completion_response, "_hidden_params", None) provider_response_model: Final = _get_hidden_str_for_cost_calc(hidden_params, "provider_response_model") explicit_pricing: Final = custom_pricing is True or base_model is not None priced_from_response: Final = provider_response_model is not None or completion_response_model is not None region_name: Final = ( _get_hidden_str_for_cost_calc(hidden_params, "region_name") if not explicit_pricing and priced_from_response else None ) if custom_pricing is True: if router_model_id is not None and router_model_id in litellm.model_cost: entry: Final = litellm.model_cost[router_model_id] if ( entry.get("input_cost_per_token") is not None or entry.get("input_cost_per_second") is not None or entry.get("tiered_pricing") is not None ): return_model = router_model_id else: return_model = model else: return_model = model elif base_model is not None or provider_response_model is not None: return_model = base_model if base_model is not None else provider_response_model elif completion_response_model is None and hidden_params is not None: if hidden_params.get("model", None) is not None and len(hidden_params["model"]) > 0: return_model = hidden_params.get("model", model) if return_model is None and completion_response_model is not None: return_model = completion_response_model if return_model is None and model is not None: return_model = model if ( return_model is not None and custom_llm_provider is not None and not _model_contains_known_llm_provider(return_model) ): # add provider prefix if not already present, to match model_cost provider_prefix: Final = custom_llm_provider if region_name is None else f"{custom_llm_provider}/{region_name}" return_model = _strip_unregistered_leading_segments(f"{provider_prefix}/{return_model}", region_name) return return_model def _strip_unregistered_leading_segments(model: str, region_name: str | None) -> str: """Resolve a provider-prefixed slash alias like "vertex_ai/vertex/claude-opus-5" to the registered cost key ("vertex_ai/claude-opus-5"), keeping the model unchanged when it already resolves downstream (custom-priced router ids) or no stripped candidate is registered (#38069).""" segments: Final = model.split("/") if "/".join(segments[1:]) in litellm.model_cost: return model head_len: Final = 2 if region_name is not None and len(segments) > 2 and segments[1] == region_name else 1 head: Final = "/".join(segments[:head_len]) tail: Final = segments[head_len:] strippable: Final = next((index for index, segment in enumerate(tail) if segment in LlmProvidersSet), len(tail)) candidates: Final = (f"{head}/{'/'.join(tail[start:])}" for start in range(min(strippable, len(tail) - 1) + 1)) return next((candidate for candidate in candidates if candidate in litellm.model_cost), model) @lru_cache(maxsize=DEFAULT_MAX_LRU_CACHE_SIZE) def _model_contains_known_llm_provider(model: str) -> bool: """ Check if the model contains a known llm provider """ _provider_prefix: Final = model.split("/")[0] return _provider_prefix in LlmProvidersSet def _get_response_model(completion_response: object) -> str | None: """ Extract the model name from a completion response object. Used as a fallback for cost calculation when the input model name doesn't exist in model_cost (e.g., Azure Model Router). """ if completion_response is None: return None if isinstance(completion_response, BaseModel): return getattr(completion_response, "model", None) elif isinstance(completion_response, dict): return completion_response.get("model", None) return None _GEMINI_TRAFFIC_TYPE_TO_SERVICE_TIER: Final[dict] = { # ON_DEMAND_PRIORITY maps to "priority" — selects input_cost_per_token_priority, etc. "ON_DEMAND_PRIORITY": "priority", # FLEX / BATCH / ON_DEMAND_FLEX maps to "flex" — selects input_cost_per_token_flex, etc. # Vertex AI reports flex/shared-capacity traffic as ON_DEMAND_FLEX, not FLEX. "FLEX": "flex", "BATCH": "flex", "ON_DEMAND_FLEX": "flex", # ON_DEMAND is standard pricing — no service_tier suffix applied "ON_DEMAND": None, } def _map_traffic_type_to_service_tier(traffic_type: str | None) -> str | None: """ Map a Gemini usageMetadata.trafficType value to a LiteLLM service_tier string. This allows the same `_priority` / `_flex` cost-key suffix logic used for OpenAI/Azure to work for Gemini and Vertex AI models. trafficType values seen in practice ------------------------------------ ON_DEMAND -> standard pricing (service_tier = None) ON_DEMAND_PRIORITY -> priority pricing (service_tier = "priority") FLEX / BATCH / ON_DEMAND_FLEX -> batch/flex pricing (service_tier = "flex") """ if traffic_type is None: return None service_tier: Final = _GEMINI_TRAFFIC_TYPE_TO_SERVICE_TIER.get(str(traffic_type).upper()) return service_tier def _normalize_service_tier(service_tier: object) -> str | None: """ Reduce a service_tier value to a concrete billable tier string or None. "auto" is a routing preference and any non-string value is not a billable tier, so both defer to standard pricing (or to the tier the provider reports on the response usage) instead of crashing the downstream cost-key lookup, which calls service_tier.lower() """ if not isinstance(service_tier, str) or service_tier.lower() == ServiceTier.AUTO.value: return None return service_tier def _extract_service_tier(source: object) -> str | None: """Read a raw ``service_tier`` off a response body or usage object, dict or pydantic model alike.""" if isinstance(source, BaseModel): return getattr(source, "service_tier", None) elif isinstance(source, dict): return source.get("service_tier") return None def _get_usage_object( completion_response: object, ) -> Usage | None: usage_obj: Final = cast( Usage | ResponseAPIUsage | dict | BaseModel, ( completion_response.get("usage") if isinstance(completion_response, dict) else getattr(completion_response, "get", lambda x: None)("usage") ), ) if usage_obj is None: return None if isinstance(usage_obj, Usage): return usage_obj elif isinstance(usage_obj, dict) and litellm.AnthropicConfig.is_anthropic_usage_object(usage_obj): return litellm.AnthropicConfig().calculate_usage(usage_object=usage_obj, reasoning_content=None) elif ( usage_obj is not None and (isinstance(usage_obj, dict) or isinstance(usage_obj, ResponseAPIUsage)) and ResponseAPILoggingUtils._is_response_api_usage(usage_obj) ): return ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(usage_obj) elif TranscriptionUsageObjectTransformation.is_transcription_usage_object(usage_obj): return TranscriptionUsageObjectTransformation.transform_transcription_usage_object( cast( TranscriptionUsageDurationObject | TranscriptionUsageTokensObject, usage_obj, ) ) elif isinstance(usage_obj, dict) and InteractionsUsageObjectTransformation.is_interactions_usage_object(usage_obj): return InteractionsUsageObjectTransformation.transform_interactions_usage_object(usage_obj) elif isinstance(usage_obj, dict): return Usage(**usage_obj) elif isinstance(usage_obj, BaseModel): return Usage(**usage_obj.model_dump()) else: verbose_logger.debug("Unknown usage object type: %s, usage_obj: %s", type(usage_obj), usage_obj) return None def _is_known_usage_objects(usage_obj): """Returns True if the usage obj is a known Usage type""" return ( isinstance(usage_obj, litellm.Usage) or isinstance(usage_obj, ResponseAPIUsage) or TranscriptionUsageObjectTransformation.is_transcription_usage_object(usage_obj) ) def _infer_call_type(call_type: CallTypesLiteral | None, completion_response: Any) -> CallTypesLiteral | None: if call_type is not None: return call_type if completion_response is None: return None if isinstance(completion_response, ModelResponse) or isinstance(completion_response, ModelResponseStream): return "completion" elif isinstance(completion_response, EmbeddingResponse): return "embedding" elif isinstance(completion_response, TranscriptionResponse): return "transcription" elif isinstance(completion_response, HttpxBinaryResponseContent): return "speech" elif isinstance(completion_response, RerankResponse): return "rerank" elif isinstance(completion_response, ImageResponse): return "image_generation" elif isinstance(completion_response, TextCompletionResponse): return "text_completion" elif isinstance(completion_response, LiteLLMSendMessageResponse): return "send_message" return call_type def _apply_cost_discount( base_cost: float, custom_llm_provider: str | None, ) -> tuple[float, float, float]: """ Apply provider-specific cost discount from module-level config. Args: base_cost: The base cost before discount custom_llm_provider: The LLM provider name Returns: Tuple of (final_cost, discount_percent, discount_amount) """ original_cost: Final = base_cost discount_percent = 0.0 discount_amount = 0.0 if custom_llm_provider and custom_llm_provider in litellm.cost_discount_config: discount_percent = litellm.cost_discount_config[custom_llm_provider] discount_amount = original_cost * discount_percent final_cost: Final = original_cost - discount_amount if verbose_logger.isEnabledFor(logging.DEBUG): verbose_logger.debug( f"Applied {discount_percent * 100}% discount to {custom_llm_provider}: " f"${original_cost:.6f} -> ${final_cost:.6f} (saved ${discount_amount:.6f})" ) return final_cost, discount_percent, discount_amount return base_cost, discount_percent, discount_amount def _apply_cost_margin( base_cost: float, custom_llm_provider: str | None, ) -> tuple[float, float, float, float]: """ Apply provider-specific or global cost margin from module-level config. Args: base_cost: The base cost before margin (after discount if applicable) custom_llm_provider: The LLM provider name Returns: Tuple of (final_cost, margin_percent, margin_fixed_amount, margin_total_amount) """ original_cost: Final = base_cost margin_percent = 0.0 margin_fixed_amount = 0.0 margin_total_amount = 0.0 # Get margin config - check provider-specific first, then global margin_config = None if custom_llm_provider and custom_llm_provider in litellm.cost_margin_config: margin_config = litellm.cost_margin_config[custom_llm_provider] if verbose_logger.isEnabledFor(logging.DEBUG): verbose_logger.debug("Found provider-specific margin config for %s: %s", custom_llm_provider, margin_config) elif "global" in litellm.cost_margin_config: margin_config = litellm.cost_margin_config["global"] if verbose_logger.isEnabledFor(logging.DEBUG): verbose_logger.debug("Using global margin config: %s", margin_config) else: if verbose_logger.isEnabledFor(logging.DEBUG): verbose_logger.debug( "No margin config found. Provider: %s, Available configs: %s", custom_llm_provider, list(litellm.cost_margin_config.keys()), ) if margin_config is not None: # Handle different margin config formats if isinstance(margin_config, (int, float)): # Simple percentage: {"openai": 0.10} margin_percent = float(margin_config) margin_total_amount = original_cost * margin_percent elif isinstance(margin_config, dict): # Complex config: {"percentage": 0.08, "fixed_amount": 0.0005} if "percentage" in margin_config: margin_percent = float(margin_config["percentage"]) margin_total_amount += original_cost * margin_percent if "fixed_amount" in margin_config: margin_fixed_amount = float(margin_config["fixed_amount"]) margin_total_amount += margin_fixed_amount final_cost: Final = original_cost + margin_total_amount if verbose_logger.isEnabledFor(logging.DEBUG): verbose_logger.debug( f"Applied margin to {custom_llm_provider or 'global'}: " f"${original_cost:.6f} -> ${final_cost:.6f} " f"(margin: {margin_percent * 100 if margin_percent > 0 else 0}% + ${margin_fixed_amount:.6f} = ${margin_total_amount:.6f})" ) return final_cost, margin_percent, margin_fixed_amount, margin_total_amount return base_cost, margin_percent, margin_fixed_amount, margin_total_amount def _store_cost_breakdown_in_logging_obj( litellm_logging_obj: LitellmLoggingObject | None, prompt_tokens_cost_usd_dollar: float, completion_tokens_cost_usd_dollar: float, cost_for_built_in_tools_cost_usd_dollar: float, total_cost_usd_dollar: float, additional_costs: dict | None = None, original_cost: float | None = None, discount_percent: float | None = None, discount_amount: float | None = None, margin_percent: float | None = None, margin_fixed_amount: float | None = None, margin_total_amount: float | None = None, cache_read_cost: float | None = None, cache_creation_cost: float | None = None, reasoning_cost: float | None = None, service_tier: str | None = None, data_residency: str | None = None, vertex_location: str | None = None, ) -> None: """ Helper function to store cost breakdown in the logging object. Args: litellm_logging_obj: The logging object to store breakdown in prompt_tokens_cost_usd_dollar: Cost of input tokens completion_tokens_cost_usd_dollar: Cost of completion tokens (includes reasoning if applicable) cost_for_built_in_tools_cost_usd_dollar: Cost of built-in tools total_cost_usd_dollar: Total cost of request additional_costs: Free-form additional costs dict (e.g., {"azure_model_router_flat_cost": 0.00014}) original_cost: Cost before discount discount_percent: Discount percentage applied (0.05 = 5%) discount_amount: Discount amount in USD margin_percent: Margin percentage applied (0.10 = 10%) margin_fixed_amount: Fixed margin amount in USD margin_total_amount: Total margin added in USD service_tier: Tier the costs above were priced on, already resolved data_residency: Region uplift the costs above were priced on, already resolved vertex_location: Vertex AI location the costs above were priced on, already resolved """ if litellm_logging_obj is None: return try: # Store the cost breakdown litellm_logging_obj.set_cost_breakdown( input_cost=prompt_tokens_cost_usd_dollar, output_cost=completion_tokens_cost_usd_dollar, total_cost=total_cost_usd_dollar, cost_for_built_in_tools_cost_usd_dollar=cost_for_built_in_tools_cost_usd_dollar, additional_costs=additional_costs, original_cost=original_cost, discount_percent=discount_percent, discount_amount=discount_amount, margin_percent=margin_percent, margin_fixed_amount=margin_fixed_amount, margin_total_amount=margin_total_amount, cache_read_cost=cache_read_cost, cache_creation_cost=cache_creation_cost, reasoning_cost=reasoning_cost, service_tier=service_tier, data_residency=data_residency, vertex_location=vertex_location, ) except Exception as breakdown_error: verbose_logger.debug("Error storing cost breakdown: %s", breakdown_error) # Don't fail the main cost calculation if breakdown storage fails def completion_cost( completion_response: object | None = None, model: str | None = None, prompt="", messages: list = [], completion="", total_time: float | None = 0.0, # used for replicate, sagemaker call_type: CallTypesLiteral | None = None, ### REGION ### custom_llm_provider=None, region_name=None, # used for bedrock pricing ### IMAGE GEN ### size: str | None = None, quality: str | None = None, n: int | None = None, # number of images ### CUSTOM PRICING ### custom_cost_per_token: CostPerToken | None = None, custom_cost_per_second: float | None = None, optional_params: dict | None = None, custom_pricing: bool | None = None, base_model: str | None = None, standard_built_in_tools_params: StandardBuiltInToolsParams | None = None, litellm_model_name: str | None = None, router_model_id: str | None = None, litellm_logging_obj: LitellmLoggingObject | None = None, ### SERVICE TIER ### service_tier: str | None = None, # for OpenAI service tier pricing ### DATA RESIDENCY ### data_residency: str | None = None, # for OpenAI regional-processing uplift (e.g. "eu", "us") ### VERTEX LOCATION ### vertex_location: str | None = None, # for Vertex AI regional-endpoint uplift (e.g. "us-east5", "global") ) -> float: """ Calculate the cost of a given completion call fot GPT-3.5-turbo, llama2, any litellm supported llm. Parameters: completion_response (litellm.ModelResponses): [Required] The response received from a LiteLLM completion request. [OPTIONAL PARAMS] model (str): Optional. The name of the language model used in the completion calls prompt (str): Optional. The input prompt passed to the llm completion (str): Optional. The output completion text from the llm total_time (float, int): Optional. (Only used for Replicate LLMs) The total time used for the request in seconds custom_cost_per_token: Optional[CostPerToken]: the cost per input + output token for the llm api call. custom_cost_per_second: Optional[float]: the cost per second for the llm api call. Returns: float: The cost in USD dollars for the completion based on the provided parameters. Exceptions: Raises exception if model not in the litellm model cost map. Register model, via custom pricing or PR - https://github.com/BerriAI/litellm/blob/main/model_prices_and_context_window.json Note: - If completion_response is provided, the function extracts token information and the model name from it. - If completion_response is not provided, the function calculates token counts based on the model and input text. - The cost is calculated based on the model, prompt tokens, and completion tokens. - For certain models containing "togethercomputer" in the name, prices are based on the model size. - For un-mapped Replicate models, the cost is calculated based on the total time used for the request. """ try: call_type = _infer_call_type(call_type, completion_response) or "completion" if ( (call_type == "aimage_generation" or call_type == "image_generation") and model is not None and isinstance(model, str) and len(model) == 0 and custom_llm_provider == "azure" ): model = "dall-e-2" # for dall-e-2, azure expects an empty model name # Handle Inputs to completion_cost prompt_tokens = 0 prompt_characters: int | None = None completion_tokens = 0 completion_characters: int | None = None cache_creation_input_tokens: int | None = None cache_read_input_tokens: int | None = None audio_transcription_file_duration: float = 0.0 cost_per_token_usage_object: Final[Usage | None] = _get_usage_object(completion_response=completion_response) rerank_billed_units: RerankBilledUnits | None = None # Extract service_tier from optional_params if not provided directly if service_tier is None and optional_params is not None: service_tier = optional_params.get("service_tier") service_tier = _normalize_service_tier(service_tier) # Extract service_tier from completion_response if not provided if service_tier is None and completion_response is not None: service_tier = _extract_service_tier(completion_response) service_tier = _normalize_service_tier(service_tier) # Extract service_tier from usage object if not provided if service_tier is None and cost_per_token_usage_object is not None: service_tier = _extract_service_tier(cost_per_token_usage_object) service_tier = _normalize_service_tier(service_tier) selected_model: Final = _select_model_name_for_cost_calc( model=model, completion_response=completion_response, custom_llm_provider=custom_llm_provider, custom_pricing=custom_pricing, base_model=base_model, router_model_id=router_model_id, ) potential_model_names: Final = [ selected_model, _get_response_model(completion_response), ] if model is not None: potential_model_names.append(model) for idx, model in enumerate(potential_model_names): try: if verbose_logger.isEnabledFor(logging.DEBUG): verbose_logger.debug("selected model name for cost calculation: %s", model) if completion_response is not None and ( isinstance(completion_response, BaseModel) or isinstance(completion_response, dict) ): # tts returns a custom class if isinstance(completion_response, dict): usage_obj: dict | Usage | None = completion_response.get("usage", {}) else: usage_obj = getattr(completion_response, "usage", {}) if isinstance(usage_obj, BaseModel) and not _is_known_usage_objects(usage_obj=usage_obj): _usage_for_dump = cast(BaseModel, usage_obj) setattr( completion_response, "usage", litellm.Usage(**_usage_for_dump.model_dump()), ) if usage_obj is None: _usage = {} elif isinstance(usage_obj, BaseModel): _usage = cast(BaseModel, usage_obj).model_dump() else: _usage = usage_obj if litellm.AnthropicConfig.is_anthropic_usage_object(_usage): _usage = ( litellm.AnthropicConfig() .calculate_usage(usage_object=_usage, reasoning_content=None) .model_dump() ) elif ResponseAPILoggingUtils._is_response_api_usage(_usage): _usage = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage( _usage ).model_dump() elif TranscriptionUsageObjectTransformation.is_transcription_usage_object(_usage): tr_usage = TranscriptionUsageObjectTransformation.transform_transcription_usage_object( cast( TranscriptionUsageDurationObject | TranscriptionUsageTokensObject, _usage, ) ) if tr_usage is not None: _usage = tr_usage.model_dump() elif InteractionsUsageObjectTransformation.is_interactions_usage_object(_usage): _usage = InteractionsUsageObjectTransformation.transform_interactions_usage_object( _usage ).model_dump() else: _usage = _usage # get input/output tokens from completion_response prompt_tokens = _usage.get("prompt_tokens", 0) completion_tokens = _usage.get("completion_tokens", 0) cache_creation_input_tokens = _usage.get("cache_creation_input_tokens", 0) cache_read_input_tokens = _usage.get("cache_read_input_tokens", 0) if ( "prompt_tokens_details" in _usage and _usage["prompt_tokens_details"] != {} and _usage["prompt_tokens_details"] ): prompt_tokens_details = _usage.get("prompt_tokens_details") or {} cache_read_input_tokens = prompt_tokens_details.get("cached_tokens", 0) total_time = getattr(completion_response, "_response_ms", 0) hidden_params = getattr(completion_response, "_hidden_params", None) if hidden_params is not None: custom_llm_provider = hidden_params.get("custom_llm_provider", custom_llm_provider or None) region_name = hidden_params.get("region_name", region_name) # For Gemini/Vertex AI responses, trafficType is stored in # provider_specific_fields. Map it to the service_tier used # by the cost key lookup (_priority / _flex suffixes) so that # ON_DEMAND_PRIORITY requests are billed at priority prices. if service_tier is None: provider_specific = hidden_params.get("provider_specific_fields") or {} raw_traffic_type = provider_specific.get("traffic_type") if raw_traffic_type: service_tier = _map_traffic_type_to_service_tier(raw_traffic_type) else: if model is None: raise ValueError( f"Model is None and does not exist in passed completion_response. Passed completion_response={completion_response}, model={model}" ) if len(messages) > 0: prompt_tokens = token_counter(model=model, messages=messages) elif len(prompt) > 0: prompt_tokens = token_counter(model=model, text=prompt) completion_tokens = token_counter(model=model, text=completion) # Handle A2A calls before model check - A2A doesn't require a model if call_type in _A2A_CALL_TYPES: from litellm.a2a_protocol.cost_calculator import A2ACostCalculator return A2ACostCalculator.calculate_a2a_cost(litellm_logging_obj=litellm_logging_obj) if model is None: raise ValueError( f"Model is None and does not exist in passed completion_response. Passed completion_response={completion_response}, model={model}" ) if custom_llm_provider is None: try: model, custom_llm_provider, _, _ = litellm.get_llm_provider( model=model ) # strip the llm provider from the model name -> for image gen cost calculation except Exception as e: verbose_logger.debug( "litellm.cost_calculator.py::completion_cost() - Error inferring custom_llm_provider - %s", e, ) if CostCalculatorUtils._call_type_has_image_response(call_type) and isinstance( completion_response, ImageResponse ): ### IMAGE GENERATION COST CALCULATION ### return CostCalculatorUtils.route_image_generation_cost_calculator( model=model, custom_llm_provider=custom_llm_provider, completion_response=completion_response, quality=quality, n=n, size=size, optional_params=optional_params, call_type=call_type, ) elif call_type in _VIDEO_CALL_TYPES: ### VIDEO GENERATION COST CALCULATION ### # Extract custom model_info for deployment-specific pricing _video_model_info: ModelInfo | None = None if custom_pricing and litellm_logging_obj is not None: _litellm_params = getattr(litellm_logging_obj, "litellm_params", None) if _litellm_params is not None: _video_model_info = next( ( model_info for _metadata_key in ("metadata", "litellm_metadata") if (model_info := (_litellm_params.get(_metadata_key) or {}).get("model_info")) is not None ), None, ) usage_obj = getattr(completion_response, "usage", None) duration_seconds: float | None = None video_resolution: str | None = None provider_reported_cost: float | None = None if completion_response is not None and usage_obj: # Handle both dict and Pydantic Usage object if isinstance(usage_obj, dict): duration_seconds = usage_obj.get("duration_seconds", None) _vr = usage_obj.get("video_resolution", None) provider_reported_cost = usage_obj.get("provider_reported_cost_usd", None) else: duration_seconds = getattr(usage_obj, "duration_seconds", None) _vr = getattr(usage_obj, "video_resolution", None) provider_reported_cost = getattr(usage_obj, "provider_reported_cost_usd", None) if _vr is not None: video_resolution = str(_vr).strip().lower() if _video_model_info is None and provider_reported_cost is not None: return float(provider_reported_cost) if duration_seconds is not None: # Calculate cost based on video duration using video-specific cost calculation from litellm.llms.openai.cost_calculation import ( video_generation_cost, ) return video_generation_cost( model=model, duration_seconds=duration_seconds, custom_llm_provider=custom_llm_provider, model_info=_video_model_info, video_resolution=video_resolution, ) # Fallback to default video cost calculation if no duration available return default_video_cost_calculator( model=model, duration_seconds=0.0, # Default to 0 if no duration available custom_llm_provider=custom_llm_provider, model_info=_video_model_info, video_resolution=video_resolution, ) elif call_type in _SPEECH_CALL_TYPES: prompt_characters = litellm.utils._count_characters(text=prompt) elif call_type in _TRANSCRIPTION_CALL_TYPES: # Check _hidden_params first (duration stored there to # avoid polluting the response body), then fall back to # the response attribute (for verbose_json responses that # naturally include duration from the provider). _hidden = getattr(completion_response, "_hidden_params", {}) or {} audio_transcription_file_duration = _hidden.get( "audio_transcription_duration", getattr(completion_response, "duration", 0.0), ) elif call_type in _RERANK_CALL_TYPES: if completion_response is not None and isinstance(completion_response, RerankResponse): meta_obj = completion_response.meta if meta_obj is not None: billed_units: RerankBilledUnits = meta_obj.get("billed_units") or {} else: billed_units = {} rerank_billed_units = RerankBilledUnits( search_units=billed_units.get("search_units"), total_tokens=billed_units.get("total_tokens"), ) search_units = billed_units.get("search_units") or 1 # cohere charges per request by default. completion_tokens = search_units elif call_type in _SEARCH_CALL_TYPES: from litellm.search import search_provider_cost_per_query # Extract number_of_queries from optional_params or default to 1 number_of_queries = 1 if optional_params is not None: # Check if query is a list (multiple queries) query = optional_params.get("query") if isinstance(query, list): number_of_queries = len(query) elif query is not None: number_of_queries = 1 search_model = model or "" if custom_llm_provider and "/" not in search_model: # If model is like "tavily-search", construct "tavily/search" for cost lookup search_model = f"{custom_llm_provider}/search" ( prompt_cost, completion_cost_result, ) = search_provider_cost_per_query( model=search_model, custom_llm_provider=custom_llm_provider, number_of_queries=number_of_queries, optional_params=optional_params, ) # Return the total cost (prompt_cost + completion_cost, but for search it's just prompt_cost) _final_cost = prompt_cost + completion_cost_result # Apply discount original_cost = _final_cost ( _final_cost, discount_percent, discount_amount, ) = _apply_cost_discount( base_cost=_final_cost, custom_llm_provider=custom_llm_provider, ) # Apply margin from module-level config if configured ( _final_cost, margin_percent, margin_fixed_amount, margin_total_amount, ) = _apply_cost_margin( base_cost=_final_cost, custom_llm_provider=custom_llm_provider, ) # Store cost breakdown in logging object if available _store_cost_breakdown_in_logging_obj( litellm_logging_obj=litellm_logging_obj, prompt_tokens_cost_usd_dollar=prompt_cost, completion_tokens_cost_usd_dollar=completion_cost_result, cost_for_built_in_tools_cost_usd_dollar=0.0, total_cost_usd_dollar=_final_cost, original_cost=original_cost, discount_percent=discount_percent, discount_amount=discount_amount, margin_percent=margin_percent, margin_fixed_amount=margin_fixed_amount, margin_total_amount=margin_total_amount, service_tier=service_tier, data_residency=data_residency, ) return _final_cost elif call_type == _AREALTIME_CALL_TYPE and isinstance( completion_response, LiteLLMRealtimeStreamLoggingObject ): if cost_per_token_usage_object is None or custom_llm_provider is None: raise ValueError( f"usage object and custom_llm_provider must be provided for realtime stream cost calculation. Got cost_per_token_usage_object={cost_per_token_usage_object}, custom_llm_provider={custom_llm_provider}" ) return handle_realtime_stream_cost_calculation( results=completion_response.results, combined_usage_object=cost_per_token_usage_object, custom_llm_provider=custom_llm_provider, litellm_model_name=model, data_residency=data_residency, litellm_logging_obj=litellm_logging_obj, ) elif call_type == _MCP_CALL_TYPE: from litellm.proxy._experimental.mcp_server.cost_calculator import ( MCPCostCalculator, ) return MCPCostCalculator.calculate_mcp_tool_call_cost(litellm_logging_obj=litellm_logging_obj) # Calculate cost based on prompt_tokens, completion_tokens if ( "togethercomputer" in model or "together_ai" in model or custom_llm_provider == "together_ai" ) and not has_together_registry_pricing(model, litellm.model_cost): model = get_model_params_and_category(model, call_type=CallTypes(call_type)) # replicate llms are calculate based on time for request running # see https://replicate.com/pricing elif (model in litellm.replicate_models or "replicate" in model) and model not in litellm.model_cost: # for unmapped replicate model, default to replicate's time tracking logic return get_replicate_completion_pricing(completion_response, total_time) if model is None: raise ValueError( f"Model is None and does not exist in passed completion_response. Passed completion_response={completion_response}, model={model}" ) if custom_llm_provider is not None and custom_llm_provider == "vertex_ai": # Calculate the prompt characters + response characters if len(messages) > 0: prompt_string = litellm.utils.get_formatted_prompt( data={"messages": messages}, call_type="completion" ) prompt_characters = litellm.utils._count_characters(text=prompt_string) if completion_response is not None and isinstance(completion_response, ModelResponse): completion_string = litellm.utils.get_response_string(response_obj=completion_response) completion_characters = litellm.utils._count_characters(text=completion_string) # Get the original request model for router detection request_model_for_cost = None if litellm_logging_obj is not None: request_model_for_cost = litellm_logging_obj.model ( prompt_tokens_cost_usd_dollar, completion_tokens_cost_usd_dollar, ) = cost_per_token( model=model, prompt_tokens=prompt_tokens or 0, completion_tokens=completion_tokens or 0, custom_llm_provider=custom_llm_provider, response_time_ms=total_time, region_name=region_name, custom_cost_per_second=custom_cost_per_second, custom_cost_per_token=custom_cost_per_token, prompt_characters=prompt_characters, completion_characters=completion_characters, cache_creation_input_tokens=cache_creation_input_tokens, cache_read_input_tokens=cache_read_input_tokens, usage_object=cost_per_token_usage_object, call_type=call_type, audio_transcription_file_duration=audio_transcription_file_duration, rerank_billed_units=rerank_billed_units, service_tier=service_tier, data_residency=data_residency, vertex_location=vertex_location, response=completion_response, request_model=request_model_for_cost, ) # Get additional costs from provider (e.g., routing fees, infrastructure costs) if custom_llm_provider == "azure_ai": model_for_additional_costs = request_model_for_cost if completion_response is not None: hidden_params = getattr(completion_response, "_hidden_params", None) or {} hidden_model = hidden_params.get("model") or hidden_params.get("litellm_model_name") if ( hidden_model and ( "model_router" in (hidden_model or "").lower() or "model-router" in (hidden_model or "").lower() ) or model_for_additional_costs is None ): model_for_additional_costs = hidden_model if model_for_additional_costs is None: model_for_additional_costs = model additional_costs = _get_additional_costs( model=model_for_additional_costs, custom_llm_provider=custom_llm_provider, prompt_tokens=prompt_tokens or 0, completion_tokens=completion_tokens or 0, ) else: additional_costs = None _final_cost = prompt_tokens_cost_usd_dollar + completion_tokens_cost_usd_dollar cost_for_built_in_tools = StandardBuiltInToolCostTracking.get_cost_for_built_in_tools( model=model, response_object=completion_response, usage=cost_per_token_usage_object, standard_built_in_tools_params=standard_built_in_tools_params, custom_llm_provider=custom_llm_provider, ) _final_cost += cost_for_built_in_tools if additional_costs: _final_cost += sum(additional_costs.values()) original_cost = _final_cost if litellm.cost_discount_config: ( _final_cost, discount_percent, discount_amount, ) = _apply_cost_discount( base_cost=_final_cost, custom_llm_provider=custom_llm_provider, ) else: discount_percent = 0.0 discount_amount = 0.0 # Apply margin from module-level config if configured if litellm.cost_margin_config: ( _final_cost, margin_percent, margin_fixed_amount, margin_total_amount, ) = _apply_cost_margin( base_cost=_final_cost, custom_llm_provider=custom_llm_provider, ) else: margin_percent = 0.0 margin_fixed_amount = 0.0 margin_total_amount = 0.0 # Store cost breakdown in logging object if available if litellm_logging_obj is not None: _reasoning_cost: float | None = None _cache_read_cost: float | None = None _cache_creation_cost: float | None = None if cost_per_token_usage_object is not None and model: _breakdown_provider: str | None = ( custom_llm_provider if isinstance(custom_llm_provider, str) else None ) _token_type_breakdown = get_token_type_cost_breakdown( model=model, custom_llm_provider=_breakdown_provider, usage=cost_per_token_usage_object, service_tier=service_tier, data_residency=data_residency, vertex_location=vertex_location, ) _reasoning_cost = _token_type_breakdown.reasoning_cost _cache_read_cost = _token_type_breakdown.cache_read_cost _cache_creation_cost = _token_type_breakdown.cache_creation_cost _store_cost_breakdown_in_logging_obj( litellm_logging_obj=litellm_logging_obj, prompt_tokens_cost_usd_dollar=prompt_tokens_cost_usd_dollar, completion_tokens_cost_usd_dollar=completion_tokens_cost_usd_dollar, cost_for_built_in_tools_cost_usd_dollar=cost_for_built_in_tools, total_cost_usd_dollar=_final_cost, original_cost=original_cost, additional_costs=additional_costs, discount_percent=discount_percent, discount_amount=discount_amount, margin_percent=margin_percent, margin_fixed_amount=margin_fixed_amount, margin_total_amount=margin_total_amount, cache_read_cost=_cache_read_cost, cache_creation_cost=_cache_creation_cost, reasoning_cost=_reasoning_cost, service_tier=service_tier, data_residency=data_residency, vertex_location=vertex_location, ) return _final_cost except Exception as e: verbose_logger.debug( "litellm.cost_calculator.py::completion_cost() - Error calculating cost for model=%s - %s", model, e ) if idx == len(potential_model_names) - 1: raise e raise Exception(f"Unable to calculat cost for received potential model names - {potential_model_names}") except Exception as e: raise e def get_response_cost_from_hidden_params( hidden_params: dict | BaseModel, ) -> float | None: if isinstance(hidden_params, BaseModel): _hidden_params_dict = cast(BaseModel, hidden_params).model_dump() else: _hidden_params_dict = hidden_params additional_headers: Final = _hidden_params_dict.get("additional_headers", {}) if additional_headers and "llm_provider-x-litellm-response-cost" in additional_headers: response_cost: Final = additional_headers["llm_provider-x-litellm-response-cost"] if response_cost is None: return None return float(additional_headers["llm_provider-x-litellm-response-cost"]) return None def response_cost_calculator( response_object: ModelResponse | EmbeddingResponse | ImageResponse | TranscriptionResponse | TextCompletionResponse | HttpxBinaryResponseContent | RerankResponse | ResponsesAPIResponse | LiteLLMRealtimeStreamLoggingObject | OpenAIModerationResponse | Response | SearchResponse, model: str, custom_llm_provider: str | None, call_type: Literal[ "embedding", "aembedding", "completion", "acompletion", "atext_completion", "text_completion", "image_generation", "aimage_generation", "moderation", "amoderation", "atranscription", "transcription", "aspeech", "speech", "rerank", "arerank", "search", "asearch", ], optional_params: dict, cache_hit: bool | None = None, base_model: str | None = None, custom_pricing: bool | None = None, prompt: str = "", standard_built_in_tools_params: StandardBuiltInToolsParams | None = None, litellm_model_name: str | None = None, router_model_id: str | None = None, litellm_logging_obj: LitellmLoggingObject | None = None, ### SERVICE TIER ### service_tier: str | None = None, # for OpenAI service tier pricing ### DATA RESIDENCY ### data_residency: str | None = None, # for OpenAI regional-processing uplift (e.g. "eu", "us") ### VERTEX LOCATION ### vertex_location: str | None = None, # for Vertex AI regional-endpoint uplift (e.g. "us-east5", "global") ) -> float: """ Returns - float or None: cost of response """ try: response_cost: float = 0.0 if cache_hit is not None and cache_hit is True: response_cost = 0.0 else: if isinstance(response_object, BaseModel): if hasattr(response_object, "_hidden_params"): response_object._hidden_params["optional_params"] = optional_params provider_response_cost: Final = get_response_cost_from_hidden_params(response_object._hidden_params) if provider_response_cost is not None: return provider_response_cost response_cost = completion_cost( completion_response=response_object, model=model, call_type=call_type, custom_llm_provider=custom_llm_provider, optional_params=optional_params, custom_pricing=custom_pricing, base_model=base_model, prompt=prompt, standard_built_in_tools_params=standard_built_in_tools_params, litellm_model_name=litellm_model_name, router_model_id=router_model_id, litellm_logging_obj=litellm_logging_obj, service_tier=service_tier, data_residency=data_residency, vertex_location=vertex_location, ) return response_cost except Exception as e: raise e def ocr_cost( model: str, custom_llm_provider: str | None, response: object | None = None, ) -> tuple[float, float]: """ Args: model: str - model name custom_llm_provider: Optional[str] - custom LLM provider response: Optional[Any] - response object Returns: Tuple[float, float]: cost of OCR processing (Parent function requires a tuple, so we return a tuple. Cost is only in the first element.) """ from litellm.llms.base_llm.ocr.transformation import OCRResponse ######################################################### # validate it's an OCR response ######################################################### if response is None or not isinstance(response, OCRResponse): raise ValueError(f"response must be of type OCRResponse got type={type(response)}") if response.usage_info is None: raise ValueError("OCR response usage_info is None") try: model_info: ModelInfo | None = litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider) except Exception: model_info = None credits: Final = getattr(response.usage_info, "credits", None) cost_per_credit = None if model_info is not None: cost_per_credit = model_info.get("ocr_cost_per_credit") if credits is not None and cost_per_credit is not None: return cost_per_credit * credits, 0.0 ocr_cost_per_page: float | None = None if model_info is not None: ocr_cost_per_page = model_info.get("ocr_cost_per_page") pages_processed: Final = response.usage_info.pages_processed if pages_processed is None: if cost_per_credit is not None or ocr_cost_per_page is None: # Surface missing usage data instead of silently under-reporting # cost. The previous behavior raised ValueError; we now return 0.0 # for credit-priced or unpriced models, so log a warning to keep # the regression visible to operators. verbose_logger.warning( "OCR cost: model=%s custom_llm_provider=%s response.usage_info." "pages_processed is None and credits=%s; returning 0.0 cost.", model, custom_llm_provider, credits, ) return 0.0, 0.0 raise ValueError("OCR response pages_processed is None") if ocr_cost_per_page is None: # No per-page pricing configured. Either the model is on credit-based # pricing (and credits weren't returned, so the credit branch above did # not match) or the model has no OCR pricing entry at all. Surface a # warning so that missing pricing entries are visible rather than # silently producing zero cost for billable usage. verbose_logger.warning( "OCR cost: model=%s custom_llm_provider=%s reported " "pages_processed=%s but no ocr_cost_per_page is configured; " "returning 0.0 cost.", model, custom_llm_provider, pages_processed, ) return 0.0, 0.0 total_ocr_processing_cost: Final[float] = ocr_cost_per_page * pages_processed return total_ocr_processing_cost, 0.0 def vector_store_search_cost( model: str | None, custom_llm_provider: str, response: VectorStoreSearchResponse, ) -> tuple[float, float]: """ Returns - float or None: cost of vector store search """ api_type: str | None = None if custom_llm_provider is None: custom_llm_provider = "openai" if model is not None and "/" in model: api_type, custom_llm_provider, _, _ = litellm.get_llm_provider( model=model, ) config: Final = ProviderConfigManager.get_provider_vector_stores_config( provider=LlmProviders(custom_llm_provider), api_type=api_type, ) if config is None: verbose_logger.debug("Vector store search is not supported for %s", custom_llm_provider) return 0.0, 0.0 return config.calculate_vector_store_cost( response=response, ) def rerank_cost( model: str, custom_llm_provider: str | None, billed_units: RerankBilledUnits | None = None, ) -> tuple[float, float]: """ Returns - float or None: cost of response OR none if error. """ _, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model, custom_llm_provider=custom_llm_provider) try: config: Final = ProviderConfigManager.get_provider_rerank_config( model=model, api_base=None, present_version_params=[], provider=LlmProviders(custom_llm_provider), ) try: model_info: ModelInfo | None = litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider) except Exception: model_info = None return config.calculate_rerank_cost( model=model, custom_llm_provider=custom_llm_provider, billed_units=billed_units, model_info=model_info, ) except Exception as e: raise e def transcription_cost(model: str, custom_llm_provider: str | None, duration: float) -> tuple[float, float]: return openai_cost_per_second(model=model, custom_llm_provider=custom_llm_provider, duration=duration) def default_image_cost_calculator( model: str, custom_llm_provider: str | None = None, quality: str | None = None, n: int | None = 1, # Default to 1 image size: str | None = "1024-x-1024", # OpenAI default optional_params: dict | None = None, ) -> float: """ Default image cost calculator for image generation Args: model (str): Model name image_response (ImageResponse): Response from image generation quality (Optional[str]): Image quality setting n (Optional[int]): Number of images generated size (Optional[str]): Image size (e.g. "1024x1024" or "1024-x-1024") Returns: float: Cost in USD for the image generation Raises: Exception: If model pricing not found in cost map """ # Standardize size format to use "-x-" size_str: str = size or "1024-x-1024" size_str = size_str.replace("x", "-x-") if "x" in size_str and "-x-" not in size_str else size_str # Parse dimensions height, width = map(int, size_str.split("-x-")) # Build model names for cost lookup base_model_name = f"{size_str}/{model}" model_name_without_custom_llm_provider: str | None = None if custom_llm_provider and model.startswith(f"{custom_llm_provider}/"): model_name_without_custom_llm_provider = model.replace(f"{custom_llm_provider}/", "") base_model_name = f"{custom_llm_provider}/{size_str}/{model_name_without_custom_llm_provider}" model_name_with_quality: Final = f"{quality}/{base_model_name}" if quality else base_model_name # gpt-image-1 models use low, medium, high quality. If user did not specify quality, use medium fot gpt-image-1 model family model_name_with_v2_quality: Final = f"{ImageGenerationRequestQuality.HIGH.value}/{base_model_name}" verbose_logger.debug("Looking up cost for models: %s, %s", model_name_with_quality, base_model_name) model_without_provider: Final = f"{size_str}/{model.split('/')[-1]}" model_with_quality_without_provider = f"{quality}/{model_without_provider}" if quality else model_without_provider # Try model with quality first, fall back to base model name cost_info: dict | None = None models_to_check: Final[list[str | None]] = [ model_name_with_quality, base_model_name, model_name_with_v2_quality, model_with_quality_without_provider, model_without_provider, model, model_name_without_custom_llm_provider, ] for _model in models_to_check: if _model is not None and _model in litellm.model_cost: cost_info = litellm.model_cost[_model] break if cost_info is None: raise Exception(f"Model not found in cost map. Tried checking {models_to_check}") # Priority 1: Use per-image pricing if available (for gpt-image-1 and similar models) if "input_cost_per_image" in cost_info and cost_info["input_cost_per_image"] is not None: return cost_info["input_cost_per_image"] * n # Priority 2: Fall back to per-pixel pricing for backward compatibility elif "input_cost_per_pixel" in cost_info and cost_info["input_cost_per_pixel"] is not None: return cost_info["input_cost_per_pixel"] * height * width * n else: raise Exception(f"No pricing information found for model {model}. Tried checking {models_to_check}") def default_video_cost_calculator( model: str, duration_seconds: float, custom_llm_provider: str | None = None, model_info: ModelInfo | None = None, video_resolution: str | None = None, ) -> float: """ Default video cost calculator for video generation Args: model (str): Model name duration_seconds (float): Duration of the generated video in seconds custom_llm_provider (Optional[str]): Custom LLM provider model_info (Optional[ModelInfo]): Deployment-level model info containing custom video pricing. When provided, used before falling back to the global litellm.model_cost lookup. video_resolution (Optional[str]): From usage (e.g. ``720p``, ``1080p``) for tiered per-second pricing. Returns: float: Cost in USD for the video generation Raises: Exception: If model pricing not found in cost map """ # Use custom model_info pricing if provided (deployment-specific pricing) cost_info: dict | None = None if model_info is not None: cost_info = dict(model_info) else: # Build model names for cost lookup base_model_name = model model_name_without_custom_llm_provider: str | None = None if custom_llm_provider and model.startswith(f"{custom_llm_provider}/"): model_name_without_custom_llm_provider = model.replace(f"{custom_llm_provider}/", "") base_model_name = f"{custom_llm_provider}/{model_name_without_custom_llm_provider}" verbose_logger.debug("Looking up cost for video model: %s", base_model_name) model_without_provider: Final = model.split("/")[-1] # Try model with provider first, fall back to base model name models_to_check: Final[list[str | None]] = [ base_model_name, model, model_without_provider, model_name_without_custom_llm_provider, ] for _model in models_to_check: if _model is not None and _model in litellm.model_cost: cost_info = litellm.model_cost[_model] break # If still not found, try with custom_llm_provider prefix if cost_info is None and custom_llm_provider: prefixed_model: Final = f"{custom_llm_provider}/{model}" if prefixed_model in litellm.model_cost: cost_info = litellm.model_cost[prefixed_model] if cost_info is None: raise Exception(f"Model not found in cost map for model={model}") # Check for video-specific cost per second first video_cost_per_second: Final = cost_info.get("output_cost_per_video_per_second") if video_cost_per_second is not None: return video_cost_per_second * duration_seconds output_cost_per_second: Final = _video_output_cost_per_second(cost_info, video_resolution) if output_cost_per_second is not None: return output_cost_per_second * duration_seconds # If no cost information found, return 0 verbose_logger.info( "No cost information found for video model %s. Please add pricing to model_prices_and_context_window.json", model, ) return 0.0 def batch_cost_calculator( usage: Usage, model: str, custom_llm_provider: str | None = None, model_info: ModelInfo | None = None, data_residency: str | None = None, ) -> tuple[float, float]: """ Calculate the cost of a batch job. Args: model_info: Optional deployment-level model info containing custom batch pricing (e.g. input_cost_per_token_batches). When provided, skips the global litellm.get_model_info() lookup so that deployment-specific pricing is used. """ _, custom_llm_provider, _, _ = litellm.get_llm_provider(model=model, custom_llm_provider=custom_llm_provider) verbose_logger.debug( "Calculating batch cost per token. model=%s, custom_llm_provider=%s", model, custom_llm_provider, ) if model_info is None: try: model_info = litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider) except Exception: model_info = None elif not any( model_info.get(k) is not None for k in ( "input_cost_per_token_batches", "input_cost_per_token", "output_cost_per_token_batches", "output_cost_per_token", ) ): # model_info was provided (e.g. deployment metadata with only id/db_model) # but carries no pricing fields. Fall back to the global pricing table so # that standard model pricing is used instead of silently returning $0. try: global_info: Final = litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider) if global_info: model_info = global_info except Exception: pass if not model_info: return 0.0, 0.0 input_cost_per_token_batches: Final = model_info.get("input_cost_per_token_batches") input_cost_per_token: Final = model_info.get("input_cost_per_token") output_cost_per_token_batches: Final = model_info.get("output_cost_per_token_batches") output_cost_per_token: Final = model_info.get("output_cost_per_token") total_prompt_cost = 0.0 total_completion_cost = 0.0 if input_cost_per_token_batches is not None: total_prompt_cost = usage.prompt_tokens * input_cost_per_token_batches elif input_cost_per_token: details: Final = parse_prompt_tokens_details(usage) cache_read_tokens: Final = details["cache_hit_tokens"] cache_creation_tokens: Final = details["cache_creation_tokens"] # Subtract cached tokens from prompt_tokens before calculating cost # Fixes issue where cached tokens are being charged again base_input_tokens: Final = get_billable_input_tokens(usage) - cache_creation_tokens total_prompt_cost = ( base_input_tokens * (input_cost_per_token) / 2 ) # batch cost is usually half of the regular token cost # Add cache read cost if applicable cache_read_cost_key: Final = _get_service_tier_cost_key("cache_read_input_token_cost", None) total_prompt_cost += calculate_cost_component(model_info, cache_read_cost_key, cache_read_tokens) / 2 cache_creation_cost: Final = model_info.get("cache_creation_input_token_cost") or input_cost_per_token total_prompt_cost += cache_creation_tokens * cache_creation_cost / 2 if output_cost_per_token_batches is not None: total_completion_cost = usage.completion_tokens * output_cost_per_token_batches elif output_cost_per_token: total_completion_cost = ( usage.completion_tokens * (output_cost_per_token) / 2 ) # batch cost is usually half of the regular token cost uplift: Final = _get_regional_uplift_multiplier(model_info, data_residency) if uplift != 1.0: total_prompt_cost *= uplift total_completion_cost *= uplift return total_prompt_cost, total_completion_cost def _summable_prompt_token_fields(prompt_tokens_details: BaseModel) -> list[str]: field_names: Final = list(type(prompt_tokens_details).model_fields) if getattr(prompt_tokens_details, "cache_write_tokens", None) is None: return field_names return [attr for attr in field_names if attr != "cache_creation_tokens"] class BaseTokenUsageProcessor: @staticmethod def combine_usage_objects(usage_objects: list[Usage]) -> Usage: """ Combine multiple Usage objects into a single Usage object, checking model keys for nested values. """ from litellm.types.utils import ( CompletionTokensDetailsWrapper, PromptTokensDetailsWrapper, Usage, ) combined: Final = Usage() # Sum basic token counts for usage in usage_objects: # Handle direct attributes by checking what exists in the model for attr in dir(usage): if not attr.startswith("_") and not callable(getattr(usage, attr)): current_val = getattr(combined, attr, 0) new_val = getattr(usage, attr, 0) if ( new_val is not None and isinstance(new_val, (int, float)) and isinstance(current_val, (int, float)) ): setattr(combined, attr, current_val + new_val) # Handle nested prompt_tokens_details if hasattr(usage, "prompt_tokens_details") and usage.prompt_tokens_details: if not hasattr(combined, "prompt_tokens_details") or not combined.prompt_tokens_details: combined.prompt_tokens_details = PromptTokensDetailsWrapper() # Check what keys exist in the model's prompt_tokens_details # Access model_fields on the class, not the instance, to avoid Pydantic 2.11+ deprecation warnings for attr in _summable_prompt_token_fields(usage.prompt_tokens_details): if ( hasattr(usage.prompt_tokens_details, attr) and not attr.startswith("_") and not callable(getattr(usage.prompt_tokens_details, attr)) ): current_val = getattr(combined.prompt_tokens_details, attr, 0) or 0 new_val = getattr(usage.prompt_tokens_details, attr, 0) or 0 if new_val is not None and isinstance(new_val, (int, float)): setattr( combined.prompt_tokens_details, attr, current_val + new_val, ) # Handle nested completion_tokens_details if hasattr(usage, "completion_tokens_details") and usage.completion_tokens_details: if not hasattr(combined, "completion_tokens_details") or not combined.completion_tokens_details: combined.completion_tokens_details = CompletionTokensDetailsWrapper() # Check what keys exist in the model's completion_tokens_details # Access model_fields on the class, not the instance, to avoid Pydantic 2.11+ deprecation warnings for attr in type(usage.completion_tokens_details).model_fields: if not attr.startswith("_") and not callable(getattr(usage.completion_tokens_details, attr)): current_val = getattr(combined.completion_tokens_details, attr, 0) or 0 new_val = getattr(usage.completion_tokens_details, attr, 0) or 0 if isinstance(new_val, (int, float)): setattr( combined.completion_tokens_details, attr, current_val + new_val, ) return combined class RealtimeAPITokenUsageProcessor(BaseTokenUsageProcessor): @staticmethod def collect_usage_from_realtime_stream_results( results: OpenAIRealtimeStreamList, ) -> list[Usage]: """ Collect usage from realtime stream results """ response_done_events: Final[list[OpenAIRealtimeStreamResponseBaseObject]] = cast( list[OpenAIRealtimeStreamResponseBaseObject], [result for result in results if result["type"] == "response.done"], ) usage_objects: Final[list[Usage]] = [] for result in response_done_events: usage_object = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage( result["response"].get("usage", {}) ) usage_objects.append(usage_object) return usage_objects @staticmethod def collect_and_combine_usage_from_realtime_stream_results( results: OpenAIRealtimeStreamList, ) -> Usage: """ Collect and combine usage from realtime stream results """ collected_usage_objects = RealtimeAPITokenUsageProcessor.collect_usage_from_realtime_stream_results(results) combined_usage_object: Final = RealtimeAPITokenUsageProcessor.combine_usage_objects(collected_usage_objects) return combined_usage_object @staticmethod def create_logging_realtime_object( usage: Usage, results: OpenAIRealtimeStreamList ) -> LiteLLMRealtimeStreamLoggingObject: return LiteLLMRealtimeStreamLoggingObject( usage=usage, results=results, ) _TRANSCRIPTION_COMPLETED_EVENT_TYPE: Final = "conversation.item.input_audio_transcription.completed" def _candidate_realtime_token_costs( model_name: str, combined_usage_object: Usage, custom_llm_provider: str, data_residency: str | None, ) -> tuple[float, float] | None: try: return generic_cost_per_token( model=model_name, usage=combined_usage_object, custom_llm_provider=custom_llm_provider, data_residency=data_residency, ) except Exception: return None def _cost_map_entry_declares_pricing(model_name: str, custom_llm_provider: str) -> bool: entries: Final = ( litellm.model_cost.get(model_name), litellm.model_cost.get(f"{custom_llm_provider}/{model_name}"), ) return any( entry is not None and any("cost_per" in field and value is not None for field, value in entry.items()) for entry in entries ) def _first_priced_realtime_token_costs( potential_model_names: Sequence[str | None], combined_usage_object: Usage, custom_llm_provider: str, data_residency: str | None, ) -> tuple[float, float]: candidate_costs: Final = ( (model_name, costs) for model_name in potential_model_names if model_name is not None and ( costs := _candidate_realtime_token_costs( model_name=model_name, combined_usage_object=combined_usage_object, custom_llm_provider=custom_llm_provider, data_residency=data_residency, ) ) is not None ) return next( ( costs for model_name, costs in candidate_costs if sum(costs) > 0 or _cost_map_entry_declares_pricing(model_name, custom_llm_provider) ), (0.0, 0.0), ) def handle_realtime_stream_cost_calculation( results: OpenAIRealtimeStreamList, combined_usage_object: Usage, custom_llm_provider: str, litellm_model_name: str, data_residency: str | None = None, litellm_logging_obj: LitellmLoggingObject | None = None, ) -> float: """ Handles the cost calculation for realtime stream responses. Pick the 'response.done' events. Calculate total cost across all 'response.done' events. Args: results: A list of OpenAIRealtimeStreamBaseObject objects """ received_model = None potential_model_names: Final = [] for result in results: if result["type"] == "session.created": received_model = cast(OpenAIRealtimeStreamSessionEvents, result)["session"].get("model", None) potential_model_names.append(received_model) potential_model_names.append(litellm_model_name) input_cost_per_token, output_cost_per_token = _first_priced_realtime_token_costs( potential_model_names=potential_model_names, combined_usage_object=combined_usage_object, custom_llm_provider=custom_llm_provider, data_residency=data_residency, ) transcription_cost: Final = ( handle_realtime_transcription_cost_calculation( results=results, custom_llm_provider=custom_llm_provider, litellm_model_name=litellm_model_name, ) if any(r.get("type") == _TRANSCRIPTION_COMPLETED_EVENT_TYPE for r in results) else 0.0 ) total_cost: Final = input_cost_per_token + output_cost_per_token + transcription_cost _store_cost_breakdown_in_logging_obj( litellm_logging_obj=litellm_logging_obj, prompt_tokens_cost_usd_dollar=input_cost_per_token, completion_tokens_cost_usd_dollar=output_cost_per_token, cost_for_built_in_tools_cost_usd_dollar=0.0, total_cost_usd_dollar=total_cost, additional_costs={"transcription_cost": transcription_cost} if transcription_cost > 0 else None, data_residency=data_residency, ) return total_cost def handle_realtime_transcription_cost_calculation( results: OpenAIRealtimeStreamList, custom_llm_provider: str, litellm_model_name: str, ) -> float: """ Cost for realtime transcription sessions (e.g. gpt-realtime-whisper). Transcription sessions emit no `response.done` events; instead each `conversation.item.input_audio_transcription.completed` event carries a `usage` object billed by the ASR model. The usage is one of: - {"type": "duration", "seconds": } → priced via input_cost_per_second - {"type": "tokens", "input_tokens": ...} → priced via input/audio token cost """ completed_events: Final = [ cast(dict, result) for result in results if result.get("type") == _TRANSCRIPTION_COMPLETED_EVENT_TYPE ] if not completed_events: return 0.0 model_name: Final = _get_transcription_model_name_from_results(results) or litellm_model_name try: model_info = litellm.get_model_info(model=model_name, custom_llm_provider=custom_llm_provider) except Exception: model_info = None total_cost = 0.0 for event in completed_events: usage = event.get("usage") or {} total_cost += _transcription_usage_cost(usage, model_info) return total_cost def _get_transcription_model_name_from_results( results: OpenAIRealtimeStreamList, ) -> str | None: """Resolve the ASR model from a transcription_session.* / session.* event.""" for result in results: if result.get("type") in ( "transcription_session.created", "transcription_session.updated", "session.created", "session.updated", ): session = cast(dict, result).get("session", {}) or {} transcription = ((session.get("audio", {}) or {}).get("input", {}) or {}).get( "transcription", {} ) or session.get("input_audio_transcription", {}) model = (transcription or {}).get("model") or session.get("model") if model: return model return None def _transcription_usage_cost(usage: dict, model_info: ModelInfo | None) -> float: if model_info is None: return 0.0 usage_type: Final = usage.get("type") if usage_type == "duration": seconds: Final = usage.get("seconds") or 0.0 per_second: Final = model_info.get("input_cost_per_second") or 0.0 return float(seconds) * float(per_second) if usage_type == "tokens": input_token_details: Final = usage.get("input_token_details") or {} audio_tokens: Final = input_token_details.get("audio_tokens") or 0 text_tokens: Final = input_token_details.get("text_tokens") or 0 output_tokens: Final = usage.get("output_tokens") or 0 audio_cost: Final = float(audio_tokens) * float( model_info.get("input_cost_per_audio_token") or model_info.get("input_cost_per_token") or 0.0 ) text_cost: Final = float(text_tokens) * float(model_info.get("input_cost_per_token") or 0.0) output_cost: Final = float(output_tokens) * float(model_info.get("output_cost_per_token") or 0.0) return audio_cost + text_cost + output_cost return 0.0