chore(cost-map): remove models past their deprecation date (#42435)

* chore(cost-map): remove models past their deprecation date

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(cost-calc): drop the empty parametrize left behind by the gemini web search removal

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* fix(cost-map): drop merge base block left by conflict resolution

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

* test(cost-calc): drop gemini image cost tests pinned on removed model

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

---------

Co-authored-by: kerry <kerry@berri.ai>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-09-22 21:19:26 +00:00 committed by GitHub
parent 989d7b87b2
commit 075536eca1
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
37 changed files with 18 additions and 14155 deletions

File diff suppressed because it is too large Load diff

File diff suppressed because it is too large Load diff

View file

@ -1664,7 +1664,7 @@ class TestEnableAnthropicPromptCaching:
points = self._points(model="us.anthropic.claude-sonnet-4-5-20250929-v1:0", provider="bedrock")
assert [p["index"] for p in points] == [None, -1]
@pytest.mark.parametrize("model, provider", [("gpt-4o", "openai"), ("gemini-2.0-flash", "gemini")])
@pytest.mark.parametrize("model, provider", [("gpt-4o", "openai")])
def test_non_anthropic_providers_never_injected(self, monkeypatch, model, provider):
"""These report supports_prompt_caching=True but never consume cache_control markers."""
from litellm.utils import supports_prompt_caching

View file

@ -299,42 +299,6 @@ def test_reasoning_tokens_gemini(_local_model_cost_map):
)
def test_reasoning_tokens_gemini_3_1_flash_lite(_local_model_cost_map):
"""Test cost calculation for gemini-3.1-flash-lite-preview with reasoning tokens"""
model = "gemini-3.1-flash-lite-preview"
custom_llm_provider = "gemini"
usage = Usage(
completion_tokens=1000,
prompt_tokens=500,
total_tokens=1500,
completion_tokens_details=CompletionTokensDetailsWrapper(
accepted_prediction_tokens=None,
audio_tokens=None,
reasoning_tokens=400,
rejected_prediction_tokens=None,
text_tokens=600,
),
prompt_tokens_details=PromptTokensDetailsWrapper(
audio_tokens=None, cached_tokens=None, text_tokens=500, image_tokens=None
),
)
model_cost_map = litellm.model_cost[model]
prompt_cost, completion_cost = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider=custom_llm_provider,
)
assert round(prompt_cost, 10) == round(
model_cost_map["input_cost_per_token"] * usage.prompt_tokens,
10,
)
assert round(completion_cost, 10) == round(
(model_cost_map["output_cost_per_token"] * usage.completion_tokens_details.text_tokens)
+ (model_cost_map["output_cost_per_reasoning_token"] * usage.completion_tokens_details.reasoning_tokens),
10,
)
def test_image_tokens_with_custom_pricing():
@ -2221,65 +2185,8 @@ def test_vertex_image_generation_cost_falls_back_to_flat_image_pricing(_local_mo
assert round(cost, 10) == round(expected_cost, 10)
def test_gemini_image_generation_cost_prefers_token_usage_metadata(_local_model_cost_map):
"""
When usage metadata exists on image responses, Gemini image generation cost
should be calculated from token pricing, not flat output_cost_per_image.
"""
model = "gemini/gemini-3-pro-image-preview"
model_info = litellm.get_model_info(model=model, custom_llm_provider="gemini")
input_text_tokens = 20
input_image_tokens = 1120
output_image_tokens = 1120
prompt_tokens = input_text_tokens + input_image_tokens
image_response = ImageResponse(
data=[ImageObject(b64_json="img1"), ImageObject(b64_json="img2")],
usage=ImageUsage(
input_tokens=prompt_tokens,
input_tokens_details=ImageUsageInputTokensDetails(
text_tokens=input_text_tokens,
image_tokens=input_image_tokens,
),
output_tokens=output_image_tokens,
total_tokens=prompt_tokens + output_image_tokens,
),
)
cost = gemini_image_generation_cost_calculator(
model=model,
image_response=image_response,
)
expected_prompt_cost = prompt_tokens * model_info["input_cost_per_token"]
expected_completion_cost = output_image_tokens * model_info["output_cost_per_image_token"]
expected_total_cost = expected_prompt_cost + expected_completion_cost
assert round(cost, 10) == round(expected_total_cost, 10)
# Ensure this is not falling back to flat per-image pricing.
assert cost != len(image_response.data) * model_info["output_cost_per_image"]
def test_gemini_image_generation_cost_falls_back_to_flat_image_pricing(_local_model_cost_map):
"""
Without usage metadata, Gemini image generation cost should fall back to
output_cost_per_image * number_of_images.
"""
model = "gemini/gemini-3-pro-image-preview"
model_info = litellm.get_model_info(model=model, custom_llm_provider="gemini")
image_response = ImageResponse(data=[ImageObject(b64_json="img1"), ImageObject(b64_json="img2")])
cost = gemini_image_generation_cost_calculator(
model=model,
image_response=image_response,
)
expected_cost = len(image_response.data) * model_info["output_cost_per_image"]
assert round(cost, 10) == round(expected_cost, 10)
def test_query_count_is_free_without_a_per_query_price(_local_model_cost_map):
@ -2460,23 +2367,6 @@ def test_vertex_global_or_absent_location_no_uplift(vertex_location, _local_mode
assert base == located
@pytest.mark.parametrize("model", ["claude-opus-4-1", "gemini-2.0-flash-001"])
def test_vertex_location_no_uplift_for_uniformly_priced_model(model, _local_model_cost_map):
"""Models Google prices uniformly across endpoints (Gemini 2.x, Claude Opus 4.1
and older) carry no multiplier and must not move with the location."""
from litellm.types.utils import Usage
usage = Usage(prompt_tokens=1000, completion_tokens=500, total_tokens=1500)
base = generic_cost_per_token(model=model, usage=usage, custom_llm_provider="vertex_ai")
regional = generic_cost_per_token(
model=model,
usage=usage,
custom_llm_provider="vertex_ai",
vertex_location="us-east5",
)
assert base == regional, f"{model} should not have a regional-endpoint uplift"
def test_vertex_uplift_invalid_multiplier_defaults_to_one():
@ -3695,47 +3585,6 @@ def test_route_image_generation_cost_openai_honors_deployment_input_cost_per_ima
assert cost == pytest.approx(0.07)
def test_route_image_generation_cost_gemini_adds_grounding_to_deployment_image_price(
_local_model_cost_map: None,
) -> None:
usage = ImageUsage(
input_tokens=0,
input_tokens_details=ImageUsageInputTokensDetails(image_tokens=0, text_tokens=0),
output_tokens=0,
total_tokens=0,
web_search_requests=3,
)
cost = CostCalculatorUtils.route_image_generation_cost_calculator(
model="gemini/gemini-3.1-flash-image-preview",
completion_response=_image_response(usage=usage),
custom_llm_provider="gemini",
call_type="image_generation",
model_info={"output_cost_per_image": 0.1},
)
assert cost == pytest.approx(0.1 + 3 * 0.014)
def test_route_image_generation_cost_gemini_bills_tokens_when_no_image_returned(
_local_model_cost_map: None,
) -> None:
usage = ImageUsage(
input_tokens=10,
input_tokens_details=ImageUsageInputTokensDetails(image_tokens=0, text_tokens=10),
output_tokens=1290,
total_tokens=1300,
)
cost = CostCalculatorUtils.route_image_generation_cost_calculator(
model="gemini/gemini-3.1-flash-image-preview",
completion_response=ImageResponse(data=[], usage=usage),
custom_llm_provider="gemini",
call_type="image_generation",
model_info={"output_cost_per_image": 0.08},
)
assert cost == pytest.approx(10 * 5e-07 + 1290 * 6e-05)
@pytest.mark.parametrize(

View file

@ -109,26 +109,6 @@ def test_get_cost_for_built_in_tools_file_search():
assert cost == 0.00
def test_get_cost_for_anthropic_web_search():
"""
Test that Anthropic web search cost is tracked when usage.server_tool_use.web_search_requests
is set. Use claude-3-7-sonnet-20250219 (has search_context_cost_per_query) and
custom_llm_provider=anthropic so get_cost_for_anthropic_web_search is invoked.
"""
from litellm.types.utils import ServerToolUse, Usage
model = "claude-3-7-sonnet-20250219"
usage = Usage(server_tool_use=ServerToolUse(web_search_requests=1))
cost = StandardBuiltInToolCostTracking.get_cost_for_built_in_tools(
model=model,
usage=usage,
response_object=None,
standard_built_in_tools_params=None,
custom_llm_provider="anthropic",
)
assert cost > 0.0
def test_get_cost_for_anthropic_web_search_with_server_tool_use_dict():
"""
Anthropic-compatible passthrough responses can construct Usage from a raw
@ -145,88 +125,6 @@ def test_get_cost_for_anthropic_web_search_with_server_tool_use_dict():
)
def test_anthropic_web_search_cost_from_raw_response_dict_when_usage_drops_server_tool_use():
"""
Regression: on the Anthropic /v1/messages sync cost path the response is the raw
Anthropic dict while the reconstructed OpenAI-shape Usage drops server_tool_use.
The web-search fee must still be charged by reading the count off the raw dict,
and the passed-in Usage must not be mutated.
"""
from litellm.types.utils import Usage
model = "claude-3-7-sonnet-20250219"
web_search_requests = 3
raw_response = {
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": model,
"content": [{"type": "text", "text": "hi"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {
"input_tokens": 100,
"output_tokens": 50,
"server_tool_use": {"web_search_requests": web_search_requests},
},
}
usage = Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150)
assert getattr(usage, "server_tool_use", None) is None
cost = StandardBuiltInToolCostTracking.get_cost_for_built_in_tools(
model=model,
usage=usage,
response_object=raw_response,
custom_llm_provider="anthropic",
standard_built_in_tools_params=None,
)
per_query_cost = litellm.get_model_info(model)["search_context_cost_per_query"][
"search_context_size_medium"
]
assert cost == per_query_cost * web_search_requests
assert cost > 0.0
assert getattr(usage, "server_tool_use", None) is None
def test_anthropic_web_search_cost_from_raw_response_dict_when_usage_is_none():
"""
Regression: when a caller hands the cost tracker a raw Anthropic dict without a
parallel Usage object, the web-search fee must still be priced per request from
usage.server_tool_use.web_search_requests on the dict instead of falling back to
the flat search_context_size_medium tier.
"""
model = "claude-3-7-sonnet-20250219"
web_search_requests = 4
raw_response = {
"id": "msg_1",
"type": "message",
"role": "assistant",
"model": model,
"content": [{"type": "text", "text": "hi"}],
"stop_reason": "end_turn",
"stop_sequence": None,
"usage": {
"input_tokens": 100,
"output_tokens": 50,
"server_tool_use": {"web_search_requests": web_search_requests},
},
}
cost = StandardBuiltInToolCostTracking.get_cost_for_built_in_tools(
model=model,
usage=None,
response_object=raw_response,
custom_llm_provider="anthropic",
standard_built_in_tools_params=None,
)
per_query_cost = litellm.get_model_info(model)["search_context_cost_per_query"][
"search_context_size_medium"
]
assert cost == per_query_cost * web_search_requests
def test_anthropic_web_search_zero_requests_from_raw_response_charges_zero():
"""
Regression: a raw Anthropic dict reporting zero web search requests must price
@ -287,27 +185,6 @@ def test_anthropic_response_usage_block_preserves_server_tool_use():
assert dumped_usage["server_tool_use"] == {"web_search_requests": 2}
@pytest.mark.parametrize(
"model", ["gemini/gemini-2.0-flash-001", "gemini-2.0-flash-001"]
)
def test_get_cost_for_gemini_web_search(model):
"""
Test that the cost for a web search is 0.00 when no response object is provided
"""
from litellm.types.utils import PromptTokensDetailsWrapper, Usage
usage = Usage(
prompt_tokens_details=PromptTokensDetailsWrapper(web_search_requests=1)
)
cost = StandardBuiltInToolCostTracking.get_cost_for_built_in_tools(
model=model,
usage=usage,
response_object=None,
standard_built_in_tools_params=None,
)
assert cost > 0.0
def test_completion_cost_includes_web_search_without_standard_built_in_tools_params():
"""
Test that completion_cost includes web search cost even when

View file

@ -979,10 +979,6 @@ def test_shipped_tool_search_rule_fills_mapped_claude_entries_without_flag(shipp
assert "supports_tool_search" not in litellm.model_cost[key]
assert litellm.get_model_info(model, custom_llm_provider=provider)["supports_tool_search"] is True
assert "supports_tool_search" not in litellm.model_cost["claude-opus-4-1"]
opus_4_1_info = litellm.get_model_info("claude-opus-4-1", custom_llm_provider="anthropic")
assert opus_4_1_info.get("supports_tool_search") is None
assert "supports_tool_search" not in litellm.model_cost["azure_ai/claude-opus-5"]
azure_opus_5_info = litellm.get_model_info("claude-opus-5", custom_llm_provider="azure_ai")
assert azure_opus_5_info.get("supports_tool_search") is None

View file

@ -218,7 +218,6 @@ def test_shipped_backup_marks_claude_4_6_plus_adaptive_not_4_0():
assert backup[adaptive]["supports_adaptive_thinking"] is True, adaptive
for non_adaptive in [
"claude-opus-4-20250514",
"us.anthropic.claude-opus-4-20250514-v1:0",
"claude-opus-4-5",
]:

View file

@ -8389,21 +8389,6 @@ def test_get_assembled_streaming_response_bills_a_provider_reported_usage_cost()
assert logging_obj._response_cost_calculator(result=assembled) == 0.0042
def test_get_assembled_streaming_response_without_usage_cost_leaves_pricing_to_the_price_map():
logging_obj = _responses_stream_logging_obj()
now = datetime.datetime.now()
assembled = logging_obj._get_assembled_streaming_response(
result=_completed_responses_event(ResponseAPIUsage(input_tokens=12, output_tokens=2, total_tokens=14)),
start_time=now,
end_time=now,
is_async=True,
streaming_chunks=[],
)
assert "additional_headers" not in assembled._hidden_params
price_map_cost = logging_obj._response_cost_calculator(result=assembled)
assert price_map_cost is not None and 0 < price_map_cost != 0.0042
def test_response_cost_calculator_prices_terminal_responses_event_from_its_response():

View file

@ -99,29 +99,3 @@ def test_stream_chunk_builder_coerces_server_tool_use_to_pydantic():
assert server_tool_use.web_search_requests == 3
def test_completion_cost_does_not_raise_on_streaming_web_search_response():
"""
Regression: completion_cost(...) must not raise AttributeError when the
response was reconstructed by stream_chunk_builder from a streaming
Anthropic web_search call.
"""
chunks = [
_make_text_chunk("hello"),
_make_finish_chunk_with_usage_dict_server_tool_use(),
]
rebuilt = stream_chunk_builder(chunks)
assert rebuilt is not None
# The exact dollar amount depends on the model-pricing table; what matters
# for this regression is that it does NOT raise AttributeError on
# `dict has no attribute 'web_search_requests'`.
try:
cost = completion_cost(completion_response=rebuilt)
except AttributeError as e: # pragma: no cover - regression guard
pytest.fail(
"completion_cost raised AttributeError after stream_chunk_builder "
f"(issue #26153 regression): {e}"
)
assert isinstance(cost, (int, float))

View file

@ -633,9 +633,6 @@ def test_openai_token_with_image_and_text():
"model, base_model, input_tokens, user_max_tokens, expected_value",
[
("random-model", "random-model", 1024, 1024, 1024),
("command", "command", 1000000, None, None), # model max = 4096
("command", "command", 4000, 256, 96), # model max = 4096
("command", "command", 4000, 10, 10), # model max = 4096
("gpt-3.5-turbo", "gpt-3.5-turbo", 4000, 5000, 4096), # model max output = 4096
],
)

View file

@ -5580,43 +5580,6 @@ def test_tool_config_cachepoint_not_placed_or_credited_for_model_without_prompt_
assert "litellm_gateway_injected_cache" not in bucket
def test_translate_response_format_json_schema_still_injects_tool():
"""
response_format with an explicit json_schema should still use the
synthetic tool call approach (for models that don't support native
structured outputs).
"""
config = AmazonConverseConfig()
response_format = {
"type": "json_schema",
"json_schema": {
"name": "FactResult",
"schema": {
"type": "object",
"properties": {
"facts": {
"type": "array",
"items": {"type": "string"},
},
},
"required": ["facts"],
},
},
}
optional_params: dict = {}
result = config._translate_response_format_param(
value=response_format,
model="anthropic.claude-3-haiku-20240307-v1:0",
optional_params=optional_params,
non_default_params={"response_format": response_format},
is_thinking_enabled=False,
)
assert result["json_mode"] is True
assert "tools" in result
assert "tool_choice" in result
def test_transform_response_finish_reason_stop_when_json_mode_filters_all_tools():

View file

@ -153,14 +153,6 @@ def test_uncached_request_bills_every_prompt_token_at_the_input_rate(local_model
assert completion_cost == pytest.approx(200 * info["output_cost_per_token"])
def test_legacy_endpoint_names_still_resolve(local_model_cost_map: None) -> None:
info: Final = _model_info("databricks/databricks-mixtral-8x7b-instruct")
usage: Final = Usage(prompt_tokens=100, completion_tokens=100, total_tokens=200)
prompt_cost, completion_cost = cost_per_token(model="databricks/mixtral-8x7b-instruct-v0.1", usage=usage)
assert prompt_cost == pytest.approx(100 * info["input_cost_per_token"])
assert completion_cost == pytest.approx(100 * info["output_cost_per_token"])
@pytest.mark.parametrize("model", NEW_MODELS)

View file

@ -200,172 +200,12 @@ def test_maps_no_usage_details():
assert cost_per_google_maps_grounding_request(usage=usage, model_info=model_info) == 0.0
def test_gemini_image_edit_cost_prefers_token_usage_metadata(monkeypatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
litellm.model_cost = litellm.get_model_cost_map(url="")
model = "gemini/gemini-3-pro-image-preview"
model_info = litellm.get_model_info(model=model, custom_llm_provider="gemini")
input_text_tokens = 20
input_image_tokens = 1120
output_image_tokens = 1120
prompt_tokens = input_text_tokens + input_image_tokens
image_response = ImageResponse(
data=[ImageObject(b64_json="img1"), ImageObject(b64_json="img2")],
usage=ImageUsage(
input_tokens=prompt_tokens,
input_tokens_details=ImageUsageInputTokensDetails(
text_tokens=input_text_tokens,
image_tokens=input_image_tokens,
),
output_tokens=output_image_tokens,
total_tokens=prompt_tokens + output_image_tokens,
),
)
cost = gemini_image_edit_cost_calculator(
model=model,
image_response=image_response,
)
expected_cost = (
prompt_tokens * model_info["input_cost_per_token"]
+ output_image_tokens * model_info["output_cost_per_image_token"]
)
flat_image_cost = (
len(image_response.data or []) * model_info["output_cost_per_image"]
)
assert round(cost, 10) == round(expected_cost, 10)
assert cost != flat_image_cost
def test_gemini_image_edit_cost_uses_output_token_details(monkeypatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
litellm.model_cost = litellm.get_model_cost_map(url="")
model = "gemini/gemini-3-pro-image-preview"
model_info = litellm.get_model_info(model=model, custom_llm_provider="gemini")
input_text_tokens = 20
output_text_tokens = 213
output_image_tokens = 1120
output_tokens = output_text_tokens + output_image_tokens
image_response = ImageResponse(
data=[ImageObject(b64_json="img1")],
usage=ImageUsage(
input_tokens=input_text_tokens,
input_tokens_details=ImageUsageInputTokensDetails(
text_tokens=input_text_tokens,
image_tokens=0,
),
output_tokens=output_tokens,
total_tokens=input_text_tokens + output_tokens,
prompt_tokens=input_text_tokens,
completion_tokens=output_tokens,
prompt_tokens_details={
"text_tokens": input_text_tokens,
"image_tokens": 0,
},
completion_tokens_details={
"text_tokens": output_text_tokens,
"image_tokens": output_image_tokens,
},
output_tokens_details={
"text_tokens": output_text_tokens,
"image_tokens": output_image_tokens,
},
),
)
cost = gemini_image_edit_cost_calculator(
model=model,
image_response=image_response,
)
expected_cost = (
input_text_tokens * model_info["input_cost_per_token"]
+ output_text_tokens * model_info["output_cost_per_token"]
+ output_image_tokens * model_info["output_cost_per_image_token"]
)
all_output_as_image_cost = (
input_text_tokens * model_info["input_cost_per_token"]
+ (output_text_tokens + output_image_tokens)
* model_info["output_cost_per_image_token"]
)
assert round(cost, 10) == round(expected_cost, 10)
assert cost != all_output_as_image_cost
def test_gemini_image_generation_cost_uses_output_token_details(monkeypatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
litellm.model_cost = litellm.get_model_cost_map(url="")
model = "gemini/gemini-3-pro-image-preview"
model_info = litellm.get_model_info(model=model, custom_llm_provider="gemini")
input_text_tokens = 20
output_text_tokens = 213
output_image_tokens = 1120
output_tokens = output_text_tokens + output_image_tokens
image_response = ImageResponse(
data=[ImageObject(b64_json="img1")],
usage=ImageUsage(
input_tokens=input_text_tokens,
input_tokens_details=ImageUsageInputTokensDetails(
text_tokens=input_text_tokens,
image_tokens=0,
),
output_tokens=output_tokens,
total_tokens=input_text_tokens + output_tokens,
prompt_tokens=input_text_tokens,
completion_tokens=output_tokens,
prompt_tokens_details={
"text_tokens": input_text_tokens,
"image_tokens": 0,
},
completion_tokens_details={
"text_tokens": output_text_tokens,
"image_tokens": output_image_tokens,
},
output_tokens_details={
"text_tokens": output_text_tokens,
"image_tokens": output_image_tokens,
},
),
)
cost = gemini_image_generation_cost_calculator(
model=model,
image_response=image_response,
)
expected_cost = (
input_text_tokens * model_info["input_cost_per_token"]
+ output_text_tokens * model_info["output_cost_per_token"]
+ output_image_tokens * model_info["output_cost_per_image_token"]
)
all_output_as_image_cost = (
input_text_tokens * model_info["input_cost_per_token"]
+ (output_text_tokens + output_image_tokens)
* model_info["output_cost_per_image_token"]
)
assert round(cost, 10) == round(expected_cost, 10)
assert cost != all_output_as_image_cost
def test_gemini_image_edit_cost_falls_back_to_flat_image_pricing(monkeypatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
litellm.model_cost = litellm.get_model_cost_map(url="")
model = "gemini/gemini-3-pro-image-preview"
model_info = litellm.get_model_info(model=model, custom_llm_provider="gemini")
image_response = ImageResponse(
data=[ImageObject(b64_json="img1"), ImageObject(b64_json="img2")]
)
cost = gemini_image_edit_cost_calculator(
model=model,
image_response=image_response,
)
assert cost == len(image_response.data or []) * model_info["output_cost_per_image"]
def _image_response_with_web_search(web_search_requests):
@ -383,43 +223,8 @@ def _image_response_with_web_search(web_search_requests):
return ImageResponse(data=[ImageObject(b64_json="img1")], usage=usage)
def test_gemini_image_generation_cost_adds_web_search_grounding(monkeypatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
litellm.model_cost = litellm.get_model_cost_map(url="")
model = "gemini/gemini-3-pro-image-preview"
model_info = litellm.get_model_info(model=model, custom_llm_provider="gemini")
grounded = gemini_image_generation_cost_calculator(
model=model,
image_response=_image_response_with_web_search(2),
)
ungrounded = gemini_image_generation_cost_calculator(
model=model,
image_response=_image_response_with_web_search(None),
)
expected_web_search_cost = cost_per_web_search_request(
usage=_make_usage(2), model_info=model_info
)
assert expected_web_search_cost > 0
assert round(grounded - ungrounded, 10) == round(expected_web_search_cost, 10)
def test_gemini_image_generation_cost_no_web_search_when_absent(monkeypatch):
monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True")
litellm.model_cost = litellm.get_model_cost_map(url="")
model = "gemini/gemini-3-pro-image-preview"
cost_zero = gemini_image_generation_cost_calculator(
model=model,
image_response=_image_response_with_web_search(0),
)
cost_none = gemini_image_generation_cost_calculator(
model=model,
image_response=_image_response_with_web_search(None),
)
assert cost_zero == cost_none
@pytest.mark.parametrize(

View file

@ -371,62 +371,8 @@ def test_x_initiator_header_system_only_messages():
assert headers["X-Initiator"] == "user"
def test_get_supported_openai_params_claude_model():
"""Test that Claude models with extended thinking support have thinking and reasoning parameters."""
config = GithubCopilotConfig()
# Test Claude 4 model supports thinking and reasoning_effort parameters
supported_params = config.get_supported_openai_params("claude-sonnet-4-20250514")
assert "thinking" in supported_params
assert "reasoning_effort" in supported_params
# Test Claude 3-7 model supports thinking and reasoning_effort parameters
supported_params_claude37 = config.get_supported_openai_params(
"claude-3-7-sonnet-20250219"
)
assert "thinking" in supported_params_claude37
assert "reasoning_effort" in supported_params_claude37
# Test Claude 3.5 model does NOT support thinking parameters (no extended thinking)
supported_params_claude35 = config.get_supported_openai_params("claude-3.5-sonnet")
assert "thinking" not in supported_params_claude35
assert "reasoning_effort" not in supported_params_claude35
# Test non-Claude model doesn't include thinking parameters but may include reasoning_effort
supported_params_gpt = config.get_supported_openai_params("gpt-4o")
assert "thinking" not in supported_params_gpt
# gpt-4o should NOT have reasoning_effort (not a reasoning model)
assert "reasoning_effort" not in supported_params_gpt
# Test O-series reasoning models include reasoning_effort but not thinking
supported_params_o3 = config.get_supported_openai_params("o3-mini")
assert "thinking" not in supported_params_o3
# o3-mini should have reasoning_effort (it's an O-series reasoning model)
assert "reasoning_effort" in supported_params_o3
def test_get_supported_openai_params_case_insensitive():
"""Test that Claude model detection is case-insensitive for models with extended thinking."""
config = GithubCopilotConfig()
# Test uppercase Claude 4 model with full model name
supported_params_upper = config.get_supported_openai_params(
"CLAUDE-SONNET-4-20250514"
)
assert "thinking" in supported_params_upper
assert "reasoning_effort" in supported_params_upper
# Test mixed case Claude 3-7 model (has extended thinking) with full model name
supported_params_mixed = config.get_supported_openai_params(
"Claude-3-7-Sonnet-20250219"
)
assert "thinking" in supported_params_mixed
assert "reasoning_effort" in supported_params_mixed
# Test that Claude 3.5 models don't have thinking support (case insensitive)
supported_params_35 = config.get_supported_openai_params("CLAUDE-3.5-SONNET")
assert "thinking" not in supported_params_35
assert "reasoning_effort" not in supported_params_35
def test_copilot_vision_request_header_with_image():

View file

@ -38,10 +38,6 @@ def test_gpt5_supports_reasoning_effort(config: OpenAIConfig):
assert "reasoning_effort" in config.get_supported_openai_params(model="gpt-5-mini")
def test_gpt5_chat_does_not_support_reasoning_effort(config: OpenAIConfig):
assert "reasoning_effort" not in config.get_supported_openai_params(
model="gpt-5-chat-latest"
)
def test_gpt5_chat_supports_temperature(config: OpenAIConfig):
@ -174,10 +170,6 @@ def test_gpt5_codex_unsupported_params_drop(config: OpenAIConfig):
assert param not in config.get_supported_openai_params(model="gpt-5-codex")
def test_gpt5_codex_supports_tool_choice(gpt5_config: OpenAIGPT5Config):
"""Test that GPT-5-Codex supports tool_choice parameter."""
supported_params = gpt5_config.get_supported_openai_params(model="gpt-5-codex")
assert "tool_choice" in supported_params
def test_gpt5_codex_supports_function_calling(config: OpenAIConfig):
@ -246,14 +238,6 @@ def test_gpt5_1_reasoning_effort_none(config: OpenAIConfig):
assert params["reasoning_effort"] == effort
def test_gpt5_1_codex_max_allows_reasoning_effort_xhigh(config: OpenAIConfig):
params = config.map_openai_params(
non_default_params={"reasoning_effort": "xhigh"},
optional_params={},
model="gpt-5.1-codex-max",
drop_params=False,
)
assert params["reasoning_effort"] == "xhigh"
def test_gpt5_rejects_reasoning_effort_xhigh_for_other_models(config: OpenAIConfig):

View file

@ -5948,8 +5948,8 @@ def test_calculate_web_search_requests_counts_unique_queries():
@pytest.mark.parametrize("custom_llm_provider", ["gemini", "vertex_ai"])
@pytest.mark.parametrize(
"model",
["gemini-2.5-flash", "gemini-3-pro-preview"],
ids=["thinking_budget_mapper", "thinking_level_mapper"],
["gemini-2.5-flash"],
ids=["thinking_budget_mapper"],
)
@pytest.mark.parametrize("reasoning_effort", ["banana", "xhigh"])
def test_invalid_reasoning_effort_is_a_400_not_a_500(custom_llm_provider, model, reasoning_effort):

View file

@ -238,31 +238,3 @@ def test_audio_predict_response_supports_bytes_base64_encoded(
assert logging_obj.model_call_details["response_cost"] == pytest.approx(0.06)
def test_image_predict_response_is_not_billed_as_audio(
local_model_cost_map: None,
) -> None:
logging_obj = MagicMock()
logging_obj.model_call_details = {}
response = httpx.Response(
status_code=200,
json={"predictions": [{"bytesBase64Encoded": "frame", "mimeType": "image/png"}]},
)
result = VertexPassthroughLoggingHandler.vertex_passthrough_handler(
httpx_response=response,
logging_obj=logging_obj,
url_route=(
"/v1/projects/test/locations/us-central1/publishers/google/models/imagen-4.0-generate-001:predict"
),
result=response.text,
start_time=datetime.now(),
end_time=datetime.now(),
cache_hit=False,
request_body={"instances": [{"prompt": "a red cube"}]},
)
assert isinstance(result["result"], litellm.ImageResponse)
assert logging_obj.call_type == PassthroughCallTypes.passthrough_image_generation.value
assert result["kwargs"]["response_cost"] == pytest.approx(
litellm.model_cost["vertex_ai/imagen-4.0-generate-001"]["output_cost_per_image"]
)

View file

@ -37,10 +37,6 @@ WANDB_REASONING_MODELS: Final = (
"Qwen/Qwen3.5-35B-A3B",
"zai-org/GLM-5.2",
"moonshotai/Kimi-K2.5",
"MiniMaxAI/MiniMax-M2.5",
"zai-org/GLM-4.5",
"Qwen/Qwen3-235B-A22B-Thinking-2507",
"deepseek-ai/DeepSeek-R1-0528",
)

View file

@ -152,83 +152,10 @@ class TestXAICostCalculator:
setattr(reported, "server_side_tool_usage_details", {"web_search_calls": 3})
assert get_cost_for_web_search_request("xai", reported, {}) == 0.0
def test_no_reported_cost_falls_back_to_token_math(self):
"""Absent the provider figure, nothing changes for existing callers."""
usage = Usage(prompt_tokens=100, completion_tokens=200, total_tokens=300)
prompt_cost, completion_cost = cost_per_token(model="grok-4-latest", usage=usage)
assert prompt_cost > 0.0
assert completion_cost > 0.0
def test_malformed_reported_cost_falls_back_to_token_math(self):
"""A junk value must not fail the request, fall back to calculating."""
usage = Usage(prompt_tokens=100, completion_tokens=200, total_tokens=300)
setattr(usage, "cost", "not-a-number")
prompt_cost, completion_cost = cost_per_token(model="grok-4-latest", usage=usage)
assert prompt_cost > 0.0
assert completion_cost > 0.0
def test_boolean_reported_cost_falls_back_to_token_math(self):
"""True is an int in python and would otherwise be billed as $1."""
usage = Usage(prompt_tokens=100, completion_tokens=200, total_tokens=300)
setattr(usage, "cost", True)
prompt_cost, completion_cost = cost_per_token(model="grok-4-latest", usage=usage)
assert prompt_cost > 0.0
assert completion_cost > 0.0
assert completion_cost != 1.0
def test_negative_reported_cost_is_rejected(self):
"""A negative amount must never reach spend tracking.
A caller who can set api_base controls the response body, so trusting a
negative figure would let them subtract from their own recorded spend and
slip past a budget. Fall back to token pricing instead, and keep charging
the web search surcharge, since no trustworthy total was reported.
"""
usage = Usage(
prompt_tokens=100,
completion_tokens=200,
total_tokens=300,
cost=-0.0037756,
)
setattr(usage, "server_side_tool_usage_details", {"web_search_calls": 3})
prompt_cost, completion_cost = cost_per_token(model="grok-4-latest", usage=usage)
assert prompt_cost > 0.0
assert completion_cost > 0.0
assert cost_per_web_search_request(usage=usage, model_info={}) > 0.0
def test_non_finite_reported_cost_is_rejected(self):
"""NaN compares false against every budget threshold.
Usage stores a provider supplied cost without validating it, so a caller who
controls the response body could report NaN and leave spend >= max_budget
false for the life of the key rather than mispricing one request. The
infinities are refused alongside it. Fall back to token pricing and keep
charging the web search surcharge, since no trustworthy total was reported.
"""
for reported_cost in (float("nan"), float("inf"), float("-inf")):
usage = Usage(
prompt_tokens=100,
completion_tokens=200,
total_tokens=300,
cost=reported_cost,
)
setattr(usage, "server_side_tool_usage_details", {"web_search_calls": 3})
prompt_cost, completion_cost = cost_per_token(model="grok-4-latest", usage=usage)
assert math.isfinite(prompt_cost), reported_cost
assert math.isfinite(completion_cost), reported_cost
assert prompt_cost > 0.0, reported_cost
assert completion_cost > 0.0, reported_cost
assert cost_per_web_search_request(usage=usage, model_info={}) > 0.0, reported_cost
def test_zero_reported_cost_is_honoured(self):
"""A reported zero is a real answer, not a missing value."""

View file

@ -1,102 +0,0 @@
"""
xAI retired eight slugs on 2026-05-15 but kept them resolvable: chat slugs redirect to
grok-4.3 and bill at grok-4.3's rates, while the grok-code-fast slugs are aliases of
grok-build-0.1 and bill at its rates, so the registry must price them that way or spend
tracking is wrong. The grok-3-beta, grok-3-fast, grok-3-mini, and grok-4-1-fast slugs
are absent from /v1/language-models and resolve to grok-4.3 the same way (the chat
response names grok-4.3 as the served model), so they carry grok-4.3's rates too.
https://docs.x.ai/developers/migration/may-15-retirement
https://docs.x.ai/developers/models/grok-build-0.1
"""
from __future__ import annotations
import json
from pathlib import Path
import pytest
REPO_ROOT = Path(__file__).parents[4]
PRICES_PATH = REPO_ROOT / "model_prices_and_context_window.json"
BACKUP_PRICES_PATH = REPO_ROOT / "litellm" / "model_prices_and_context_window_backup.json"
MAP_PATHS = (PRICES_PATH, BACKUP_PRICES_PATH)
REDIRECT_TARGET = "xai/grok-4.3"
GROK_3_MINI_SLUGS = (
"xai/grok-3-mini",
"xai/grok-3-mini-beta",
"xai/grok-3-mini-fast",
"xai/grok-3-mini-fast-beta",
"xai/grok-3-mini-fast-latest",
"xai/grok-3-mini-latest",
)
REDIRECTED_SLUGS = (
"xai/grok-3",
"xai/grok-3-beta",
"xai/grok-3-fast-beta",
"xai/grok-3-fast-latest",
"xai/grok-3-latest",
*GROK_3_MINI_SLUGS,
"xai/grok-4",
"xai/grok-4-0709",
"xai/grok-4-1-fast",
"xai/grok-4-1-fast-non-reasoning",
"xai/grok-4-1-fast-non-reasoning-latest",
"xai/grok-4-1-fast-reasoning",
"xai/grok-4-1-fast-reasoning-latest",
"xai/grok-4-fast-non-reasoning",
"xai/grok-4-fast-reasoning",
"xai/grok-4-latest",
)
CODE_REDIRECT_TARGET = "xai/grok-build-0.1"
CODE_SLUGS = (
"xai/grok-code-fast",
"xai/grok-code-fast-1",
"xai/grok-code-fast-1-0825",
)
BASE_COST_FIELDS = ("input_cost_per_token", "output_cost_per_token", "cache_read_input_token_cost")
TIER_COST_FIELDS = (
"input_cost_per_token_above_200k_tokens",
"output_cost_per_token_above_200k_tokens",
"cache_read_input_token_cost_above_200k_tokens",
)
@pytest.fixture(scope="module", params=[p.name for p in MAP_PATHS])
def cost_map(request: pytest.FixtureRequest) -> dict:
path = next(p for p in MAP_PATHS if p.name == request.param)
return json.loads(path.read_text(encoding="utf-8"))
@pytest.mark.parametrize("slug", REDIRECTED_SLUGS)
def test_redirected_slug_bills_at_the_target_rate(cost_map: dict, slug: str):
target = cost_map[REDIRECT_TARGET]
entry = cost_map[slug]
for field in BASE_COST_FIELDS:
assert entry[field] == target[field], field
@pytest.mark.parametrize("slug", CODE_SLUGS)
def test_code_slug_bills_at_grok_build_rate(cost_map: dict, slug: str):
"""grok-code-fast* are aliases of grok-build-0.1, not grok-4.3 redirects."""
target = cost_map[CODE_REDIRECT_TARGET]
entry = cost_map[slug]
for field in (*BASE_COST_FIELDS, *TIER_COST_FIELDS):
assert entry[field] == target[field], field
@pytest.mark.parametrize("slug", REDIRECTED_SLUGS)
def test_redirected_slug_carries_the_target_tier_rates(cost_map: dict, slug: str):
"""The request executes as grok-4.3, so it is tiered at grok-4.3's 200k boundary."""
target = cost_map[REDIRECT_TARGET]
entry = cost_map[slug]
for field in TIER_COST_FIELDS:
assert entry[field] == target[field], field
assert {k for k in entry if "_above_" in k} == {k for k in target if "_above_" in k}
def test_both_cost_maps_agree_on_the_redirected_slugs():
prices = json.loads(PRICES_PATH.read_text(encoding="utf-8"))
backup = json.loads(BACKUP_PRICES_PATH.read_text(encoding="utf-8"))
for slug in (*REDIRECTED_SLUGS, *CODE_SLUGS, REDIRECT_TARGET, CODE_REDIRECT_TARGET):
assert prices[slug] == backup[slug], slug

View file

@ -8057,7 +8057,6 @@ def test_model_has_no_cost_mapping_no_model_or_router_is_false():
[
"azure/speech/azure-tts",
"mistral/mistral-ocr-latest",
"vertex_ai/imagen-3.0-generate-001",
"dashscope/qwen-flash",
],
)

View file

@ -588,43 +588,6 @@ class TestAzureAnthropicCostCalculation:
== "claude-3-5-haiku-20241022"
)
def test_passthrough_logging_sets_response_cost_with_server_tool_use_dict(self):
from litellm.types.utils import Choices, Message, ModelResponse
logging_obj = self._create_mock_logging_obj(model="claude-3-7-sonnet-20250219")
logging_obj.get_router_model_id.return_value = None
logging_obj.litellm_params = {}
response = ModelResponse(
id="test-id",
choices=[
Choices(
finish_reason="stop",
index=0,
message=Message(content="test", role="assistant"),
)
],
created=1234567890,
model="claude-3-7-sonnet-20250219",
usage={
"prompt_tokens": 10,
"completion_tokens": 5,
"total_tokens": 15,
"server_tool_use": {"web_search_requests": 1},
},
)
kwargs = AnthropicPassthroughLoggingHandler._create_anthropic_response_logging_payload(
litellm_model_response=response,
model="claude-3-7-sonnet-20250219",
kwargs={},
start_time=datetime.now(),
end_time=datetime.now(),
logging_obj=logging_obj,
)
assert "response_cost" in kwargs
assert kwargs["response_cost"] > 0
class TestAnthropicBatchPassthroughCostTracking:
@ -2355,42 +2318,6 @@ class TestAnthropicResponseCostRecordedOnModelCallDetails:
model_call_details["response_cost"], not from kwargs, so the streaming payload
builder must record it there or streaming pass-through logs $0."""
def test_create_payload_records_response_cost_on_model_call_details(self):
from litellm.types.utils import Choices, Message, ModelResponse
logging_obj = MagicMock()
logging_obj.model_call_details = {}
logging_obj.get_router_model_id.return_value = None
logging_obj.litellm_params = {}
logging_obj.litellm_call_id = "test-call-id"
response = ModelResponse(
id="test-id",
choices=[
Choices(
finish_reason="stop",
index=0,
message=Message(content="hello", role="assistant"),
)
],
created=1234567890,
model="claude-3-7-sonnet-20250219",
usage={"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
)
kwargs = AnthropicPassthroughLoggingHandler._create_anthropic_response_logging_payload(
litellm_model_response=response,
model="claude-3-7-sonnet-20250219",
kwargs={},
start_time=datetime.now(),
end_time=datetime.now(),
logging_obj=logging_obj,
)
assert (
logging_obj.model_call_details["response_cost"] == kwargs["response_cost"]
)
assert logging_obj.model_call_details["response_cost"] > 0
class TestAnthropicPassthroughFastMode:

View file

@ -462,30 +462,6 @@ class TestVertexAIBatchPassthroughHandler:
assert mock_store.call_args[1]["unified_object_id"]
assert mock_store.call_args[1]["is_batch_create"] is expected
def test_batch_cost_calculation_integration(self):
"""Single Vertex AI response → non-zero cost with correct token counts."""
from litellm.batches.batch_utils import calculate_vertex_ai_batch_cost_and_usage
vertex_ai_batch_responses = [
{
"response": {
"usageMetadata": {
"promptTokenCount": 10,
"candidatesTokenCount": 5,
"totalTokenCount": 15,
}
}
}
]
result = calculate_vertex_ai_batch_cost_and_usage(
vertex_ai_batch_responses, model_name="gemini-2.0-flash-001"
)
assert result.usage.total_tokens == 15
assert result.usage.prompt_tokens == 10
assert result.usage.completion_tokens == 5
assert result.cost > 0, "batch_cost_calculator should return a non-zero cost"
def test_batch_response_transformation(self):
"""Test transformation of Vertex AI batch responses to OpenAI format"""
@ -639,76 +615,7 @@ class TestVertexAIBatchCostCalculation:
batch_cost_calculator no VertexGeminiConfig transformation involved.
"""
def test_should_aggregate_cost_and_usage_across_responses(self):
"""Two successful responses → costs and token counts are summed."""
from litellm.batches.batch_utils import calculate_vertex_ai_batch_cost_and_usage
responses = [
{
"response": {
"usageMetadata": {
"promptTokenCount": 10,
"candidatesTokenCount": 5,
"totalTokenCount": 15,
}
}
},
{
"response": {
"usageMetadata": {
"promptTokenCount": 8,
"candidatesTokenCount": 3,
"totalTokenCount": 11,
}
}
},
]
result = calculate_vertex_ai_batch_cost_and_usage(
responses, model_name="gemini-2.0-flash-001"
)
assert result.usage.prompt_tokens == 18
assert result.usage.completion_tokens == 8
assert result.usage.total_tokens == 26
assert result.cost > 0, "batch_cost_calculator should return a non-zero cost"
def test_should_skip_responses_with_null_response_body(self):
"""Failed lines (response: None) are skipped without error."""
from litellm.batches.batch_utils import calculate_vertex_ai_batch_cost_and_usage
responses = [
{
"response": {
"usageMetadata": {
"promptTokenCount": 10,
"candidatesTokenCount": 5,
"totalTokenCount": 15,
}
}
},
{"status": "JOB_STATE_FAILED", "response": None},
{
"response": {
"usageMetadata": {
"promptTokenCount": 8,
"candidatesTokenCount": 3,
"totalTokenCount": 11,
}
}
},
]
result = calculate_vertex_ai_batch_cost_and_usage(
responses, model_name="gemini-2.0-flash-001"
)
assert result.usage.prompt_tokens == 18
assert result.usage.completion_tokens == 8
assert result.usage.total_tokens == 26
assert result.cost > 0
assert result.successful_requests == 2
assert result.failed_requests == 1
def test_should_return_zeros_for_empty_response_list(self):
"""Empty input → zero cost and zero usage."""
@ -739,143 +646,4 @@ class TestVertexAIBatchCostCalculation:
assert result.usage.completion_tokens == 0
assert result.usage.total_tokens == 0
@pytest.mark.asyncio
async def test_openai_shaped_output_records_nonzero_cost_and_usage(self):
"""
Regression test for the bug where Vertex batch cost/usage was always 0.
After PR #25627 (transform_file_content_response), the GCS predictions.jsonl
is rewritten into OpenAI batch shape before the cost-tracking path sees it.
With disable_vertex_batch_output_transformation=False (default), the cost
dispatch must fall through to the generic aggregation path rather than
calling calculate_vertex_ai_batch_cost_and_usage (which only reads raw
usageMetadata fields).
"""
import litellm
from litellm.batches.batch_utils import calculate_batch_cost_and_usage
openai_shaped_responses = [
{
"id": "batch_req_abc123",
"custom_id": "request-1",
"response": {
"status_code": 200,
"request_id": "chatcmpl-xyz",
"body": {
"id": "chatcmpl-xyz",
"object": "chat.completion",
"model": "gemini-2.0-flash-001",
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": "Hello!"},
"finish_reason": "stop",
}
],
"usage": {
"prompt_tokens": 10,
"completion_tokens": 5,
"total_tokens": 15,
},
},
},
"error": None,
},
{
"id": "batch_req_def456",
"custom_id": "request-2",
"response": {
"status_code": 200,
"request_id": "chatcmpl-uvw",
"body": {
"id": "chatcmpl-uvw",
"object": "chat.completion",
"model": "gemini-2.0-flash-001",
"choices": [
{
"index": 0,
"message": {"role": "assistant", "content": "World!"},
"finish_reason": "stop",
}
],
"usage": {
"prompt_tokens": 8,
"completion_tokens": 3,
"total_tokens": 11,
},
},
},
"error": None,
},
]
original_flag = getattr(
litellm, "disable_vertex_batch_output_transformation", False
)
try:
litellm.disable_vertex_batch_output_transformation = False
result = await calculate_batch_cost_and_usage(
file_content_dictionary=openai_shaped_responses,
custom_llm_provider="vertex_ai",
model_name="gemini-2.0-flash-001",
)
finally:
litellm.disable_vertex_batch_output_transformation = original_flag
assert (
result.usage.prompt_tokens == 18
), f"expected 18 prompt tokens, got {result.usage.prompt_tokens}"
assert (
result.usage.completion_tokens == 8
), f"expected 8 completion tokens, got {result.usage.completion_tokens}"
assert (
result.usage.total_tokens == 26
), f"expected 26 total tokens, got {result.usage.total_tokens}"
assert (
result.cost > 0
), f"expected non-zero cost for completed Vertex batch, got {result.cost}"
@pytest.mark.asyncio
async def test_raw_vertex_output_still_works_when_transformation_disabled(self):
"""
When disable_vertex_batch_output_transformation=True the GCS file is returned
as raw Vertex predictions.jsonl; the specialized reader must be used.
"""
import litellm
from litellm.batches.batch_utils import calculate_batch_cost_and_usage
raw_vertex_responses = [
{
"request": {"contents": [{"role": "user", "parts": [{"text": "hi"}]}]},
"status": "",
"response": {
"candidates": [{"content": {"parts": [{"text": "Hello!"}]}}],
"usageMetadata": {
"promptTokenCount": 10,
"candidatesTokenCount": 5,
"totalTokenCount": 15,
},
},
"processed_time": "2026-01-01T00:00:00Z",
},
]
original_flag = getattr(
litellm, "disable_vertex_batch_output_transformation", False
)
try:
litellm.disable_vertex_batch_output_transformation = True
result = await calculate_batch_cost_and_usage(
file_content_dictionary=raw_vertex_responses,
custom_llm_provider="vertex_ai",
model_name="gemini-2.0-flash-001",
)
finally:
litellm.disable_vertex_batch_output_transformation = original_flag
assert result.usage.prompt_tokens == 10
assert result.usage.completion_tokens == 5
assert result.usage.total_tokens == 15
assert result.cost > 0, "raw Vertex shape should also produce non-zero cost"

