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 remote-tracking branch 'origin/litellm_internal_staging' into litellm_zero_ruff_lit_headroom

Browse source 
# Conflicts:
#	litellm/llms/anthropic/experimental_pass_through/responses_adapters/streaming_iterator.py
This commit is contained in:
mateo-berri 2026-08-05 02:53:06 -07:00
commit 20d297a151
17 changed files with 498 additions and 81 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()

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

@ -746,7 +746,7 @@ def generic_cost_per_token(
# Check for double-counting: sum of details > prompt_tokens means overlap
total_details: Final = text_tokens + cache_hit + audio_tokens + cache_creation + image_tokens + video_tokens
has_double_counting: Final = cache_hit > 0 and total_details > usage.prompt_tokens
has_double_counting: Final = (cache_hit > 0 or cache_creation > 0) and total_details > usage.prompt_tokens
if (text_tokens == 0 and prompt_tokens_details["image_count"] == 0) or has_double_counting:
text_tokens = usage.prompt_tokens - cache_hit - audio_tokens - cache_creation - image_tokens - video_tokens

71
litellm/litellm_core_utils/streaming_chunk_builder_utils.py
View file

@ -39,6 +39,34 @@ if TYPE_CHECKING:
)
def capture_cache_creation_token_details(
prompt_tokens_details: PromptTokensDetailsWrapper | None,
current: CacheCreationTokenDetails | None,
) -> CacheCreationTokenDetails | None:
incoming: Final = cast(
CacheCreationTokenDetails | None,
getattr(prompt_tokens_details, "cache_creation_token_details", None),
)
if incoming is not None:
return incoming
return current
def attach_cache_creation_token_details(
prompt_tokens_details: PromptTokensDetailsWrapper | None,
cache_creation_token_details: CacheCreationTokenDetails | None,
) -> PromptTokensDetailsWrapper | None:
if prompt_tokens_details is None or cache_creation_token_details is None:
return prompt_tokens_details
existing: Final = cast(
CacheCreationTokenDetails | None,
getattr(prompt_tokens_details, "cache_creation_token_details", None),
)
if existing is not None:
return prompt_tokens_details
return prompt_tokens_details.model_copy(update={"cache_creation_token_details": cache_creation_token_details})
class ChunkProcessor:
def __init__(self, chunks: list, messages: list | None = None):
self.chunks = self._sort_chunks(chunks)
@ -693,21 +721,22 @@ class ChunkProcessor:
"web_search_requests",
)
prompt_tokens_details = cast(
PromptTokensDetailsWrapper | None,
usage_chunk_dict["prompt_tokens_details"],
prompt_tokens_details = (
cast(
PromptTokensDetailsWrapper | None,
usage_chunk_dict["prompt_tokens_details"],
)
or prompt_tokens_details
)
cache_creation_token_details = self._capture_cache_creation_token_details(
cache_creation_token_details = capture_cache_creation_token_details(
prompt_tokens_details, cache_creation_token_details
)
if usage_chunk_dict["cost"] is not None:
cost = usage_chunk_dict["cost"]
prompt_tokens_details = self._attach_cache_creation_token_details(
prompt_tokens_details, cache_creation_token_details
)
prompt_tokens_details = attach_cache_creation_token_details(prompt_tokens_details, cache_creation_token_details)
completion_tokens = self._reset_anthropic_cursor_completion_tokens(
chunks=chunks,
@ -727,34 +756,6 @@ class ChunkProcessor:
cost=cost,
)
@staticmethod
def _capture_cache_creation_token_details(
prompt_tokens_details: PromptTokensDetailsWrapper | None,
current: CacheCreationTokenDetails | None,
) -> CacheCreationTokenDetails | None:
incoming: Final = cast(
CacheCreationTokenDetails | None,
getattr(prompt_tokens_details, "cache_creation_token_details", None),
)
if incoming is not None:
return incoming
return current
@staticmethod
def _attach_cache_creation_token_details(
prompt_tokens_details: PromptTokensDetailsWrapper | None,
cache_creation_token_details: CacheCreationTokenDetails | None,
) -> PromptTokensDetailsWrapper | None:
if prompt_tokens_details is None or cache_creation_token_details is None:
return prompt_tokens_details
existing: Final = cast(
CacheCreationTokenDetails | None,
getattr(prompt_tokens_details, "cache_creation_token_details", None),
)
if existing is not None:
return prompt_tokens_details
return prompt_tokens_details.model_copy(update={"cache_creation_token_details": cache_creation_token_details})
@staticmethod
def _reset_anthropic_cursor_completion_tokens(
chunks: list[dict[str, Any] | ModelResponse],

40
litellm/litellm_core_utils/streaming_handler.py
View file

@ -8,7 +8,7 @@ import time
import traceback
from collections.abc import AsyncIterator, Callable, Iterator
from dataclasses import dataclass
from typing import Any, Final, NoReturn, Union, cast
from typing import Any, Final, NoReturn, TypeVar, Union, cast
import anyio
import httpx
@ -25,10 +25,13 @@ from litellm.litellm_core_utils.thread_pool_executor import executor
from litellm.types.llms.openai import OpenAIChatCompletionChunk
from litellm.types.router import GenericLiteLLMParams
from litellm.types.utils import (
CacheCreationTokenDetails,
CompletionTokensDetailsWrapper,
Delta,
LlmProviders,
ModelResponse,
ModelResponseStream,
PromptTokensDetailsWrapper,
StreamingChoices,
Usage,
)
@ -2228,11 +2231,33 @@ class CustomStreamWrapper:
return chunk
_TokenDetails = TypeVar("_TokenDetails", PromptTokensDetailsWrapper, CompletionTokensDetailsWrapper)
def _coerce_token_details(
usage: dict | BaseModel, field: str, details_type: type[_TokenDetails]
) -> _TokenDetails | None:
raw = usage.get(field) if isinstance(usage, dict) else getattr(usage, field, None)
if raw is None:
return None
if isinstance(raw, details_type):
return raw
return details_type(**(raw if isinstance(raw, dict) else raw.model_dump()))
def calculate_total_usage(chunks: list[ModelResponse]) -> Usage:
"""Assume most recent usage chunk has total usage uptil then."""
from litellm.litellm_core_utils.streaming_chunk_builder_utils import (
attach_cache_creation_token_details,
capture_cache_creation_token_details,
)
prompt_tokens: int = 0
completion_tokens: int = 0
latest_usage_chunk = None
prompt_tokens_details: PromptTokensDetailsWrapper | None = None
completion_tokens_details: CompletionTokensDetailsWrapper | None = None
cache_creation_token_details: CacheCreationTokenDetails | None = None
for chunk in chunks:
if "usage" in chunk and chunk["usage"] is not None:
@ -2242,11 +2267,24 @@ def calculate_total_usage(chunks: list[ModelResponse]) -> Usage:
prompt_tokens = usage.get("prompt_tokens", 0) or 0
if "completion_tokens" in usage:
completion_tokens = usage.get("completion_tokens", 0) or 0
incoming_prompt_tokens_details = _coerce_token_details(
usage, "prompt_tokens_details", PromptTokensDetailsWrapper
)
cache_creation_token_details = capture_cache_creation_token_details(
incoming_prompt_tokens_details, cache_creation_token_details
)
prompt_tokens_details = incoming_prompt_tokens_details or prompt_tokens_details
completion_tokens_details = (
_coerce_token_details(usage, "completion_tokens_details", CompletionTokensDetailsWrapper)
or completion_tokens_details
)
returned_usage_chunk: Final = Usage(
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
prompt_tokens_details=attach_cache_creation_token_details(prompt_tokens_details, cache_creation_token_details),
completion_tokens_details=completion_tokens_details,
)
if latest_usage_chunk is not None:

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

@ -2102,6 +2102,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)
cache_read_tokens = getattr(usage, "output_tokens_details", None)
# 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,

6
litellm/responses/litellm_completion_transformation/transformation.py
View file

@ -2019,6 +2019,12 @@ class LiteLLMCompletionResponsesConfig:
if hasattr(prompt_details, "audio_tokens") and prompt_details.audio_tokens is not None:
input_details_dict["audio_tokens"] = prompt_details.audio_tokens
cache_write_tokens = getattr(prompt_details, "cache_write_tokens", None) or getattr(
prompt_details, "cache_creation_tokens", None
)
if cache_write_tokens is not None:
input_details_dict["cache_write_tokens"] = cache_write_tokens
if input_details_dict:
response_usage.input_tokens_details = InputTokensDetails(**input_details_dict)

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

@ -2233,6 +2233,32 @@ def test_generic_cost_per_token_openai_cache_write_tokens_gpt_5_6():
assert prompt_cost > 1000 * info["input_cost_per_token"]
def test_generic_cost_per_token_backs_out_cache_write_tokens_from_text_tokens():
"""
Regression for #34801: when a provider reports text_tokens covering the whole
prompt alongside cache-write tokens (and no cache reads), the cache-write tokens
must be backed out of the text total instead of being billed twice.
"""
os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True"
litellm.model_cost = litellm.get_model_cost_map(url="")
model = "gpt-5.6"
usage = Usage(
prompt_tokens=1000,
completion_tokens=10,
total_tokens=1010,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=0, cache_write_tokens=800, text_tokens=1000
),
)
prompt_cost, _ = generic_cost_per_token(model=model, usage=usage, custom_llm_provider="openai")
info = litellm.get_model_info(model=model, custom_llm_provider="openai")
expected_prompt = 200 * info["input_cost_per_token"] + 800 * info["cache_creation_input_token_cost"]
assert prompt_cost == pytest.approx(expected_prompt)
def test_token_type_cost_breakdown_reconciles_with_generic_total():
"""
Both-ways check: the reasoning subset must sum with the remaining (text) output
@ -2492,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.

43
tests/test_litellm/litellm_core_utils/test_streaming_chunk_builder_utils.py
View file

@ -994,6 +994,49 @@ def test_cost_field_in_usage_chunks():
assert usage.completion_tokens == 5
def test_prompt_tokens_details_survive_later_usage_chunk_without_details():
"""Regression for #34801: a trailing usage chunk that omits
`prompt_tokens_details` must not wipe the OpenAI cache-read/cache-write split,
otherwise those tokens get re-priced at the uncached input rate."""
from litellm.types.utils import PromptTokensDetailsWrapper
chunk_with_details = ModelResponseStream(
id="chatcmpl-1",
created=1745513206,
model="openai/gpt-5.6-sol",
choices=[
StreamingChoices(finish_reason=None, index=0, delta=Delta(content="Hi"))
],
usage=Usage(
prompt_tokens=6017,
completion_tokens=4,
total_tokens=6021,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=6004, cache_write_tokens=10
),
),
)
chunk_without_details = ModelResponseStream(
id="chatcmpl-1",
created=1745513207,
model="openai/gpt-5.6-sol",
choices=[
StreamingChoices(finish_reason="stop", index=0, delta=Delta(content=""))
],
usage=Usage(prompt_tokens=6017, completion_tokens=4, total_tokens=6021),
)
chunks = [chunk_with_details, chunk_without_details]
usage = ChunkProcessor(chunks=chunks).calculate_usage(
chunks=chunks, model="openai/gpt-5.6-sol", completion_output="Hi"
)
assert usage.prompt_tokens == 6017
assert usage.prompt_tokens_details is not None
assert usage.prompt_tokens_details.cached_tokens == 6004
assert usage.prompt_tokens_details.cache_write_tokens == 10
def test_get_combined_tool_content_custom_tool_call():
from litellm.litellm_core_utils.streaming_chunk_builder_utils import ChunkProcessor
from litellm.types.utils import ChatCompletionMessageCustomToolCall

97
tests/test_litellm/litellm_core_utils/test_streaming_handler.py
View file

@ -1449,6 +1449,103 @@ def test_calculate_total_usage_with_dict_usage_cost():
assert getattr(usage, "cost", None) == 0.00025
def test_calculate_total_usage_preserves_prompt_cache_token_details():
"""Regression for #34801: dropping `prompt_tokens_details` here re-prices OpenAI
cache-read tokens at the uncached input rate, overstating spend."""
from litellm.litellm_core_utils.streaming_handler import calculate_total_usage
usage_with_details = Usage(
prompt_tokens=6017,
completion_tokens=4,
total_tokens=6021,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=6004, cache_write_tokens=10
),
completion_tokens_details=CompletionTokensDetailsWrapper(reasoning_tokens=2),
)
chunk_with_details = ModelResponseStream(
id="chatcmpl-1",
created=1745513206,
model="openai/gpt-5.6-sol",
choices=[
StreamingChoices(finish_reason=None, index=0, delta=Delta(content="Hi"))
],
usage=usage_with_details,
)
chunk_without_details = ModelResponseStream(
id="chatcmpl-1",
created=1745513207,
model="openai/gpt-5.6-sol",
choices=[
StreamingChoices(finish_reason="stop", index=0, delta=Delta(content=""))
],
usage=Usage(prompt_tokens=6017, completion_tokens=4, total_tokens=6021),
)
usage = calculate_total_usage([chunk_with_details, chunk_without_details])
assert usage.prompt_tokens == 6017
assert usage.prompt_tokens_details is not None
assert usage.prompt_tokens_details.cached_tokens == 6004
assert usage.prompt_tokens_details.cache_write_tokens == 10
assert usage.completion_tokens_details is not None
assert usage.completion_tokens_details.reasoning_tokens == 2
def test_calculate_total_usage_preserves_anthropic_cache_creation_ttl_breakdown():
"""Anthropic sends the 5m/1h cache-write split only on `message_start`; the later
`message_delta` repeats the flat count without the split. Losing it here bills 1h
cache writes at the cheaper 5m rate."""
from litellm.litellm_core_utils.streaming_handler import calculate_total_usage
from litellm.types.utils import CacheCreationTokenDetails
message_start_chunk = ModelResponseStream(
id="chatcmpl-1",
created=1745513206,
model="claude-sonnet-5",
choices=[
StreamingChoices(finish_reason=None, index=0, delta=Delta(content="Hi"))
],
usage=Usage(
prompt_tokens=120,
completion_tokens=1,
total_tokens=121,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=0,
cache_creation_tokens=100,
cache_creation_token_details=CacheCreationTokenDetails(
ephemeral_5m_input_tokens=20, ephemeral_1h_input_tokens=80
),
),
),
)
message_delta_chunk = ModelResponseStream(
id="chatcmpl-1",
created=1745513207,
model="claude-sonnet-5",
choices=[
StreamingChoices(finish_reason="stop", index=0, delta=Delta(content=""))
],
usage=Usage(
prompt_tokens=120,
completion_tokens=4,
total_tokens=124,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=0, cache_creation_tokens=100
),
),
)
usage = calculate_total_usage([message_start_chunk, message_delta_chunk])
assert usage.prompt_tokens_details is not None
assert usage.prompt_tokens_details.cache_creation_tokens == 100
ttl_breakdown = usage.prompt_tokens_details.cache_creation_token_details
assert ttl_breakdown is not None
assert ttl_breakdown.ephemeral_5m_input_tokens == 20
assert ttl_breakdown.ephemeral_1h_input_tokens == 80
@pytest.mark.asyncio
async def test_openrouter_streaming_cost_after_finish_reason(logging_obj: Logging):
from litellm.utils import ModelResponseListIterator

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(

21
tests/test_litellm/responses/litellm_completion_transformation/test_litellm_completion_responses.py
View file

@ -1772,6 +1772,27 @@ class TestUsageTransformation:
assert response_usage.input_tokens_details.cached_tokens == 3
assert response_usage.input_tokens_details.text_tokens == 6
def test_transform_usage_preserves_cache_write_tokens(self):
"""Regression for #34801: the chat-completions to Responses bridge dropped
cache-write tokens, so cache-creation billing disappeared on that route."""
usage = Usage(
prompt_tokens=1000,
completion_tokens=10,
total_tokens=1010,
prompt_tokens_details=PromptTokensDetailsWrapper(
cached_tokens=100,
cache_write_tokens=800,
),
)
response_usage = LiteLLMCompletionResponsesConfig._transform_chat_completion_usage_to_responses_usage(
chat_completion_response=usage
)
assert response_usage.input_tokens_details is not None
assert response_usage.input_tokens_details.cached_tokens == 100
assert getattr(response_usage.input_tokens_details, "cache_write_tokens", None) == 800
def test_transform_usage_with_reasoning_tokens_gemini(self):
"""Test that reasoning_tokens from Gemini are properly transformed to output_tokens_details"""
# Setup: Simulate Gemini usage with thoughtsTokenCount

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)

5
ui/litellm-dashboard/src/app/(dashboard)/cost-optimization/_components/CacheLeakageCard.tsx
View file

@ -114,10 +114,7 @@ const CacheLeakageCard: React.FC<CacheLeakageCardProps> = ({ activity }) => {
<AdvancedDatePicker value={dateValue} onValueChange={onDateChange} />
</div>
</div>
<Tabs
value={dimension}
onValueChange={(value) => setDimension(value === "model" ? "model" : "key")}
>
<Tabs value={dimension} onValueChange={(value) => setDimension(value === "model" ? "model" : "key")}>
<TabsList>
<TabsTrigger value="key">By virtual key</TabsTrigger>
<TabsTrigger value="model">By model</TabsTrigger>