diff --git a/litellm/cost_calculator.py b/litellm/cost_calculator.py index c6a785b82be..f59cdf68def 100644 --- a/litellm/cost_calculator.py +++ b/litellm/cost_calculator.py @@ -10,7 +10,7 @@ from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias, cast from httpx import Response -from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError, with_config +from pydantic import BaseModel, ConfigDict, TypeAdapter, ValidationError from typing_extensions import ReadOnly, TypedDict, assert_never import litellm @@ -150,6 +150,7 @@ from litellm.utils import ( TextCompletionResponse, TranscriptionResponse, cached_get_model_info_helper, + model_info_with_pricing_overrides, runtime_model_cost_registrations, token_counter, unmapped_model_generalization, @@ -741,22 +742,14 @@ def cost_per_token( elif custom_llm_provider == "anthropic": return anthropic_cost_per_token(model=model, usage=usage_block, service_tier=service_tier) elif custom_llm_provider == "bedrock": - if bedrock_model_info is not None: + resolved_bedrock_info: Final = 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, usage=usage_block, custom_llm_provider="bedrock", service_tier=service_tier, - model_info=bedrock_model_info, - ) - registered_pricing_model: Final = _bedrock_registered_pricing_model(model) - if registered_pricing_model is not None: - return generic_cost_per_token( - model=model, - usage=usage_block, - custom_llm_provider="bedrock", - service_tier=service_tier, - model_info=_cost_map_model_info(registered_pricing_model, None), + model_info=resolved_bedrock_info, ) return bedrock_cost_per_token(model=model, usage=usage_block, service_tier=service_tier) elif custom_llm_provider == "openai": @@ -887,52 +880,28 @@ def _bedrock_registration_has_prices(entry: Mapping[str, object]) -> bool: ) -def _bedrock_registered_pricing_model(model: str) -> str | None: - return next( - ( - key - for key, entry in runtime_model_cost_registrations(model, "bedrock") - if _bedrock_registration_has_prices(entry) - ), - None, - ) - - -@with_config(ConfigDict(extra="allow")) -class _BedrockPricingInfo(ModelInfo): - pass - - -_BEDROCK_PRICING_INFO: Final = TypeAdapter(_BedrockPricingInfo) - - -def _bedrock_cache_overlay( +def _bedrock_pricing_info( model: str | None, invoked_model_id: str | None, custom_llm_provider: str | None, explicit_pricing: bool, ) -> ModelInfo | None: - if explicit_pricing or custom_llm_provider != "bedrock" or not model or not invoked_model_id: + if explicit_pricing or custom_llm_provider != "bedrock" or not model: return None registration: Final = next( ( - entry - for _, entry in runtime_model_cost_registrations(model, "bedrock") + (key, entry) + for key, entry in runtime_model_cost_registrations(model, "bedrock") if _bedrock_registration_has_prices(entry) ), None, ) - if registration is None or _cost_map_entry_prices_anything(registration): - return None - profile: Final = _cost_map_model_info(invoked_model_id, "bedrock", prefer_registration=False) + profile: Final = _cost_map_model_info(invoked_model_id, "bedrock") if invoked_model_id else None if profile is None or not _declares_rates(profile): - return None - overrides: Final = { - field: value - for field, value in registration.items() - if value is not None and field.startswith("cache_") and is_server_derived_pricing_key(field) - } - return _BEDROCK_PRICING_INFO.validate_python({**profile, **overrides}) + return _cost_map_model_info(registration[0], None) if registration is not None else None + if registration is None or registration[0] == profile["key"]: + return profile + return model_info_with_pricing_overrides(profile, registration[1]) def _declares_rates(model_info: ModelInfo | None) -> bool: @@ -1000,25 +969,13 @@ def select_model_name_for_cost_calc( 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 - invoked_model_info: Final = ( - _cost_map_model_info(invoked_model_id, "bedrock", prefer_registration=False) - if invoked_model_id and custom_llm_provider == "bedrock" and not explicit_pricing - else None - ) configured_model: Final = model or ( completion_response_model if isinstance(completion_response_model, str) else None ) - registered_pricing_model: Final = ( - _bedrock_registered_pricing_model(configured_model) - if invoked_model_info is not None and isinstance(configured_model, str) - else None - ) - cache_overlay: Final = _bedrock_cache_overlay( + bedrock_model_info: Final = _bedrock_pricing_info( configured_model, invoked_model_id, custom_llm_provider, explicit_pricing ) - bedrock_pricing_model: Final = ( - cache_overlay["key"] if cache_overlay is not None else registered_pricing_model - ) or (invoked_model_info["key"] if invoked_model_info is not None and _declares_rates(invoked_model_info) else None) + bedrock_pricing_model: Final = bedrock_model_info["key"] if bedrock_model_info is not None else None priced_from_response: Final = provider_response_model is not None or completion_response_model is not None priced_region: Final = ( _get_hidden_str_for_cost_calc(hidden_params, "region_name") or region_name @@ -1666,18 +1623,11 @@ def completion_cost( invoked_model_id=invoked_model_id, ) - bedrock_model_info: Final = _bedrock_cache_overlay( + bedrock_model_info: Final = _bedrock_pricing_info( model or _get_response_model(completion_response), invoked_model_id, _get_provider_for_cost_calc(model, custom_llm_provider), explicit_pricing, - ) or ( - _cost_map_model_info(selected_model, "bedrock") - if selected_model is not None - and not explicit_pricing - and _get_provider_for_cost_calc(model, custom_llm_provider) == "bedrock" - and _bedrock_registered_pricing_model(selected_model) is not None - else None ) requested_model: Final = model request_pricing_provider: Final = custom_llm_provider if isinstance(custom_llm_provider, str) else None @@ -2409,17 +2359,8 @@ def _layered_ocr_pricing(*sources: Mapping[str, object] | None) -> OCRPricing: ) -def _cost_map_model_info( - model: str, custom_llm_provider: str | None, *, prefer_registration: bool = True -) -> ModelInfo | None: +def _cost_map_model_info(model: str, custom_llm_provider: str | None) -> ModelInfo | None: try: - registered_pricing_model: Final = ( - _bedrock_registered_pricing_model(model) - if custom_llm_provider == "bedrock" and prefer_registration - else None - ) - if registered_pricing_model is not None: - return litellm.get_model_info(model=registered_pricing_model) return litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider) except Exception: return None @@ -2473,14 +2414,14 @@ def pricing_entry_for_cost_calc( region_name=region_name, invoked_model_id=_invoked_model_id_for_cost(None, litellm_logging_obj), ) - cache_overlay: Final = _bedrock_cache_overlay( + bedrock_model_info: Final = _bedrock_pricing_info( model or _get_response_model(completion_response), _invoked_model_id_for_cost(None, litellm_logging_obj), _get_provider_for_cost_calc(model, custom_llm_provider), custom_pricing is True or base_model is not None, ) - if cache_overlay is not None: - return cache_overlay["key"], cache_overlay + if bedrock_model_info is not None: + return bedrock_model_info["key"], bedrock_model_info candidates: Final = (selected_model, _get_response_model(completion_response), model) resolved: Final = next( (info for info in (_cost_map_model_info(name, custom_llm_provider) for name in candidates if name) if info), diff --git a/litellm/litellm_core_utils/llm_cost_calc/utils.py b/litellm/litellm_core_utils/llm_cost_calc/utils.py index 63f4dfa2259..e7987e6221d 100644 --- a/litellm/litellm_core_utils/llm_cost_calc/utils.py +++ b/litellm/litellm_core_utils/llm_cost_calc/utils.py @@ -285,7 +285,7 @@ def _parse_token_threshold(threshold: str) -> float: def _parse_above_token_threshold(key: str) -> float: - return _parse_token_threshold(key.split("_above_")[1].split("_tokens")[0]) + return _parse_token_threshold(key.rsplit("_above_", 1)[1].split("_tokens")[0]) def _prompt_exceeds_threshold(prompt_tokens: int, threshold: float, inclusive: bool) -> bool: @@ -683,10 +683,12 @@ def _get_token_base_cost( # Optimization: collect threshold keys first to avoid sorting all model_info keys. # Standard thresholds and thresholds suffixed for this request's service tier both count. tier_key_suffix: Final = get_service_tier_cost_key("", service_tier) + threshold_prefixes: Final = tuple(f"{prefix}_above_" for prefix in _BATCH_RATE_PREFIXES) threshold_keys: Final = [ k for k in model_info - if k.startswith("input_cost_per_token_above_") + if k.startswith(threshold_prefixes) + and "_tokens" in k and ( not k.endswith(_NON_STANDARD_THRESHOLD_SUFFIXES) or (tier_key_suffix != "" and k.endswith(tier_key_suffix)) ) @@ -699,7 +701,7 @@ def _get_token_base_cost( if value is not None: try: # Handle both formats: _above_128k_tokens and _above_128_tokens - threshold_str = key.split("_above_")[1].split("_tokens")[0] + threshold_str = key.rsplit("_above_", 1)[1].split("_tokens")[0] threshold = _parse_above_token_threshold(key) if _prompt_exceeds_threshold(usage.prompt_tokens, threshold, threshold_is_inclusive): # Prefer a service_tier-specific above-threshold key when available, @@ -712,7 +714,7 @@ def _get_token_base_cost( service_tier, ) if service_tier - else key + else f"input_cost_per_token_above_{threshold_str}_tokens" ) prompt_base_cost = cast( float, diff --git a/litellm/router.py b/litellm/router.py index 5bf46c68b20..80182cd96bc 100644 --- a/litellm/router.py +++ b/litellm/router.py @@ -396,12 +396,14 @@ from litellm.types.utils import ( StandardLoggingRoutingDecision, Usage, all_litellm_params, + is_server_derived_pricing_key, shared_backend_model_info, ) from litellm.types.utils import ModelInfo as ModelMapInfo from litellm.utils import ( CustomStreamWrapper, EmbeddingResponse, + ModelInfoLookupMiss, ModelResponse, Rules, TextCompletionResponse, @@ -412,6 +414,8 @@ from litellm.utils import ( get_utc_datetime, is_generalized_model_info, is_region_allowed, + lookup_model_info_for_registration, + model_info_with_pricing_overrides, provider_rejectable_params, set_live_deployment_replay, ) @@ -9836,6 +9840,17 @@ class Router: params_base_model: Final = deployment.litellm_params.get("base_model") if isinstance(params_base_model, str) and params_base_model: return params_base_model + params: Final = _LITELLM_PARAMS_ADAPTER.validate_python(deployment.litellm_params.model_dump(exclude_none=True)) + invoked_model: Final = params.get("model_id") + if ( + isinstance(invoked_model, str) + and invoked_model + and ( + deployment.litellm_params.custom_llm_provider == "bedrock" + or deployment.litellm_params.model.startswith("bedrock/") + ) + ): + return invoked_model if invoked_model.startswith("bedrock/") else f"bedrock/{invoked_model}" return deployment.litellm_params.model @staticmethod @@ -11075,7 +11090,7 @@ class Router: return (backend_key,) @staticmethod - def _deployment_model_cost_payload(deployment: Deployment) -> dict: # mutable-ok: cost-map entry + def _deployment_model_cost_payload(deployment: Deployment) -> dict[str, object]: # mutable-ok: cost-map entry """The ``model_info`` a deployment contributes to ``litellm.model_cost``. Custom pricing lives on ``litellm_params`` rather than ``model_info``, and @@ -11083,11 +11098,31 @@ class Router: both are folded back in here. That keeps this reproducible from a deployment alone, which is what lets a refresh rebuild the same entries. """ - model_info: Final[dict] = deployment.model_info.model_dump(exclude_none=True) # mutable-ok: built in place + model_info: Final = dict( # mutable-ok: existing inheritance helpers fill missing rates in place + _MODEL_INFO_ADAPTER.validate_python(deployment.model_info.model_dump(exclude_none=True)) + ) for field in CustomPricingLiteLLMParams.model_fields: field_value = deployment.litellm_params.get(field) if field_value is not None: model_info[field] = field_value + backend_model: Final = Router._cost_map_backend_model(deployment) + params: Final = _LITELLM_PARAMS_ADAPTER.validate_python(deployment.litellm_params.model_dump(exclude_none=True)) + if ( + params.get("model_id") + and backend_model != deployment.litellm_params.model + and ( + deployment.litellm_params.custom_llm_provider == "bedrock" + or deployment.litellm_params.model.startswith("bedrock/") + ) + ): + profile_info: Final = lookup_model_info_for_registration(model=backend_model, custom_llm_provider="bedrock") + if not isinstance(profile_info, ModelInfoLookupMiss): + explicit_info: Final = _MODEL_INFO_ADAPTER.validate_python(model_info) + priced_info: Final = model_info_with_pricing_overrides(profile_info, explicit_info) + return { + **{key: value for key, value in priced_info.items() if is_server_derived_pricing_key(key)}, + **explicit_info, + } Router._inherit_builtin_base_rates_for_off_peak( model_info=model_info, backend_model=deployment.litellm_params.model, diff --git a/litellm/utils.py b/litellm/utils.py index b61b1825888..581e6d4a441 100644 --- a/litellm/utils.py +++ b/litellm/utils.py @@ -3399,6 +3399,25 @@ def _without_overridden_price_variants( ) +@with_config(ConfigDict(extra="allow")) +class _PricingModelInfo(ModelInfo): + pass + + +_PRICING_MODEL_INFO: Final = TypeAdapter(_PricingModelInfo) + + +def model_info_with_pricing_overrides(model_info: ModelInfo, registration: Mapping[str, object]) -> ModelInfo: + from litellm.types.utils import is_server_derived_pricing_key + + overrides: Final = { + name: value for name, value in registration.items() if value is not None and is_server_derived_pricing_key(name) + } + return _PRICING_MODEL_INFO.validate_python( + update_dictionary(dict(_without_overridden_price_variants(model_info, overrides)), overrides) + ) + + @dataclass(frozen=True, slots=True) class _BuiltinCostMatch: key: str @@ -3407,7 +3426,7 @@ class _BuiltinCostMatch: @dataclass(frozen=True, slots=True) -class _LookupMiss: +class ModelInfoLookupMiss: tripped: bool @@ -3494,17 +3513,19 @@ def is_generalized_model_info(model_info: ModelInfo) -> bool: _REGISTRATION_ATTEMPTS: Final = 3 -def _lookup_for_registration(model: str, custom_llm_provider: str | None) -> ModelInfo | _LookupMiss: +def lookup_model_info_for_registration(model: str, custom_llm_provider: str | None) -> ModelInfo | ModelInfoLookupMiss: try: info: Final = get_model_info(model=model, custom_llm_provider=custom_llm_provider) except ModelNotMappedError: - return _LookupMiss(tripped=False) + return ModelInfoLookupMiss(tripped=False) except Exception: - return _LookupMiss(tripped=True) - return _LookupMiss(tripped=False) if is_generalized_model_info(info) else info + return ModelInfoLookupMiss(tripped=True) + return ModelInfoLookupMiss(tripped=False) if is_generalized_model_info(info) else info -def _get_builtin_model_info_for_registration(model: str, custom_llm_provider: str | None) -> ModelInfo | _LookupMiss: +def _get_builtin_model_info_for_registration( + model: str, custom_llm_provider: str | None +) -> ModelInfo | ModelInfoLookupMiss: """Resolve ``model`` to its built-in cost-map entry for registration merging. Misses when the lookup cannot map the model or resolved it via a @@ -3513,10 +3534,12 @@ def _get_builtin_model_info_for_registration(model: str, custom_llm_provider: st changing the map mid-walk, is retried, so a transient failure registers exactly where the untripped lookup would. """ - lookups: Final = (_lookup_for_registration(model, custom_llm_provider) for _ in range(_REGISTRATION_ATTEMPTS)) + lookups: Final = ( + lookup_model_info_for_registration(model, custom_llm_provider) for _ in range(_REGISTRATION_ATTEMPTS) + ) return next( - (lookup for lookup in lookups if not (isinstance(lookup, _LookupMiss) and lookup.tripped)), - _LookupMiss(tripped=True), + (lookup for lookup in lookups if not (isinstance(lookup, ModelInfoLookupMiss) and lookup.tripped)), + ModelInfoLookupMiss(tripped=True), ) @@ -3551,7 +3574,7 @@ def _registration_base( builtin_model_info: Final = _get_builtin_model_info_for_registration( model=key, custom_llm_provider=custom_llm_provider ) - if not isinstance(builtin_model_info, _LookupMiss): + if not isinstance(builtin_model_info, ModelInfoLookupMiss): return _RegistrationWrite( key=builtin_model_info["key"], entry=dict(_without_synthesized_costs(builtin_model_info, key=key, value=value)), diff --git a/tests/unit/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py b/tests/unit/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py index 4b0ec78f015..3a3daf85a16 100644 --- a/tests/unit/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py +++ b/tests/unit/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py @@ -2523,6 +2523,33 @@ def test_get_cost_per_unit_falls_back_from_service_tier_key_to_base(): assert get_cost_per_unit(model_info_direct, "input_cost_per_token_priority") == 5e-6 +@pytest.mark.parametrize( + "field", ("output_cost_per_token", "cache_read_input_token_cost", "cache_creation_input_token_cost") +) +def test_flat_input_price_keeps_other_token_thresholds(field: str) -> None: + from typing import Final, cast + from litellm.litellm_core_utils.llm_cost_calc.utils import _get_token_base_cost + + info: Final = cast( + ModelInfo, + { + "input_cost_per_token": 0.001, + "output_cost_per_token": 0.002, + "cache_read_input_token_cost": 0.003, + "cache_creation_input_token_cost": 0.004, + f"{field}_above_300k_tokens": 0.005, + }, + ) + rates: Final = _get_token_base_cost( + model_info=info, + usage=Usage(prompt_tokens=300_001, completion_tokens=3, total_tokens=300_004), + ) + assert rates[0] == 0.001 + assert rates[1] == (0.005 if field == "output_cost_per_token" else 0.002) + assert rates[2] == (0.005 if field == "cache_creation_input_token_cost" else 0.004) + assert rates[4] == (0.005 if field == "cache_read_input_token_cost" else 0.003) + + def test_threshold_keys_exclude_service_tier_variants(): from typing import cast diff --git a/tests/unit/test_cost_calculator.py b/tests/unit/test_cost_calculator.py index c6f93a001f3..1909ec2fe5d 100644 --- a/tests/unit/test_cost_calculator.py +++ b/tests/unit/test_cost_calculator.py @@ -10129,10 +10129,73 @@ def test_bedrock_cache_registration_with_unpriced_profile_keeps_registered_rates ) == pytest.approx(10 * 0.01 + 3 * 0.02 + 5 * 0.001) +@pytest.mark.parametrize("direction", ("input", "output")) +@pytest.mark.parametrize("rate", (0.0, 0.004)) +@pytest.mark.parametrize("prompt_tokens", (100, 300_001)) +def test_bedrock_partial_registration_overrides_only_its_profile_direction( + _local_model_cost_map: None, + monkeypatch: pytest.MonkeyPatch, + direction: str, + rate: float, + prompt_tokens: int, +) -> None: + from litellm.cost_calculator import pricing_entry_for_cost_calc + + model: Final = "openai.partial-pricing-test" + profile: Final = f"us.{model}" + profile_entry: Final = { + "litellm_provider": "bedrock", + "mode": "chat", + "input_cost_per_token": 0.01, + "output_cost_per_token": 0.02, + "input_cost_per_token_above_300k_tokens": 0.03, + "output_cost_per_token_above_300k_tokens": 0.06, + "cache_read_input_token_cost": 0.001, + } + monkeypatch.setattr(litellm, "model_cost", {**litellm.model_cost, profile: dict(profile_entry)}) + litellm.register_model({model: {f"{direction}_cost_per_token": rate}}, persist_across_reloads=False) + response: Final = ModelResponse( + model=model, + usage=Usage( + prompt_tokens=prompt_tokens, + completion_tokens=3, + total_tokens=prompt_tokens + 3, + prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=5), + ), + ) + logging_obj: Final = Logging( + model=model, + messages=[], + stream=False, + call_type="completion", + start_time=datetime.datetime(2026, 1, 1), + litellm_call_id="partial-pricing", + function_id="partial-pricing", + ) + logging_obj.model_call_details["model_id"] = profile + input_rate: Final = rate if direction == "input" else (0.03 if prompt_tokens > 300_000 else 0.01) + output_rate: Final = rate if direction == "output" else (0.06 if prompt_tokens > 300_000 else 0.02) + assert completion_cost( + completion_response=response, + model=model, + custom_llm_provider="bedrock", + litellm_logging_obj=logging_obj, + ) == pytest.approx((prompt_tokens - 5) * input_rate + 3 * output_rate + 5 * 0.001) + assert logging_obj.billed_token_rates is not None + assert logging_obj.billed_token_rates.input_cost_per_token == input_rate + assert logging_obj.billed_token_rates.output_cost_per_token == output_rate + assert logging_obj.cost_breakdown is not None + assert logging_obj.cost_breakdown["cache_read_cost"] == pytest.approx(5 * 0.001) + entry: Final = pricing_entry_for_cost_calc(model, response, "bedrock", False, None, None, None, logging_obj) + assert entry is not None + assert entry[1][f"{direction}_cost_per_token"] == rate + assert litellm.model_cost[profile] == profile_entry + + @pytest.mark.parametrize("prompt_tokens", (299_999, 300_000, 300_001)) @pytest.mark.parametrize("cache_field", ("cache_read_input_token_cost", "cache_creation_input_token_cost")) @pytest.mark.parametrize("tier_override", (None, 0.0, 0.003)) -def test_bedrock_cache_overlay_preserves_dynamic_threshold_rates( +def test_bedrock_registered_cache_prices_preserve_dynamic_threshold_rates( _local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch, prompt_tokens: int, diff --git a/tests/unit/test_router_model_cost_isolation.py b/tests/unit/test_router_model_cost_isolation.py index 094a13c37b2..b18d9a3a15a 100644 --- a/tests/unit/test_router_model_cost_isolation.py +++ b/tests/unit/test_router_model_cost_isolation.py @@ -832,6 +832,107 @@ def test_should_not_downgrade_chatgpt_shared_key_mode_with_alias_override(): _restore_model_cost_entries(model_keys) +@pytest.mark.parametrize("field", ("input_cost_per_token", "output_cost_per_token", "cache_read_input_token_cost")) +@pytest.mark.parametrize("rate", (0.0, 0.004)) +@pytest.mark.parametrize("explicit_base", (False, True)) +@pytest.mark.parametrize("provider", (None, "bedrock")) +def test_bedrock_deployment_inherits_invoked_profile_around_custom_prices( + monkeypatch: pytest.MonkeyPatch, + field: str, + rate: float, + explicit_base: bool, + provider: str | None, +) -> None: + from litellm.cost_calculator import completion_cost + from litellm.types.utils import ModelResponse, Usage, PromptTokensDetailsWrapper + + model: Final = "openai.gpt-5.6-sol" + profile: Final = f"us.{model}" + base_model: Final = f"global.{model}" + profile_prices: Final = { + "litellm_provider": "bedrock", + "mode": "chat", + "input_cost_per_token": 0.01, + "output_cost_per_token": 0.02, + "cache_read_input_token_cost": 0.001, + } + base_prices: Final = {**profile_prices, "input_cost_per_token": 0.03, "output_cost_per_token": 0.04} + monkeypatch.setattr( + litellm, "model_cost", {**litellm.model_cost, profile: dict(profile_prices), base_model: dict(base_prices)} + ) + deployment: Final = Deployment( + model_name="profile-priced", + litellm_params=LiteLLM_Params( + model=f"converse/{model}" if provider else f"bedrock/converse/{model}", + model_id=profile, + custom_llm_provider=provider, + ), + model_info=ModelInfo( + **{ + "id": "profile-priced-deployment", + field: rate, + **({"base_model": f"bedrock/{base_model}"} if explicit_base else {}), + } + ), + ) + Router._register_deployment_pricing(deployment) + response: Final = ModelResponse( + model=model, + usage=Usage( + prompt_tokens=100, + completion_tokens=3, + total_tokens=103, + prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=5), + ), + ) + input_rate: Final = rate if field == "input_cost_per_token" else (0.03 if explicit_base else 0.01) + output_rate: Final = rate if field == "output_cost_per_token" else (0.04 if explicit_base else 0.02) + cache_rate: Final = rate if field == "cache_read_input_token_cost" else 0.001 + assert completion_cost( + model=model, + completion_response=response, + custom_llm_provider="bedrock", + custom_pricing=True, + router_model_id="profile-priced-deployment", + ) == pytest.approx(95 * input_rate + 3 * output_rate + 5 * cache_rate) + assert litellm.model_cost[profile] == profile_prices + + +@pytest.mark.parametrize("catalog_entry", (None, {"litellm_provider": 1, "mode": "chat"})) +def test_bedrock_deployment_keeps_custom_prices_when_profile_lookup_fails( + monkeypatch: pytest.MonkeyPatch, + catalog_entry: dict[str, object] | None, +) -> None: + from litellm.cost_calculator import completion_cost + from litellm.types.utils import ModelResponse, Usage + + model: Final = "openai.unmapped-pricing-test" + profile: Final = f"us.{model}" + monkeypatch.setattr( + litellm, + "model_cost", + { + **litellm.model_cost, + **({profile: catalog_entry} if catalog_entry is not None else {}), + }, + ) + deployment: Final = Deployment( + model_name="custom-priced", + litellm_params=LiteLLM_Params(model=f"bedrock/converse/{model}", model_id=profile), + model_info=ModelInfo(id="unmapped-profile-deployment", input_cost_per_token=0.01, output_cost_per_token=0.02), + ) + Router._register_deployment_pricing(deployment) + assert completion_cost( + model=model, + completion_response=ModelResponse( + model=model, usage=Usage(prompt_tokens=10, completion_tokens=3, total_tokens=13) + ), + custom_llm_provider="bedrock", + custom_pricing=True, + router_model_id="unmapped-profile-deployment", + ) == pytest.approx(10 * 0.01 + 3 * 0.02) + + def test_partial_custom_pricing_inherits_builtin_cache_pricing(): """A deployment that overrides only input/output cost on a cache-supporting model must still bill cache_read and cache_creation tokens. Before the