scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

Merge pull request #34957 from BerriAI/litellm_gpt56_cache_token_pricing

Browse source 
fix(cost): bill gpt-5.6 prompt cache reads at the cache read rate
This commit is contained in:
Mateo Wang 2026-08-05 02:30:41 -07:00 • committed by GitHub
commit 64f4bedde1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
9 changed files with 228 additions and 40 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
litellm/cost_calculator.py
View file

@ -878,6 +878,8 @@ def _get_usage_object(
return None
if isinstance(usage_obj, Usage):
return usage_obj
elif isinstance(usage_obj, dict) and litellm.AnthropicConfig.is_anthropic_usage_object(usage_obj):
return litellm.AnthropicConfig().calculate_usage(usage_object=usage_obj, reasoning_content=None)
elif (
usage_obj is not None
and (isinstance(usage_obj, dict) or isinstance(usage_obj, ResponseAPIUsage))
@ -1249,7 +1251,13 @@ def completion_cost(
else:
_usage = usage_obj
if ResponseAPILoggingUtils._is_response_api_usage(_usage):
if litellm.AnthropicConfig.is_anthropic_usage_object(_usage):
_usage = (
litellm.AnthropicConfig()
.calculate_usage(usage_object=_usage, reasoning_content=None)
.model_dump()
)
elif ResponseAPILoggingUtils._is_response_api_usage(_usage):
_usage = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(
_usage
).model_dump()

15
litellm/llms/anthropic/chat/transformation.py
View file

@ -2104,6 +2104,21 @@ class AnthropicConfig(AnthropicModelInfo, BaseConfig):
compaction_blocks,
)
@staticmethod
def is_anthropic_usage_object(usage_object: dict) -> bool:
"""Anthropic reports prompt cache tokens as top-level ``cache_read_input_tokens`` /
``cache_creation_input_tokens``; no other API surface uses those keys, and the
Responses API mapping would silently drop them.
Requiring a cache key is deliberate: Responses API usage also carries top-level
``input_tokens``, so the cache keys are the only shape discriminator between the
two. A cache-free Anthropic payload falls through to the Responses API mapping,
which is safe because both mappings agree whenever no cache tokens are present.
"""
if "prompt_tokens" in usage_object or "input_tokens" not in usage_object:
return False
return any(key in usage_object for key in ("cache_read_input_tokens", "cache_creation_input_tokens"))
def calculate_usage(
self,
usage_object: dict,

33
litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py
View file

@ -8,6 +8,9 @@ from typing import Any, Final
from litellm import verbose_logger
from litellm._uuid import uuid
from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicUsage
from .transformation import LiteLLMAnthropicToResponsesAPIAdapter
class AnthropicResponsesStreamWrapper:
@ -227,24 +230,17 @@ class AnthropicResponsesStreamWrapper:
event.get("response") if isinstance(event, dict) else None
)
stop_reason = "end_turn"
input_tokens = 0
output_tokens = 0
cache_creation_tokens = 0
cache_read_tokens = 0
anthropic_usage: AnthropicUsage = AnthropicUsage(input_tokens=0, output_tokens=0)
if response_obj is not None:
status: Final = getattr(response_obj, "status", None)
if status == "incomplete":
stop_reason = "max_tokens"
usage: Final = getattr(response_obj, "usage", None)
if usage is not None:
input_tokens = getattr(usage, "input_tokens", 0) or 0
output_tokens = getattr(usage, "output_tokens", 0) or 0
cache_creation_tokens = getattr(usage, "input_tokens_details", None) # type: ignore[assignment]
cache_read_tokens = getattr(usage, "output_tokens_details", None) # type: ignore[assignment]
# Prefer direct cache fields if present
cache_creation_tokens = int(getattr(usage, "cache_creation_input_tokens", 0) or 0)
cache_read_tokens = int(getattr(usage, "cache_read_input_tokens", 0) or 0)
anthropic_usage = (
LiteLLMAnthropicToResponsesAPIAdapter.translate_responses_api_usage_to_anthropic_usage(
getattr(response_obj, "usage", None)
)
)
# Check if tool_use was in the output to override stop_reason
if response_obj is not None:
@ -257,20 +253,11 @@ class AnthropicResponsesStreamWrapper:
stop_reason = "tool_use"
break
usage_delta: Final[dict[str, Any]] = {
"input_tokens": input_tokens,
"output_tokens": output_tokens,
}
if cache_creation_tokens:
usage_delta["cache_creation_input_tokens"] = cache_creation_tokens
if cache_read_tokens:
usage_delta["cache_read_input_tokens"] = cache_read_tokens
self._chunk_queue.append(
{
"type": "message_delta",
"delta": {"stop_reason": stop_reason, "stop_sequence": None},
"usage": usage_delta,
"usage": dict(anthropic_usage),
}
)
self._chunk_queue.append({"type": "message_stop"})

32
litellm/llms/anthropic/experimental_pass_through/responses_adapters/transformation.py
View file

@ -29,7 +29,7 @@ from litellm.types.llms.anthropic_messages.anthropic_response import (
AnthropicMessagesResponse,
AnthropicUsage,
)
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIResponse
class LiteLLMAnthropicToResponsesAPIAdapter:
@ -38,6 +38,24 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
converts Responses API responses back to Anthropic format.
"""
@staticmethod
def translate_responses_api_usage_to_anthropic_usage(
raw_usage: ResponseAPIUsage | None,
) -> AnthropicUsage:
"""Map Responses API usage onto Anthropic usage, where ``input_tokens``
excludes the cache-read and cache-write tokens reported alongside it.
"""
if raw_usage is None:
return AnthropicUsage(input_tokens=0, output_tokens=0)
from litellm.llms.anthropic.experimental_pass_through.adapters.transformation import (
LiteLLMAnthropicMessagesAdapter,
)
from litellm.responses.utils import ResponseAPILoggingUtils
chat_usage = ResponseAPILoggingUtils._transform_response_api_usage_to_chat_usage(raw_usage)
return LiteLLMAnthropicMessagesAdapter._translate_openai_usage_to_anthropic_usage(chat_usage)
# ------------------------------------------------------------------ #
# Request translation: Anthropic -> Responses API #
# ------------------------------------------------------------------ #
@ -386,8 +404,6 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
ResponseReasoningItem,
)
from litellm.types.llms.openai import ResponseAPIUsage
content: Final[list[dict[str, Any]]] = []
stop_reason: AnthropicFinishReason = "end_turn"
@ -453,15 +469,7 @@ class LiteLLMAnthropicToResponsesAPIAdapter:
if response.status == "incomplete":
stop_reason = "max_tokens"
# usage
raw_usage: Final[ResponseAPIUsage | None] = response.usage
input_tokens: Final = int(getattr(raw_usage, "input_tokens", 0) or 0)
output_tokens: Final = int(getattr(raw_usage, "output_tokens", 0) or 0)
anthropic_usage: Final = AnthropicUsage(
input_tokens=input_tokens,
output_tokens=output_tokens,
)
anthropic_usage: Final = self.translate_responses_api_usage_to_anthropic_usage(response.usage)
return AnthropicMessagesResponse(
id=response.id,

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

@ -2518,6 +2518,47 @@ def test_generic_cost_per_token_gemini_35_flash_lite():
assert completion_cost == pytest.approx(0.00125)
@pytest.mark.parametrize(
"service_tier,input_rate,cache_read_rate,cache_write_rate,output_rate",
[
("flex", 2.5e-6, 2.5e-7, 3.125e-6, 1.5e-5),
("priority", 1e-5, 1e-6, 1.25e-5, 6e-5),
],
)
def test_service_tier_cache_creation_rates_for_gpt_5_6(
_local_model_cost_map,
service_tier,
input_rate,
cache_read_rate,
cache_write_rate,
output_rate,
):
"""Regression: gpt-5.6 publishes cache_creation_input_token_cost_flex/_priority, so a
flex or priority request must bill cache writes at that tier's rate instead of falling
back to the standard 6.25e-6 rate."""
usage = Usage(
prompt_tokens=10_000,
completion_tokens=500,
total_tokens=10_500,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=6_000,
cache_write_tokens=3_000,
text_tokens=1_000,
),
)
prompt_cost, completion_cost = generic_cost_per_token(
model="gpt-5.6-sol",
usage=usage,
custom_llm_provider="openai",
service_tier=service_tier,
)
expected_prompt = 1_000 * input_rate + 6_000 * cache_read_rate + 3_000 * cache_write_rate
assert prompt_cost == pytest.approx(expected_prompt, rel=1e-9)
assert completion_cost == pytest.approx(500 * output_rate, rel=1e-9)
def test_fast_service_tier_bills_at_the_priority_rate(_local_model_cost_map):
"""Regression: OpenAI's Fast mode replaced Priority Processing and costs 2x standard.

40
tests/test_litellm/llms/anthropic/chat/test_anthropic_chat_transformation.py
View file

@ -23,7 +23,7 @@ from litellm.llms.anthropic.experimental_pass_through.messages.transformation im
AnthropicMessagesConfig,
)
from litellm.types.llms.anthropic import ANTHROPIC_BETA_HEADER_VALUES
from litellm.types.utils import ServerToolUse
from litellm.types.utils import ServerToolUse, Usage
def test_response_format_transformation_unit_test():
@ -5845,3 +5845,41 @@ def test_top_k_forwarded_at_transform_on_models_that_accept_it():
)
assert result["top_k"] == 40
def test_is_anthropic_usage_object_distinguishes_chat_usage():
"""Chat-shaped Usage mirrors cache_read_input_tokens alongside prompt_tokens that already
include the cache tokens, so treating it as Anthropic usage would re-add them and
double-count the prompt. Only the Anthropic shape, where input_tokens excludes cache
tokens, may take the Anthropic mapping."""
assert AnthropicConfig.is_anthropic_usage_object(
{"input_tokens": 3, "output_tokens": 5, "cache_read_input_tokens": 4014}
)
assert AnthropicConfig.is_anthropic_usage_object(
{"input_tokens": 3, "output_tokens": 5, "cache_creation_input_tokens": 10}
)
assert not AnthropicConfig.is_anthropic_usage_object(
Usage(
prompt_tokens=4017,
completion_tokens=5,
total_tokens=4022,
cache_read_input_tokens=4014,
).model_dump()
)
assert not AnthropicConfig.is_anthropic_usage_object({"input_tokens": 3, "output_tokens": 5})
def test_is_anthropic_usage_object_rejects_responses_api_usage():
"""completion_cost checks the Anthropic shape before the Responses API shape, so a
Responses API usage payload, whose cache reads live in nested input_tokens_details,
must never match; matching would route it past the converter that reads the nested
field and its cache reads would be billed at the full input rate."""
assert not AnthropicConfig.is_anthropic_usage_object(
{
"input_tokens": 4017,
"output_tokens": 5,
"total_tokens": 4022,
"input_tokens_details": {"cached_tokens": 4014},
"output_tokens_details": {"reasoning_tokens": 0},
}
)

