diff --git a/litellm/cost_calculator.py b/litellm/cost_calculator.py index 5b4a1d30ff0..233ff48e8a3 100644 --- a/litellm/cost_calculator.py +++ b/litellm/cost_calculator.py @@ -964,6 +964,19 @@ def _bedrock_registration_has_prices(entry: Mapping[str, object]) -> bool: ) +def _bedrock_cost_registration(model: str | None) -> tuple[str, Mapping[str, object]] | None: + if model is None: + return None + return next( + ( + (key, entry) + for key, entry in runtime_model_cost_registrations(model, "bedrock") + if _bedrock_registration_has_prices(entry) + ), + None, + ) + + def _bedrock_pricing_info( model: str | None, invoked_model_id: str | None, @@ -972,14 +985,7 @@ def _bedrock_pricing_info( ) -> ModelInfo | None: if explicit_pricing or custom_llm_provider != "bedrock" or not model: return None - registration: Final = next( - ( - (key, entry) - for key, entry in runtime_model_cost_registrations(model, "bedrock") - if _bedrock_registration_has_prices(entry) - ), - None, - ) + registration: Final = _bedrock_cost_registration(model) profile: Final = _cost_map_model_info(invoked_model_id, "bedrock") if invoked_model_id else None if profile is None or not _declares_rates(profile): return _cost_map_model_info(registration[0], None) if registration is not None else None @@ -2770,7 +2776,14 @@ def pricing_entry_for_cost_calc( custom_pricing is True or base_model is not None, ) if bedrock_model_info is not None: - return bedrock_model_info["key"], bedrock_model_info + key: Final = bedrock_model_info["key"] + declared_pricing: Final = declared_cost_map_pricing(key, model or key, custom_llm_provider) + registration: Final = _bedrock_cost_registration(model) + declared_keys: Final = frozenset(declared_pricing) | ( + frozenset(registration[1]) if registration is not None else frozenset[str]() + ) + selected_entry: Final[Mapping[str, object]] = bedrock_model_info + return key, MappingProxyType({name: value for name, value in selected_entry.items() if name in declared_keys}) candidates: Final = (selected_model, _get_response_model(completion_response), model) resolved: Final = next( (info for info in (_cost_map_model_info(name, custom_llm_provider) for name in candidates if name) if info), @@ -2778,7 +2791,7 @@ def pricing_entry_for_cost_calc( ) if resolved is None: return None - return resolved["key"], _raw_cost_map_entry(resolved["key"]) or resolved + return resolved["key"], declared_cost_map_pricing(resolved["key"], model or resolved["key"], custom_llm_provider) def ocr_cost( diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index d88c8876ae0..8f3101915be 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -135,6 +135,7 @@ from litellm.litellm_core_utils.llm_cost_calc.usage_object_transformation import ) from litellm.litellm_core_utils.llm_cost_calc.zero_cost_diagnostic import ( diagnose_zero_cost, + is_explicitly_free_usage, zero_cost_warning, ) from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response import LiteLLMResponseObjectHandler @@ -2417,6 +2418,33 @@ class Logging(LiteLLMLoggingBaseClass): usage=usage, ) + def is_usage_explicitly_free(self, result: object) -> bool: + usage: Final = get_usage_object(completion_response=result) + if usage is None or self.model_call_details.get("response_cost_failure_debug_information") is not None: + return False + custom_pricing: Final = self.custom_pricing_for(result) + if custom_pricing and self.custom_model_info is not None: + return is_explicitly_free_usage(usage, self.custom_model_info) + if custom_pricing and self.custom_cost_per_token is not None: + return is_explicitly_free_usage(usage, self.custom_cost_per_token) + custom_llm_provider: Final = self.model_call_details.get("custom_llm_provider") + pricing: Final = pricing_entry_for_cost_calc( + model=self.model, + completion_response=result, + custom_llm_provider=self.get_router_pricing_provider() or custom_llm_provider, + custom_pricing=custom_pricing, + base_model=get_base_model_from_metadata(model_call_details=self.model_call_details), + router_model_id=self.get_router_model_id(), + region_name=_resolve_mantle_region_for_cost( + custom_llm_provider=custom_llm_provider, + litellm_params=self.model_call_details.get("litellm_params"), + ), + litellm_logging_obj=self, + ) + if pricing is None: + return False + return is_explicitly_free_usage(usage, pricing[1]) + def spilled_over_ptu_for(self, result: object) -> bool: result_hidden_params: Final = getattr(result, HIDDEN_PARAMS_ATTR, None) or MappingProxyType({}) additional_headers: Final = ( diff --git a/litellm/litellm_core_utils/llm_cost_calc/zero_cost_diagnostic.py b/litellm/litellm_core_utils/llm_cost_calc/zero_cost_diagnostic.py index 3ed598b5e97..78912d0a938 100644 --- a/litellm/litellm_core_utils/llm_cost_calc/zero_cost_diagnostic.py +++ b/litellm/litellm_core_utils/llm_cost_calc/zero_cost_diagnostic.py @@ -1,10 +1,12 @@ from collections.abc import Mapping from functools import reduce +from types import MappingProxyType from typing import Final from pydantic import TypeAdapter, ValidationError from typing_extensions import assert_never +from litellm.litellm_core_utils.llm_cost_calc.tiered_pricing import select_tier_for_input from litellm.litellm_core_utils.llm_cost_calc.utils import parse_prompt_tokens_details from litellm.types.utils import ( CompletionTokensDetailsWrapper, @@ -23,6 +25,7 @@ _TEXT_OUTPUT_RATE: Final = "output_cost_per_token" _AUDIO_OUTPUT_RATE: Final = "output_cost_per_audio_token" _RATE_KEY_MARKERS: Final = ("cost", "pricing") _NESTED_PRICING: Final = TypeAdapter(Mapping[str, object] | tuple[object, ...]) +_PRICING_TIERS: Final = TypeAdapter(tuple[Mapping[str, object], ...]) _MAX_PRICING_DEPTH: Final = 4 @@ -105,6 +108,33 @@ def _is_explicit_zero(value: object) -> bool: return not isinstance(value, bool) and isinstance(value, (int, float)) and value == 0 +def _pricing_for_usage(usage: Usage, pricing_entry: Mapping[str, object]) -> Mapping[str, object]: + tiered_pricing: Final = pricing_entry.get("tiered_pricing") + if not isinstance(tiered_pricing, list): + return pricing_entry + try: + tiers: Final = _PRICING_TIERS.validate_python(tiered_pricing) + except ValidationError: + return pricing_entry + tier: Final = select_tier_for_input(tiers, usage.prompt_tokens) + if tier is None or _TEXT_INPUT_RATE not in tier: + return pricing_entry + cache_rate: Final = tier.get(_CACHE_READ_INPUT_RATE) + return MappingProxyType( + { + **pricing_entry, + **tier, + _CACHE_READ_INPUT_RATE: cache_rate if cache_rate is not None else tier[_TEXT_INPUT_RATE], + } + ) + + +def is_explicitly_free_usage(usage: Usage, pricing_entry: Mapping[str, object]) -> bool: + selected_pricing: Final = _pricing_for_usage(usage, pricing_entry) + used_keys: Final = tuple(_effective_pricing_key(key, selected_pricing) for key in used_pricing_keys(usage)) + return bool(used_keys) and all(_is_explicit_zero(selected_pricing.get(key)) for key in used_keys) + + def diagnose_zero_cost( usage: Usage, pricing_model: str, diff --git a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py index c87357122ce..e2b651f8885 100644 --- a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py +++ b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py @@ -1500,7 +1500,11 @@ def _stamp_streaming_usage_cost(usage: Usage, response: ModelResponse, logging_o computed_cost: Final = logging_obj.response_cost_calculator(result=response) if isinstance(computed_cost, bool) or not isinstance(computed_cost, (int, float)) or computed_cost < 0: return - if computed_cost > 0 or isinstance(getattr(usage, "cost", None), (int, float)): + if ( + computed_cost > 0 + or isinstance(getattr(usage, "cost", None), (int, float)) + or logging_obj.is_usage_explicitly_free(response) + ): stamp_litellm_bill_on_usage(response, usage, float(computed_cost)) diff --git a/litellm/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 361e576f3e1..956e202a65a 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -114,7 +114,12 @@ def _stamp_usage_cost( usage: Final[object] = getattr(usage_chunk, "usage", None) if usage is None or isinstance(bill, bool) or not isinstance(bill, (int, float)): return - if bill <= 0 and not isinstance(getattr(usage, "cost", None), (int, float)): + complete_usage: Final = getattr(complete_response, "usage", None) + if ( + bill <= 0 + and not isinstance(getattr(usage, "cost", None), (int, float)) + and not (bill == 0 and getattr(complete_usage, "cost", None) == 0) + ): return chunk_hidden: Final = get_hidden_params(usage_chunk) if chunk_hidden is not None and UPSTREAM_USAGE_COST in complete_hidden: diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index 28a82b500c9..c2191375cb4 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -1931,7 +1931,11 @@ def stamp_responses_usage_cost( return if isinstance(bill, bool) or not isinstance(bill, (int, float)): return - if bill > 0 or isinstance(usage_obj.cost, (int, float)): + if ( + bill > 0 + or isinstance(usage_obj.cost, (int, float)) + or (bill == 0 and logging_obj.is_usage_explicitly_free(priced)) + ): stamp_litellm_bill_on_usage(response_obj, usage_obj, float(bill)) diff --git a/tests/unit/litellm_core_utils/llm_cost_calc/test_zero_cost_diagnostic.py b/tests/unit/litellm_core_utils/llm_cost_calc/test_zero_cost_diagnostic.py index bdc14c4934b..d325a4677f7 100644 --- a/tests/unit/litellm_core_utils/llm_cost_calc/test_zero_cost_diagnostic.py +++ b/tests/unit/litellm_core_utils/llm_cost_calc/test_zero_cost_diagnostic.py @@ -6,6 +6,7 @@ import pytest from litellm.litellm_core_utils.llm_cost_calc.zero_cost_diagnostic import ( ZERO_COST_COUNTER_NAME, diagnose_zero_cost, + is_explicitly_free_usage, used_pricing_keys, zero_cost_warning, ) @@ -18,6 +19,85 @@ PRICED_ENTRY: Final = {"input_cost_per_token": 1e-06, "output_cost_per_token": 2 TEXT_USAGE: Final = Usage(prompt_tokens=10, completion_tokens=20, total_tokens=30) +@pytest.mark.parametrize( + ("usage", "rates", "expected"), + [ + (TEXT_USAGE, FREE_ENTRY, True), + (TEXT_USAGE, PRICED_ENTRY, False), + (TEXT_USAGE, {}, False), + (TEXT_USAGE, {"input_cost_per_token": 0.0}, False), + (TEXT_USAGE, {"input_cost_per_token": False, "output_cost_per_token": 0.0}, False), + (Usage(prompt_tokens=10, completion_tokens=0), {"input_cost_per_token": 0.0}, True), + (Usage(prompt_tokens=0, completion_tokens=0), FREE_ENTRY, False), + ( + Usage(prompt_tokens=10, completion_tokens=0, prompt_tokens_details={"cached_tokens": 10}), + FREE_ENTRY, + False, + ), + ( + Usage(prompt_tokens=10, completion_tokens=0, prompt_tokens_details={"cached_tokens": 10}), + {"input_cost_per_token": 0.0}, + True, + ), + ], +) +def test_free_usage_requires_explicit_zero_rates_for_every_used_token_kind( + usage: Usage, rates: Mapping[str, object], expected: bool +) -> None: + assert is_explicitly_free_usage(usage, rates) is expected + + +@pytest.mark.parametrize( + ("input_tokens", "tier_output_rate", "expected"), + [(10, 0.0, True), (11, 0.0, False), (10, 0.001, False), (10, None, False)], +) +def test_free_usage_uses_the_selected_tier_and_preserves_missing_output_prices( + input_tokens: int, tier_output_rate: float | None, expected: bool +) -> None: + rates: Final = { + "input_cost_per_token": 0.01, + "tiered_pricing": [ + {"range": [0, 10], "input_cost_per_token": 0.0, "output_cost_per_token": tier_output_rate}, + {"range": [10, 20], "input_cost_per_token": 0.01, "output_cost_per_token": 0.02}, + ], + } + + assert is_explicitly_free_usage(Usage(prompt_tokens=input_tokens, completion_tokens=5), rates) is expected + + +@pytest.mark.parametrize("cache_rate", (None, 0.01)) +def test_free_cached_usage_uses_the_selected_tier_cache_rate_or_its_input_rate(cache_rate: float | None) -> None: + rates: Final = { + "input_cost_per_token": 0.01, + "cache_read_input_token_cost": 0.02, + "tiered_pricing": [ + {"range": [0, 20], "input_cost_per_token": 0.0, "cache_read_input_token_cost": cache_rate} + ], + } + usage: Final = Usage(prompt_tokens=10, completion_tokens=0, prompt_tokens_details={"cached_tokens": 10}) + + assert is_explicitly_free_usage(usage, rates) is (cache_rate is None) + + +@pytest.mark.parametrize( + "tiers", + [ + ["invalid tier"], + [], + [{"range": [0, 20], "output_cost_per_token": 0.0}], + [{"range": "invalid range", "input_cost_per_token": 0.0, "output_cost_per_token": 0.0}], + ], +) +def test_invalid_or_inactive_tiers_do_not_make_paid_usage_free(tiers: object) -> None: + rates: Final = { + "input_cost_per_token": 0.01, + "output_cost_per_token": 0.02, + "tiered_pricing": tiers, + } + + assert is_explicitly_free_usage(Usage(prompt_tokens=10, completion_tokens=5), rates) is False + + def test_missing_pricing_key_names_every_rate_the_usage_needs() -> None: diagnostic = diagnose_zero_cost( usage=TEXT_USAGE, pricing_model="dep-1", pricing_entry=PER_SECOND_ENTRY, calculation_failed=False diff --git a/tests/unit/litellm_core_utils/test_litellm_logging.py b/tests/unit/litellm_core_utils/test_litellm_logging.py index e8f90dc0146..c1d878d5b09 100644 --- a/tests/unit/litellm_core_utils/test_litellm_logging.py +++ b/tests/unit/litellm_core_utils/test_litellm_logging.py @@ -79,6 +79,64 @@ from typing import List, Optional from litellm.types.utils import LiteLLMCommonStrings +@pytest.mark.parametrize("metadata_key", ("metadata", "litellm_metadata")) +def test_free_usage_uses_deployment_metadata_prices(metadata_key: str) -> None: + obj: Final = Logging( + model="metadata-free-deployment", + messages=[], + stream=True, + call_type="completion", + start_time=1.0, + litellm_call_id="metadata-free", + function_id="test", + ) + obj.litellm_params = {metadata_key: {"model_info": {"input_cost_per_token": 0.0, "output_cost_per_token": 0.0}}} + response: Final = ModelResponse( + model="metadata-free-deployment", usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15) + ) + + assert obj.is_usage_explicitly_free(response) is True + + +@pytest.mark.parametrize("pricing_source", ("model_info", "token_rates")) +@pytest.mark.parametrize("rate", (0.0, 0.001)) +@pytest.mark.parametrize("calculation_failed", (False, True)) +def test_per_call_free_prices_override_catalog_unless_pricing_failed( + monkeypatch: pytest.MonkeyPatch, pricing_source: str, rate: float, calculation_failed: bool +) -> None: + model: Final = "per-call-free-price-test" + monkeypatch.setitem( + litellm.model_cost, + model, + {"litellm_provider": "openai", "mode": "chat", "input_cost_per_token": 0.01, "output_cost_per_token": 0.02}, + ) + obj: Final = Logging( + model=model, + messages=[], + stream=True, + call_type="completion", + start_time=1.0, + litellm_call_id="per-call-free", + function_id="test", + ) + obj.litellm_params = {"input_cost_per_token": rate, "output_cost_per_token": rate} + if pricing_source == "model_info": + obj.set_per_call_pricing( + { + **litellm.get_model_info(model=model, custom_llm_provider="openai"), + "input_cost_per_token": rate, + "output_cost_per_token": rate, + } + ) + else: + obj.custom_cost_per_token = {"input_cost_per_token": rate, "output_cost_per_token": rate} + if calculation_failed: + obj.model_call_details["response_cost_failure_debug_information"] = {"error": "pricing failed"} + response: Final = ModelResponse(model=model, usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15)) + + assert obj.is_usage_explicitly_free(response) is (rate == 0.0 and not calculation_failed) + + @pytest.fixture def logging_obj(): return LitellmLogging( diff --git a/tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py b/tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py index 527c83d5760..858c0f90930 100644 --- a/tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py +++ b/tests/unit/litellm_core_utils/test_streaming_chunk_builder_utils.py @@ -10,6 +10,7 @@ import respx import litellm from litellm import ChatCompletionUsageBlock, stream_chunk_builder +from litellm.litellm_core_utils.litellm_logging import Logging from litellm.litellm_core_utils.streaming_chunk_builder_utils import ( ChunkProcessor, stream_chunk_builder_text_completion, @@ -31,6 +32,84 @@ from litellm.types.utils import ( ) +@pytest.mark.parametrize("routed_provider", (False, True)) +@pytest.mark.parametrize( + ("profile_rates", "registered_rates", "expected_cost"), + [ + ( + {"input_cost_per_token": 0.01, "output_cost_per_token": 0.02}, + {"input_cost_per_token": 0.0, "output_cost_per_token": 0.0}, + 0.0, + ), + ( + {"input_cost_per_token": 0.01, "output_cost_per_token": 0.02}, + {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002}, + 0.02, + ), + ({"input_cost_per_token": 0.01, "output_cost_per_token": 0.02}, {"input_cost_per_token": 0.0}, 0.1), + ({"input_cost_per_token": 0.0}, {"input_cost_per_token": 0.0}, None), + ], +) +def test_bedrock_stream_cost_uses_registered_rates_over_profile_prices( + monkeypatch: pytest.MonkeyPatch, + profile_rates: dict[str, float], + registered_rates: dict[str, float], + expected_cost: float | None, + routed_provider: bool, +) -> None: + model: Final = "openai.stream-pricing-test" + profile: Final = f"us.{model}" + monkeypatch.setattr( + litellm, + "model_cost", + { + **litellm.model_cost, + profile: {"litellm_provider": "bedrock", "mode": "chat", **profile_rates}, + }, + ) + litellm.register_model({model: registered_rates}, persist_across_reloads=False) + logging_obj: Final = Logging( + model=model, + messages=[], + stream=True, + call_type="completion", + start_time=1.0, + litellm_call_id="bedrock-free-stream", + function_id="test", + ) + logging_obj.optional_params = {} + logging_obj.litellm_params = ( + {"metadata": {"model_info": {"pricing_provider": "bedrock"}}} if routed_provider else {} + ) + logging_obj.model_call_details.update( + custom_llm_provider="openai" if routed_provider else "bedrock", model_id=profile + ) + chunks: Final = [ + ModelResponseStream( + id="bedrock-free-stream", + created=1, + model=model, + choices=[StreamingChoices(index=0, delta=Delta(content="hi"), finish_reason="stop")], + ), + ModelResponseStream( + id="bedrock-free-stream", + created=1, + model=model, + choices=[], + usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15), + ), + ] + + response: Final = stream_chunk_builder(chunks, logging_obj=logging_obj) + + assert response is not None + assert response.usage.prompt_tokens == 10 + assert response.usage.completion_tokens == 5 + assert getattr(response.usage, "cost", None) == ( + pytest.approx(expected_cost) if expected_cost is not None else None + ) + + def test_get_combined_tool_content(): chunks = [ ModelResponseStream( diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index 3b798b35096..cb0ef9a9a33 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -7649,6 +7649,85 @@ def test_chunk_creator_preserves_hidden_provider_specific_fields_from_parsed_chu assert assembled._hidden_params["provider_specific_fields"] == {"traffic_type": "ON_DEMAND_FLEX"} +@pytest.mark.parametrize("sync_mode", [True, False]) +@pytest.mark.parametrize( + ("rates", "expected_cost"), + [ + ({"input_cost_per_token": 0.0, "output_cost_per_token": 0.0}, 0.0), + ({"input_cost_per_token": 1e-6, "output_cost_per_token": 2e-6}, 20e-6), + ({"input_cost_per_token": 0.0}, None), + ({}, None), + ( + { + "input_cost_per_token": 0.01, + "output_cost_per_token": 0.02, + "tiered_pricing": [ + {"range": [0, 128000], "input_cost_per_token": 0.0, "output_cost_per_token": 0.0} + ], + }, + 0.0, + ), + ( + { + "tiered_pricing": [ + {"range": [0, 128000], "input_cost_per_token": 0.0, "output_cost_per_token": 0.0} + ] + }, + 0.0, + ), + ], +) +async def test_stream_usage_distinguishes_free_from_unpriced_calls( + monkeypatch: pytest.MonkeyPatch, sync_mode: bool, rates: dict[str, object], expected_cost: float | None +) -> None: + model: Final = "stream-zero-cost-regression" + monkeypatch.setitem(litellm.model_cost, model, {"litellm_provider": "openai", "mode": "chat", **rates}) + logging_obj: Final = Logging( + model=model, + messages=[{"role": "user", "content": "hi"}], + stream=True, + call_type="completion" if sync_mode else "acompletion", + start_time=1.0, + litellm_call_id="free-stream", + function_id="test", + ) + logging_obj.optional_params = {} + logging_obj.litellm_params = {} + logging_obj.stream_options = {"include_usage": True} + logging_obj.model_call_details["custom_llm_provider"] = "openai" + chunks: Final = [ + ModelResponseStream( + id="free-stream", + created=1, + model=model, + choices=[StreamingChoices(index=0, delta=Delta(content="hi"), finish_reason="stop")], + ), + ModelResponseStream( + id="free-stream", + created=1, + model=model, + choices=[], + usage=Usage(prompt_tokens=10, completion_tokens=5, total_tokens=15), + ), + ] + stream: Final = CustomStreamWrapper( + completion_stream=ModelResponseListIterator(model_responses=chunks), + model=model, + custom_llm_provider="openai", + logging_obj=logging_obj, + stream_options={"include_usage": True}, + ) + emitted: Final = tuple(stream) if sync_mode else tuple([chunk async for chunk in stream]) + usage_chunks: Final = tuple(chunk for chunk in emitted if getattr(chunk, "usage", None) is not None) + + assert len(usage_chunks) == 1 + assert usage_chunks[0].usage.prompt_tokens == 10 + assert usage_chunks[0].usage.completion_tokens == 5 + assert getattr(usage_chunks[0].usage, "cost", None) == ( + pytest.approx(expected_cost) if expected_cost is not None else None + ) + + def test_the_usage_chunk_carries_the_bill_the_builder_priced_for_the_spend_row() -> None: from litellm.litellm_core_utils.streaming_handler import ( _stamp_usage_cost, # pyright: ignore[reportPrivateUsage] # the helper under test diff --git a/tests/unit/responses/test_streaming_iterator.py b/tests/unit/responses/test_streaming_iterator.py index 11f4e78619e..17a5112ce6f 100644 --- a/tests/unit/responses/test_streaming_iterator.py +++ b/tests/unit/responses/test_streaming_iterator.py @@ -861,6 +861,61 @@ def _responses_api_response_with_usage() -> ResponsesAPIResponse: ) +@pytest.mark.parametrize( + ("rates", "expected_cost"), + [ + ({"input_cost_per_token": 0.0, "output_cost_per_token": 0.0}, 0.0), + ({"input_cost_per_token": 1e-6, "output_cost_per_token": 2e-6}, 140e-6), + ({"input_cost_per_token": 0.0}, None), + ({}, None), + ( + { + "input_cost_per_token": 0.01, + "output_cost_per_token": 0.02, + "tiered_pricing": [ + {"range": [0, 128000], "input_cost_per_token": 0.0, "output_cost_per_token": 0.0} + ], + }, + 0.0, + ), + ( + { + "tiered_pricing": [ + {"range": [0, 128000], "input_cost_per_token": 0.0, "output_cost_per_token": 0.0} + ] + }, + 0.0, + ), + ], +) +def test_responses_stream_usage_distinguishes_free_from_unpriced_calls( + monkeypatch: pytest.MonkeyPatch, rates: dict[str, object], expected_cost: float | None +) -> None: + from litellm.responses.streaming_iterator import stamp_responses_usage_cost + + model: Final = "responses-zero-cost-regression" + monkeypatch.setitem(litellm.model_cost, model, {"litellm_provider": "openai", "mode": "chat", **rates}) + response: Final = _responses_api_response_with_usage().model_copy(update={"model": model}) + logging_obj: Final = LiteLLMLoggingObj( + model=model, + messages=[], + stream=True, + call_type="responses", + start_time=1.0, + litellm_call_id="free-response", + function_id="test", + ) + logging_obj.optional_params = {} + logging_obj.litellm_params = {} + logging_obj.model_call_details["custom_llm_provider"] = "openai" + + stamp_responses_usage_cost(response, logging_obj) + + assert getattr(response.usage, "cost", None) == ( + pytest.approx(expected_cost) if expected_cost is not None else None + ) + + def test_stamp_responses_usage_cost_stamps_computed_cost(): from litellm.responses.streaming_iterator import stamp_responses_usage_cost