View file

@ -634,25 +634,6 @@ def test_equal_modeled_usage_is_zero_under_equivalent_model_names() -> None:
assert _savings("claude-opus-5", "anthropic/claude-opus-5", usage, usage) == 0.0
def test_baseline_is_priced_under_its_own_provider():
"""Two providers can serve the same bare model name at different rates, so dropping
the provider prices the baseline against a vendor the operator never named. Here it
decides whether routing reads as a saving or a loss."""
usage = Usage(prompt_tokens=100_000, completion_tokens=10_000, total_tokens=110_000)
azure = compute_autorouter_savings(
baseline_model="azure_ai/deepseek-r1",
selected_model="claude-haiku-4-5",
selected_provider="anthropic",
usage=usage,
)
deepseek = compute_autorouter_savings(
baseline_model="deepseek/deepseek-r1",
selected_model="claude-haiku-4-5",
selected_provider="anthropic",
usage=usage,
)
assert azure != pytest.approx(deepseek)
assert azure > 0 > deepseek
def test_unresolvable_baseline_remains_unknown():

View file

@ -3887,179 +3887,7 @@ class TestSpendLogsPayload:
}
return mock_response
@pytest.mark.asyncio
async def test_spend_logs_payload_success_log_with_api_base(self, monkeypatch):
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
# Clear any env overrides that would change the recorded api_base
monkeypatch.delenv("ANTHROPIC_BASE_URL", raising=False)
monkeypatch.delenv("ANTHROPIC_API_BASE", raising=False)
litellm.callbacks = [_ProxyDBLogger(message_logging=False)]
# litellm._turn_on_debug()
client = AsyncHTTPHandler()
with (
patch.object(
litellm.proxy.db.db_spend_update_writer.DBSpendUpdateWriter,
"_insert_spend_log_to_db",
) as mock_client,
patch.object(litellm.proxy.proxy_server, "prisma_client"),
patch.object(client, "post", side_effect=self.mock_anthropic_response),
):
response = await litellm.acompletion(
model="claude-4-sonnet-20250514",
messages=[{"role": "user", "content": "Hello, world!"}],
metadata={"user_api_key_end_user_id": "test_user_1"},
client=client,
)
assert response.choices[0].message.content == "Hi! My name is Claude."
await _wait_for_mock_call(mock_client)
kwargs = mock_client.call_args.kwargs
payload: SpendLogsPayload = kwargs["payload"]
expected_payload = SpendLogsPayload(
**{
"request_id": "chatcmpl-34df56d5-4807-45c1-bb99-61e52586b802",
"call_type": "acompletion",
"api_key": "",
"cache_hit": "None",
"startTime": datetime.datetime(
2025, 3, 24, 22, 2, 42, 975883, tzinfo=datetime.timezone.utc
),
"endTime": datetime.datetime(
2025, 3, 24, 22, 2, 42, 989132, tzinfo=datetime.timezone.utc
),
"completionStartTime": datetime.datetime(
2025, 3, 24, 22, 2, 42, 989132, tzinfo=datetime.timezone.utc
),
"model": "claude-4-sonnet-20250514",
"user": "",
"team_id": "",
"metadata": '{"applied_guardrails": [], "attempted_fallbacks": null, "original_model_group": null, "batch_models": null, "batch_successful_requests": null, "batch_failed_requests": null, "mcp_tool_call_metadata": null, "vector_store_request_metadata": null, "routing_decision": null, "internal_call_origin": null, "guardrail_information": null, "compression_savings": null, "litellm_gateway_injected_cache": null, "router_metadata": null, "autorouter_savings_estimate": null, "autorouter_baseline_observation": null, "azure_spillover": null, "usage_object": {"completion_tokens": 503, "prompt_tokens": 2095, "total_tokens": 2598, "completion_tokens_details": null, "prompt_tokens_details": {"audio_tokens": null, "cached_tokens": 0}, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0}, "model_map_information": {"model_map_key": "claude-4-sonnet-20250514", "model_map_value": {"key": "claude-4-sonnet-20250514", "max_tokens": 128000, "max_input_tokens": 200000, "max_output_tokens": 128000, "input_cost_per_token": 3e-06, "cache_creation_input_token_cost": 3.75e-06, "cache_read_input_token_cost": 3e-07, "input_cost_per_character": null, "input_cost_per_token_above_128k_tokens": null, "input_cost_per_token_above_200k_tokens": null, "input_cost_per_query": null, "input_cost_per_second": null, "input_cost_per_audio_token": null, "input_cost_per_token_batches": null, "output_cost_per_token_batches": null, "output_cost_per_token": 1.5e-05, "output_cost_per_audio_token": null, "output_cost_per_character": null, "output_cost_per_token_above_128k_tokens": null, "output_cost_per_character_above_128k_tokens": null, "output_cost_per_token_above_200k_tokens": null, "output_cost_per_second": null, "output_cost_per_image": null, "output_vector_size": null, "litellm_provider": "anthropic", "mode": "chat", "supports_system_messages": null, "supports_response_schema": true, "supports_vision": true, "supports_function_calling": true, "supports_tool_choice": true, "supports_assistant_prefill": true, "supports_prompt_caching": true, "supports_audio_input": false, "supports_audio_output": false, "supports_pdf_input": true, "supports_embedding_image_input": false, "supports_native_streaming": null, "supports_web_search": false, "supports_reasoning": true, "search_context_cost_per_query": null, "tpm": null, "rpm": null, "supported_openai_params": ["stream", "stop", "temperature", "top_p", "max_tokens", "max_completion_tokens", "tools", "tool_choice", "extra_headers", "parallel_tool_calls", "response_format", "user", "reasoning_effort", "thinking"]}}, "additional_usage_values": {"completion_tokens_details": {"accepted_prediction_tokens": null, "audio_tokens": null, "reasoning_tokens": null, "rejected_prediction_tokens": null, "text_tokens": 503, "image_tokens": null}, "prompt_tokens_details": {"audio_tokens": null, "cached_tokens": 0, "text_tokens": null, "image_tokens": null}, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0}}',
"cache_key": "Cache OFF",
"spend": 0.01383,
"total_tokens": 2598,
"prompt_tokens": 2095,
"completion_tokens": 503,
"request_tags": "[]",
"end_user": "test_user_1",
"api_base": "https://api.anthropic.com/v1/messages",
"model_group": "",
"model_id": "",
"requester_ip_address": None,
"custom_llm_provider": "anthropic",
"messages": "{}",
"response": "{}",
"proxy_server_request": "{}",
"status": "success",
"mcp_namespaced_tool_name": None,
"agent_id": None,
}
)
differences = _compare_nested_dicts(
payload, expected_payload, ignore_keys=ignored_keys
)
if differences:
pytest.fail(f"Dictionary mismatch: {differences}")
@pytest.mark.asyncio
async def test_spend_logs_payload_success_log_with_router(self, monkeypatch):
from litellm.llms.custom_httpx.http_handler import AsyncHTTPHandler
# Clear any env overrides that would change the recorded api_base
monkeypatch.delenv("ANTHROPIC_BASE_URL", raising=False)
monkeypatch.delenv("ANTHROPIC_API_BASE", raising=False)
litellm.callbacks = [_ProxyDBLogger(message_logging=False)]
# litellm._turn_on_debug()
client = AsyncHTTPHandler()
router = Router(
model_list=[
{
"model_name": "my-anthropic-model-group",
"litellm_params": {
"model": "claude-4-sonnet-20250514",
},
"model_info": {
"id": "my-unique-model-id",
},
}
]
)
with (
patch.object(
litellm.proxy.db.db_spend_update_writer.DBSpendUpdateWriter,
"_insert_spend_log_to_db",
) as mock_client,
patch.object(litellm.proxy.proxy_server, "prisma_client"),
patch.object(client, "post", side_effect=self.mock_anthropic_response),
):
response = await router.acompletion(
model="my-anthropic-model-group",
messages=[{"role": "user", "content": "Hello, world!"}],
metadata={"user_api_key_end_user_id": "test_user_1"},
client=client,
)
assert response.choices[0].message.content == "Hi! My name is Claude."
await _wait_for_mock_call(mock_client)
kwargs = mock_client.call_args.kwargs
payload: SpendLogsPayload = kwargs["payload"]
expected_payload = SpendLogsPayload(
**{
"request_id": "chatcmpl-34df56d5-4807-45c1-bb99-61e52586b802",
"call_type": "acompletion",
"api_key": "",
"cache_hit": "None",
"startTime": datetime.datetime(
2025, 3, 24, 22, 2, 42, 975883, tzinfo=datetime.timezone.utc
),
"endTime": datetime.datetime(
2025, 3, 24, 22, 2, 42, 989132, tzinfo=datetime.timezone.utc
),
"completionStartTime": datetime.datetime(
2025, 3, 24, 22, 2, 42, 989132, tzinfo=datetime.timezone.utc
),
"model": "claude-4-sonnet-20250514",
"user": "",
"team_id": "",
"metadata": '{"applied_guardrails": [], "attempted_fallbacks": 0, "original_model_group": "my-anthropic-model-group", "batch_models": null, "batch_successful_requests": null, "batch_failed_requests": null, "mcp_tool_call_metadata": null, "vector_store_request_metadata": null, "routing_decision": null, "internal_call_origin": null, "guardrail_information": null, "compression_savings": null, "litellm_gateway_injected_cache": null, "router_metadata": null, "autorouter_savings_estimate": null, "autorouter_baseline_observation": null, "azure_spillover": null, "usage_object": {"completion_tokens": 503, "prompt_tokens": 2095, "total_tokens": 2598, "completion_tokens_details": null, "prompt_tokens_details": {"audio_tokens": null, "cached_tokens": 0}, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0}, "model_map_information": {"model_map_key": "claude-4-sonnet-20250514", "model_map_value": {"key": "claude-4-sonnet-20250514", "max_tokens": 128000, "max_input_tokens": 200000, "max_output_tokens": 128000, "input_cost_per_token": 3e-06, "cache_creation_input_token_cost": 3.75e-06, "cache_read_input_token_cost": 3e-07, "input_cost_per_character": null, "input_cost_per_token_above_128k_tokens": null, "input_cost_per_token_above_200k_tokens": null, "input_cost_per_query": null, "input_cost_per_second": null, "input_cost_per_audio_token": null, "input_cost_per_token_batches": null, "output_cost_per_token_batches": null, "output_cost_per_token": 1.5e-05, "output_cost_per_audio_token": null, "output_cost_per_character": null, "output_cost_per_token_above_128k_tokens": null, "output_cost_per_character_above_128k_tokens": null, "output_cost_per_token_above_200k_tokens": null, "output_cost_per_second": null, "output_cost_per_image": null, "output_vector_size": null, "litellm_provider": "anthropic", "mode": "chat", "supports_system_messages": null, "supports_response_schema": true, "supports_vision": true, "supports_function_calling": true, "supports_tool_choice": true, "supports_assistant_prefill": true, "supports_prompt_caching": true, "supports_audio_input": false, "supports_audio_output": false, "supports_pdf_input": true, "supports_embedding_image_input": false, "supports_native_streaming": null, "supports_web_search": false, "supports_reasoning": true, "search_context_cost_per_query": null, "tpm": null, "rpm": null, "supported_openai_params": ["stream", "stop", "temperature", "top_p", "max_tokens", "max_completion_tokens", "tools", "tool_choice", "extra_headers", "parallel_tool_calls", "response_format", "user", "reasoning_effort", "thinking"]}}, "additional_usage_values": {"completion_tokens_details": {"accepted_prediction_tokens": null, "audio_tokens": null, "reasoning_tokens": null, "rejected_prediction_tokens": null, "text_tokens": 503, "image_tokens": null}, "prompt_tokens_details": {"audio_tokens": null, "cached_tokens": 0, "text_tokens": null, "image_tokens": null}, "cache_creation_input_tokens": 0, "cache_read_input_tokens": 0}}',
"cache_key": "Cache OFF",
"spend": 0.01383,
"total_tokens": 2598,
"prompt_tokens": 2095,
"completion_tokens": 503,
"request_tags": "[]",
"end_user": "test_user_1",
"api_base": "https://api.anthropic.com/v1/messages",
"model_group": "my-anthropic-model-group",
"model_id": "my-unique-model-id",
"requester_ip_address": None,
"custom_llm_provider": "anthropic",
"messages": "{}",
"response": "{}",
"proxy_server_request": "{}",
"status": "success",
"mcp_namespaced_tool_name": None,
"agent_id": None,
}
)
differences = _compare_nested_dicts(
payload, expected_payload, ignore_keys=ignored_keys
)
if differences:
pytest.fail(f"Dictionary mismatch: {differences}")
def _compare_nested_dicts(

View file

@ -48,72 +48,6 @@ class MetadataCaptureCallback(CustomLogger):
self.event.set()
@pytest.mark.asyncio
async def test_metadata_passed_to_custom_callback_codex_models():
"""
Test that metadata passed to completion() is available in custom callback
when using codex models (responses API bridge path).
Codex models have mode=responses and route through responses_api_bridge,
which passes litellm_metadata. The fix ensures this is preserved as
litellm_params.metadata for callback compatibility.
"""
from litellm.types.llms.openai import ResponsesAPIResponse
mock_response = ResponsesAPIResponse.model_construct(
id="resp-test",
created_at=0,
output=[
{
"type": "message",
"id": "msg-1",
"status": "completed",
"role": "assistant",
"content": [{"type": "output_text", "text": "Hello!"}],
}
],
object="response",
model="gpt-5.1-codex",
status="completed",
usage={
"input_tokens": 5,
"output_tokens": 10,
"total_tokens": 15,
},
)
test_metadata = {"foo": "bar", "trace_id": "test-123"}
callback = MetadataCaptureCallback()
original_callbacks = litellm.callbacks.copy() if litellm.callbacks else []
litellm.callbacks = [callback]
try:
with patch(
"litellm.llms.custom_httpx.http_handler.AsyncHTTPHandler.post",
new_callable=AsyncMock,
) as mock_post:
mock_post.return_value = _make_mock_http_response(
mock_response.model_dump()
)
# gpt-5.1-codex has mode=responses - routes through responses bridge
await litellm.acompletion(
model="gpt-5.1-codex",
messages=[{"role": "user", "content": "Hello"}],
metadata=test_metadata,
)
await asyncio.wait_for(callback.event.wait(), timeout=5.0)
assert callback.captured_kwargs is not None, "Callback should have been invoked"
litellm_params = callback.captured_kwargs.get("litellm_params", {})
metadata = litellm_params.get("metadata") or {}
assert "foo" in metadata, "metadata['foo'] should be accessible in callback"
assert metadata["foo"] == "bar"
assert metadata.get("trace_id") == "test-123"
finally:
litellm.callbacks = original_callbacks
@pytest.mark.asyncio

View file

@ -148,7 +148,6 @@ async def test_each_tokenizer_gets_its_own_cached_counter(fake_tokenizers: None)
("gpt-4o", "o200k_base"),
("gpt-4o-mini", "o200k_base"),
("gpt-4o-2024-08-06", "o200k_base"),
("chatgpt-4o-latest", "o200k_base"),
("gpt-4.1", "o200k_base"),
("gpt-5", "o200k_base"),
("gpt-5-mini", "o200k_base"),

View file

@ -159,131 +159,8 @@ def test_cost_calculator_with_response_cost_in_additional_headers():
assert result == 1000
def test_cost_calculator_with_usage(_local_model_cost_map, monkeypatch):
usage = Usage(
prompt_tokens=120,
completion_tokens=100,
prompt_tokens_details=PromptTokensDetailsWrapper(
text_tokens=10,
audio_tokens=90,
image_tokens=20,
),
)
mr = ModelResponse(usage=usage, model="gemini-2.0-flash-001")
result = response_cost_calculator(
response_object=mr,
model="",
custom_llm_provider="vertex_ai",
call_type="acompletion",
optional_params={},
cache_hit=None,
base_model=None,
)
model_info = litellm.model_cost["gemini-2.0-flash-001"]
# Step 1: Test a model where input_cost_per_image_token is not set.
# In this case the calculation should use input_cost_per_token as fallback.
assert model_info.get("input_cost_per_image_token") is None, (
"Test case expects that input_cost_per_image_token is not set"
)
expected_cost = (
usage.prompt_tokens_details.audio_tokens * model_info["input_cost_per_audio_token"]
+ usage.prompt_tokens_details.text_tokens * model_info["input_cost_per_token"]
+ usage.prompt_tokens_details.image_tokens * model_info["input_cost_per_token"]
+ usage.completion_tokens * model_info["output_cost_per_token"]
)
assert result == expected_cost, f"Got {result}, Expected {expected_cost}"
# Step 2: Set input_cost_per_image_token.
# In this case the explicit cost information should be used.
temp_model_info_object = dict(model_info)
temp_model_info_object["input_cost_per_image_token"] = 0.5
monkeypatch.setattr(
litellm,
"model_cost",
{"gemini-2.0-flash-001": temp_model_info_object},
)
# Invalidate caches after modifying litellm.model_cost
from litellm.utils import _invalidate_model_cost_lowercase_map
_invalidate_model_cost_lowercase_map()
result = response_cost_calculator(
response_object=mr,
model="",
custom_llm_provider="vertex_ai",
call_type="acompletion",
optional_params={},
cache_hit=None,
base_model=None,
)
expected_cost = (
usage.prompt_tokens_details.audio_tokens * temp_model_info_object["input_cost_per_audio_token"]
+ usage.prompt_tokens_details.text_tokens * temp_model_info_object["input_cost_per_token"]
+ usage.prompt_tokens_details.image_tokens * temp_model_info_object["input_cost_per_image_token"]
+ usage.completion_tokens * temp_model_info_object["output_cost_per_token"]
)
assert result == expected_cost, f"Got {result}, Expected {expected_cost}"
def test_handle_realtime_stream_cost_calculation_stores_cost_breakdown():
"""Regression: realtime cost must populate logging_obj.cost_breakdown so the
spend logs / UI show input vs output cost (issue: cost_breakdown was None for
/v1/realtime even though a total spend was computed)."""
from datetime import datetime
from litellm.litellm_core_utils.litellm_logging import Logging
results: OpenAIRealtimeStreamList = [
{"type": "session.created", "session": {"model": "gpt-4o-realtime-preview"}},
{
"type": "response.done",
"response": {
"usage": {
"input_tokens": 100,
"output_tokens": 50,
"total_tokens": 150,
}
},
},
]
combined_usage_object = RealtimeAPITokenUsageProcessor.collect_and_combine_usage_from_realtime_stream_results(
results=results,
)
logging_obj = Logging(
model="gpt-4o-realtime-preview",
messages=[],
stream=False,
call_type="_arealtime",
start_time=datetime.now(),
litellm_call_id="realtime-cost-breakdown-test",
function_id="realtime-cost-breakdown-test",
)
total_cost = handle_realtime_stream_cost_calculation(
results=results,
combined_usage_object=combined_usage_object,
custom_llm_provider="openai",
litellm_model_name="gpt-4o-realtime-preview",
litellm_logging_obj=logging_obj,
)
assert total_cost > 0
assert logging_obj.cost_breakdown is not None
assert logging_obj.cost_breakdown["input_cost"] > 0
assert logging_obj.cost_breakdown["output_cost"] > 0
assert abs(logging_obj.cost_breakdown["input_cost"] + logging_obj.cost_breakdown["output_cost"] - total_cost) < 1e-9
assert abs(logging_obj.cost_breakdown["total_cost"] - total_cost) < 1e-9
def test_realtime_stream_combines_text_and_audio_token_details():
@ -1124,126 +1001,6 @@ def test_bedrock_cost_calculator_comparison_with_without_cache():
print(f"Cost with cache: {cost_with_cache}")
def test_log_context_cost_calculation():
"""
Test that log context cost calculation works correctly with tiered pricing.
This test verifies that when using extended context (above 200k tokens),
the log context costs are calculated using the appropriate tiered rates.
"""
from litellm import completion_cost
from litellm.types.utils import (
Choices,
Message,
ModelResponse,
PromptTokensDetailsWrapper,
Usage,
)
# Create a mock response with extended context usage
extended_context_response = ModelResponse(
id="test-extended-context-response",
created=1750733889,
model="claude-4-sonnet-20250514",
object="chat.completion",
system_fingerprint=None,
choices=[
Choices(
finish_reason="stop",
index=0,
message=Message(
content="This is a test response for extended context cost calculation.",
role="assistant",
tool_calls=None,
function_call=None,
),
)
],
usage=Usage(
total_tokens=350000, # Above 200k threshold
prompt_tokens=301000, # Above 200k threshold
completion_tokens=50000,
prompt_tokens_details=PromptTokensDetailsWrapper(
text_tokens=300000,
cached_tokens=0, # No cache hits
audio_tokens=None,
image_tokens=None,
character_count=None,
video_length_seconds=None,
cache_creation_tokens=1000,
),
completion_tokens_details=None,
_cache_creation_input_tokens=1000, # Some tokens added to cache
),
)
# Calculate the cost using the extended context model
result = completion_cost(
completion_response=extended_context_response,
model="claude-4-sonnet-20250514",
custom_llm_provider="anthropic",
)
# Debug: Print the actual result
print(f"DEBUG: Actual cost result: ${result:.6f}")
# Get model info to understand the pricing
from litellm import get_model_info
model_info = get_model_info(model="claude-4-sonnet-20250514", custom_llm_provider="anthropic")
# Calculate expected cost based on actual model pricing
input_cost_per_token = model_info.get("input_cost_per_token", 0)
output_cost_per_token = model_info.get("output_cost_per_token", 0)
cache_creation_cost_per_token = model_info.get("cache_creation_input_token_cost", 0)
# Check if tiered pricing is applied
input_cost_above_200k = model_info.get("input_cost_per_token_above_200k_tokens", input_cost_per_token)
output_cost_above_200k = model_info.get("output_cost_per_token_above_200k_tokens", output_cost_per_token)
cache_creation_above_200k = model_info.get(
"cache_creation_input_token_cost_above_200k_tokens",
cache_creation_cost_per_token,
)
print(f"DEBUG: Base input cost per token: ${input_cost_per_token:.2e}")
print(f"DEBUG: Base output cost per token: ${output_cost_per_token:.2e}")
print(f"DEBUG: Base cache creation cost per token: ${cache_creation_cost_per_token:.2e}")
# Handle tiered pricing - if not available, use base pricing
if input_cost_above_200k is not None:
print(f"DEBUG: Tiered input cost per token (>200k): ${input_cost_above_200k:.2e}")
else:
print("DEBUG: No tiered input pricing available, using base pricing")
input_cost_above_200k = input_cost_per_token
if output_cost_above_200k is not None:
print(f"DEBUG: Tiered output cost per token (>200k): ${output_cost_above_200k:.2e}")
else:
print("DEBUG: No tiered output pricing available, using base pricing")
output_cost_above_200k = output_cost_per_token
if cache_creation_above_200k is not None:
print(f"DEBUG: Tiered cache creation cost per token (>200k): ${cache_creation_above_200k:.2e}")
else:
print("DEBUG: No tiered cache creation pricing available, using base pricing")
cache_creation_above_200k = cache_creation_cost_per_token
# Since we're above 200k tokens, we should use tiered pricing if available
expected_input_cost = 300000 * input_cost_above_200k
expected_output_cost = 50000 * output_cost_above_200k
expected_cache_cost = 1000 * cache_creation_above_200k
expected_total = expected_input_cost + expected_output_cost + expected_cache_cost
print(f"DEBUG: Expected total: ${expected_total:.6f}")
# Allow for small floating point differences
assert abs(result - expected_total) < 1e-6, f"Expected cost ${expected_total:.6f}, but got ${result:.6f}"
print(f"✓ Log context cost calculation with tiered pricing is correct: ${result:.6f}")
print(f" - Input tokens (300k): ${expected_input_cost:.6f}")
print(f" - Output tokens (50k): ${expected_output_cost:.6f}")
print(f" - Cache creation (1k): ${expected_cache_cost:.6f}")
print(f" - Total: ${result:.6f}")
def test_gemini_25_explicit_caching_cost_direct_usage():
@ -1814,56 +1571,6 @@ def test_cost_margin_with_discount(monkeypatch):
print(f" - Expected: ${expected_cost:.6f}")
def test_azure_image_generation_cost_calculator():
from unittest.mock import MagicMock
from litellm.types.utils import (
ImageObject,
ImageResponse,
ImageUsage,
ImageUsageInputTokensDetails,
)
response_cost_calculator_kwargs = {
"response_object": ImageResponse(
created=1761785270,
background=None,
data=[
ImageObject(
b64_json=None,
revised_prompt="A futuristic, techno-inspired green duck wearing cool modern sunglasses. The duck has a sleek, metallic appearance with glowing neon green accents, standing on a high-tech urban background with holographic billboards and illuminated city lights in the distance. The duck's feathers have a glossy, high-tech sheen, resembling a robotic design but still maintaining its avian features. The scene has a vibrant, cyberpunk aesthetic with a neon color palette.",
url="test-azure-blob-url-with-sas-token",
)
],
output_format=None,
quality="hd",
size=None,
usage=ImageUsage(
input_tokens=0,
input_tokens_details=ImageUsageInputTokensDetails(image_tokens=0, text_tokens=0),
output_tokens=0,
total_tokens=0,
),
),
"model": "azure/dall-e-3",
"cache_hit": False,
"custom_llm_provider": "azure",
"base_model": "azure/dall-e-3",
"call_type": "aimage_generation",
"optional_params": {},
"custom_pricing": False,
"prompt": "",
"standard_built_in_tools_params": {
"web_search_options": None,
"file_search": None,
},
"router_model_id": "6738c432ffc9b733597c6b86613ca20dc5f49bde591fd3d03e7cd6aa25bb241e",
"litellm_logging_obj": MagicMock(),
"service_tier": None,
}
cost = response_cost_calculator(**response_cost_calculator_kwargs)
assert cost > 0.079
def test_completion_cost_extracts_service_tier_from_response(_local_model_cost_map):
@ -2616,87 +2323,6 @@ def test_gemini_without_cache_tokens_details():
print("✅ Gemini without cacheTokensDetails works correctly")
def test_gemini_implicit_caching_cost_calculation():
"""
Test for Issue #16341: Gemini implicit cached tokens not counted in spend log
When Gemini uses implicit caching, it returns cachedContentTokenCount but NOT
cacheTokensDetails. In this case, we should subtract cachedContentTokenCount
from text_tokens to correctly calculate costs.
See: https://github.com/BerriAI/litellm/issues/16341
"""
from litellm import completion_cost
from litellm.llms.vertex_ai.gemini.vertex_and_google_ai_studio_gemini import (
VertexGeminiConfig,
)
from litellm.types.utils import Choices, Message, ModelResponse
# Simulate Gemini response with implicit caching (cachedContentTokenCount only)
completion_response = {
"usageMetadata": {
"promptTokenCount": 10000,
"candidatesTokenCount": 5,
"totalTokenCount": 10005,
"cachedContentTokenCount": 8000, # Implicit caching - no cacheTokensDetails
"promptTokensDetails": [{"modality": "TEXT", "tokenCount": 10000}],
"candidatesTokensDetails": [{"modality": "TEXT", "tokenCount": 5}],
}
}
usage = VertexGeminiConfig._calculate_usage(completion_response)
# Verify parsing
assert usage.cache_read_input_tokens == 8000, (
f"cache_read_input_tokens should be 8000, got {usage.cache_read_input_tokens}"
)
assert usage.prompt_tokens_details.cached_tokens == 8000, (
f"cached_tokens should be 8000, got {usage.prompt_tokens_details.cached_tokens}"
)
# CRITICAL: text_tokens should be (10000 - 8000) = 2000, NOT 10000
# This is the fix for issue #16341
assert usage.prompt_tokens_details.text_tokens == 2000, (
f"text_tokens should be 2000 (10000 - 8000), got {usage.prompt_tokens_details.text_tokens}"
)
# Verify cost calculation uses cached token pricing
response = ModelResponse(
id="mock-id",
model="gemini-2.0-flash",
choices=[
Choices(
index=0,
message=Message(role="assistant", content="Hello!"),
finish_reason="stop",
)
],
usage=usage,
)
cost = completion_cost(
completion_response=response,
model="gemini-2.0-flash",
custom_llm_provider="gemini",
)
# Get model pricing for verification
import litellm
model_info = litellm.get_model_info("gemini/gemini-2.0-flash")
input_cost = model_info.get("input_cost_per_token", 0)
cache_read_cost = model_info.get("cache_read_input_token_cost", input_cost)
output_cost = model_info.get("output_cost_per_token", 0)
# Expected cost: (2000 * input) + (8000 * cache_read) + (5 * output)
expected_cost = (2000 * input_cost) + (8000 * cache_read_cost) + (5 * output_cost)
assert abs(cost - expected_cost) < 1e-9, (
f"Cost calculation is wrong. Got ${cost:.6f}, expected ${expected_cost:.6f}. "
f"Cached tokens may not be using reduced pricing."
)
print("✅ Issue #16341 fix verified: Gemini implicit caching cost calculated correctly")
def test_additional_costs_only_for_azure_ai(_local_model_cost_map):

View file

@ -157,21 +157,3 @@ def test_acount_tokens_no_api_key_falls_back(monkeypatch):
assert result.tokenizer_type == "local_tokenizer"
async def test_acount_tokens_local_fallback_counts_off_the_event_loop():
from tests.large_text import text
from tests.test_litellm.litellm_core_utils.event_loop_lag import (
assert_loop_stayed_free,
timed_with_loop_lags,
warm_tokenizer,
)
model = "together_ai/meta-llama/Llama-3-8b-chat-hf"
warm_tokenizer(model)
result, took, lags = await timed_with_loop_lags(
lambda: litellm.acount_tokens(model=model, messages=[{"role": "user", "content": text * 100}])
)
assert result.tokenizer_type == "local_tokenizer"
assert result.total_tokens > 100_000
assert_loop_stayed_free(took, lags)

View file

@ -90,27 +90,6 @@ class TestGPTImageCostCalculator:
class TestGPTImageCostRouting:
"""Test that gpt-image models are properly routed to the token-based calculator"""
def test_openai_dalle_routes_to_pixel_calculator(self):
"""Test that OpenAI DALL-E still routes to pixel-based calculator"""
from litellm.litellm_core_utils.llm_cost_calc.utils import CostCalculatorUtils
image_response = ImageResponse(
created=1234567890,
data=[ImageObject(url="http://example.com/image.jpg")],
)
image_response.size = "1024x1024"
image_response.quality = "standard"
cost = CostCalculatorUtils.route_image_generation_cost_calculator(
model="dall-e-3",
completion_response=image_response,
custom_llm_provider="openai",
size="1024x1024",
quality="standard",
n=1,
)
assert cost >= 0
class TestGPTImage15OutputImageTokens:

View file

@ -850,36 +850,6 @@ async def test_arouter_async_get_healthy_deployments():
assert result[0]["litellm_params"]["model"] == "gpt-3.5-turbo"
@pytest.mark.asyncio
@patch("litellm.amoderation")
async def test_arouter_amoderation_with_credential_name(mock_amoderation):
"""
Test that router.amoderation passes litellm_credential_name to the underlying litellm.amoderation call
"""
mock_amoderation.return_value = AsyncMock()
router = litellm.Router(
model_list=[
{
"model_name": "text-moderation-stable",
"litellm_params": {
"model": "text-moderation-stable",
"litellm_credential_name": "my-custom-auth",
},
},
],
)
await router.amoderation(input="I love everyone!", model="text-moderation-stable")
mock_amoderation.assert_called_once()
call_kwargs = mock_amoderation.call_args[1] # Get the kwargs of the call
print(
"call kwargs for router.amoderation=",
json.dumps(call_kwargs, indent=4, default=str),
)
assert call_kwargs["litellm_credential_name"] == "my-custom-auth"
assert call_kwargs["model"] == "text-moderation-stable"
def test_arouter_test_team_model():

View file

@ -95,15 +95,6 @@ def _successor(info: dict[str, object]) -> str | None:
return successor if isinstance(successor, str) else None
def test_together_successor_metadata_points_at_known_models(cost_map: CostMap):
successors = {
model: successor
for model, info in cost_map.items()
if model.startswith("together_ai/") and (successor := _successor(info)) is not None
}
assert len(successors) >= 10
for model, successor in successors.items():
assert successor in cost_map, f"{model} names successor {successor} that is not in the map"
def test_together_backup_cost_map_in_sync(cost_map: CostMap):

View file

@ -1465,12 +1465,6 @@ class TestProxyFunctionCalling:
("gemini/gemini-2.5-pro", "litellm_proxy/gemini/gemini-2.5-pro", True),
("gemini/gemini-2.5-flash", "litellm_proxy/gemini/gemini-2.5-flash", True),
# Groq models (mixed support)
("groq/gemma-7b-it", "litellm_proxy/groq/gemma-7b-it", True),
(
"groq/llama-3.3-70b-versatile",
"litellm_proxy/groq/llama-3.3-70b-versatile",
True,
),
# Cohere models (generally don't support function calling)
("command-nightly", "litellm_proxy/command-nightly", False),
],