29
tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_streaming_iterator.py
View file

@ -6,6 +6,7 @@ Tests for AnthropicResponsesStreamWrapper
import asyncio
import os
import sys
from types import SimpleNamespace
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "../../../../../..")))
@ -130,3 +131,31 @@ class TestProcessEventTextDeltaWithoutOutputItemAdded:
("content_block_start", 0),
("content_block_delta", 0),
]
class TestResponseCompletedUsage:
"""The Anthropic ``message_delta`` usage must report cache reads/writes and
exclude them from ``input_tokens``, so spend is not billed at the uncached
input rate."""
def test_response_completed_usage_carries_cache_tokens(self):
from litellm.types.llms.openai import ResponseAPIUsage
response = SimpleNamespace(
status="completed",
output=[],
usage=ResponseAPIUsage(
input_tokens=4017,
input_tokens_details={"cached_tokens": 4004, "cache_write_tokens": 10},
output_tokens=5,
total_tokens=4022,
),
)
chunks = _process_all([{"type": "response.completed", "response": response}])
message_delta = next(c for c in chunks if c["type"] == "message_delta")
assert message_delta["usage"] == {
"input_tokens": 3,
"output_tokens": 5,
"cache_creation_input_tokens": 10,
"cache_read_input_tokens": 4004,
}

