scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-30 01:52:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 70

fix(cost): read cache tokens from Anthropic-style usage fields as fallback

Browse source 
When prompt_tokens_details is absent or missing cache fields,
generic_cost_per_token now falls back to cache_read_input_tokens
and cache_creation_input_tokens on the Usage object.  This fixes
silent $0 billing for cache_creation and cache_read token costs
on bedrock_converse models that use prompt caching.

Closes #29145

Assisted-by: Claude Opus 4.6
This commit is contained in:
Jonathan Wrede 2026-05-29 05:49:53 +00:00
parent 9918a9c78c
commit b372fc1328
No known key found for this signature in database
2 changed files with 160 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

23
litellm/litellm_core_utils/llm_cost_calc/utils.py
View file

@ -487,6 +487,20 @@ def _parse_prompt_tokens_details(usage: Usage) -> PromptTokensDetailsResult:
or 0.0
)
# Fallback: Anthropic/Bedrock-style usage objects may carry cache token
# counts as top-level fields (cache_read_input_tokens,
# cache_creation_input_tokens) even when prompt_tokens_details does not
# include them. Use these as a fallback so cache costs are not silently
# dropped.
if cache_hit_tokens == 0:
_fallback_read = getattr(usage, "cache_read_input_tokens", None)
if _fallback_read is not None and isinstance(_fallback_read, int):
cache_hit_tokens = _fallback_read
if cache_creation_tokens == 0:
_fallback_create = getattr(usage, "cache_creation_input_tokens", None)
if _fallback_create is not None and isinstance(_fallback_create, int):
cache_creation_tokens = _fallback_create
return PromptTokensDetailsResult(
cache_hit_tokens=cache_hit_tokens,
cache_creation_tokens=cache_creation_tokens,
@ -697,6 +711,15 @@ def generic_cost_per_token( # noqa: PLR0915
)
if usage.prompt_tokens_details:
prompt_tokens_details = _parse_prompt_tokens_details(usage)
else:
# Fallback: Anthropic/Bedrock-style usage objects carry cache token
# counts as top-level fields even when prompt_tokens_details is absent.
_fallback_read = getattr(usage, "cache_read_input_tokens", None)
_fallback_create = getattr(usage, "cache_creation_input_tokens", None)
if _fallback_read is not None and isinstance(_fallback_read, int):
prompt_tokens_details["cache_hit_tokens"] = _fallback_read
if _fallback_create is not None and isinstance(_fallback_create, int):
prompt_tokens_details["cache_creation_tokens"] = _fallback_create
## EDGE CASE - text tokens not set or includes cached tokens (double-counting)
## Some providers (like xAI) report text_tokens = prompt_tokens (including cached)

137
tests/test_litellm/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py
View file

@ -1538,3 +1538,140 @@ def test_data_residency_composes_with_service_tier(_local_model_cost_map):
assert priority_base_total > 0
assert priority_eu_total == pytest.approx(priority_base_total * 1.10, rel=1e-9)
class TestBedrockCacheTokenCost:
"""
Verify that cache_creation and cache_read token costs are correctly
calculated for bedrock_converse models.
Regression tests for https://github.com/BerriAI/litellm/issues/29145
"""
MODEL = "eu.anthropic.claude-sonnet-4-5-20250929-v1:0"
PROVIDER = "bedrock"
# Rates from the model cost map for this model
INPUT_RATE = 3.3e-6
OUTPUT_RATE = 1.65e-5
CACHE_CREATE_RATE = 4.125e-6
CACHE_READ_RATE = 3.3e-7
TEXT_TOKENS = 1
CACHE_CREATE_TOKENS = 1190
CACHE_READ_TOKENS = 32634
COMPLETION_TOKENS = 672
EXPECTED_PROMPT_COST = (
TEXT_TOKENS * INPUT_RATE
+ CACHE_CREATE_TOKENS * CACHE_CREATE_RATE
+ CACHE_READ_TOKENS * CACHE_READ_RATE
)
EXPECTED_COMPLETION_COST = COMPLETION_TOKENS * OUTPUT_RATE
def test_bedrock_cache_cost_with_prompt_tokens_details(self, _local_model_cost_map):
"""Normal non-streaming path: prompt_tokens_details is populated."""
usage = Usage(
prompt_tokens=(
self.TEXT_TOKENS + self.CACHE_READ_TOKENS + self.CACHE_CREATE_TOKENS
),
completion_tokens=self.COMPLETION_TOKENS,
total_tokens=(
self.TEXT_TOKENS
+ self.CACHE_READ_TOKENS
+ self.CACHE_CREATE_TOKENS
+ self.COMPLETION_TOKENS
),
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=self.CACHE_READ_TOKENS,
cache_creation_tokens=self.CACHE_CREATE_TOKENS,
text_tokens=self.TEXT_TOKENS,
),
cache_creation_input_tokens=self.CACHE_CREATE_TOKENS,
cache_read_input_tokens=self.CACHE_READ_TOKENS,
)
prompt_cost, completion_cost = generic_cost_per_token(
model=self.MODEL,
usage=usage,
custom_llm_provider=self.PROVIDER,
)
assert prompt_cost == pytest.approx(self.EXPECTED_PROMPT_COST, rel=1e-6)
assert completion_cost == pytest.approx(self.EXPECTED_COMPLETION_COST, rel=1e-6)
def test_bedrock_cache_cost_without_prompt_tokens_details(
self, _local_model_cost_map
):
"""Fallback path: prompt_tokens_details is absent but Anthropic-style
cache fields are set on the Usage object (e.g. streaming chunk builder
output)."""
usage = Usage(
prompt_tokens=(
self.TEXT_TOKENS + self.CACHE_READ_TOKENS + self.CACHE_CREATE_TOKENS
),
completion_tokens=self.COMPLETION_TOKENS,
total_tokens=(
self.TEXT_TOKENS
+ self.CACHE_READ_TOKENS
+ self.CACHE_CREATE_TOKENS
+ self.COMPLETION_TOKENS
),
)
# Mimic the streaming chunk builder: set cache fields via setattr
setattr(usage, "cache_creation_input_tokens", self.CACHE_CREATE_TOKENS)
setattr(usage, "cache_read_input_tokens", self.CACHE_READ_TOKENS)
usage._cache_creation_input_tokens = self.CACHE_CREATE_TOKENS
usage._cache_read_input_tokens = self.CACHE_READ_TOKENS
assert usage.prompt_tokens_details is None
prompt_cost, completion_cost = generic_cost_per_token(
model=self.MODEL,
usage=usage,
custom_llm_provider=self.PROVIDER,
)
assert prompt_cost == pytest.approx(self.EXPECTED_PROMPT_COST, rel=1e-6)
assert completion_cost == pytest.approx(self.EXPECTED_COMPLETION_COST, rel=1e-6)
def test_bedrock_cache_cost_prompt_details_missing_cache_fields(
self, _local_model_cost_map
):
"""prompt_tokens_details exists but lacks cache fields; top-level
Anthropic-style fields should be used as fallback."""
usage = Usage(
prompt_tokens=(
self.TEXT_TOKENS + self.CACHE_READ_TOKENS + self.CACHE_CREATE_TOKENS
),
completion_tokens=self.COMPLETION_TOKENS,
total_tokens=(
self.TEXT_TOKENS
+ self.CACHE_READ_TOKENS
+ self.CACHE_CREATE_TOKENS
+ self.COMPLETION_TOKENS
),
prompt_tokens_details=PromptTokensDetailsWrapper(
text_tokens=self.TEXT_TOKENS,
),
cache_creation_input_tokens=self.CACHE_CREATE_TOKENS,
cache_read_input_tokens=self.CACHE_READ_TOKENS,
)
prompt_cost, completion_cost = generic_cost_per_token(
model=self.MODEL,
usage=usage,
custom_llm_provider=self.PROVIDER,
)
assert prompt_cost == pytest.approx(self.EXPECTED_PROMPT_COST, rel=1e-6)
assert completion_cost == pytest.approx(self.EXPECTED_COMPLETION_COST, rel=1e-6)
def test_bedrock_no_cache_tokens_unchanged(self, _local_model_cost_map):
"""When no cache tokens are present, cost should be straightforward."""
usage = Usage(
prompt_tokens=1000,
completion_tokens=500,
total_tokens=1500,
)
prompt_cost, completion_cost = generic_cost_per_token(
model=self.MODEL,
usage=usage,
custom_llm_provider=self.PROVIDER,
)
assert prompt_cost == pytest.approx(1000 * self.INPUT_RATE, rel=1e-6)
assert completion_cost == pytest.approx(500 * self.OUTPUT_RATE, rel=1e-6)