fix(anthropic): aggregate 5m/1h cache-write split across iterations path

The iterations branch in AnthropicConfig.calculate_usage summed
cache_creation_input_tokens but never aggregated the per-iteration
cache_creation 5m/1h breakdown, leaving cache_creation_token_details
as None. As a result all cache-creation tokens fell back to the flat
5m write rate, underbilling 1h cache writes by up to 2x.

Aggregate the ephemeral_5m/ephemeral_1h split across iterations so 1h
writes are priced at the 1h rate.

Fixes LIT-4868

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
shivam 2026-07-28 00:07:06 +00:00 committed by Devin AI
parent 1d3b64c66f
commit 66d9752db5
2 changed files with 69 additions and 1 deletions

View file

@ -1,6 +1,7 @@
import json
import re
import time
from collections.abc import Iterable, Mapping
from typing import TYPE_CHECKING, Any, Final, NoReturn, cast
import httpx
@ -2117,6 +2118,18 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
return False
return any(key in usage_object for key in ("cache_read_input_tokens", "cache_creation_input_tokens"))
@staticmethod
def _aggregate_cache_creation_token_details(
cache_creation_objects: Iterable[Mapping[str, Any] | None],
) -> CacheCreationTokenDetails | None:
breakdowns: Final = tuple(c for c in cache_creation_objects if isinstance(c, Mapping))
if not breakdowns:
return None
return CacheCreationTokenDetails(
ephemeral_5m_input_tokens=sum(int(c.get("ephemeral_5m_input_tokens") or 0) for c in breakdowns),
ephemeral_1h_input_tokens=sum(int(c.get("ephemeral_1h_input_tokens") or 0) for c in breakdowns),
)
def calculate_usage(
self,
usage_object: dict,
@ -2150,6 +2163,9 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
cache_creation_input_tokens = sum(it.get("cache_creation_input_tokens", 0) or 0 for it in iterations)
cache_read_input_tokens = sum(it.get("cache_read_input_tokens", 0) or 0 for it in iterations)
prompt_tokens += cache_creation_input_tokens + cache_read_input_tokens
cache_creation_token_details = self._aggregate_cache_creation_token_details(
it.get("cache_creation") for it in iterations
)
if not iterations:
if "cache_creation_input_tokens" in _usage and _usage["cache_creation_input_tokens"] is not None:
@ -2182,7 +2198,7 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
if tool_search_count > 0:
tool_search_requests = tool_search_count
if "cache_creation" in _usage and _usage["cache_creation"] is not None:
if cache_creation_token_details is None and "cache_creation" in _usage and _usage["cache_creation"] is not None:
cache_creation_token_details = CacheCreationTokenDetails(
ephemeral_5m_input_tokens=_usage["cache_creation"].get("ephemeral_5m_input_tokens"),
ephemeral_1h_input_tokens=_usage["cache_creation"].get("ephemeral_1h_input_tokens"),

View file

@ -105,6 +105,58 @@ def test_calculate_usage():
assert usage._cache_read_input_tokens == 0
def test_calculate_usage_aggregates_cache_creation_split_across_iterations():
"""
In the iterations path each iteration can carry the 5m/1h cache_creation
breakdown. calculate_usage must aggregate it into cache_creation_token_details
so 1h writes are priced at the 1h rate instead of silently falling back to 5m.
Regression for LIT-4868.
"""
from litellm.llms.anthropic.cost_calculation import cost_per_token
config = AnthropicConfig()
usage_object = {
"input_tokens": 0,
"output_tokens": 5,
"iterations": [
{
"type": "message",
"input_tokens": 0,
"output_tokens": 3,
"cache_creation_input_tokens": 10000,
"cache_read_input_tokens": 0,
"cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 10000},
},
{
"type": "message",
"input_tokens": 0,
"output_tokens": 2,
"cache_creation_input_tokens": 10000,
"cache_read_input_tokens": 0,
"cache_creation": {"ephemeral_5m_input_tokens": 0, "ephemeral_1h_input_tokens": 10000},
},
],
}
usage = config.calculate_usage(usage_object=usage_object, reasoning_content=None)
details = usage.prompt_tokens_details.cache_creation_token_details
assert details is not None
assert details.ephemeral_5m_input_tokens == 0
assert details.ephemeral_1h_input_tokens == 20000
assert usage.prompt_tokens_details.cache_creation_tokens == 20000
info = litellm.get_model_info(model="claude-opus-4-8", custom_llm_provider="anthropic")
rate_5m = info["cache_creation_input_token_cost"]
rate_1h = info["cache_creation_input_token_cost_above_1hr"]
assert rate_1h > rate_5m
prompt_cost, _ = cost_per_token(model="claude-opus-4-8", usage=usage)
assert prompt_cost == pytest.approx(20000 * rate_1h)
assert prompt_cost != pytest.approx(20000 * rate_5m)
def test_calculate_usage_clamps_text_tokens_when_reasoning_estimate_exceeds_output():
config = AnthropicConfig()