41
tests/test_litellm/llms/anthropic/experimental_pass_through/responses_adapters/test_responses_adapters_transformation.py
View file

@ -20,6 +20,7 @@ from litellm.llms.anthropic.experimental_pass_through.responses_adapters.transfo
LiteLLMAnthropicToResponsesAPIAdapter,
)
from litellm.types.llms.anthropic import AnthropicMessagesRequest
from litellm.types.llms.openai import ResponseAPIUsage
def _make_request(**overrides) -> AnthropicMessagesRequest:
@ -823,11 +824,19 @@ def _make_mock_response(
model: str = "gpt-4o",
input_tokens: int = 100,
output_tokens: int = 50,
cached_tokens: int = 0,
cache_write_tokens: int = 0,
) -> MagicMock:
"""Build a minimal mock ResponsesAPIResponse."""
usage = MagicMock()
usage.input_tokens = input_tokens
usage.output_tokens = output_tokens
usage = ResponseAPIUsage(
input_tokens=input_tokens,
input_tokens_details={
"cached_tokens": cached_tokens,
"cache_write_tokens": cache_write_tokens,
},
output_tokens=output_tokens,
total_tokens=input_tokens + output_tokens,
)
resp = MagicMock()
resp.id = response_id
@ -961,6 +970,32 @@ class TestTranslateResponse:
assert result["usage"]["input_tokens"] == 200
assert result["usage"]["output_tokens"] == 75
def test_cache_tokens_mapped_to_anthropic_usage(self):
"""Cache reads/writes reported by the Responses API must survive the
Anthropic mapping, and input_tokens must exclude them so spend is not
billed at the uncached input rate."""
response = _make_mock_response(
output=[_make_output_message(["OK"])],
input_tokens=4017,
output_tokens=5,
cached_tokens=4004,
cache_write_tokens=10,
)
result: Any = _ADAPTER.translate_response(response)
assert result["usage"] == {
"input_tokens": 3,
"output_tokens": 5,
"cache_creation_input_tokens": 10,
"cache_read_input_tokens": 4004,
}
def test_missing_usage_maps_to_zero_tokens(self):
"""A response without a usage object must map to zeroed Anthropic usage."""
assert LiteLLMAnthropicToResponsesAPIAdapter.translate_responses_api_usage_to_anthropic_usage(None) == {
"input_tokens": 0,
"output_tokens": 0,
}
def test_model_and_id_preserved(self):
"""Model and response ID from the Responses API are forwarded."""
response = _make_mock_response(

27
tests/test_litellm/test_cost_calculator.py
View file

@ -3511,3 +3511,30 @@ def test_combine_usage_objects_sums_mirrored_cache_write_fields_once():
assert combined_pair.prompt_tokens_details is not None
assert combined_pair.prompt_tokens_details.cache_write_tokens == 100
assert combined_pair.prompt_tokens_details.cache_creation_tokens == 100
def test_completion_cost_prices_anthropic_shaped_cache_read_tokens():
"""Regression: an Anthropic /v1/messages response reports cache reads as top-level
cache_read_input_tokens with input_tokens excluding them. Reading that usage as
Responses API usage dropped the cache tokens and billed the whole prompt at the
uncached input rate, overstating spend on cache hits."""
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
response = {
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": "gpt-5.6-sol",
"stop_reason": "end_turn",
"content": [{"type": "text", "text": "1"}],
"usage": {"input_tokens": 3, "output_tokens": 5, "cache_read_input_tokens": 4014},
}
cost = litellm.completion_cost(
completion_response=response,
model="gpt-5.6-sol",
custom_llm_provider="openai",
)
assert cost == pytest.approx(3 * 5e-6 + 4014 * 5e-7 + 5 * 3e-5, rel=1e-9)