scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-06 08:16:43 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 78

Merge e04c9a12b9 into 3a52ae0a3f

Browse source
This commit is contained in:
limdeng 2026-08-29 01:10:06 +08:00 • committed by GitHub
commit 5c4e5a1db9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 1054 additions and 8 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

274
litellm/cost_calculator.py
View file

@ -2,8 +2,9 @@
## File for 'response_cost' calculation in Logging
import logging
import time
from collections.abc import Sequence
from collections.abc import Mapping, Sequence
from functools import lru_cache
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final, Literal, cast
from httpx import Response
@ -239,6 +240,259 @@ def _cost_per_token_custom_pricing_helper(
return None
def _litellm_params_as_mapping(litellm_params: object | None) -> Mapping[str, object] | None:
if litellm_params is None:
return None
if isinstance(litellm_params, Mapping):
return litellm_params
dump: Final = getattr(litellm_params, "model_dump", None)
if not callable(dump):
return None
dumped: Final = dump()
if not isinstance(dumped, Mapping):
return None
return dumped
def _model_info_from_params(params: Mapping[str, object], metadata_key: str) -> Mapping[str, object] | None:
metadata: Final = params.get(metadata_key)
if not isinstance(metadata, Mapping):
return None
return _litellm_params_as_mapping(metadata.get("model_info"))
def _as_token_rate(value: object) -> float | None:
if isinstance(value, bool) or value is None:
return None
if isinstance(value, (int, float)):
return float(value)
return None
def _custom_rates_from_mapping(source: Mapping[str, object] | None) -> Mapping[str, float] | None:
if source is None:
return None
input_cost: Final = source.get("input_cost_per_token")
output_cost: Final = source.get("output_cost_per_token")
if input_cost is None and output_cost is None:
return None
cache_read: Final = source.get("cache_read_input_token_cost")
cache_creation: Final = source.get("cache_creation_input_token_cost")
pairs: Final = (
("input_cost_per_token", input_cost),
("output_cost_per_token", output_cost),
("cache_read_input_token_cost", cache_read),
("cache_creation_input_token_cost", cache_creation),
)
return MappingProxyType({key: rate for key, value in pairs if (rate := _as_token_rate(value)) is not None})
def extract_custom_cost_per_token(
litellm_params: object | None,
) -> Mapping[str, float] | None:
"""Return deployment token rates from litellm_params when input and/or output is set.
Rates may sit on litellm_params itself (UI / model_list) or under
metadata.model_info / litellm_metadata.model_info (/v1/messages, /v1/responses).
One-sided rates are returned as-is; callers that need a complete CostPerToken
fill the missing side from the published price map.
Optional cache rates are copied when present so the custom-pricing helper can
apply them instead of falling back to the input rate.
"""
params: Final = _litellm_params_as_mapping(litellm_params)
if params is None:
return None
return (
_custom_rates_from_mapping(params)
or _custom_rates_from_mapping(_model_info_from_params(params, "metadata"))
or _custom_rates_from_mapping(_model_info_from_params(params, "litellm_metadata"))
)
def _published_model_info(
model: str | None,
custom_llm_provider: str | None,
) -> Mapping[str, object] | None:
if not model:
return None
try:
info: Final = litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider)
except Exception: # noqa: BLE001 # get_model_info raises Exception for unmapped models
return None
return MappingProxyType({str(key): value for key, value in info.items()})
def _rate_from_model_info(info: Mapping[str, object] | None, field: str) -> float | None:
if info is None:
return None
return _as_token_rate(info.get(field))
def _published_token_rate(
model: str | None,
custom_llm_provider: str | None,
field: str,
) -> float | None:
return _rate_from_model_info(_published_model_info(model, custom_llm_provider), field)
def _unique_model_names(*names: str | None) -> tuple[str, ...]:
return tuple(
dict.fromkeys(
part
for name in names
if isinstance(name, str) and name
for part in ((name,) if "/" not in name else (name, name.split("/", 1)[1]))
if part
)
)
def _cost_map_rate(key: str | None, field: str) -> float | None:
if not key:
return None
raw: Final = litellm.model_cost.get(key)
if not isinstance(raw, Mapping):
return None
return _as_token_rate(raw.get(field))
def _declared_token_rate(
model: str | None,
custom_llm_provider: str | None,
field: str,
) -> float | None:
"""Return a price-map rate that was actually declared on the entry.
``get_model_info`` synthesizes ``input_cost_per_token`` / ``output_cost_per_token``
to 0 when they are missing. A custom ``router_model_id`` entry typically has
only those two fields; treating the zeros or missing cache keys as published
would skip the backend model that does have cache-specific rates.
"""
if not model:
return None
from_map: Final = _cost_map_rate(model, field)
if from_map is not None:
return from_map
if custom_llm_provider:
from_prefixed: Final = _cost_map_rate(f"{custom_llm_provider}/{model}", field)
if from_prefixed is not None:
return from_prefixed
info: Final = _published_model_info(model, custom_llm_provider)
if info is None:
return None
info_key: Final = info.get("key")
from_resolved: Final = _cost_map_rate(info_key if isinstance(info_key, str) else None, field)
if from_resolved is not None:
return from_resolved
if field in ("input_cost_per_token", "output_cost_per_token"):
return None
return _rate_from_model_info(info, field)
def _first_declared_token_rate(
models: Sequence[str | None],
custom_llm_provider: str | None,
field: str,
) -> float | None:
for candidate in _unique_model_names(*models):
rate = _declared_token_rate(candidate, custom_llm_provider, field)
if rate is not None:
return rate
return None
def _complete_custom_cost_per_token(
rates: Mapping[str, float] | None,
*,
model: str | None,
custom_llm_provider: str | None,
fallback_models: Sequence[str | None] = (),
) -> CostPerToken | None:
"""Fill missing sides of a partial custom CostPerToken from declared price-map rates.
``model`` is often a custom ``router_model_id`` that only stores input/output.
``fallback_models`` should include the backend model so cache-specific rates
come from that published entry instead of the normal input rate.
"""
if rates is None:
return None
input_cost: Final = rates.get("input_cost_per_token")
output_cost: Final = rates.get("output_cost_per_token")
if input_cost is None and output_cost is None:
return None
lookup_models: Final = (model, *fallback_models)
resolved_input: Final = (
float(input_cost)
if input_cost is not None
else (_first_declared_token_rate(lookup_models, custom_llm_provider, "input_cost_per_token") or 0.0)
)
resolved_output: Final = (
float(output_cost)
if output_cost is not None
else (_first_declared_token_rate(lookup_models, custom_llm_provider, "output_cost_per_token") or 0.0)
)
cache_read: Final = rates.get("cache_read_input_token_cost")
cache_creation: Final = rates.get("cache_creation_input_token_cost")
published_cache_read: Final = _first_declared_token_rate(
lookup_models, custom_llm_provider, "cache_read_input_token_cost"
)
published_cache_creation: Final = _first_declared_token_rate(
lookup_models, custom_llm_provider, "cache_creation_input_token_cost"
)
completed: Final[CostPerToken] = {
"input_cost_per_token": resolved_input,
"output_cost_per_token": resolved_output,
"cache_read_input_token_cost": (
float(cache_read)
if cache_read is not None
else (published_cache_read if published_cache_read is not None else resolved_input)
),
"cache_creation_input_token_cost": (
float(cache_creation)
if cache_creation is not None
else (published_cache_creation if published_cache_creation is not None else resolved_input)
),
}
return completed
def _custom_cost_per_token_from_logging_obj(
litellm_logging_obj: LitellmLoggingObject | None,
) -> Mapping[str, float] | None:
if litellm_logging_obj is None:
return None
from_attr: Final = extract_custom_cost_per_token(getattr(litellm_logging_obj, "litellm_params", None))
if from_attr is not None:
return from_attr
details: Final = getattr(litellm_logging_obj, "model_call_details", None)
nested: Final = details.get("litellm_params") if isinstance(details, Mapping) else None
return extract_custom_cost_per_token(nested)
def _backend_model_from_logging_obj(
litellm_logging_obj: LitellmLoggingObject | None,
) -> str | None:
if litellm_logging_obj is None:
return None
attr_params: Final = _litellm_params_as_mapping(getattr(litellm_logging_obj, "litellm_params", None))
if attr_params is not None:
attr_model: Final = attr_params.get("model")
if isinstance(attr_model, str) and attr_model:
return attr_model
details: Final = getattr(litellm_logging_obj, "model_call_details", None)
nested: Final = details.get("litellm_params") if isinstance(details, Mapping) else None
nested_params: Final = _litellm_params_as_mapping(nested)
if nested_params is not None:
nested_model: Final = nested_params.get("model")
if isinstance(nested_model, str) and nested_model:
return nested_model
logging_model: Final = getattr(litellm_logging_obj, "model", None)
if isinstance(logging_model, str) and logging_model:
return logging_model
return None
def _get_additional_costs(
model: str,
custom_llm_provider: str | None,
@ -1268,6 +1522,22 @@ def completion_cost(
if model is not None:
potential_model_names.append(model)
resolved_custom_cost_per_token: Final = (
custom_cost_per_token
if custom_cost_per_token is not None
else _complete_custom_cost_per_token(
_custom_cost_per_token_from_logging_obj(litellm_logging_obj),
model=selected_model,
custom_llm_provider=custom_llm_provider if isinstance(custom_llm_provider, str) else None,
fallback_models=(
model if isinstance(model, str) else None,
_get_response_model(completion_response),
base_model,
_backend_model_from_logging_obj(litellm_logging_obj),
),
)
)
for idx, model in enumerate(potential_model_names):
try:
if verbose_logger.isEnabledFor(logging.DEBUG):
@ -1616,7 +1886,7 @@ def completion_cost(
response_time_ms=total_time,
region_name=region_name,
custom_cost_per_second=custom_cost_per_second,
custom_cost_per_token=custom_cost_per_token,
custom_cost_per_token=resolved_custom_cost_per_token,
prompt_characters=prompt_characters,
completion_characters=completion_characters,
cache_creation_input_tokens=cache_creation_input_tokens,

9
litellm/proxy/litellm_pre_call_utils.py
View file

@ -538,6 +538,15 @@ def _strip_client_pricing_overrides(data: dict[str, Any]) -> None:
)
def strip_unauthorized_client_pricing(
data: dict[str, Any], # mutable-ok: in-place strip of the caller request body
user_api_key_dict: UserAPIKeyAuth,
) -> None:
"""Drop client pricing overrides unless the key or team allows them."""
if not _key_or_team_allows_client_pricing_override(user_api_key_dict):
_strip_client_pricing_overrides(data)
def _get_metadata_variable_name(request: Request) -> str:
"""
Helper to return what the "metadata" field should be called in the request data

1
litellm/proxy/pass_through_endpoints/llm_provider_handlers/anthropic_passthrough_logging_handler.py
View file

@ -287,6 +287,7 @@ class AnthropicPassthroughLoggingHandler:
custom_llm_provider=custom_llm_provider,
custom_pricing=custom_pricing,
router_model_id=router_model_id,
litellm_logging_obj=logging_obj,
)
)

6
litellm/proxy/pass_through_endpoints/pass_through_endpoints.py
View file

@ -77,7 +77,10 @@ from litellm.proxy.common_utils.http_parsing_utils import (
from litellm.proxy.common_utils.sse_keepalive import (
wrap_passthrough_sse_bytes_with_keepalive_pings,
)
from litellm.proxy.litellm_pre_call_utils import LiteLLMProxyRequestSetup
from litellm.proxy.litellm_pre_call_utils import (
LiteLLMProxyRequestSetup,
strip_unauthorized_client_pricing,
)
from litellm.proxy.utils import normalize_route_for_root_path
from litellm.repositories.team_repository import TeamRepository
from litellm.secret_managers.main import get_secret_str
@ -549,6 +552,7 @@ class HttpPassThroughEndpointHelpers(BasePassthroughUtils):
from litellm.types.utils import all_litellm_params
_parsed_body = _parsed_body or {}
strip_unauthorized_client_pricing(_parsed_body, user_api_key_dict)
litellm_params_in_body: Final = {}
for k in all_litellm_params:

118
tests/pass_through_unit_tests/test_pass_through_unit_tests.py
View file

@ -11,6 +11,7 @@ import httpx
import pytest
import litellm
from typing import AsyncGenerator
from litellm.cost_calculator import extract_custom_cost_per_token
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.types.passthrough_endpoints.pass_through_endpoints import EndpointType
from litellm.proxy.pass_through_endpoints.success_handler import (
@ -235,6 +236,113 @@ def test_init_kwargs_with_litellm_metadata(mock_request, mock_user_api_key_dict)
assert metadata["user_api_key"] == "test-key"
def _passthrough_logging_obj():
return LiteLLMLoggingObj(
model="test-model",
messages=[],
stream=False,
call_type="test-call-type",
start_time=datetime.now(),
litellm_call_id="test-call-id",
function_id="test-function-id",
)
def test_init_kwargs_strips_client_token_rates(mock_request, mock_user_api_key_dict):
"""Client-supplied 0 rates must not land in litellm_params (budget bypass)."""
request = mock_request()
parsed_body = {
"model": "claude-sonnet-4-5-20250929",
"input_cost_per_token": 0.0,
"output_cost_per_token": 0.0,
"messages": [{"role": "user", "content": "hi"}],
}
passthrough_payload = PassthroughStandardLoggingPayload(
url="https://test.com",
request_body={},
)
result = HttpPassThroughEndpointHelpers._init_kwargs_for_pass_through_endpoint(
request=request,
user_api_key_dict=mock_user_api_key_dict,
passthrough_logging_payload=passthrough_payload,
_parsed_body=parsed_body,
litellm_call_id="test-call-id",
logging_obj=_passthrough_logging_obj(),
)
assert "input_cost_per_token" not in result["litellm_params"]
assert "output_cost_per_token" not in result["litellm_params"]
assert extract_custom_cost_per_token(result["litellm_params"]) is None
def test_init_kwargs_strips_client_model_info_pricing(
mock_request, mock_user_api_key_dict
):
request = mock_request()
parsed_body = {
"litellm_metadata": {
"tags": ["keep-me"],
"model_info": {
"input_cost_per_token": 0.0,
"output_cost_per_token": 0.0,
},
}
}
passthrough_payload = PassthroughStandardLoggingPayload(
url="https://test.com",
request_body={},
)
result = HttpPassThroughEndpointHelpers._init_kwargs_for_pass_through_endpoint(
request=request,
user_api_key_dict=mock_user_api_key_dict,
passthrough_logging_payload=passthrough_payload,
_parsed_body=parsed_body,
litellm_call_id="test-call-id",
logging_obj=_passthrough_logging_obj(),
)
metadata = result["litellm_params"]["metadata"]
assert metadata["tags"] == ["keep-me"]
assert "model_info" not in metadata
def test_init_kwargs_keeps_client_pricing_when_key_allows_override(mock_request):
request = mock_request()
user_api_key_dict = UserAPIKeyAuth(
api_key="test-key",
user_id="test-user",
team_id="test-team",
end_user_id="test-user",
metadata={"allow_client_pricing_override": True},
)
parsed_body = {
"input_cost_per_token": 0.0,
"output_cost_per_token": 0.0,
}
passthrough_payload = PassthroughStandardLoggingPayload(
url="https://test.com",
request_body={},
)
result = HttpPassThroughEndpointHelpers._init_kwargs_for_pass_through_endpoint(
request=request,
user_api_key_dict=user_api_key_dict,
passthrough_logging_payload=passthrough_payload,
_parsed_body=parsed_body,
litellm_call_id="test-call-id",
logging_obj=_passthrough_logging_obj(),
)
assert result["litellm_params"]["input_cost_per_token"] == 0.0
assert result["litellm_params"]["output_cost_per_token"] == 0.0
assert extract_custom_cost_per_token(result["litellm_params"]) == {
"input_cost_per_token": 0.0,
"output_cost_per_token": 0.0,
}
def test_init_kwargs_with_tags_in_header(mock_request, mock_user_api_key_dict):
"""
Tags should be added to metadata if they exist in headers
@ -574,11 +682,13 @@ def test_init_kwargs_filters_pricing_params(mock_request, mock_user_api_key_dict
assert parsed_body["temperature"] == 0.7
assert parsed_body["max_tokens"] == 100
# Verify pricing parameters are stored in litellm_params for internal use
# Unauthorized keys must not keep client rates in litellm_params; otherwise
# extract_custom_cost_per_token would bill from the request body (budget bypass).
# Authorized keys are covered by test_init_kwargs_keeps_client_pricing_when_key_allows_override.
litellm_params = result["litellm_params"]
assert litellm_params["input_cost_per_token"] == 0.00002
assert litellm_params["output_cost_per_token"] == 0.00002
# Note: Other pricing params are also stored but we test the key ones that caused the regression
assert "input_cost_per_token" not in litellm_params
assert "output_cost_per_token" not in litellm_params
assert extract_custom_cost_per_token(litellm_params) is None
def test_custom_pricing_used_in_cost_calculation():

50
tests/test_litellm/litellm_core_utils/test_litellm_logging.py
View file

@ -17,7 +17,7 @@ from litellm.constants import SENTRY_DENYLIST, SENTRY_PII_DENYLIST
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.litellm_logging import Logging as LitellmLogging
from litellm.litellm_core_utils.litellm_logging import set_callbacks
from litellm.types.utils import ModelResponse, TextCompletionResponse
from litellm.types.utils import ModelResponse, TextCompletionResponse, Usage
@pytest.fixture
@ -278,6 +278,54 @@ def test_response_cost_calculator_uses_router_model_id_from_litellm_metadata():
litellm.model_cost.pop(custom_model_id, None)
def test_response_cost_calculator_unknown_anthropic_model_uses_litellm_params_rates():
"""Native /v1/messages cost calc should apply deployment rates for an
unmapped anthropic model. Do not register_model — that is the
completions-only workaround and is not the /messages path.
Regression for #25204.
"""
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
unknown_model = "litellm-unmapped-custom-priced-qwen"
input_cost = 1.2e-05
output_cost = 3.6e-05
logging_obj = LiteLLMLoggingObj(
model=unknown_model,
messages=[{"role": "user", "content": "Hi"}],
stream=False,
call_type="anthropic_messages",
start_time=time.time(),
litellm_call_id="test-messages-custom-pricing",
function_id="test-fn",
)
logging_obj.update_environment_variables(
model=unknown_model,
user="",
optional_params={},
litellm_params={
"custom_llm_provider": "anthropic",
"input_cost_per_token": input_cost,
"output_cost_per_token": output_cost,
},
)
logging_obj.model_call_details["custom_llm_provider"] = "anthropic"
response_obj = ModelResponse(
id="msg_test",
model=unknown_model,
choices=[],
usage=Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120),
)
cost = logging_obj._response_cost_calculator(result=response_obj)
assert cost is not None
expected_cost = (100 * input_cost) + (20 * output_cost)
assert cost == pytest.approx(expected_cost)
assert cost > 0
class TestGetRouterModelId:
"""Tests for the get_router_model_id helper method."""

604
tests/test_litellm/test_cost_calculator.py
View file

@ -11,14 +11,19 @@ import litellm
from litellm.cost_calculator import (
BaseTokenUsageProcessor,
RealtimeAPITokenUsageProcessor,
_complete_custom_cost_per_token,
_custom_cost_per_token_from_logging_obj,
_published_token_rate,
completion_cost,
cost_per_token,
extract_custom_cost_per_token,
handle_realtime_stream_cost_calculation,
response_cost_calculator,
)
from litellm.types.llms.openai import OpenAIRealtimeStreamList
from litellm.types.utils import (
CacheCreationTokenDetails,
CustomPricingLiteLLMParams,
ModelInfo,
ModelResponse,
PromptTokensDetailsWrapper,
@ -978,6 +983,605 @@ def test_custom_pricing_cost_calc_uses_router_model_id_from_litellm_metadata():
assert custom_model_id not in (selected_model_no_custom or "")
def test_extract_custom_cost_per_token_from_litellm_params_and_model_info():
assert extract_custom_cost_per_token(None) is None
assert extract_custom_cost_per_token({"custom_llm_provider": "anthropic"}) is None
assert extract_custom_cost_per_token({"input_cost_per_token": 1.2e-05}) == {
"input_cost_per_token": 1.2e-05,
}
assert extract_custom_cost_per_token({"output_cost_per_token": 3.6e-05}) == {
"output_cost_per_token": 3.6e-05,
}
assert extract_custom_cost_per_token(
{
"input_cost_per_token": 1.2e-05,
"output_cost_per_token": 3.6e-05,
"cache_read_input_token_cost": 1.2e-06,
}
) == {
"input_cost_per_token": 1.2e-05,
"output_cost_per_token": 3.6e-05,
"cache_read_input_token_cost": 1.2e-06,
}
assert extract_custom_cost_per_token(
{
"metadata": {
"model_info": {
"id": "deploy-meta",
"input_cost_per_token": 0.0002,
"output_cost_per_token": 0.0008,
},
},
}
) == {
"input_cost_per_token": 0.0002,
"output_cost_per_token": 0.0008,
}
assert extract_custom_cost_per_token(
{
"litellm_metadata": {
"model_info": {
"id": "deploy-1",
"input_cost_per_token": 0.0003,
"output_cost_per_token": 0.0015,
},
},
}
) == {
"input_cost_per_token": 0.0003,
"output_cost_per_token": 0.0015,
}
def test_extract_custom_cost_per_token_from_pydantic_params():
both_sides = CustomPricingLiteLLMParams(
input_cost_per_token=1.2e-05,
output_cost_per_token=3.6e-05,
)
assert extract_custom_cost_per_token(both_sides) == {
"input_cost_per_token": 1.2e-05,
"output_cost_per_token": 3.6e-05,
}
input_only = CustomPricingLiteLLMParams(input_cost_per_token=1.2e-05)
assert extract_custom_cost_per_token(input_only) == {
"input_cost_per_token": 1.2e-05,
}
def test_extract_custom_cost_per_token_rejects_non_mapping_sources():
assert extract_custom_cost_per_token("not-params") is None
assert extract_custom_cost_per_token([1, 2]) is None
class _UncallableDump:
model_dump = "not-callable"
assert extract_custom_cost_per_token(_UncallableDump()) is None
class _NonDictDump:
def model_dump(self):
return ["not", "a", "mapping"]
assert extract_custom_cost_per_token(_NonDictDump()) is None
assert extract_custom_cost_per_token({"metadata": "not-a-dict"}) is None
assert extract_custom_cost_per_token({"metadata": {"model_info": "x"}}) is None
def test_complete_custom_cost_per_token_defensive_branches(_local_model_cost_map):
assert _complete_custom_cost_per_token(None, model="gpt-4o-mini", custom_llm_provider="openai") is None
assert _complete_custom_cost_per_token({}, model="gpt-4o-mini", custom_llm_provider="openai") is None
assert _custom_cost_per_token_from_logging_obj(None) is None
assert _published_token_rate(None, "openai", "input_cost_per_token") is None
assert _published_token_rate("", "openai", "output_cost_per_token") is None
assert _published_token_rate("gpt-4o-mini", "openai", "this_field_does_not_exist") is None
assert _published_token_rate(
"litellm-unmapped-custom-priced-qwen", "anthropic", "input_cost_per_token"
) is None
def test_complete_output_only_keeps_published_cache_rates(_local_model_cost_map):
"""Output-only custom pricing must not bill cache at the normal input rate."""
model = "claude-sonnet-4-5-20250929"
published = litellm.get_model_info(model=model, custom_llm_provider="anthropic")
custom_output = 5e-06
assert published["cache_read_input_token_cost"] != published["input_cost_per_token"]
assert published["cache_creation_input_token_cost"] != published["input_cost_per_token"]
completed = _complete_custom_cost_per_token(
{"output_cost_per_token": custom_output},
model=model,
custom_llm_provider="anthropic",
)
assert completed is not None
assert completed["output_cost_per_token"] == custom_output
assert completed["input_cost_per_token"] == published["input_cost_per_token"]
assert completed["cache_read_input_token_cost"] == published["cache_read_input_token_cost"]
assert completed["cache_creation_input_token_cost"] == published["cache_creation_input_token_cost"]
def test_completion_cost_output_only_custom_rate_uses_published_cache_rates(
_local_model_cost_map,
):
import time
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
model = "claude-sonnet-4-5-20250929"
published = litellm.get_model_info(model=model, custom_llm_provider="anthropic")
custom_output = 5e-06
regular_prompt = 20
cache_read = 80
completion_tokens = 10
logging_obj = LiteLLMLoggingObj(
model=model,
messages=[{"role": "user", "content": "Hi"}],
stream=False,
call_type="anthropic_messages",
start_time=time.time(),
litellm_call_id="test-output-only-cache-rates",
function_id="test-fn",
)
logging_obj.update_environment_variables(
model=model,
user="",
optional_params={},
litellm_params={
"custom_llm_provider": "anthropic",
"output_cost_per_token": custom_output,
},
)
logging_obj.model_call_details["custom_llm_provider"] = "anthropic"
response = ModelResponse(
id="test-id",
model=model,
choices=[],
usage=Usage(
prompt_tokens=regular_prompt + cache_read,
completion_tokens=completion_tokens,
total_tokens=regular_prompt + cache_read + completion_tokens,
prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=cache_read),
),
)
cost = completion_cost(
completion_response=response,
model=model,
custom_llm_provider="anthropic",
call_type="anthropic_messages",
litellm_logging_obj=logging_obj,
)
expected = (
regular_prompt * published["input_cost_per_token"]
+ cache_read * published["cache_read_input_token_cost"]
+ completion_tokens * custom_output
)
assert cost == pytest.approx(expected)
billed_cache_at_input = (
regular_prompt * published["input_cost_per_token"]
+ cache_read * published["input_cost_per_token"]
+ completion_tokens * custom_output
)
assert cost != pytest.approx(billed_cache_at_input)
def test_complete_output_only_router_id_uses_backend_cache_rates(_local_model_cost_map):
"""A custom router_model_id usually stores only input/output. Missing cache
rates must come from the backend Anthropic model, not the normal input rate.
"""
backend = "claude-sonnet-4-5-20250929"
router_id = "71ad2e1c-71db-4246-a558-d01480578941"
published = litellm.get_model_info(model=backend, custom_llm_provider="anthropic")
custom_output = 5e-06
litellm.register_model(
{
router_id: {
"input_cost_per_token": 1e-06,
"output_cost_per_token": custom_output,
"litellm_provider": "anthropic",
"mode": "chat",
}
},
persist_across_reloads=False,
)
assert litellm.model_cost[router_id].get("cache_read_input_token_cost") is None
assert published["cache_read_input_token_cost"] != published["input_cost_per_token"]
without_backend = _complete_custom_cost_per_token(
{"output_cost_per_token": custom_output},
model=f"anthropic/{router_id}",
custom_llm_provider="anthropic",
)
assert without_backend is not None
assert without_backend["cache_read_input_token_cost"] == without_backend["input_cost_per_token"]
completed = _complete_custom_cost_per_token(
{"output_cost_per_token": custom_output},
model=f"anthropic/{router_id}",
custom_llm_provider="anthropic",
fallback_models=(backend,),
)
assert completed is not None
assert completed["output_cost_per_token"] == custom_output
assert completed["input_cost_per_token"] == 1e-06
assert completed["cache_read_input_token_cost"] == published["cache_read_input_token_cost"]
assert completed["cache_creation_input_token_cost"] == published["cache_creation_input_token_cost"]
def test_completion_cost_router_id_uses_backend_cache_rates(_local_model_cost_map):
import time
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
backend = "claude-sonnet-4-5-20250929"
router_id = "test-router-custom-cache-uuid"
published = litellm.get_model_info(model=backend, custom_llm_provider="anthropic")
custom_input = 1e-06
custom_output = 5e-06
regular_prompt = 20
cache_read = 80
completion_tokens = 10
litellm.register_model(
{
router_id: {
"input_cost_per_token": custom_input,
"output_cost_per_token": custom_output,
"litellm_provider": "anthropic",
"mode": "chat",
}
},
persist_across_reloads=False,
)
logging_obj = LiteLLMLoggingObj(
model=backend,
messages=[{"role": "user", "content": "Hi"}],
stream=False,
call_type="anthropic_messages",
start_time=time.time(),
litellm_call_id="test-router-id-cache-rates",
function_id="test-fn",
)
logging_obj.update_environment_variables(
model=backend,
user="",
optional_params={},
litellm_params={
"model": backend,
"custom_llm_provider": "anthropic",
"input_cost_per_token": custom_input,
"output_cost_per_token": custom_output,
},
)
logging_obj.model_call_details["custom_llm_provider"] = "anthropic"
response = ModelResponse(
id="test-id",
model=backend,
choices=[],
usage=Usage(
prompt_tokens=regular_prompt + cache_read,
completion_tokens=completion_tokens,
total_tokens=regular_prompt + cache_read + completion_tokens,
prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=cache_read),
),
)
cost = completion_cost(
completion_response=response,
model=backend,
custom_llm_provider="anthropic",
call_type="anthropic_messages",
custom_pricing=True,
router_model_id=router_id,
litellm_logging_obj=logging_obj,
)
expected = (
regular_prompt * custom_input
+ cache_read * published["cache_read_input_token_cost"]
+ completion_tokens * custom_output
)
billed_cache_at_custom_input = (
regular_prompt * custom_input + cache_read * custom_input + completion_tokens * custom_output
)
assert cost == pytest.approx(expected)
assert cost != pytest.approx(billed_cache_at_custom_input)
def test_completion_cost_unknown_anthropic_model_uses_litellm_params_rates():
"""Unknown anthropic models logged $0 on /v1/messages even when the
deployment set input/output rates in litellm_params.
The public price map has no entry, so provider dispatch must not run
before custom_cost_per_token is applied. Regression for #25204.
"""
import time
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
unknown_model = "litellm-unmapped-custom-priced-qwen"
input_cost = 1.2e-05
output_cost = 3.6e-05
prompt_tokens = 100
completion_tokens = 20
assert unknown_model not in litellm.model_cost
assert f"anthropic/{unknown_model}" not in litellm.model_cost
logging_obj = LiteLLMLoggingObj(
model=unknown_model,
messages=[{"role": "user", "content": "Hi"}],
stream=False,
call_type="anthropic_messages",
start_time=time.time(),
litellm_call_id="test-unmapped-custom-pricing",
function_id="test-fn",
)
logging_obj.update_environment_variables(
model=unknown_model,
user="",
optional_params={},
litellm_params={
"custom_llm_provider": "anthropic",
"input_cost_per_token": input_cost,
"output_cost_per_token": output_cost,
},
)
logging_obj.model_call_details["custom_llm_provider"] = "anthropic"
response = ModelResponse(
id="test-id",
model=unknown_model,
choices=[],
usage=Usage(
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
),
)
cost = completion_cost(
completion_response=response,
model=unknown_model,
custom_llm_provider="anthropic",
call_type="anthropic_messages",
custom_pricing=True,
litellm_logging_obj=logging_obj,
)
expected = prompt_tokens * input_cost + completion_tokens * output_cost
assert cost == pytest.approx(expected)
assert cost > 0
def test_anthropic_passthrough_unknown_model_spend_uses_litellm_params_rates():
"""Passthrough /v1/messages must pass the logging object into
completion_cost so unmapped models pick up deployment rates.
Regression for #25204.
"""
import time
from datetime import datetime
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
from litellm.proxy.pass_through_endpoints.llm_provider_handlers.anthropic_passthrough_logging_handler import (
AnthropicPassthroughLoggingHandler,
)
unknown_model = "litellm-unmapped-custom-priced-qwen"
input_cost = 1.2e-05
output_cost = 3.6e-05
logging_obj = LiteLLMLoggingObj(
model=unknown_model,
messages=[{"role": "user", "content": "Hi"}],
stream=False,
call_type="anthropic_messages",
start_time=time.time(),
litellm_call_id="test-passthrough-custom-pricing",
function_id="test-fn",
)
logging_obj.update_environment_variables(
model=unknown_model,
user="",
optional_params={},
litellm_params={
"custom_llm_provider": "anthropic",
"input_cost_per_token": input_cost,
"output_cost_per_token": output_cost,
},
)
logging_obj.model_call_details["custom_llm_provider"] = "anthropic"
response = ModelResponse(
id="msg_test",
model=unknown_model,
choices=[],
usage=Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120),
)
kwargs = AnthropicPassthroughLoggingHandler._create_anthropic_response_logging_payload(
litellm_model_response=response,
model=unknown_model,
kwargs={},
start_time=datetime.now(),
end_time=datetime.now(),
logging_obj=logging_obj,
)
expected = 100 * input_cost + 20 * output_cost
assert kwargs["response_cost"] == pytest.approx(expected)
assert logging_obj.model_call_details["response_cost"] == pytest.approx(
expected
)
assert kwargs["response_cost"] > 0
@pytest.mark.parametrize(
"declared",
[
{"input_cost_per_token": 1e-06},
{"output_cost_per_token": 5e-06},
],
ids=["input-only", "output-only"],
)
def test_completion_cost_one_sided_custom_rate_keeps_published_other_side(
_local_model_cost_map, declared
):
"""A deployment may configure only one direction.
The missing side must keep the published price-map rate, not 0.
"""
import time
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
model = "gpt-4o-mini"
published = litellm.get_model_info(model=model)
prompt_tokens = 100
completion_tokens = 20
input_cost = declared.get(
"input_cost_per_token", published["input_cost_per_token"]
)
output_cost = declared.get(
"output_cost_per_token", published["output_cost_per_token"]
)
logging_obj = LiteLLMLoggingObj(
model=model,
messages=[{"role": "user", "content": "Hi"}],
stream=False,
call_type="completion",
start_time=time.time(),
litellm_call_id="test-one-sided-custom-pricing",
function_id="test-fn",
)
logging_obj.update_environment_variables(
model=model,
user="",
optional_params={},
litellm_params={"custom_llm_provider": "openai", **declared},
)
logging_obj.model_call_details["custom_llm_provider"] = "openai"
response = ModelResponse(
id="test-id",
model=model,
choices=[],
usage=Usage(
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
),
)
cost = completion_cost(
completion_response=response,
model=model,
custom_llm_provider="openai",
litellm_logging_obj=logging_obj,
)
expected = prompt_tokens * input_cost + completion_tokens * output_cost
assert cost == pytest.approx(expected)
def test_completion_cost_one_sided_unknown_model_uses_zero_for_missing_side():
"""Unmapped models have no published other-side rate, so that side is 0."""
import time
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
unknown_model = "litellm-unmapped-custom-priced-qwen-onesided"
input_cost = 1.2e-05
prompt_tokens = 100
completion_tokens = 20
logging_obj = LiteLLMLoggingObj(
model=unknown_model,
messages=[{"role": "user", "content": "Hi"}],
stream=False,
call_type="anthropic_messages",
start_time=time.time(),
litellm_call_id="test-unmapped-one-sided",
function_id="test-fn",
)
logging_obj.update_environment_variables(
model=unknown_model,
user="",
optional_params={},
litellm_params={
"custom_llm_provider": "anthropic",
"input_cost_per_token": input_cost,
},
)
logging_obj.model_call_details["custom_llm_provider"] = "anthropic"
response = ModelResponse(
id="test-id",
model=unknown_model,
choices=[],
usage=Usage(
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
),
)
cost = completion_cost(
completion_response=response,
model=unknown_model,
custom_llm_provider="anthropic",
call_type="anthropic_messages",
custom_pricing=True,
litellm_logging_obj=logging_obj,
)
assert cost == pytest.approx(prompt_tokens * input_cost)
def test_completion_cost_reads_nested_litellm_params_from_model_call_details():
import time
from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj
unknown_model = "litellm-unmapped-nested-litellm-params"
input_cost = 1.2e-05
output_cost = 3.6e-05
prompt_tokens = 100
completion_tokens = 20
logging_obj = LiteLLMLoggingObj(
model=unknown_model,
messages=[{"role": "user", "content": "Hi"}],
stream=False,
call_type="anthropic_messages",
start_time=time.time(),
litellm_call_id="test-nested-litellm-params",
function_id="test-fn",
)
logging_obj.litellm_params = None
logging_obj.model_call_details["litellm_params"] = {
"custom_llm_provider": "anthropic",
"input_cost_per_token": input_cost,
"output_cost_per_token": output_cost,
}
logging_obj.model_call_details["custom_llm_provider"] = "anthropic"
response = ModelResponse(
id="test-id",
model=unknown_model,
choices=[],
usage=Usage(
prompt_tokens=prompt_tokens,
completion_tokens=completion_tokens,
total_tokens=prompt_tokens + completion_tokens,
),
)
cost = completion_cost(
completion_response=response,
model=unknown_model,
custom_llm_provider="anthropic",
call_type="anthropic_messages",
custom_pricing=True,
litellm_logging_obj=logging_obj,
)
expected = prompt_tokens * input_cost + completion_tokens * output_cost
assert cost == pytest.approx(expected)
def test_per_request_custom_pricing_with_router():
"""When custom pricing is passed as per-request kwargs (not in model_list),
_select_model_name_for_cost_calc should fall back to the model name