View file

@ -46,34 +46,6 @@ class TestXAIResponsesAutoRouting:
assert model_info.get("mode") != "responses"
assert updated_model == model
def test_responses_api_bridge_check_with_tools(self):
"""Test that with tools, xAI automatically routes to Responses API"""
model = "grok-3"
custom_llm_provider = "xai"
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get the weather",
"parameters": {
"type": "object",
"properties": {"location": {"type": "string"}},
},
},
}
]
web_search_options = None
model_info, updated_model = responses_api_bridge_check(
model=model,
custom_llm_provider=custom_llm_provider,
web_search_options=web_search_options,
)
# Should auto-route to responses mode when tools are present
assert model_info.get("mode") == "chat"
assert updated_model == model
def test_responses_api_bridge_check_with_empty_tools(self):
"""Test that with empty tools list, xAI does not route to Responses API"""
@ -134,57 +106,8 @@ class TestXAIResponsesAutoRouting:
assert model_info.get("mode") == "responses"
assert updated_model == "grok-3" # prefix removed
def test_responses_api_bridge_check_with_code_interpreter_tool(self):
"""Test auto-routing with code_interpreter tool"""
model = "grok-3"
custom_llm_provider = "xai"
tools = [{"type": "code_interpreter"}]
web_search_options = None
model_info, updated_model = responses_api_bridge_check(
model=model,
custom_llm_provider=custom_llm_provider,
web_search_options=web_search_options,
)
# Should auto-route with code_interpreter tool
assert model_info.get("mode") == "chat"
assert updated_model == model
def test_responses_api_bridge_check_with_web_search_tool(self):
"""Test auto-routing with web_search tool"""
model = "grok-4"
custom_llm_provider = "xai"
tools = [
{"type": "web_search", "filters": {"allowed_domains": ["wikipedia.org"]}}
]
web_search_options = None
model_info, updated_model = responses_api_bridge_check(
model=model,
custom_llm_provider=custom_llm_provider,
web_search_options=web_search_options,
)
# Should auto-route with web_search tool
assert model_info.get("mode") == "chat"
assert updated_model == model
def test_responses_api_bridge_check_with_x_search_tool(self):
"""Test auto-routing with x_search tool"""
model = "grok-4"
custom_llm_provider = "xai"
tools = [{"type": "x_search", "allowed_x_handles": ["@elonmusk"]}]
web_search_options = None
model_info, updated_model = responses_api_bridge_check(
model=model,
custom_llm_provider=custom_llm_provider,
web_search_options=web_search_options,
)
# Should auto-route with x_search tool
assert model_info.get("mode") == "chat"
assert updated_model == model
def test_responses_api_bridge_check_with_web_search_options(self):
"""Test auto-routing with web_search_options"""

