fix(cost_calculator): mirror the anthropic geo uplift in the token-type cost breakdown

This commit is contained in:
mateo-berri 2026-08-14 16:44:22 -07:00
parent d330b64943
commit e94a97fcfc
3 changed files with 96 additions and 5 deletions

View file

@ -694,6 +694,23 @@ def _get_regional_uplift_multiplier(model_info: ModelInfo, data_residency: str |
return 1.0
def get_provider_specific_geo_multiplier(model_info: ModelInfo, usage: Usage) -> float:
"""
Resolve the provider-specific regional pricing multiplier for the geo the
request was served from (``usage.inference_geo``), e.g. Anthropic's ``us: 1.1``
stored under ``provider_specific_entry``. The regional surcharge applies to
every token type, so per-type cost breakdowns must scale by it too.
Returns 1.0 when the request was served globally or the model carries no
multiplier for the geo.
"""
inference_geo: Final = getattr(usage, "inference_geo", None)
if not isinstance(inference_geo, str) or inference_geo.lower() in ("global", "not_available"):
return 1.0
provider_specific_entry: Final[dict[str, float]] = model_info.get("provider_specific_entry") or {}
return float(provider_specific_entry.get(inference_geo.lower(), 1.0))
def _resolve_reasoning_token_cost(
model_info: ModelInfo,
service_tier: str | None,
@ -981,6 +998,14 @@ def get_token_type_cost_breakdown(
cache_read_cost *= uplift
cache_creation_cost *= uplift
# Mirror the provider-specific geo uplift (e.g. Anthropic us: 1.1) the totals
# apply, so cache and reasoning line items stay reconciled with them.
geo_multiplier: Final = get_provider_specific_geo_multiplier(model_info=model_info, usage=usage)
if geo_multiplier != 1.0:
reasoning_cost *= geo_multiplier
cache_read_cost *= geo_multiplier
cache_creation_cost *= geo_multiplier
return TokenTypeCostBreakdown(
reasoning_cost=reasoning_cost,
cache_read_cost=cache_read_cost,

View file

@ -13,6 +13,7 @@ from litellm.litellm_core_utils.llm_cost_calc.utils import (
_parse_prompt_tokens_details,
calculate_cache_writing_cost,
generic_cost_per_token,
get_provider_specific_geo_multiplier,
)
if TYPE_CHECKING:
@ -82,11 +83,7 @@ def cost_per_token(model: str, usage: "Usage", service_tier: str | None = None)
model_info: Final = litellm.get_model_info(model=model, custom_llm_provider="anthropic")
provider_specific_entry: Final[dict] = model_info.get("provider_specific_entry") or {}
geo_multiplier: Final = (
provider_specific_entry.get(usage.inference_geo.lower(), 1.0)
if getattr(usage, "inference_geo", None) and usage.inference_geo.lower() not in ("global", "not_available")
else 1.0
)
geo_multiplier: Final = get_provider_specific_geo_multiplier(model_info=model_info, usage=usage)
speed_multiplier: Final = (
provider_specific_entry.get("fast", 1.0) if getattr(usage, "speed", None) == "fast" else 1.0
)

View file

@ -2558,6 +2558,75 @@ def test_token_type_cost_breakdown_applies_regional_uplift():
assert text_input_cost + eu.cache_read_cost == pytest.approx(prompt_cost)
def test_token_type_cost_breakdown_applies_anthropic_geo_multiplier(monkeypatch):
"""
Anthropic's regional (geo) uplift lives in provider_specific_entry and is
applied to every token type in the totals, so the per-type breakdown must
scale its cache and reasoning line items by it too. Otherwise the logged
cache costs stay at the base rate and the cache uplift is misattributed to
plain input for exactly the cache-heavy regional traffic the uplift targets.
"""
from litellm.llms.anthropic.cost_calculation import (
cost_per_token as anthropic_cost_per_token,
)
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
litellm.model_cost = litellm.get_model_cost_map(url="")
model = "claude-test-geo-breakdown-model"
litellm.register_model(
model_cost={
model: {
"input_cost_per_token": 5e-6,
"output_cost_per_token": 25e-6,
"cache_creation_input_token_cost": 6.25e-6,
"cache_read_input_token_cost": 0.5e-6,
"litellm_provider": "anthropic",
"max_tokens": 8192,
"provider_specific_entry": {"us": 1.1},
}
}
)
def make_usage() -> Usage:
return Usage(
prompt_tokens=10_000,
completion_tokens=500,
total_tokens=10_500,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=2_000,
cache_creation_tokens=6_000,
),
completion_tokens_details=CompletionTokensDetailsWrapper(
reasoning_tokens=200, text_tokens=300
),
)
base_usage = make_usage()
geo_usage = make_usage()
geo_usage.inference_geo = "us"
base = get_token_type_cost_breakdown(
model=model, custom_llm_provider="anthropic", usage=base_usage
)
geo = get_token_type_cost_breakdown(
model=model, custom_llm_provider="anthropic", usage=geo_usage
)
assert base.cache_read_cost == pytest.approx(2_000 * 0.5e-6)
assert base.cache_creation_cost == pytest.approx(6_000 * 6.25e-6)
assert geo.cache_read_cost == pytest.approx(base.cache_read_cost * 1.1)
assert geo.cache_creation_cost == pytest.approx(base.cache_creation_cost * 1.1)
assert geo.reasoning_cost == pytest.approx(base.reasoning_cost * 1.1)
# The uplifted breakdown must still reconcile with the uplifted totals.
prompt_cost, completion_cost = anthropic_cost_per_token(model=model, usage=geo_usage)
text_input_cost = 2_000 * 5e-6 * 1.1
text_output_cost = 300 * 25e-6 * 1.1
assert text_input_cost + geo.cache_read_cost + geo.cache_creation_cost == pytest.approx(prompt_cost)
assert text_output_cost + geo.reasoning_cost == pytest.approx(completion_cost)
@pytest.mark.parametrize("details_as_dict", [True, False])
def test_image_response_input_image_tokens_priced_at_image_rate(details_as_dict):
"""