scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-15 09:12:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 311

Merge babe47bafb into a7acb82a39

Browse source
This commit is contained in:
yuneng-jiang 2026-11-03 22:13:09 +00:00 • committed by GitHub
commit 3facced960
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
10 changed files with 292 additions and 173 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

13
litellm/cost_calculator.py
View file

@ -147,7 +147,8 @@ from litellm.utils import (
ProviderConfigManager,
TextCompletionResponse,
TranscriptionResponse,
cached_get_model_info_helper,
get_model_info_for_pricing,
get_model_info_helper_for_pricing,
model_info_with_pricing_overrides,
runtime_model_cost_registrations,
token_counter,
@ -362,7 +363,7 @@ def _per_second_pricing_cost(
audio_seconds: float = 0.0,
) -> tuple[float, float] | None:
try:
model_info: Final = cached_get_model_info_helper(model=model, custom_llm_provider=custom_llm_provider)
model_info: Final = get_model_info_helper_for_pricing(model=model, custom_llm_provider=custom_llm_provider)
except Exception: # noqa: BLE001 # the lookup raises plain Exception for an unmapped model
return None
if _has_token_or_tiered_pricing(model_info) or not _bills_wall_clock_seconds(model_info):
@ -583,7 +584,9 @@ def cost_per_token(
model = model_without_prefix
if is_decisions_call_type(call_type):
decisions_model_info: Final = cached_get_model_info_helper(model=model, custom_llm_provider=custom_llm_provider)
decisions_model_info: Final = get_model_info_helper_for_pricing(
model=model, custom_llm_provider=custom_llm_provider
)
decisions_overrides: Final = decisions_pricing_fields(
litellm.model_cost.get(decisions_model_info["key"], {}) # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType] # model cost map is dynamically loaded
)
@ -787,7 +790,7 @@ def cost_per_token(
service_tier=service_tier,
)
else:
model_info: Final = cached_get_model_info_helper(model=model, custom_llm_provider=custom_llm_provider)
model_info: Final = get_model_info_helper_for_pricing(model=model, custom_llm_provider=custom_llm_provider)
if _has_token_or_tiered_pricing(model_info):
return generic_cost_per_token(
model=model,
@ -2353,7 +2356,7 @@ def _layered_ocr_pricing(*sources: Mapping[str, object] | None) -> OCRPricing:
def _cost_map_model_info(model: str, custom_llm_provider: str | None) -> ModelInfo | None:
try:
return litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider)
return get_model_info_for_pricing(model=model, custom_llm_provider=custom_llm_provider)
except Exception:
return None

6
litellm/litellm_core_utils/get_llm_provider_logic.py
View file

@ -146,6 +146,10 @@ def handle_anthropic_text_model_custom_llm_provider(
return model, custom_llm_provider
def declared_provider(model: str | None, custom_llm_provider: str | None = None) -> str | None:
return custom_llm_provider or (model.split("/", 1)[0] if model and "/" in model else None)
def declared_authenticating_provider(model: str | None, custom_llm_provider: str | None = None) -> str | None:
"""The authenticating provider this pair already names, or None.
@ -154,7 +158,7 @@ def declared_authenticating_provider(model: str | None, custom_llm_provider: str
and for a declared pair the resolver's answer is the declaration itself, so metadata callers
adopt the declaration instead of resolving.
"""
declared: Final = custom_llm_provider or (model.split("/", 1)[0] if model and "/" in model else None)
declared: Final = declared_provider(model, custom_llm_provider)
return declared if declared in PROVIDERS_THAT_AUTHENTICATE_ON_PROVIDER_INFO else None

4
litellm/litellm_core_utils/litellm_logging.py
View file

@ -181,7 +181,7 @@ from litellm.types.utils import (
Usage,
)
from litellm.types.videos.main import VideoObject
from litellm.utils import get_base_model_from_metadata, print_verbose
from litellm.utils import get_base_model_from_metadata, get_model_info_for_pricing, print_verbose
from ..integrations.argilla import ArgillaLogger
from ..integrations.arize.arize_phoenix import ArizePhoenixLogger
@ -6324,7 +6324,7 @@ class StandardLoggingPayloadSetup:
model_cost_information = StandardLoggingModelInformation(model_map_key="", model_map_value=None)
else:
try:
_model_cost_information: Final = litellm.get_model_info(
_model_cost_information: Final = get_model_info_for_pricing(
model=pricing_model_name or model_cost_name,
custom_llm_provider=custom_llm_provider,
api_base=api_base,

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

@ -34,7 +34,7 @@ from litellm.types.utils import (
Usage,
text_tokens_without_nested_reasoning,
)
from litellm.utils import get_model_info
from litellm.utils import get_model_info, get_model_info_for_pricing
# Pre-resolved CallTypes enum values for fast membership checks
_IMAGE_RESPONSE_CALL_TYPES: Final = frozenset(
@ -1385,7 +1385,9 @@ def generic_cost_per_token(
# per-deployment override: those are registered under the deployment id and kept off
# the shared model-name key, so resolving from the name here reads the public rate.
resolved_model_info: Final = apply_provider_cache_read_default(
get_model_info(model=model, custom_llm_provider=custom_llm_provider) if model_info is None else model_info,
get_model_info_for_pricing(model=model, custom_llm_provider=custom_llm_provider)
if model_info is None
else model_info,
custom_llm_provider,
)
@ -1708,7 +1710,7 @@ def get_billed_token_rates(
resolved_model_info: Final = apply_provider_cache_read_default(
model_info
if model_info is not None
else get_model_info(model=model, custom_llm_provider=custom_llm_provider),
else get_model_info_for_pricing(model=model, custom_llm_provider=custom_llm_provider),
custom_llm_provider,
)
except Exception: # noqa: BLE001 # get_model_info raises a bare Exception for an unmapped model: no rates

39
litellm/utils.py
View file

@ -6358,6 +6358,8 @@ def get_model_info_helper(
custom_llm_provider: str | None = None,
api_base: str | None = None,
api_key: str | None = None,
*,
catalog_only: bool = False,
) -> ModelInfoBase:
"""
Helper for 'get_model_info'. Separated out to avoid infinite loop caused by returning 'supported_openai_param's
@ -6393,7 +6395,7 @@ def get_model_info_helper(
model_cost_custom_llm_provider: Final = custom_llm_provider
#########################
provider_config: BaseLLMModelInfo | None = None
if custom_llm_provider and custom_llm_provider in LlmProvidersSet:
if not catalog_only and custom_llm_provider and custom_llm_provider in LlmProvidersSet:
provider_config = ProviderConfigManager.get_provider_model_info(
model=model, provider=LlmProviders(custom_llm_provider)
)
@ -6417,7 +6419,7 @@ def get_model_info_helper(
e,
)
if custom_llm_provider == "huggingface":
if not catalog_only and custom_llm_provider == "huggingface":
max_tokens: Final = _get_max_position_embeddings(model_name=model)
return ModelInfoBase(
key=model,
@ -6894,6 +6896,39 @@ get_model_info.cache_clear = _cached_get_model_info.cache_clear
get_model_info.cache_info = _cached_get_model_info.cache_info
def get_model_info_for_pricing(
model: str, custom_llm_provider: str | None = None, api_base: str | None = None
) -> ModelInfo:
from litellm.litellm_core_utils.get_llm_provider_logic import declared_provider
provider: Final = declared_provider(model, custom_llm_provider)
if provider not in ("ollama", "ollama_chat"):
return get_model_info(model=model, custom_llm_provider=custom_llm_provider, api_base=api_base)
try:
catalog: Final = get_model_info_helper(model=model, custom_llm_provider=provider, catalog_only=True)
except ModelNotMappedError:
return ModelInfo(
key=model.removeprefix(f"{provider}/"),
litellm_provider="ollama",
mode="chat",
input_cost_per_token=0.0,
output_cost_per_token=0.0,
max_tokens=None,
max_input_tokens=None,
max_output_tokens=None,
supported_openai_params=None,
)
return ModelInfo(**catalog, supported_openai_params=None)
def get_model_info_helper_for_pricing(model: str, custom_llm_provider: str | None) -> ModelInfoBase:
from litellm.litellm_core_utils.get_llm_provider_logic import declared_provider
if declared_provider(model, custom_llm_provider) in ("ollama", "ollama_chat"):
return get_model_info_for_pricing(model=model, custom_llm_provider=custom_llm_provider)
return cached_get_model_info_helper(model=model, custom_llm_provider=custom_llm_provider)
def json_schema_type(python_type_name: str):
"""Converts standard python types to json schema types

193
tests/unit/litellm_core_utils/llm_cost_calc/test_llm_cost_calc_utils.py
View file

@ -324,20 +324,15 @@ def test_reasoning_tokens_gemini(_local_model_cost_map):
)
def test_image_tokens_with_custom_pricing():
def test_image_tokens_with_custom_pricing() -> None:
"""Test that image_tokens in completion are properly costed with output_cost_per_image_token."""
from unittest.mock import patch
# Mock model info with image token pricing
mock_model_info = {
model_info: Final = {
"input_cost_per_token": 1e-6,
"output_cost_per_token": 2e-6,
"output_cost_per_image_token": 5e-6, # Custom pricing for image tokens in output
}
usage = Usage(
usage: Final = Usage(
completion_tokens=1720, # text_tokens (600) + image_tokens (1120)
prompt_tokens=14,
total_tokens=1734,
@ -354,36 +349,32 @@ def test_image_tokens_with_custom_pricing():
),
)
with patch(
"litellm.litellm_core_utils.llm_cost_calc.utils.get_model_info",
return_value=mock_model_info,
):
prompt_cost, completion_cost = generic_cost_per_token(
model="test-model", usage=usage, custom_llm_provider="gemini"
)
prompt_cost, completion_cost = generic_cost_per_token(
model="test-model",
usage=usage,
custom_llm_provider="gemini",
model_info=cast(ModelInfo, model_info),
)
# Expected costs:
# Prompt: 14 * 1e-6
# Completion: (600 * 2e-6) + (1120 * 5e-6)
expected_prompt_cost = 14 * 1e-6
expected_completion_cost = (600 * 2e-6) + (1120 * 5e-6)
expected_prompt_cost: Final = 14 * 1e-6
expected_completion_cost: Final = (600 * 2e-6) + (1120 * 5e-6)
assert round(prompt_cost, 12) == round(expected_prompt_cost, 12)
assert round(completion_cost, 12) == round(expected_completion_cost, 12)
def test_image_tokens_fallback_to_base_cost():
def test_image_tokens_fallback_to_base_cost() -> None:
"""Test that image_tokens fall back to base cost when output_cost_per_image_token is not set."""
from unittest.mock import patch
# Mock model info without image token pricing
mock_model_info = {
model_info: Final = {
"input_cost_per_token": 1e-6,
"output_cost_per_token": 2e-6,
# No output_cost_per_image_token defined
}
usage = Usage(
usage: Final = Usage(
completion_tokens=1720,
prompt_tokens=14,
total_tokens=1734,
@ -400,19 +391,18 @@ def test_image_tokens_fallback_to_base_cost():
),
)
with patch(
"litellm.litellm_core_utils.llm_cost_calc.utils.get_model_info",
return_value=mock_model_info,
):
prompt_cost, completion_cost = generic_cost_per_token(
model="test-model", usage=usage, custom_llm_provider="gemini"
)
prompt_cost, completion_cost = generic_cost_per_token(
model="test-model",
usage=usage,
custom_llm_provider="gemini",
model_info=cast(ModelInfo, model_info),
)
# Expected costs:
# Prompt: 14 * 1e-6
# Completion: (600 * 2e-6) + (1120 * 2e-6) # image_tokens use base cost
expected_prompt_cost = 14 * 1e-6
expected_completion_cost = (600 * 2e-6) + (1120 * 2e-6)
expected_prompt_cost: Final = 14 * 1e-6
expected_completion_cost: Final = (600 * 2e-6) + (1120 * 2e-6)
assert round(prompt_cost, 12) == round(expected_prompt_cost, 12)
assert round(completion_cost, 12) == round(expected_completion_cost, 12)
@ -443,16 +433,14 @@ def test_video_input_tokens_gemini_omni_flash_preview(_local_model_cost_map):
)
def test_video_tokens_fallback_to_base_cost():
def test_video_tokens_fallback_to_base_cost() -> None:
"""Video output tokens fall back to the base output rate when output_cost_per_video_token is not set."""
from unittest.mock import patch
mock_model_info = {
model_info: Final = {
"input_cost_per_token": 1e-6,
"output_cost_per_token": 2e-6,
}
usage = Usage(
usage: Final = Usage(
completion_tokens=1720,
prompt_tokens=14,
total_tokens=1734,
@ -463,30 +451,27 @@ def test_video_tokens_fallback_to_base_cost():
prompt_tokens_details=PromptTokensDetailsWrapper(text_tokens=14),
)
with patch(
"litellm.litellm_core_utils.llm_cost_calc.utils.get_model_info",
return_value=mock_model_info,
):
prompt_cost, completion_cost = generic_cost_per_token(
model="test-model", usage=usage, custom_llm_provider="gemini"
)
prompt_cost, completion_cost = generic_cost_per_token(
model="test-model",
usage=usage,
custom_llm_provider="gemini",
model_info=cast(ModelInfo, model_info),
)
assert round(prompt_cost, 12) == round(14 * 1e-6, 12)
assert round(completion_cost, 12) == round((600 + 1120) * 2e-6, 12)
def test_prompt_audio_and_video_tokens_without_modality_rates_bill_at_tiered_input_rate():
def test_prompt_audio_and_video_tokens_without_modality_rates_bill_at_tiered_input_rate() -> None:
"""Prompt audio/video tokens with no modality rate are prompt tokens, so a >200k prompt bills them above-200k."""
from unittest.mock import patch
mock_model_info = {
model_info: Final = {
"input_cost_per_token": 2e-6,
"input_cost_per_token_above_200k_tokens": 4e-6,
"output_cost_per_token": 1.2e-5,
"output_cost_per_token_above_200k_tokens": 1.8e-5,
}
usage = Usage(
usage: Final = Usage(
prompt_tokens=816032,
completion_tokens=4559,
total_tokens=820591,
@ -494,23 +479,20 @@ def test_prompt_audio_and_video_tokens_without_modality_rates_bill_at_tiered_inp
completion_tokens_details=CompletionTokensDetailsWrapper(text_tokens=4559),
)
with patch(
"litellm.litellm_core_utils.llm_cost_calc.utils.get_model_info",
return_value=mock_model_info,
):
prompt_cost, completion_cost = generic_cost_per_token(
model="test-model", usage=usage, custom_llm_provider="vertex_ai"
)
prompt_cost, completion_cost = generic_cost_per_token(
model="test-model",
usage=usage,
custom_llm_provider="vertex_ai",
model_info=cast(ModelInfo, model_info),
)
assert round(prompt_cost, 9) == round(816032 * 4e-6, 9)
assert round(completion_cost, 9) == round(4559 * 1.8e-5, 9)
def test_prompt_modality_rates_still_win_over_tiered_input_rate():
def test_prompt_modality_rates_still_win_over_tiered_input_rate() -> None:
"""An explicit input_cost_per_audio_token / input_cost_per_video_token beats the fallback, even above 200k."""
from unittest.mock import patch
mock_model_info = {
model_info: Final = {
"input_cost_per_token": 2e-6,
"input_cost_per_token_above_200k_tokens": 4e-6,
"input_cost_per_audio_token": 3e-6,
@ -518,7 +500,7 @@ def test_prompt_modality_rates_still_win_over_tiered_input_rate():
"output_cost_per_token": 1.2e-5,
}
usage = Usage(
usage: Final = Usage(
prompt_tokens=300000,
completion_tokens=10,
total_tokens=300010,
@ -526,11 +508,12 @@ def test_prompt_modality_rates_still_win_over_tiered_input_rate():
completion_tokens_details=CompletionTokensDetailsWrapper(text_tokens=10),
)
with patch(
"litellm.litellm_core_utils.llm_cost_calc.utils.get_model_info",
return_value=mock_model_info,
):
prompt_cost, _ = generic_cost_per_token(model="test-model", usage=usage, custom_llm_provider="vertex_ai")
prompt_cost, _ = generic_cost_per_token(
model="test-model",
usage=usage,
custom_llm_provider="vertex_ai",
model_info=cast(ModelInfo, model_info),
)
assert round(prompt_cost, 9) == round(100000 * 4e-6 + 100000 * 3e-6 + 100000 * 1e-6, 9)
@ -1755,12 +1738,9 @@ def test_gpt55_dated_variants_match_base_reasoning_effort_capabilities(_local_mo
)
def test_string_cost_values():
def test_string_cost_values() -> None:
"""Test that cost values defined as strings are properly converted to floats."""
from unittest.mock import patch
# Mock model info with string cost values (as might be read from config.yaml)
mock_model_info = {
model_info: Final = {
"input_cost_per_token": "3e-7", # String representation of scientific notation
"output_cost_per_token": "6e-7", # String representation of scientific notation
"input_cost_per_audio_token": "0.000001", # String representation of decimal
@ -1772,7 +1752,7 @@ def test_string_cost_values():
# Test usage with various token types
# Note: prompt_tokens must equal sum of details to avoid double-counting adjustment
# text_tokens(700) + audio_tokens(100) + cached_tokens(200) + cache_creation_tokens(150) = 1150
usage = Usage(
usage: Final = Usage(
prompt_tokens=1150,
completion_tokens=500,
total_tokens=1650,
@ -1792,18 +1772,16 @@ def test_string_cost_values():
),
)
# Mock get_model_info to return our mock model info
with patch(
"litellm.litellm_core_utils.llm_cost_calc.utils.get_model_info",
return_value=mock_model_info,
):
prompt_cost, completion_cost = generic_cost_per_token(
model="test-model", usage=usage, custom_llm_provider="test-provider"
)
prompt_cost, completion_cost = generic_cost_per_token(
model="test-model",
usage=usage,
custom_llm_provider="test-provider",
model_info=cast(ModelInfo, model_info),
)
# Calculate expected costs manually
# Prompt cost = text_tokens * input_cost + audio_tokens * audio_cost + cached_tokens * cache_read_cost + cache_creation_tokens * cache_creation_cost
expected_prompt_cost = (
expected_prompt_cost: Final = (
700 * 3e-7 # text tokens
+ 100 * 1e-6 # audio tokens
+ 200 * 1.5e-8 # cached tokens
@ -1811,7 +1789,7 @@ def test_string_cost_values():
)
# Completion cost = text_tokens * output_cost + audio_tokens * audio_output_cost
expected_completion_cost = 450 * 6e-7 + 50 * 2e-6 # text tokens # audio tokens
expected_completion_cost: Final = 450 * 6e-7 + 50 * 2e-6 # text tokens # audio tokens
# Assert costs are calculated correctly
assert round(prompt_cost, 12) == round(expected_prompt_cost, 12)
@ -1925,19 +1903,17 @@ def test_calculate_cost_component_with_string_values():
assert cost == 0.0
def test_string_cost_values_edge_cases():
def test_string_cost_values_edge_cases() -> None:
"""Test edge cases for string cost value handling."""
from unittest.mock import patch
# Test with mixed string and float cost values
mock_model_info = {
model_info: Final = {
"input_cost_per_token": "1e-6", # String
"output_cost_per_token": 2e-6, # Float
"input_cost_per_audio_token": "invalid", # Invalid string
"output_cost_per_audio_token": None, # None value
}
usage = Usage(
usage: Final = Usage(
prompt_tokens=1000,
completion_tokens=500,
total_tokens=1500,
@ -1953,30 +1929,26 @@ def test_string_cost_values_edge_cases():
),
)
with patch(
"litellm.litellm_core_utils.llm_cost_calc.utils.get_model_info",
return_value=mock_model_info,
):
prompt_cost, completion_cost = generic_cost_per_token(
model="test-model", usage=usage, custom_llm_provider="test-provider"
)
prompt_cost, completion_cost = generic_cost_per_token(
model="test-model",
usage=usage,
custom_llm_provider="test-provider",
model_info=cast(ModelInfo, model_info),
)
# Expected costs:
# Prompt: 1000 * 1e-6 + 100 * 0 (invalid string becomes 0)
# Completion: 500 * 2e-6 + 50 * 2e-6 (audio tokens fall back to base cost when output_cost_per_audio_token is None)
expected_prompt_cost = 1000 * 1e-6
expected_completion_cost = 500 * 2e-6 + 50 * 2e-6
expected_prompt_cost: Final = 1000 * 1e-6
expected_completion_cost: Final = 500 * 2e-6 + 50 * 2e-6
assert round(prompt_cost, 12) == round(expected_prompt_cost, 12)
assert round(completion_cost, 12) == round(expected_completion_cost, 12)
def test_string_cost_values_with_threshold():
def test_string_cost_values_with_threshold() -> None:
"""Test that string cost values work correctly with threshold pricing."""
from unittest.mock import patch
# Mock model info with string cost values including threshold pricing
mock_model_info = {
model_info: Final = {
"input_cost_per_token": "1e-6", # String base cost
"output_cost_per_token": "2e-6", # String base cost
"input_cost_per_token_above_200k_tokens": "5e-7", # String threshold cost (lower)
@ -1984,23 +1956,22 @@ def test_string_cost_values_with_threshold():
}
# Test usage above threshold
usage = Usage(
usage: Final = Usage(
prompt_tokens=250000, # Above 200k threshold
completion_tokens=1000,
total_tokens=251000,
)
with patch(
"litellm.litellm_core_utils.llm_cost_calc.utils.get_model_info",
return_value=mock_model_info,
):
prompt_cost, completion_cost = generic_cost_per_token(
model="test-model", usage=usage, custom_llm_provider="test-provider"
)
prompt_cost, completion_cost = generic_cost_per_token(
model="test-model",
usage=usage,
custom_llm_provider="test-provider",
model_info=cast(ModelInfo, model_info),
)
# Expected costs using threshold pricing (string values converted to float)
expected_prompt_cost = 250000 * 5e-7 # threshold cost
expected_completion_cost = 1000 * 1e-6 # threshold cost
expected_prompt_cost: Final = 250000 * 5e-7 # threshold cost
expected_completion_cost: Final = 1000 * 1e-6 # threshold cost
assert round(prompt_cost, 12) == round(expected_prompt_cost, 12)
assert round(completion_cost, 12) == round(expected_completion_cost, 12)

61
tests/unit/litellm_core_utils/llm_cost_calc/test_utils.py
View file

@ -9,6 +9,8 @@ import pytest
import respx
import litellm
from litellm.cost_calculator import cost_per_token, pricing_entry_for_cost_calc
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token, get_billed_token_rates
from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER
from litellm.types.utils import ModelResponse, PromptTokensDetailsWrapper, Usage
@ -237,3 +239,62 @@ async def test_router_prices_balanced_columns_by_the_same_rules_as_flex(
assert balanced_cost == pytest.approx(flex_cost)
if not overridden_columns:
assert balanced_cost == pytest.approx(_cost_at(PARITY_ROW, "_balanced"))
@pytest.mark.parametrize("provider", ("ollama", "ollama_chat"))
@pytest.mark.parametrize("registered", (False, True))
@pytest.mark.parametrize("qualified", (False, True))
def test_ollama_pricing_uses_registered_rates_without_contacting_the_server(
local_model_cost_map: None,
monkeypatch: pytest.MonkeyPatch,
respx_mock: respx.MockRouter,
provider: str,
registered: bool,
qualified: bool,
) -> None:
model: Final = "test-ollama-pricing-registered" if registered else "test-ollama-pricing-unmapped"
monkeypatch.setenv("OLLAMA_API_BASE", "https://ambient-pricing.example")
metadata: Final = respx_mock.post("https://ambient-pricing.example/api/show").respond(
json={"model_info": {"test.context_length": 32768}}
)
if registered:
monkeypatch.setitem(
litellm.model_cost,
f"{provider}/{model}",
{
"litellm_provider": provider,
"mode": "chat",
"input_cost_per_token": 2e-06,
"output_cost_per_token": 5e-06,
"input_cost_per_token_decisions": 3e-06,
"output_cost_per_token_decisions": 7e-06,
},
)
usage: Final = Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120)
cost_model: Final = f"{provider}/{model}" if qualified else model
cost_provider: Final = None if qualified else provider
assert cost_per_token(
model=cost_model, custom_llm_provider=cost_provider, usage_object=usage, call_type="systemone"
) == pytest.approx((0.0003, 0.00014) if registered else (0.0, 0.0))
assert generic_cost_per_token(model=cost_model, custom_llm_provider=cost_provider, usage=usage) == pytest.approx(
(0.0002, 0.0001) if registered else (0.0, 0.0)
)
rates: Final = get_billed_token_rates(model=cost_model, custom_llm_provider=cost_provider, usage=usage)
assert rates is not None
assert (rates.input_cost_per_token, rates.output_cost_per_token) == (
(2e-06, 5e-06) if registered else (0.0, 0.0)
)
diagnostic: Final = pricing_entry_for_cost_calc(
model=cost_model,
completion_response=None,
custom_llm_provider=cost_provider,
custom_pricing=False,
base_model=None,
router_model_id=None,
region_name=None,
litellm_logging_obj=None,
)
assert diagnostic is not None
assert diagnostic[1]["input_cost_per_token"] == (2e-06 if registered else 0.0)
assert not metadata.called

21
tests/unit/litellm_core_utils/test_litellm_logging.py
View file

@ -19,6 +19,7 @@ from unittest.mock import AsyncMock, MagicMock, patch
import httpx
import pytest
import respx
import pytest_asyncio
from mcp.types import AudioContent, CallToolResult, ImageContent, TextContent
from openai import AsyncOpenAI
@ -12470,3 +12471,23 @@ def test_standard_logging_payload_reports_the_deployment_entry_the_request_was_b
)
finally:
litellm.model_cost.pop(deployment_id, None)
@pytest.mark.parametrize("provider", ("ollama", "ollama_chat"))
def test_ollama_cost_information_does_not_request_live_metadata(
local_model_cost_map: None, respx_mock: respx.MockRouter, provider: str
) -> None:
metadata: Final = respx_mock.post("https://cost-information.example/api/show").respond(
json={"model_info": {"test.context_length": 32768}}
)
info: Final = StandardLoggingPayloadSetup.get_model_cost_information(
base_model="test-ollama-cost-information",
custom_pricing=False,
custom_llm_provider=provider,
init_response_obj=litellm.ModelResponse(model="test-ollama-cost-information"),
api_base="https://cost-information.example",
)
rates: Final = info["model_map_value"]
assert rates is not None
assert (rates["input_cost_per_token"], rates["output_cost_per_token"]) == (0.0, 0.0)
assert not metadata.called

98
tests/unit/llms/reducto/test_cost.py
View file

@ -1,26 +1,28 @@
import litellm
from typing import Final
import pytest
import litellm
from litellm.cost_calculator import completion_cost
from litellm.llms.base_llm.ocr.transformation import OCRPage, OCRResponse, OCRUsageInfo
def test_ocr_cost_prefers_credit_pricing_when_pages_processed_is_none(monkeypatch):
monkeypatch.setattr(
litellm,
"get_model_info",
lambda model, custom_llm_provider=None: {"ocr_cost_per_credit": 0.003},
def test_ocr_cost_prefers_credit_pricing_when_pages_processed_is_none() -> None:
model: Final = "litellm-test-credit-ocr"
litellm.register_model(
{f"reducto/{model}": {"litellm_provider": "reducto", "mode": "ocr", "ocr_cost_per_credit": 0.003}}
)
response = OCRResponse(
response: Final = OCRResponse(
pages=[OCRPage(index=0, markdown="credit priced")],
model="parse-v3",
model=model,
usage_info=OCRUsageInfo(pages_processed=None, credits=10),
)
cost = completion_cost(
cost: Final = completion_cost(
completion_response=response,
model="reducto/parse-v3",
model=f"reducto/{model}",
custom_llm_provider="reducto",
call_type="ocr",
)
@ -28,25 +30,28 @@ def test_ocr_cost_prefers_credit_pricing_when_pages_processed_is_none(monkeypatc
assert cost == 0.03
def test_ocr_cost_prefers_zero_credit_pricing_over_page_pricing(monkeypatch):
monkeypatch.setattr(
litellm,
"get_model_info",
lambda model, custom_llm_provider=None: {
"ocr_cost_per_credit": 0.0,
"ocr_cost_per_page": 0.5,
},
def test_ocr_cost_prefers_zero_credit_pricing_over_page_pricing() -> None:
model: Final = "litellm-test-zero-credit-ocr"
litellm.register_model(
{
f"reducto/{model}": {
"litellm_provider": "reducto",
"mode": "ocr",
"ocr_cost_per_credit": 0.0,
"ocr_cost_per_page": 0.5,
}
}
)
response = OCRResponse(
response: Final = OCRResponse(
pages=[OCRPage(index=0, markdown="free credit priced")],
model="parse-v3",
model=model,
usage_info=OCRUsageInfo(pages_processed=2, credits=10),
)
cost = completion_cost(
cost: Final = completion_cost(
completion_response=response,
model="reducto/parse-v3",
model=f"reducto/{model}",
custom_llm_provider="reducto",
call_type="ocr",
)
@ -54,22 +59,21 @@ def test_ocr_cost_prefers_zero_credit_pricing_over_page_pricing(monkeypatch):
assert cost == 0.0
def test_ocr_cost_falls_back_to_page_pricing(monkeypatch):
monkeypatch.setattr(
litellm,
"get_model_info",
lambda model, custom_llm_provider=None: {"ocr_cost_per_page": 0.5},
def test_ocr_cost_falls_back_to_page_pricing() -> None:
model: Final = "litellm-test-page-ocr"
litellm.register_model(
{f"mistral/{model}": {"litellm_provider": "mistral", "mode": "ocr", "ocr_cost_per_page": 0.5}}
)
response = OCRResponse(
response: Final = OCRResponse(
pages=[OCRPage(index=0, markdown="page priced")],
model="mistral-ocr-latest",
model=model,
usage_info=OCRUsageInfo(pages_processed=2),
)
cost = completion_cost(
cost: Final = completion_cost(
completion_response=response,
model="mistral/mistral-ocr-latest",
model=f"mistral/{model}",
custom_llm_provider="mistral",
call_type="ocr",
)
@ -77,22 +81,19 @@ def test_ocr_cost_falls_back_to_page_pricing(monkeypatch):
assert cost == 1.0
def test_ocr_cost_returns_zero_when_no_pricing_and_no_pages(monkeypatch):
monkeypatch.setattr(
litellm,
"get_model_info",
lambda model, custom_llm_provider=None: {},
)
def test_ocr_cost_returns_zero_when_no_pricing_and_no_pages() -> None:
model: Final = "litellm-test-unpriced-ocr"
litellm.register_model({f"reducto/{model}": {"litellm_provider": "reducto", "mode": "ocr"}})
response = OCRResponse(
response: Final = OCRResponse(
pages=[OCRPage(index=0, markdown="unpriced")],
model="parse-v3",
model=model,
usage_info=OCRUsageInfo(pages_processed=None, credits=5),
)
cost = completion_cost(
cost: Final = completion_cost(
completion_response=response,
model="reducto/parse-v3",
model=f"reducto/{model}",
custom_llm_provider="reducto",
call_type="ocr",
)
@ -100,23 +101,22 @@ def test_ocr_cost_returns_zero_when_no_pricing_and_no_pages(monkeypatch):
assert cost == 0.0
def test_ocr_cost_raises_when_pages_processed_missing_for_page_pricing(monkeypatch):
monkeypatch.setattr(
litellm,
"get_model_info",
lambda model, custom_llm_provider=None: {"ocr_cost_per_page": 0.5},
def test_ocr_cost_raises_when_pages_processed_missing_for_page_pricing() -> None:
model: Final = "litellm-test-missing-page-ocr"
litellm.register_model(
{f"mistral/{model}": {"litellm_provider": "mistral", "mode": "ocr", "ocr_cost_per_page": 0.5}}
)
response = OCRResponse(
response: Final = OCRResponse(
pages=[OCRPage(index=0, markdown="missing pages")],
model="mistral-ocr-latest",
model=model,
usage_info=OCRUsageInfo(pages_processed=None),
)
with pytest.raises(ValueError, match="OCR response pages_processed is None"):
completion_cost(
completion_response=response,
model="mistral/mistral-ocr-latest",
model=f"mistral/{model}",
custom_llm_provider="mistral",
call_type="ocr",
)

22
tests/unit/test_utils.py
View file

@ -10009,3 +10009,25 @@ def test_get_optional_params_forwards_a_tool_config_cache_point_to_a_downstream_
)
assert optional_params["extra_body"] == {"kept": True, "cache_control_injection_points": [point]}
@pytest.mark.parametrize("provider", ("ollama", "ollama_chat"))
def test_ollama_pricing_lookup_preserves_live_model_info_discovery(
local_model_cost_map: None, respx_mock: respx.MockRouter, provider: str
) -> None:
from litellm.utils import get_model_info_for_pricing
metadata: Final = respx_mock.post("https://model-info.example/api/show").respond(
json={"model_info": {"test.context_length": 32768}}
)
pricing: Final = get_model_info_for_pricing(
model="test-live-discovery", custom_llm_provider=provider, api_base="https://model-info.example"
)
assert (pricing["input_cost_per_token"], pricing["output_cost_per_token"]) == (0.0, 0.0)
assert not metadata.called
discovered: Final = litellm.get_model_info(
model="test-live-discovery", custom_llm_provider=provider, api_base="https://model-info.example"
)
assert discovered["max_input_tokens"] == 32768
assert metadata.call_count == 1