View file

@ -157,27 +157,6 @@ def test_foundry_gpt_6_astra_keeps_sampling_params_when_reasoning_effort_is_none
assert optional_params == {"reasoning_effort": "none", "temperature": 0.2, "top_p": 0.9}
def test_a_gpt_5_name_without_a_foundry_row_keeps_reading_its_own_entry(
monkeypatch: pytest.MonkeyPatch, _local_model_cost_map
):
"""Most gpt-5-family names have no azure_ai/ row. Reading an azure_ai/ key for those finds
nothing, and an openai.azure.com base sends the name down the azure provider, which has no key
for it either, so every effort answer would silently fall back to false and take temperature,
top_p and logprobs down with it."""
monkeypatch.setenv("AZURE_AI_API_BASE", "https://example-resource.openai.azure.com")
monkeypatch.setenv("AZURE_AI_API_KEY", "placeholder")
optional_params = litellm.utils.get_optional_params(
model="gpt-5.1-chat-latest",
custom_llm_provider="azure_ai",
temperature=0.2,
top_p=0.9,
logprobs=True,
)
assert optional_params["temperature"] == 0.2
assert optional_params["top_p"] == 0.9
assert optional_params["logprobs"] is True
def test_azure_ai_grok_stop_parameter_handling():

View file

@ -1695,21 +1695,6 @@ def test_nim_vllm_extras_translated_end_to_end_in_request_body():
assert request_body["top_k"] == 40
def test_in_schema_unsupported_params_still_raise():
with pytest.raises(litellm.UnsupportedParamsError):
litellm.get_optional_params(
model="accounts/fireworks/models/llama-v3-70b-instruct",
custom_llm_provider="fireworks_ai",
drop_params=False,
store=True,
)
optional_params = litellm.get_optional_params(
model="accounts/fireworks/models/llama-v3-70b-instruct",
custom_llm_provider="fireworks_ai",
drop_params=True,
store=True,
)
assert "store" not in optional_params
def test_streaming_preserves_selected_model_for_private_accounting():

View file

@ -715,7 +715,7 @@ class TestMoonshotReasoningEffort:
def force_local_model_cost(self, monkeypatch):
monkeypatch.setattr(litellm, "model_cost", GetModelCostMap.load_local_model_cost_map())
@pytest.mark.parametrize("model", ["kimi-k3", "kimi-k2.5", "kimi-k2.6", "kimi-k2-thinking"])
@pytest.mark.parametrize("model", ["kimi-k3", "kimi-k2.5", "kimi-k2.6"])
def test_reasoning_model_supports_reasoning_effort(self, model):
assert "reasoning_effort" in MoonshotChatConfig().get_supported_openai_params(model)