fix(cost): make deployment custom pricing robust to model_cost state

Two intertwined billing bugs around deployment-level custom pricing
(e.g. zero-cost BYOK wildcard routes). First, completion() and
embedding() register per-request input/output_cost_per_token via
litellm.register_model, which resolves "<provider>/<model>" to the
canonical cost-map key and overwrites the shared entry. One request
with zero rates re-priced the canonical model entry for the whole
process, silently zeroing billing for all other traffic on that model

Second, cost calculation resolved deployment custom pricing through
litellm.model_cost[<router_model_id>]. The proxy model-cost-map reload
replaces litellm.model_cost wholesale, dropping every router-registered
deployment entry, after which _select_model_name_for_cost_calc falls
back to the concrete model name and bills custom-priced traffic at the
model's public price

Fix: thread the deployment's actual rates from litellm_params and
metadata/litellm_metadata model_info into the cost calculator as
custom_cost_per_token (cost_per_token already early-returns on it), and
only register per-request pricing when the model has no pricing in the
cost map, so unknown models can still be priced while known models keep
their canonical entries

Fixes #30081.
This commit is contained in:
Filippo Mattia Menghi 2026-06-10 09:37:19 +02:00
parent e15b37a18e
commit 75a6c0ee31
5 changed files with 210 additions and 4 deletions

View file

@ -1856,6 +1856,7 @@ def response_cost_calculator(
cache_hit: Optional[bool] = None,
base_model: Optional[str] = None,
custom_pricing: Optional[bool] = None,
custom_cost_per_token: Optional[CostPerToken] = None,
prompt: str = "",
standard_built_in_tools_params: Optional[StandardBuiltInToolsParams] = None,
litellm_model_name: Optional[str] = None,
@ -1893,6 +1894,7 @@ def response_cost_calculator(
custom_llm_provider=custom_llm_provider,
optional_params=optional_params,
custom_pricing=custom_pricing,
custom_cost_per_token=custom_cost_per_token,
base_model=base_model,
prompt=prompt,
standard_built_in_tools_params=standard_built_in_tools_params,

View file

@ -99,6 +99,7 @@ from litellm.types.utils import (
CachingDetails,
CallTypes,
CostBreakdown,
CostPerToken,
CostResponseTypes,
CustomPricingLiteLLMParams,
DynamicPromptManagementParamLiteral,
@ -1519,6 +1520,18 @@ class Logging(LiteLLMLoggingBaseClass):
)
)
# Pass the deployment's actual rates through the call, so custom pricing
# holds even when the router_model_id entry is missing from
# litellm.model_cost (model-cost-map reloads replace the map wholesale,
# after which cost calc would fall back to the model's public price).
custom_cost_per_token: Optional[CostPerToken] = None
if custom_pricing is True:
custom_cost_per_token = get_custom_cost_per_token_from_litellm_params(
litellm_params=(
self.litellm_params if hasattr(self, "litellm_params") else None
)
)
prompt = "" # use for tts cost calc
_input = self.model_call_details.get("input", None)
if _input is not None and isinstance(_input, str):
@ -1541,6 +1554,7 @@ class Logging(LiteLLMLoggingBaseClass):
"call_type": self.call_type,
"optional_params": self.optional_params,
"custom_pricing": custom_pricing,
"custom_cost_per_token": custom_cost_per_token,
"prompt": prompt,
"standard_built_in_tools_params": self.standard_built_in_tools_params,
"router_model_id": router_model_id,
@ -4774,6 +4788,45 @@ def use_custom_pricing_for_model(litellm_params: Optional[dict]) -> bool:
return False
def get_custom_cost_per_token_from_litellm_params(
litellm_params: Optional[dict],
) -> Optional[CostPerToken]:
"""
Extract explicit per-token rates from litellm_params or its
metadata/litellm_metadata model_info, mirroring the lookup order of
`use_custom_pricing_for_model`. A literal 0 is a valid rate (e.g. zero-cost
BYOK deployments). Returns None unless both base rates are set.
"""
if litellm_params is None:
return None
sources = [litellm_params]
for metadata_key in ("metadata", "litellm_metadata"):
metadata: dict = litellm_params.get(metadata_key, {}) or {}
model_info: dict = metadata.get("model_info", {}) or {}
if model_info:
sources.append(model_info)
for source in sources:
input_cost = source.get("input_cost_per_token")
output_cost = source.get("output_cost_per_token")
if input_cost is None or output_cost is None:
continue
custom_cost: CostPerToken = {
"input_cost_per_token": input_cost,
"output_cost_per_token": output_cost,
}
for cache_key in (
"cache_read_input_token_cost",
"cache_creation_input_token_cost",
):
if source.get(cache_key) is not None:
custom_cost[cache_key] = source[cache_key] # type: ignore[literal-required]
return custom_cost
return None
def is_valid_sha256_hash(value: str) -> bool:
# Check if the value is a valid SHA-256 hash (64 hexadecimal characters)
return bool(re.fullmatch(r"[a-fA-F0-9]{64}", value))

View file

@ -1083,6 +1083,28 @@ def _build_custom_pricing_entry(
return entry
def _model_has_known_pricing(model: str, custom_llm_provider: str) -> bool:
"""
Check if the model already has pricing in the global cost map.
register_model() resolves "<provider>/<model>" to the canonical cost-map
key, so registering request-scoped rates for a known model overwrites the
shared entry and re-prices every other request for that model in the
process (e.g. a zero-cost BYOK deployment zeroing the real model's
billing). Known models get their per-request rates via
custom_cost_per_token at cost-calculation time instead; registration is
only needed so unknown models can be priced at all.
"""
for key in (model, f"{custom_llm_provider}/{model}"):
entry = litellm.model_cost.get(key)
if entry is not None and (
entry.get("input_cost_per_token") is not None
or entry.get("input_cost_per_second") is not None
):
return True
return False
@tracer.wrap()
@client
def completion( # type: ignore # noqa: PLR0915
@ -1451,8 +1473,9 @@ def completion( # type: ignore # noqa: PLR0915
### REGISTER CUSTOM MODEL PRICING -- IF GIVEN ###
if (
input_cost_per_token is not None and output_cost_per_token is not None
) or input_cost_per_second is not None:
(input_cost_per_token is not None and output_cost_per_token is not None)
or input_cost_per_second is not None
) and not _model_has_known_pricing(model, custom_llm_provider):
litellm.register_model(
{
f"{custom_llm_provider}/{model}": _build_custom_pricing_entry(
@ -4988,8 +5011,9 @@ def embedding( # noqa: PLR0915
### REGISTER CUSTOM MODEL PRICING -- IF GIVEN ###
if (
input_cost_per_token is not None and output_cost_per_token is not None
) or input_cost_per_second is not None:
(input_cost_per_token is not None and output_cost_per_token is not None)
or input_cost_per_second is not None
) and not _model_has_known_pricing(model, custom_llm_provider):
litellm.register_model(
{
f"{custom_llm_provider}/{model}": _build_custom_pricing_entry(

View file

@ -2554,3 +2554,88 @@ def test_openrouter_gemini_3_1_flash_lite_stable_pricing():
assert model_info["cache_read_input_token_cost"] == 2.5e-08
assert model_info["max_input_tokens"] == 1048576
assert model_info["max_output_tokens"] == 65536
def _logging_obj_with_custom_pricing(litellm_params: dict, model: str):
import datetime
from litellm.litellm_core_utils.litellm_logging import Logging
logging_obj = Logging(
model=model,
messages=[{"role": "user", "content": "hi"}],
stream=False,
call_type="acompletion",
start_time=datetime.datetime.now(),
litellm_call_id="test-call-id",
function_id="test",
)
logging_obj.update_environment_variables(
model=model, user=None, optional_params={}, litellm_params=litellm_params
)
return logging_obj
def _response_with_usage(model: str) -> ModelResponse:
response = ModelResponse(model=model)
response.usage = Usage(
prompt_tokens=100_000, completion_tokens=10_000, total_tokens=110_000
)
return response
def test_response_cost_calculator_uses_deployment_rates_when_map_entry_missing():
"""A zero-cost BYOK deployment whose router_model_id entry is absent from
litellm.model_cost (post cost-map reload) must be priced at 0, not at the
concrete model's public price."""
logging_obj = _logging_obj_with_custom_pricing(
litellm_params={
"metadata": {
"model_info": {
"id": "byok-deployment-id-not-in-model-cost",
"input_cost_per_token": 0,
"output_cost_per_token": 0,
}
}
},
model="anthropic/claude-opus-4-6",
)
cost = logging_obj._response_cost_calculator(
result=_response_with_usage("claude-opus-4-6")
)
assert cost == 0.0
def test_response_cost_calculator_custom_pricing_survives_model_cost_reload(
monkeypatch,
):
"""Deployment custom pricing must produce the same cost before and after
litellm.model_cost is wholesale-replaced (proxy model-cost-map reload),
which drops every router-registered deployment-id entry."""
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
monkeypatch.setattr(litellm, "model_cost", dict(litellm.model_cost))
deployment_id = "custom-priced-deployment-reload-test"
rates = {"input_cost_per_token": 1e-06, "output_cost_per_token": 2e-06}
litellm.register_model({deployment_id: {"litellm_provider": "anthropic", **rates}})
logging_obj = _logging_obj_with_custom_pricing(
litellm_params={
"litellm_metadata": {"model_info": {"id": deployment_id, **rates}}
},
model="anthropic/claude-opus-4-6",
)
expected = 100_000 * 1e-06 + 10_000 * 2e-06
cost_before_reload = logging_obj._response_cost_calculator(
result=_response_with_usage("claude-opus-4-6")
)
assert cost_before_reload == pytest.approx(expected)
monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url=""))
assert deployment_id not in litellm.model_cost
cost_after_reload = logging_obj._response_cost_calculator(
result=_response_with_usage("claude-opus-4-6")
)
assert cost_after_reload == pytest.approx(expected)

View file

@ -1962,3 +1962,45 @@ class TestCallTypesOCR:
call_type = CallTypes("aocr")
assert call_type == CallTypes.aocr
def test_completion_custom_pricing_does_not_overwrite_canonical_model_cost(
monkeypatch,
):
"""Per-request custom pricing must apply to that request only. One request
with zero rates must not re-price the shared model_cost entry, which
register_model resolves "<provider>/<model>" to."""
model = "gpt-3.5-turbo"
original_input_cost = litellm.model_cost[model]["input_cost_per_token"]
assert original_input_cost > 0
monkeypatch.setitem(litellm.model_cost, model, dict(litellm.model_cost[model]))
response = litellm.completion(
model=f"openai/{model}",
messages=[{"role": "user", "content": "hi"}],
mock_response="ok",
api_key="sk-test",
input_cost_per_token=0.0,
output_cost_per_token=0.0,
)
assert litellm.model_cost[model]["input_cost_per_token"] == original_input_cost
assert response._hidden_params["response_cost"] == 0.0
def test_completion_custom_pricing_still_registers_unknown_model():
model = "openai/unknown-custom-priced-model-xyz"
litellm.model_cost.pop(model, None)
try:
litellm.completion(
model=model,
messages=[{"role": "user", "content": "hi"}],
mock_response="ok",
api_key="sk-test",
input_cost_per_token=1e-07,
output_cost_per_token=2e-07,
)
assert litellm.model_cost[model]["input_cost_per_token"] == 1e-07
assert litellm.model_cost[model]["output_cost_per_token"] == 2e-07
finally:
litellm.model_cost.pop(model, None)