From 332ba6f9fbd0faff4c9da65e3b61b013a2de50ce Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 27 Oct 2026 19:37:28 +0000 Subject: [PATCH] refactor(cost): write the bill into usage.cost once at the proxy egress instead of inside the pipeline --- litellm/cost_calculator.py | 16 +- litellm/litellm_core_utils/hidden_params.py | 29 - .../llm_response_utils/response_metadata.py | 41 +- .../streaming_chunk_builder_utils.py | 14 +- litellm/proxy/common_request_processing.py | 11 +- litellm/proxy/proxy_server.py | 20 +- litellm/proxy/usage_cost_egress.py | 77 +++ .../streaming_iterator.py | 4 +- litellm/responses/streaming_iterator.py | 30 +- tests/unit/proxy/test_usage_cost_egress.py | 200 +++++++ .../unit/responses/test_streaming_iterator.py | 182 ++---- tests/unit/test_cost_calculator.py | 519 +++++------------- 12 files changed, 530 insertions(+), 613 deletions(-) create mode 100644 litellm/proxy/usage_cost_egress.py create mode 100644 tests/unit/proxy/test_usage_cost_egress.py diff --git a/litellm/cost_calculator.py b/litellm/cost_calculator.py index 382943db244..cf28dbcd17c 100644 --- a/litellm/cost_calculator.py +++ b/litellm/cost_calculator.py @@ -25,7 +25,7 @@ from litellm.constants import ( from litellm.litellm_core_utils.core_helpers import as_str_mapping, provider_reported_usage_cost from litellm.litellm_core_utils.cost_discount import resolve_cost_discount from litellm.litellm_core_utils.decisions_pricing import decisions_pricing_fields, is_decisions_call_type -from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR, provider_stated_usage_cost +from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR from litellm.litellm_core_utils.llm_cost_calc.tool_call_cost_tracking import ( StandardBuiltInToolCostTracking, ) @@ -1560,16 +1560,6 @@ def _store_cost_breakdown_in_logging_obj( # Don't fail the main cost calculation if breakdown storage fails -def _with_provider_stated_cost(completion_response: object, usage: Usage | None) -> Usage | None: - if usage is None: - return usage - stamped_cost: Final = getattr(usage, "cost", None) - stated_cost: Final = provider_stated_usage_cost(completion_response, stamped_cost) - if stated_cost is stamped_cost: - return usage - return usage.model_copy(update=MappingProxyType({"cost": stated_cost})) - - def _without_provider_stated_cost(usage: Usage | None) -> Usage | None: if usage is None or getattr(usage, "cost", None) is None: return usage @@ -1712,9 +1702,7 @@ def completion_cost( cache_creation_input_tokens: int | None = None cache_read_input_tokens: int | None = None audio_transcription_file_duration: float = 0.0 - provider_usage_object: Final = _with_provider_stated_cost( - completion_response, get_usage_object(completion_response=completion_response) - ) + provider_usage_object: Final = get_usage_object(completion_response=completion_response) cost_per_token_usage_object: Final[Usage | None] = ( _without_provider_stated_cost(provider_usage_object) if custom_pricing or base_model is not None diff --git a/litellm/litellm_core_utils/hidden_params.py b/litellm/litellm_core_utils/hidden_params.py index 7281b138894..a74337d2549 100644 --- a/litellm/litellm_core_utils/hidden_params.py +++ b/litellm/litellm_core_utils/hidden_params.py @@ -134,32 +134,3 @@ def get_or_create_hidden_params(obj: object) -> MutableMapping[str, object]: if hidden_params_mapping is not None: return hidden_params_mapping raise TypeError(f"unsupported hidden params storage: {type(hidden_params).__name__}") - - -USAGE_COST_BEFORE_LITELLM_STAMP: Final = "usage_cost_before_litellm_stamp" - - -def stamp_billed_usage_cost(obj: object, usage: object, cost: float) -> None: - """usage.cost becomes what litellm billed; the provider's own figure is kept so a later pricing pass bills the same.""" - hidden_params: Final = get_hidden_params(obj) - if hidden_params is None or USAGE_COST_BEFORE_LITELLM_STAMP not in hidden_params: - set_hidden_param(obj, USAGE_COST_BEFORE_LITELLM_STAMP, getattr(usage, "cost", None)) - setattr(usage, "cost", cost) - - -def stamp_billable_usage_cost(obj: object, usage: object, cost: object) -> float | None: - """Stamp a priced bill, a $0 one included once the provider put a number there; an unpriceable reply keeps its usage.""" - if isinstance(cost, bool) or not isinstance(cost, (int, float)) or cost < 0: - return None - if cost == 0 and getattr(usage, "cost", None) is None: - return None - stamp_billed_usage_cost(obj, usage, float(cost)) - return float(cost) - - -def provider_stated_usage_cost(obj: object, usage_cost: object) -> object: - """What the provider itself put in usage.cost, looking past a bill litellm stamped there.""" - hidden_params: Final = get_hidden_params(obj) - if hidden_params is None or USAGE_COST_BEFORE_LITELLM_STAMP not in hidden_params: - return usage_cost - return hidden_params.get(USAGE_COST_BEFORE_LITELLM_STAMP) diff --git a/litellm/litellm_core_utils/llm_response_utils/response_metadata.py b/litellm/litellm_core_utils/llm_response_utils/response_metadata.py index ef47e2f0d9f..ace57a64aae 100644 --- a/litellm/litellm_core_utils/llm_response_utils/response_metadata.py +++ b/litellm/litellm_core_utils/llm_response_utils/response_metadata.py @@ -6,23 +6,15 @@ from typing import Final import httpx from pydantic import TypeAdapter, ValidationError -import litellm from litellm.constants import LITELLM_DETAILED_TIMING from litellm.litellm_core_utils.core_helpers import get_litellm_metadata_from_kwargs, process_response_headers -from litellm.litellm_core_utils.hidden_params import ( - USAGE_COST_BEFORE_LITELLM_STAMP, - get_hidden_params, - stamp_billable_usage_cost, -) from litellm.litellm_core_utils.llm_response_utils.get_api_base import get_api_base from litellm.litellm_core_utils.logging_utils import LiteLLMLoggingObject -from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIResponse from litellm.types.utils import ( EmbeddingResponse, HiddenParams, ModelResponse, TranscriptionResponse, - Usage, ) @@ -167,18 +159,16 @@ class ResponseMetadata: ## ADD OTHER HIDDEN PARAMS model_info: Final = kwargs.get("model_info", {}) or {} model_id: Final = model_info.get("id", None) - response_cost: Final = logging_obj.response_cost_calculator( - result=self.result, - litellm_model_name=model, - router_model_id=model_id, - cache_hit=cache_hit, - ) - self._stamp_billed_usage_cost(response_cost) new_params: Final = { "litellm_call_id": getattr(logging_obj, "litellm_call_id", None), "api_base": get_api_base(model=model or "", optional_params=kwargs), "model_id": model_id, - "response_cost": response_cost, + "response_cost": logging_obj.response_cost_calculator( + result=self.result, + litellm_model_name=model, + router_model_id=model_id, + cache_hit=cache_hit, + ), "additional_headers": process_response_headers( self._get_additional_headers_from_hidden_params() or {}, preserve_litellm_internal_headers=True, @@ -189,25 +179,6 @@ class ResponseMetadata: } self._update_hidden_params(new_params) - def _stamp_billed_usage_cost(self, response_cost: object) -> None: - """A usage.cost the upstream reported becomes what litellm billed, so a client and the spend log agree on a - priced deployment. A reply without one keeps its usage as the upstream sent it unless the operator asked for - the bill in every usage block with ``litellm.include_cost_in_streaming_usage``, the same switch that puts it - into streamed usage.""" - if not isinstance(self.result, (ModelResponse, ResponsesAPIResponse)): - return - usage: Final = getattr(self.result, "usage", None) - if not isinstance(usage, (Usage, ResponseAPIUsage)): - return - if getattr(usage, "cost", None) is None and not litellm.include_cost_in_streaming_usage: - return - if stamp_billable_usage_cost(self.result, usage, response_cost) is None: - return - stamped: Final = get_hidden_params(self.result) - if stamped is None or USAGE_COST_BEFORE_LITELLM_STAMP not in stamped: - return - self._update_hidden_params({USAGE_COST_BEFORE_LITELLM_STAMP: stamped[USAGE_COST_BEFORE_LITELLM_STAMP]}) - def _update_hidden_params(self, new_params: Mapping[str, object]) -> None: """ Update hidden params - handles when self._hidden_params is a dict or HiddenParams object diff --git a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py index b71ac13684f..7cf024be2a1 100644 --- a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py +++ b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py @@ -13,7 +13,7 @@ from typing_extensions import ReadOnly, Required import litellm from litellm._logging import verbose_logger from litellm.litellm_core_utils.core_helpers import chunk_usage, provider_reported_usage_cost -from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR, set_hidden_params, stamp_billable_usage_cost +from litellm.litellm_core_utils.hidden_params import HIDDEN_PARAMS_ATTR, set_hidden_params from litellm.litellm_core_utils.llm_cost_calc.guardrail_cost import BEDROCK_TRACE_FIELDS, object_items, object_mapping from litellm.litellm_core_utils.prompt_templates.common_utils import ( get_content_from_model_response, # pyright: ignore[reportUnknownVariableType] # response param accepts an untyped dict @@ -1411,13 +1411,13 @@ def _set_stream_builder_response_cost( _set_hidden_response_cost(response, response_cost) -def _stamp_streaming_usage_cost(usage: Usage, response: ModelResponse, logging_obj: Optional["Logging"]) -> None: +def _stamp_streaming_response_cost(response: ModelResponse, logging_obj: Optional["Logging"]) -> None: if logging_obj is None: return - billed: Final = stamp_billable_usage_cost(response, usage, logging_obj.response_cost_calculator(result=response)) - if billed is None: + computed_cost: Final = logging_obj.response_cost_calculator(result=response) + if not isinstance(computed_cost, (int, float)) or computed_cost <= 0: return - _set_hidden_response_cost(response, billed) + _set_hidden_response_cost(response, float(computed_cost)) def _field_of(value: object, name: str) -> object: @@ -1587,7 +1587,7 @@ def stream_chunk_builder( response, logging_obj, ) - _stamp_streaming_usage_cost(usage, response, logging_obj) + _stamp_streaming_response_cost(response, logging_obj) _set_stream_builder_response_cost( response, logging_obj, @@ -1775,7 +1775,7 @@ def stream_chunk_builder( ) break - _stamp_streaming_usage_cost(usage, response, logging_obj) + _stamp_streaming_response_cost(response, logging_obj) _set_stream_builder_response_cost( response, logging_obj, diff --git a/litellm/proxy/common_request_processing.py b/litellm/proxy/common_request_processing.py index 3b9c17baa02..c5843da28d9 100644 --- a/litellm/proxy/common_request_processing.py +++ b/litellm/proxy/common_request_processing.py @@ -130,6 +130,7 @@ from litellm.proxy.route_llm_request import ( route_request, ) from litellm.proxy.telemetry.request_context import bind_call_id as bind_telemetry_call_id +from litellm.proxy.usage_cost_egress import with_client_usage_cost from litellm.proxy.utils import ( # noqa: F401 # legacy module exports ProxyLogging, _check_and_merge_model_level_guardrails, # pyright: ignore[reportPrivateUsage,reportUnusedImport] # backwards-compatible package export @@ -273,6 +274,7 @@ from litellm.proxy.litellm_pre_call_utils import ( ) from litellm.proxy.policy_engine.response_retrieval import attach_post_call_pipelines_to_retrieval from litellm.types.litellm_params import STREAM_USAGE_INJECTED_FIELD +from litellm.types.llms.openai import ResponsesAPIResponse from litellm.types.utils import ( ModelResponse, ModelResponseStream, @@ -3325,7 +3327,7 @@ class ProxyBaseLLMRequestProcessing: user_api_key_dict=user_api_key_dict, ) - return response + return with_client_usage_cost(response, response_cost_for_headers) @staticmethod async def _record_container_owners_from_responses_if_needed( @@ -4710,6 +4712,7 @@ class ProxyBaseLLMRequestProcessing: completion_tokens_details=usage.get("completion_tokens_details"), prompt_tokens_details=usage.get("prompt_tokens_details"), server_tool_use=usage.get("server_tool_use"), + cost=usage.get("cost"), ) @staticmethod @@ -4724,7 +4727,7 @@ class ProxyBaseLLMRequestProcessing: @staticmethod def _completion_cost_or_none( - model_response: ModelResponse, model_name: str, service_tier: str | None + model_response: ModelResponse | ResponsesAPIResponse, model_name: str, service_tier: str | None ) -> float | None: try: return litellm.completion_cost( @@ -4735,7 +4738,7 @@ class ProxyBaseLLMRequestProcessing: @staticmethod def _logging_obj_cost_or_none( - model_response: ModelResponse, litellm_logging_obj: LiteLLMLoggingObj + model_response: ModelResponse | ResponsesAPIResponse, litellm_logging_obj: LiteLLMLoggingObj ) -> float | None: # Pricing a frame stamps cost_breakdown and, on failure, the cost-failure debug key onto # the live logging object. The pass-through handlers never recompute either one, so a @@ -4762,7 +4765,7 @@ class ProxyBaseLLMRequestProcessing: @staticmethod def _streamed_usage_cost( - model_response: ModelResponse, + model_response: ModelResponse | ResponsesAPIResponse, model_name: str, service_tier: str | None, litellm_logging_obj: LiteLLMLoggingObj | None, diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index bc79becf7bb..5c7388f7861 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -922,6 +922,7 @@ from litellm.proxy.ui_crud_endpoints.proxy_setting_endpoints import ( from litellm.proxy.ui_crud_endpoints.user_banner_endpoints import ( router as user_banner_endpoints_router, ) +from litellm.proxy.usage_cost_egress import Pricer, with_client_stream_usage_cost from litellm.proxy.utils import ( # noqa: F401, RUF100 # legacy module exports PrismaClient, ProxyLogging, @@ -10134,6 +10135,22 @@ def _fast_serialize_simple_model_response_stream( return orjson.dumps(payload) +def _stream_usage_pricer(request_data: Mapping[str, object]) -> Pricer: + """Bills a usage-bearing stream event the way the request's logging object bills the spend row.""" + logging_obj: Final = request_data.get("litellm_logging_obj") + model_name: Final = str(request_data.get("model", "")) + + def price(reply: ModelResponse | ResponsesAPIResponse) -> float | None: + return ProxyBaseLLMRequestProcessing._streamed_usage_cost( # pyright: ignore[reportPrivateUsage] # the proxy's one stream pricer + reply, + model_name, + None, + logging_obj if isinstance(logging_obj, LiteLLMLoggingObj) else None, + ) + + return price + + def _serialize_streaming_chunk(chunk: BaseModel) -> str | bytes: if isinstance(chunk, ModelResponseStream): serialized_chunk: Final = _fast_serialize_simple_model_response_stream(chunk) @@ -10472,6 +10489,7 @@ async def async_data_generator( needs_per_chunk_hook: Final = proxy_logging_obj.needs_per_chunk_streaming_hook() is_raw_sse_stream: Final = bool(request_data.get("_litellm_raw_sse_stream")) strip_stream_usage: Final = bool(request_data.get("_litellm_strip_stream_usage")) + price_stream_usage: Final = _stream_usage_pricer(request_data) raw_sse_buffer = "" # A stream can start on a deployment with keepalive off and fall back @@ -10558,7 +10576,7 @@ async def async_data_generator( error_state.observe_chunk(chunk) raw_passthrough = False if isinstance(chunk, BaseModel): - chunk = _serialize_streaming_chunk(chunk) + chunk = _serialize_streaming_chunk(with_client_stream_usage_cost(chunk, price_stream_usage)) elif isinstance(chunk, bytes): chunk = chunk.decode("utf-8", errors="replace") if is_raw_sse_stream: diff --git a/litellm/proxy/usage_cost_egress.py b/litellm/proxy/usage_cost_egress.py new file mode 100644 index 00000000000..820fce85974 --- /dev/null +++ b/litellm/proxy/usage_cost_egress.py @@ -0,0 +1,77 @@ +"""What a client reads in ``usage.cost``: litellm's bill, written once on the way out of the proxy, onto a copy. + +Inside the pipeline ``usage.cost`` stays what the upstream reported, so pricing a reply again (an MCP round, a +logging copy, a stream rebuilt from its chunks) never reads litellm's own bill back as a provider report. +""" + +from __future__ import annotations + +import math +from collections.abc import Callable +from typing import Final, TypeVar + +from pydantic import BaseModel + +import litellm +from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIResponse +from litellm.types.utils import ModelResponse, ModelResponseStream, Usage + +Pricer = Callable[[ModelResponse | ResponsesAPIResponse], float | None] +"""Prices a reply the way the request's logging object bills it (deployment rates, discount, margin).""" + +_RESPONSES_USAGE_EVENTS: Final = frozenset({"response.completed", "response.incomplete"}) +_ReplyT = TypeVar("_ReplyT", bound=BaseModel) + + +def billable(cost: object) -> float | None: + if isinstance(cost, bool) or not isinstance(cost, (int, float)) or not math.isfinite(cost) or cost < 0: + return None + return float(cost) + + +def client_usage_cost(upstream_cost: object, bill: object, *, wanted: bool = False) -> float | None: + """The bill, when the upstream put a number in usage.cost or the operator wants it in every usage block + (``litellm.include_cost_in_streaming_usage``). None leaves the usage as the upstream sent it.""" + reported: Final = isinstance(upstream_cost, (int, float)) and not isinstance(upstream_cost, bool) + if not reported and not wanted and not litellm.include_cost_in_streaming_usage: + return None + return billable(bill) + + +def _with_usage_cost(reply: _ReplyT, usage: Usage | ResponseAPIUsage, cost: float) -> _ReplyT: + return reply.model_copy(update={"usage": usage.model_copy(update={"cost": cost})}) + + +def with_client_usage_cost(response: object, bill: object) -> object: + """A plain chat or Responses reply as the client gets it: a copy with the bill in usage.cost, or the reply + itself when its usage stays as the upstream sent it.""" + if not isinstance(response, (ModelResponse, ResponsesAPIResponse)): + return response + usage: Final = getattr(response, "usage", None) + if not isinstance(usage, (Usage, ResponseAPIUsage)): + return response + cost: Final = client_usage_cost(getattr(usage, "cost", None), bill) + if cost is None: + return response + return _with_usage_cost(response, usage, cost) + + +def with_client_stream_usage_cost(chunk: BaseModel, price: Pricer) -> BaseModel: + """A streamed event as the client gets it. Only the usage-bearing events are priced: the chat chunk carrying + ``usage`` and the Responses ``response.completed`` / ``response.incomplete`` events, which carry the bill + whether or not the upstream reported one, as they did before the bill moved out here.""" + if isinstance(chunk, ModelResponseStream): + usage: Final = getattr(chunk, "usage", None) + if not isinstance(usage, Usage): + return chunk + cost: Final = client_usage_cost(getattr(usage, "cost", None), price(ModelResponse(usage=usage))) + return chunk if cost is None else _with_usage_cost(chunk, usage, cost) + if getattr(chunk, "type", None) not in _RESPONSES_USAGE_EVENTS: + return chunk + response: Final = getattr(chunk, "response", None) + if not isinstance(response, ResponsesAPIResponse) or not isinstance(response.usage, ResponseAPIUsage): + return chunk + response_cost: Final = client_usage_cost(getattr(response.usage, "cost", None), price(response), wanted=True) + if response_cost is None: + return chunk + return chunk.model_copy(update={"response": _with_usage_cost(response, response.usage, response_cost)}) diff --git a/litellm/responses/litellm_completion_transformation/streaming_iterator.py b/litellm/responses/litellm_completion_transformation/streaming_iterator.py index b5ef9e95741..dc84a955d88 100644 --- a/litellm/responses/litellm_completion_transformation/streaming_iterator.py +++ b/litellm/responses/litellm_completion_transformation/streaming_iterator.py @@ -25,7 +25,7 @@ from litellm.responses.litellm_completion_transformation.tool_search import ( from litellm.responses.litellm_completion_transformation.transformation import ( LiteLLMCompletionResponsesConfig, ) -from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator, stamp_responses_usage_cost +from litellm.responses.streaming_iterator import ResponsesAPIStreamingIterator, normalize_responses_usage from litellm.responses.utils import ResponsesAPIRequestUtils from litellm.types.llms.openai import ( PART_UNION_TYPES, @@ -1480,7 +1480,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): } ) ) - stamp_responses_usage_cost(partial, self.litellm_logging_obj) + normalize_responses_usage(partial) return ResponseIncompleteEvent(type=ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, response=partial) def _emit_terminal_response_event( diff --git a/litellm/responses/streaming_iterator.py b/litellm/responses/streaming_iterator.py index ca6bbf0046c..9daf350baa7 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -41,7 +41,6 @@ from litellm.exceptions import MidStreamFallbackError, RateLimitError from litellm.litellm_core_utils.asyncify import run_async_function from litellm.litellm_core_utils.core_helpers import process_response_headers from litellm.litellm_core_utils.exception_mapping_utils import with_deployment_api_base -from litellm.litellm_core_utils.hidden_params import USAGE_COST_BEFORE_LITELLM_STAMP, stamp_billable_usage_cost from litellm.litellm_core_utils.internal_call_metadata import without_budget_reservation from litellm.litellm_core_utils.litellm_logging import BilledRoundsSpend from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj @@ -586,7 +585,7 @@ class BaseResponsesAPIStreamingIterator: else openai_responses_api_chunk.model_copy(update={"response": _billed_response}) ) self.completed_response = _terminal_chunk - stamp_responses_usage_cost(_billed_response, self.logging_obj) + normalize_responses_usage(_billed_response) if _chunk_type == openai_types.ResponsesAPIStreamEvents.RESPONSE_FAILED: self._handle_logging_failed_response() @@ -676,17 +675,10 @@ class BaseResponsesAPIStreamingIterator: headers: Final[Mapping[str, object]] = processed if isinstance(processed, Mapping) else EMPTY_MAPPING raw_headers: Final[Mapping[str, object]] = raw if isinstance(raw, Mapping) else EMPTY_MAPPING # rebuild by value and let existing keys win: sharing the source dicts would alias what the proxy - # splats into the client's HTTP headers, and copying non-header keys would carry response_cost. - # The provider's own usage.cost travels along, or the stamped bill would be priced as a provider report - stamp: Final[Mapping[str, object]] = ( - {USAGE_COST_BEFORE_LITELLM_STAMP: source[USAGE_COST_BEFORE_LITELLM_STAMP]} - if USAGE_COST_BEFORE_LITELLM_STAMP in source - else EMPTY_MAPPING - ) + # splats into the client's HTTP headers, and copying non-header keys would carry response_cost target.hidden_params = { "additional_headers": {**headers}, "headers": {**raw_headers}, - **stamp, **existing, } @@ -1077,7 +1069,7 @@ class BaseResponsesAPIStreamingIterator: "output": _partial_output_items(self._generated_content), } ) - stamp_responses_usage_cost(partial, self.logging_obj) + normalize_responses_usage(partial) return ResponseIncompleteEvent(type=ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, response=partial) def _note_yielded_event(self, event: ResponsesAPIStreamingResponse) -> None: @@ -1749,20 +1741,14 @@ def _estimate_usage_safely( return None -def stamp_responses_usage_cost( - response_obj: ResponsesAPIResponse | None, logging_obj: LiteLLMLoggingObj | None -) -> None: - if response_obj is None or logging_obj is None: +def normalize_responses_usage(response_obj: ResponsesAPIResponse | None) -> None: + """A terminal response built off an untyped dict carries its usage as a dict, logging and the proxy read a model.""" + if response_obj is None: return usage_obj: Final[ResponseAPIUsage | None] = _usage_as_model(getattr(response_obj, "usage", None)) if usage_obj is None: return - response_obj.usage = usage_obj # rebind-ok: the stamped cost has to ride on the response the client receives - try: - cost: Final[float | None] = logging_obj.response_cost_calculator(result=response_obj) - except Exception: - return - stamp_billable_usage_cost(response_obj, usage_obj, cost) + response_obj.usage = usage_obj # rebind-ok: the typed usage has to ride on the response the client receives def opening_item_payload(payload: Mapping[str, object]) -> Mapping[str, object]: @@ -1796,7 +1782,7 @@ def build_synthetic_response_events( chunk_size: int, ) -> list[ResponsesAPIStreamingResponse]: openai_types: Final = get_openai_response_types() - stamp_responses_usage_cost(transformed, logging_obj) + normalize_responses_usage(transformed) events: Final[list[ResponsesAPIStreamingResponse]] = [ _build_response_status_event(openai_types.ResponsesAPIStreamEvents.RESPONSE_CREATED, transformed), diff --git a/tests/unit/proxy/test_usage_cost_egress.py b/tests/unit/proxy/test_usage_cost_egress.py new file mode 100644 index 00000000000..3c62ddb9308 --- /dev/null +++ b/tests/unit/proxy/test_usage_cost_egress.py @@ -0,0 +1,200 @@ +from __future__ import annotations + +from typing import Final + +import pytest + +import litellm +from litellm.proxy.usage_cost_egress import with_client_stream_usage_cost, with_client_usage_cost +from litellm.types.llms.openai import ( + ResponseAPIUsage, + ResponseCompletedEvent, + ResponseIncompleteEvent, + ResponsesAPIResponse, + ResponsesAPIStreamEvents, +) +from litellm.types.utils import Choices, Message, ModelResponse, ModelResponseStream, StreamingChoices, Usage + +UPSTREAM_COST: Final = 0.037 +OPERATOR_BILL: Final = 100 * 2e-6 + 20 * 6e-6 + + +def _chat_reply(upstream_cost: float | None) -> ModelResponse: + reply: Final = ModelResponse( + id="chatcmpl-egress", + model="omni-model", + choices=[Choices(index=0, message=Message(role="assistant", content="hi"), finish_reason="stop")], + usage=Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120, cost=upstream_cost), + ) + reply._hidden_params["response_cost"] = OPERATOR_BILL + return reply + + +def _responses_reply(upstream_cost: float | None) -> ResponsesAPIResponse: + return ResponsesAPIResponse( + id="resp_egress", + created_at=1, + status="completed", + model="omni-model", + object="response", + output=[], + usage=ResponseAPIUsage(input_tokens=100, output_tokens=20, total_tokens=120, cost=upstream_cost), + ) + + +def _usage_chunk(upstream_cost: float | None) -> ModelResponseStream: + return ModelResponseStream( + id="chatcmpl-egress", + model="omni-model", + choices=[], + usage=Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120, cost=upstream_cost), + ) + + +def _bill(_reply: object) -> float | None: + return OPERATOR_BILL + + +def _unpriceable(_reply: object) -> float | None: + return None + + +@pytest.mark.parametrize("reply", [_chat_reply(UPSTREAM_COST), _responses_reply(UPSTREAM_COST)]) +def test_a_reply_whose_upstream_reported_a_cost_tells_the_client_what_litellm_billed(reply: object) -> None: + billed: Final = with_client_usage_cost(reply, OPERATOR_BILL) + + assert billed is not reply, "the pipeline's own object keeps the upstream figure" + assert getattr(getattr(billed, "usage", None), "cost", None) == pytest.approx(OPERATOR_BILL) + assert getattr(getattr(reply, "usage", None), "cost", None) == UPSTREAM_COST + + +def test_the_billed_copy_of_a_chat_reply_keeps_its_hidden_params() -> None: + reply: Final = _chat_reply(UPSTREAM_COST) + + billed: Final = with_client_usage_cost(reply, OPERATOR_BILL) + + assert isinstance(billed, ModelResponse) + assert billed._hidden_params["response_cost"] == pytest.approx(OPERATOR_BILL) + assert billed.model_dump()["usage"]["cost"] == pytest.approx(OPERATOR_BILL) + + +@pytest.mark.parametrize("reply", [_chat_reply(None), _responses_reply(None)]) +def test_a_reply_without_an_upstream_cost_goes_out_as_the_upstream_sent_it( + monkeypatch: pytest.MonkeyPatch, reply: object +) -> None: + monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", False) + + assert with_client_usage_cost(reply, OPERATOR_BILL) is reply + + +@pytest.mark.parametrize("reply", [_chat_reply(None), _responses_reply(None)]) +def test_include_cost_in_streaming_usage_puts_the_bill_on_a_reply_without_an_upstream_cost( + monkeypatch: pytest.MonkeyPatch, reply: object +) -> None: + monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", True) + + billed: Final = with_client_usage_cost(reply, OPERATOR_BILL) + + assert getattr(getattr(billed, "usage", None), "cost", None) == pytest.approx(OPERATOR_BILL) + + +def test_a_zero_bill_replaces_the_upstream_cost() -> None: + billed: Final = with_client_usage_cost(_chat_reply(UPSTREAM_COST), 0.0) + + assert getattr(getattr(billed, "usage", None), "cost", None) == 0.0 + + +@pytest.mark.parametrize("bill", [None, -0.001, float("nan"), float("inf"), True, "0.1"]) +def test_an_unpriceable_reply_keeps_the_upstream_cost(bill: object) -> None: + reply: Final = _chat_reply(UPSTREAM_COST) + + assert with_client_usage_cost(reply, bill) is reply + assert reply.usage is not None and reply.usage.cost == UPSTREAM_COST + + +def test_a_reply_that_is_neither_chat_nor_responses_passes_through() -> None: + reply: Final = {"usage": {"cost": UPSTREAM_COST}} + + assert with_client_usage_cost(reply, OPERATOR_BILL) is reply + + +def test_the_streamed_usage_chunk_tells_the_client_what_litellm_billed() -> None: + chunk: Final = _usage_chunk(UPSTREAM_COST) + + billed: Final = with_client_stream_usage_cost(chunk, _bill) + + assert isinstance(billed, ModelResponseStream) and billed is not chunk + assert billed.usage is not None and billed.usage.cost == pytest.approx(OPERATOR_BILL) + assert chunk.usage is not None and chunk.usage.cost == UPSTREAM_COST + + +def test_a_content_chunk_is_not_priced() -> None: + chunk: Final = ModelResponseStream( + id="chatcmpl-egress", model="omni-model", choices=[StreamingChoices(index=0, delta={"content": "hi"})] + ) + priced: list[object] = [] + + def price(reply: object) -> float | None: + priced.append(reply) + return OPERATOR_BILL + + assert with_client_stream_usage_cost(chunk, price) is chunk + assert priced == [] + + +def test_a_streamed_usage_chunk_without_an_upstream_cost_goes_out_as_sent(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", False) + chunk: Final = _usage_chunk(None) + + assert with_client_stream_usage_cost(chunk, _bill) is chunk + + +def test_include_cost_in_streaming_usage_puts_the_bill_on_the_streamed_usage_chunk( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", True) + + billed: Final = with_client_stream_usage_cost(_usage_chunk(None), _bill) + + assert isinstance(billed, ModelResponseStream) + assert billed.usage is not None and billed.usage.cost == pytest.approx(OPERATOR_BILL) + + +def test_the_stream_pricer_sees_the_upstream_cost_so_the_bill_matches_the_spend_row() -> None: + seen: list[object] = [] + + def price(reply: object) -> float | None: + seen.append(getattr(getattr(reply, "usage", None), "cost", None)) + return OPERATOR_BILL + + with_client_stream_usage_cost(_usage_chunk(UPSTREAM_COST), price) + + assert seen == [UPSTREAM_COST] + + +@pytest.mark.parametrize("upstream_cost", [UPSTREAM_COST, None]) +@pytest.mark.parametrize( + "event_type", [ResponsesAPIStreamEvents.RESPONSE_COMPLETED, ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE] +) +def test_a_terminal_responses_event_carries_the_bill_with_or_without_an_upstream_cost( + monkeypatch: pytest.MonkeyPatch, upstream_cost: float | None, event_type: ResponsesAPIStreamEvents +) -> None: + monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", False) + event_cls: Final = ( + ResponseCompletedEvent if event_type == ResponsesAPIStreamEvents.RESPONSE_COMPLETED else ResponseIncompleteEvent + ) + event: Final = event_cls(type=event_type, response=_responses_reply(upstream_cost)) + + billed: Final = with_client_stream_usage_cost(event, _bill) + + assert billed is not event + assert getattr(billed, "response").usage.cost == pytest.approx(OPERATOR_BILL) + assert event.response.usage is not None and event.response.usage.cost == upstream_cost + + +def test_an_unpriceable_terminal_responses_event_goes_out_as_sent() -> None: + event: Final = ResponseCompletedEvent( + type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED, response=_responses_reply(None) + ) + + assert with_client_stream_usage_cost(event, _unpriceable) is event diff --git a/tests/unit/responses/test_streaming_iterator.py b/tests/unit/responses/test_streaming_iterator.py index 8d624a186ad..bd9122a927c 100644 --- a/tests/unit/responses/test_streaming_iterator.py +++ b/tests/unit/responses/test_streaming_iterator.py @@ -3,15 +3,18 @@ completion_start_time on the first chunk so downstream TTFT consumers (Prometheus, OTEL, SpendLogs completionStartTime) do not fall back to completion_start_time = end_time.""" -import asyncio, importlib +import asyncio import functools import gc +import importlib import json import threading import weakref from collections.abc import Callable, Mapping +from contextlib import suppress from datetime import datetime -from typing import Final, Optional +from types import SimpleNamespace +from typing import Final from unittest.mock import AsyncMock, MagicMock, Mock, patch import httpx @@ -24,7 +27,8 @@ from litellm.exceptions import MidStreamFallbackError from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLoggingObj from litellm.llms.base_llm.responses.transformation import BaseResponsesAPIConfig -from litellm.responses.streaming_iterator import( +from litellm.responses import streaming_iterator as streaming_module +from litellm.responses.streaming_iterator import ( CachedResponsesAPIStreamingIterator, MockResponsesAPIStreamingIterator, ResponsesAPIStreamingIterator, @@ -38,15 +42,12 @@ from litellm.types.llms.openai import ( ResponsesAPIStreamEvents, ResponsesAPIStreamingResponse, ) -from contextlib import suppress -from litellm.responses import streaming_iterator as streaming_module from litellm.types.utils import CallTypes from tests._vcr_conftest_common import install_live_call_probe, record_vcr_outcome -from types import SimpleNamespace def _sse_event(payload: dict) -> bytes: - return f"data: {json.dumps(payload)}\n\n".encode("utf-8") + return f"data: {json.dumps(payload)}\n\n".encode() def _mock_config() -> Mock: @@ -80,7 +81,7 @@ def _make_iterator( *, sse_events: list[bytes], logging_obj: LiteLLMLoggingObj, - trailing_error: Optional[Exception] = None, + trailing_error: Exception | None = None, config: Mock | None = None, request_data: dict | None = None, ) -> ResponsesAPIStreamingIterator: @@ -109,7 +110,7 @@ def _make_sync_iterator( *, sse_events: list[bytes], logging_obj: LiteLLMLoggingObj, - trailing_error: Optional[Exception] = None, + trailing_error: Exception | None = None, ) -> SyncResponsesAPIStreamingIterator: def iter_bytes(): for evt in sse_events: @@ -489,7 +490,6 @@ class _LoopRecordingLogger(CustomLogger): self.hook_loop = asyncio.get_running_loop() await asyncio.sleep(0.05) self.hook_finished = True - return None class _SyncOnlyRecordingLogger(CustomLogger): @@ -838,51 +838,6 @@ def _responses_api_response_with_usage() -> ResponsesAPIResponse: ) -def test_stamp_responses_usage_cost_stamps_computed_cost(): - from litellm.responses.streaming_iterator import stamp_responses_usage_cost - - response = _responses_api_response_with_usage() - logging_obj = Mock(spec=LiteLLMLoggingObj) - logging_obj.response_cost_calculator.return_value = 0.000704 - - stamp_responses_usage_cost(response, logging_obj) - - assert getattr(response.usage, "cost", None) == pytest.approx(0.000704) - logging_obj.response_cost_calculator.assert_called_once_with(result=response) - - -def test_stamp_responses_usage_cost_puts_the_bill_over_the_provider_reported_cost(): - from litellm.litellm_core_utils.hidden_params import USAGE_COST_BEFORE_LITELLM_STAMP, provider_stated_usage_cost - from litellm.responses.streaming_iterator import stamp_responses_usage_cost - - response = _responses_api_response_with_usage() - setattr(response.usage, "cost", 0.5) - logging_obj = Mock(spec=LiteLLMLoggingObj) - logging_obj.response_cost_calculator.return_value = 0.00032 - - stamp_responses_usage_cost(response, logging_obj) - - assert getattr(response.usage, "cost", None) == pytest.approx(0.00032) - assert provider_stated_usage_cost(response, response.usage.cost) == pytest.approx(0.5) - assert response._hidden_params[USAGE_COST_BEFORE_LITELLM_STAMP] == pytest.approx(0.5) - - -def test_stamp_responses_usage_cost_stamps_a_zero_bill_only_over_a_provider_reported_cost(): - from litellm.responses.streaming_iterator import stamp_responses_usage_cost - - unpriced = _responses_api_response_with_usage() - reported = _responses_api_response_with_usage() - setattr(reported.usage, "cost", 0.5) - logging_obj = Mock(spec=LiteLLMLoggingObj) - logging_obj.response_cost_calculator.return_value = 0.0 - - stamp_responses_usage_cost(unpriced, logging_obj) - stamp_responses_usage_cost(reported, logging_obj) - - assert getattr(unpriced.usage, "cost", None) is None - assert getattr(reported.usage, "cost", None) == 0.0 - - def _unvalidated_response_with_dict_usage(usage: dict) -> ResponsesAPIResponse: return ResponsesAPIResponse.model_construct( id="resp_lit7391", @@ -896,8 +851,9 @@ def _unvalidated_response_with_dict_usage(usage: dict) -> ResponsesAPIResponse: ) -def test_stamp_responses_usage_cost_keeps_provider_cost_from_dict_usage(): - from litellm.responses.streaming_iterator import stamp_responses_usage_cost +def test_normalize_responses_usage_types_a_dict_usage_and_keeps_the_provider_cost(): + from litellm.responses.streaming_iterator import normalize_responses_usage + response = _unvalidated_response_with_dict_usage( { "input_tokens": 29, @@ -907,37 +863,20 @@ def test_stamp_responses_usage_cost_keeps_provider_cost_from_dict_usage(): "cost": {"currency": "USD", "input_cost": 0, "output_cost": 3e-05, "total_cost": 3e-05}, } ) - logging_obj = Mock(spec=LiteLLMLoggingObj) - logging_obj.response_cost_calculator.return_value = 3e-05 - stamp_responses_usage_cost(response, logging_obj) + normalize_responses_usage(response) assert isinstance(response.usage, ResponseAPIUsage) assert response.usage.cost == pytest.approx(3e-05) assert response.usage.output_tokens_details.reasoning_tokens == 117 -def test_stamp_responses_usage_cost_computes_cost_for_dict_usage_without_cost(): - from litellm.responses.streaming_iterator import stamp_responses_usage_cost - response = _unvalidated_response_with_dict_usage({"input_tokens": 29, "output_tokens": 120, "total_tokens": 149}) - logging_obj = Mock(spec=LiteLLMLoggingObj) - logging_obj.response_cost_calculator.return_value = 0.000704 - - stamp_responses_usage_cost(response, logging_obj) - - assert isinstance(response.usage, ResponseAPIUsage) - assert response.usage.cost == pytest.approx(0.000704) - logging_obj.response_cost_calculator.assert_called_once_with(result=response) - - -def test_stamp_responses_usage_cost_survives_calculator_failure(): - from litellm.responses.streaming_iterator import stamp_responses_usage_cost +def test_normalize_responses_usage_never_writes_a_cost_the_provider_did_not_report(): + from litellm.responses.streaming_iterator import normalize_responses_usage response = _responses_api_response_with_usage() - logging_obj = Mock(spec=LiteLLMLoggingObj) - logging_obj.response_cost_calculator.side_effect = RuntimeError("cost map unavailable") - stamp_responses_usage_cost(response, logging_obj) + normalize_responses_usage(response) assert getattr(response.usage, "cost", None) is None @@ -959,7 +898,7 @@ def _capture_dispatch(logged: list): return _dispatch -def _headers_config(*, transform_hidden_params: Optional[dict] = None) -> Mock: +def _headers_config(*, transform_hidden_params: dict | None = None) -> Mock: """Config whose completed event carries a real ResponsesAPIResponse, so the logging copy performs a genuine model_dump/model_validate round trip.""" mock_config = Mock(spec=BaseResponsesAPIConfig) @@ -1274,10 +1213,10 @@ async def test_completed_event_without_usage_counts_tool_input_deltas(tool_delta @pytest.mark.asyncio -async def test_completed_event_with_a_dict_response_is_typed_and_billed(): +async def test_completed_event_with_a_dict_response_is_typed(): """transform_streaming_response can model_construct a terminal event whose response stays a plain dict; the iterator must type it so the estimated - usage reaches the cost stamping path.""" + usage reaches logging and the proxy egress as a model.""" dict_response: Final = { "id": "resp_dict", "model": "gpt-4o-mini", @@ -1319,8 +1258,7 @@ async def test_completed_event_with_a_dict_response_is_typed_and_billed(): assert usage is not None assert usage.input_tokens > 0 assert usage.output_tokens > 0 - assert usage.cost == pytest.approx(0.000704) - logging_obj.response_cost_calculator.assert_any_call(result=completed_response) + assert getattr(usage, "cost", None) is None, "the bill goes on at the proxy egress, never on the event logging prices" def test_billed_terminal_response_keeps_a_response_that_already_has_usage(): @@ -1390,6 +1328,7 @@ def _vcr_outcome_gate(request, vcr): yield record_vcr_outcome(request, vcr) + @pytest.fixture(scope="function") def setup_and_teardown(): """ @@ -1407,6 +1346,7 @@ def setup_and_teardown(): loop.close() asyncio.set_event_loop(None) + class _FakeLoggingObj: def __init__(self): self.success_calls = 0 @@ -1447,10 +1387,12 @@ class _FakeLoggingObj: self.completion_start_time = completion_start_time self.model_call_details["completion_start_time"] = completion_start_time + async def _await_tasks_spawned_since(existing: set["asyncio.Task[object]"]) -> None: spawned: Final = asyncio.all_tasks() - existing await asyncio.wait_for(asyncio.gather(*spawned, return_exceptions=True), timeout=5) + def _make_completed_response(response_id: str = "resp_test") -> ResponseCompletedEvent: return ResponseCompletedEvent( type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED, @@ -1478,6 +1420,7 @@ def _make_completed_response(response_id: str = "resp_test") -> ResponseComplete ), ) + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") @pytest.mark.asyncio async def test_log_background_task_failure_logs_task_exceptions(monkeypatch): @@ -1500,6 +1443,7 @@ async def test_log_background_task_failure_logs_task_exceptions(monkeypatch): task.exception(), ) + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") @pytest.mark.asyncio async def test_log_background_task_failure_ignores_cancelled_tasks(monkeypatch): @@ -1515,6 +1459,7 @@ async def test_log_background_task_failure_ignores_cancelled_tasks(monkeypatch): error_logger.assert_not_called() + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") def test_content_part_done_event_supports_refusal_and_reasoning_text(): refusal_event = streaming_module._build_content_part_done_event( @@ -1542,6 +1487,7 @@ def test_content_part_done_event_supports_refusal_and_reasoning_text(): assert reasoning_event.part.reasoning == "because" assert unsupported_event is None + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") def test_dump_response_object_handles_model_and_unknown_values(): response = ResponsesAPIResponse( @@ -1557,6 +1503,7 @@ def test_dump_response_object_handles_model_and_unknown_values(): assert streaming_module._dump_response_object({"type": "message"}) == {"type": "message"} assert streaming_module._dump_response_object(object()) == {} + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") @pytest.mark.asyncio async def test_responses_streaming_triggers_hooks(monkeypatch): @@ -1613,6 +1560,7 @@ async def test_responses_streaming_triggers_hooks(monkeypatch): assert seen["request_data"].get("litellm_params") is not None assert seen["call_type"] == CallTypes.responses + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") @pytest.mark.asyncio async def test_responses_streaming_calls_post_streaming_deployment_hook(monkeypatch): @@ -1653,6 +1601,7 @@ async def test_responses_streaming_calls_post_streaming_deployment_hook(monkeypa # reset callbacks litellm.callbacks = original_callbacks + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") @pytest.mark.asyncio async def test_responses_streaming_failure_triggers_failure_handlers(): @@ -1682,6 +1631,7 @@ async def test_responses_streaming_failure_triggers_failure_handlers(): assert logging_obj.failure_calls >= 1 assert logging_obj.async_failure_calls >= 1 + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") def test_process_chunk_requires_provider_config(): iterator = ResponsesAPIStreamingIterator( @@ -1696,6 +1646,7 @@ def test_process_chunk_requires_provider_config(): with pytest.raises(ValueError, match="responses_api_provider_config is required"): iterator._process_chunk(json.dumps({"type": "response.completed"})) + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") def test_process_chunk_wraps_encrypted_content_with_model_id(): openai_types = streaming_module._get_openai_response_types() @@ -1730,8 +1681,9 @@ def test_process_chunk_wraps_encrypted_content_with_model_id(): assert event.item.encrypted_content.startswith("litellm_enc:") assert event.item.encrypted_content.endswith(";ciphertext") + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") -def test_process_chunk_completed_response_updates_id_and_usage_cost(monkeypatch): +def test_process_chunk_completed_response_updates_id_and_leaves_usage_cost_alone(monkeypatch): original_include_cost = litellm.include_cost_in_streaming_usage litellm.include_cost_in_streaming_usage = True openai_types = streaming_module._get_openai_response_types() @@ -1780,9 +1732,10 @@ def test_process_chunk_completed_response_updates_id_and_usage_cost(monkeypatch) assert iterator.completed_response is event assert event.response.id != "resp_live" assert event.response.id.startswith("resp_") - assert event.response.usage.cost == 1.23 + assert event.response.usage.cost is None, "the bill goes on at the proxy egress, never on the event logging prices" completion_handler.assert_called_once() + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") def test_process_chunk_failed_response_triggers_failure_logging(monkeypatch): openai_types = streaming_module._get_openai_response_types() @@ -1818,6 +1771,7 @@ def test_process_chunk_failed_response_triggers_failure_logging(monkeypatch): assert iterator.completed_response is event failure_handler.assert_called_once() + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") @pytest.mark.asyncio async def test_handle_logging_failed_response_uses_response_error_message(): @@ -1850,6 +1804,7 @@ async def test_handle_logging_failed_response_uses_response_error_message(): assert logging_obj.failure_calls == 1 assert logging_obj.async_failure_calls == 1 + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") def test_process_chunk_returns_none_for_invalid_json_and_non_dict_payload(): class _NoopConfig: @@ -1868,6 +1823,7 @@ def test_process_chunk_returns_none_for_invalid_json_and_non_dict_payload(): assert iterator._process_chunk("not-json") is None assert iterator._process_chunk(json.dumps(["not", "a", "dict"])) is None + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") def test_process_chunk_cost_annotation_failure_is_nonfatal(monkeypatch): original_include_cost = litellm.include_cost_in_streaming_usage @@ -1915,6 +1871,7 @@ def test_process_chunk_cost_annotation_failure_is_nonfatal(monkeypatch): assert event.response.usage.cost is None completion_handler.assert_called_once() + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") def test_get_completed_response_object_accepts_direct_response(): logging_obj = _FakeLoggingObj() @@ -1931,6 +1888,7 @@ def test_get_completed_response_object_accepts_direct_response(): assert iterator._get_completed_response_object() is direct_response + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") @pytest.mark.asyncio async def test_responses_streaming_completed_event_persists_async_cache(): @@ -1979,6 +1937,7 @@ async def test_responses_streaming_completed_event_persists_async_cache(): assert json.loads(litellm.cache.async_add_cache.call_args.args[0])["id"] == iterator.completed_response.response.id litellm.cache = original_cache + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") def test_responses_streaming_completed_event_persists_sync_cache(): logging_obj = _FakeLoggingObj() @@ -2024,6 +1983,7 @@ def test_responses_streaming_completed_event_persists_sync_cache(): assert json.loads(litellm.cache.add_cache.call_args.args[0])["id"] == iterator.completed_response.response.id litellm.cache = original_cache + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") def test_log_completed_response_sync_direct_path(monkeypatch): hook_calls = {"post_call": 0, "metadata": 0} @@ -2065,6 +2025,7 @@ def test_log_completed_response_sync_direct_path(monkeypatch): assert hook_calls["post_call"] == 1 assert hook_calls["metadata"] == 1 + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") def test_log_completed_response_falls_back_when_model_validate_fails(monkeypatch): class _BadSerializableResponse: @@ -2094,6 +2055,7 @@ def test_log_completed_response_falls_back_when_model_validate_fails(monkeypatch assert logging_obj.success_calls == 1 assert logging_obj.async_success_calls == 1 + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") @pytest.mark.parametrize( "scenario", @@ -2158,6 +2120,7 @@ def test_persist_completed_response_to_cache_guard_branches(monkeypatch, scenari expected_cached_flag = scenario == "already_cached" assert iterator._completed_response_cached is expected_cached_flag + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") def test_build_synthetic_response_events_covers_annotations_function_calls_and_refusals(): original_include_cost = litellm.include_cost_in_streaming_usage @@ -2216,6 +2179,7 @@ def test_build_synthetic_response_events_covers_annotations_function_calls_and_r assert "response.function_call_arguments.done" in event_types assert event_types[-1] == "response.completed" + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") @pytest.mark.asyncio async def test_mock_responses_streaming_iterator_async_iteration_logs_completion( @@ -2266,6 +2230,7 @@ async def test_mock_responses_streaming_iterator_async_iteration_logs_completion assert hook_calls["post_call"] == 1 assert hook_calls["metadata"] == 1 + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") def test_mock_responses_streaming_iterator_sync_iteration_logs_completion(monkeypatch): hook_calls = {"post_call": 0, "metadata": 0} @@ -2311,6 +2276,7 @@ def test_mock_responses_streaming_iterator_sync_iteration_logs_completion(monkey assert hook_calls["post_call"] == 1 assert hook_calls["metadata"] == 1 + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") @pytest.mark.asyncio async def test_cached_responses_stream_async_hit_triggers_success_callbacks( @@ -2370,6 +2336,7 @@ async def test_cached_responses_stream_async_hit_triggers_success_callbacks( litellm.cache.add_cache.assert_not_called() litellm.cache = original_cache + @pytest.mark.usefixtures("_vcr_outcome_gate", "setup_and_teardown") def test_cached_responses_stream_sync_hit_triggers_success_callbacks(monkeypatch): hook_calls = {"post_call": 0, "metadata": 0} @@ -2468,7 +2435,9 @@ def test_faked_stream_bridged_to_chat_sends_tool_arguments_once(): tool_call_deltas: Final = tuple( tool_call for chunk in chunks - for tool_call in (chunk.choices[0].delta.tool_calls or ()) # comprehension-ok: flattening each chunk's tool calls + for tool_call in ( + chunk.choices[0].delta.tool_calls or () + ) # comprehension-ok: flattening each chunk's tool calls ) joined_arguments: Final = "".join(tool_call.function.arguments or "" for tool_call in tool_call_deltas) joined_text: Final = "".join(chunk.choices[0].delta.content or "" for chunk in chunks) @@ -2524,7 +2493,6 @@ def test_faked_stream_opens_items_in_progress_and_empty_like_a_real_stream(): assert [part.text for part in added_parts] == [""] - def _answer_with_reasoning_refusal_and_custom_tool_call() -> ResponsesAPIResponse: return ResponsesAPIResponse( id="resp_synthetic", @@ -2658,7 +2626,9 @@ async def test_shutdown_drain_waits_for_a_pending_logging_task(): @pytest.mark.asyncio async def test_shutdown_drain_gives_up_on_a_stuck_logging_task_at_its_timeout(): - stuck: Final = streaming_module.spawn_logging_task(asyncio.get_running_loop(), asyncio.Event().wait(), task_name="stuck") + stuck: Final = streaming_module.spawn_logging_task( + asyncio.get_running_loop(), asyncio.Event().wait(), task_name="stuck" + ) await asyncio.wait_for(streaming_module.drain_logging_tasks(timeout=0.05), timeout=1) @@ -2800,39 +2770,3 @@ async def test_websocket_turns_logging_survives_garbage_collection(terminal_even await _run_websocket_turn(logger, terminal_event) assert await _logging_tasks_left_after_garbage_collection(logger) == 1 - - -def test_logging_copy_keeps_the_provider_usage_cost_behind_a_stamped_bill() -> None: - from litellm.litellm_core_utils.hidden_params import USAGE_COST_BEFORE_LITELLM_STAMP - from litellm.responses.streaming_iterator import _logging_copy - - logging_obj = LiteLLMLoggingObj( - model="gpt-4o-mini", - messages=[{"role": "user", "content": "hi"}], - stream=True, - call_type="aresponses", - start_time=datetime.now(), - litellm_call_id="stamped-copy", - function_id="stamped-copy", - ) - iterator = _make_iterator(sse_events=[], logging_obj=logging_obj) - billed = ResponsesAPIResponse( - id="resp_stamped", - created_at=1, - status="completed", - model="gpt-4o-mini", - object="response", - output=[], - usage={"input_tokens": 100, "output_tokens": 20, "total_tokens": 120, "cost": 0.63}, - ) - billed.hidden_params = {USAGE_COST_BEFORE_LITELLM_STAMP: 0.42, "response_cost": 0.63} - iterator.completed_response = ResponseCompletedEvent( - type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED, response=billed - ) - logging_response = _logging_copy(iterator.completed_response) - - iterator._restore_provider_response_headers(logging_response) - - copied = logging_response.response.hidden_params - assert copied[USAGE_COST_BEFORE_LITELLM_STAMP] == 0.42 - assert "response_cost" not in copied diff --git a/tests/unit/test_cost_calculator.py b/tests/unit/test_cost_calculator.py index b6ff6cadb8d..78480dc4f31 100644 --- a/tests/unit/test_cost_calculator.py +++ b/tests/unit/test_cost_calculator.py @@ -1,48 +1,75 @@ -import json import asyncio import datetime +import json +import os +import threading import time from collections.abc import Iterator, Mapping from types import MappingProxyType, SimpleNamespace -from typing import Final, Literal, cast +from typing import Final, cast +from unittest.mock import MagicMock, patch import httpx import pytest import respx -from openai.types.completion_usage import CompletionUsage from pydantic import BaseModel import litellm from litellm import TranscriptionResponse, model_cost from litellm.constants import REALTIME_METERED_USAGE_KEY -from litellm.cost_calculator import BaseTokenUsageProcessor, RealtimeAPITokenUsageProcessor, ResponsesWebSocketTokenUsageProcessor, batch_cost_calculator, completion_cost, cost_per_token, handle_realtime_stream_cost_calculation, response_cost_calculator -from litellm.litellm_core_utils.llm_cost_calc.utils import parse_prompt_tokens_details +from litellm.cost_calculator import ( + BaseTokenUsageProcessor, + RealtimeAPITokenUsageProcessor, + ResponsesWebSocketTokenUsageProcessor, + batch_cost_calculator, + completion_cost, + cost_per_token, + handle_realtime_stream_cost_calculation, + response_cost_calculator, +) from litellm.integrations.custom_logger import CustomLogger from litellm.litellm_core_utils.fallback_generalizations import ( get_fallback_generalization_rules, set_fallback_generalizations, ) -from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER from litellm.litellm_core_utils.litellm_logging import Logging -from litellm.litellm_core_utils.llm_response_utils.response_metadata import update_response_metadata -from litellm.responses.streaming_iterator import stamp_responses_usage_cost +from litellm.litellm_core_utils.llm_cost_calc.utils import parse_prompt_tokens_details from litellm.litellm_core_utils.llm_response_utils.convert_dict_to_response import ( convert_to_model_response_object, ) +from litellm.litellm_core_utils.logging_worker import GLOBAL_LOGGING_WORKER from litellm.llms.base_llm.ocr.transformation import OCRPage, OCRResponse, OCRUsageInfo from litellm.llms.custom_httpx.http_handler import HTTPHandler from litellm.llms.fireworks_ai.cost_calculator import get_base_model_for_pricing -from litellm.llms.together_ai.cost_calculator import get_model_params_and_category from litellm.types.llms.base import CachedTokensDetails -from litellm.types.llms.openai import HttpxBinaryResponseContent, OpenAIRealtimeStreamList, ResponseAPIUsage, ResponsesAPIResponse +from litellm.types.llms.openai import ( + HttpxBinaryResponseContent, + OpenAIRealtimeStreamList, + ResponseAPIUsage, + ResponsesAPIResponse, +) from litellm.types.llms.xai import RealtimeMeteredUsage from litellm.types.rerank import RerankResponse -from litellm.types.utils import CacheCreationTokenDetails, CallTypes, Choices, CompletionTokensDetailsWrapper, EmbeddingResponse, ImageObject, ImageResponse, ImageUsage, ImageUsageInputTokensDetails, LiteLLMRealtimeStreamLoggingObject, Message, ModelInfo, ModelResponse, PromptTokensDetails, PromptTokensDetailsWrapper, Usage +from litellm.types.utils import ( + CacheCreationTokenDetails, + CallTypes, + Choices, + CompletionTokensDetailsWrapper, + EmbeddingResponse, + ImageObject, + ImageResponse, + ImageUsage, + ImageUsageInputTokensDetails, + LiteLLMRealtimeStreamLoggingObject, + Message, + ModelInfo, + ModelResponse, + PromptTokensDetails, + PromptTokensDetailsWrapper, + Usage, +) from litellm.types.videos.main import VideoObject from litellm.utils import supports_prompt_caching -import os -import threading -from unittest.mock import MagicMock, patch class ResponseCostCapture(CustomLogger): @@ -518,9 +545,7 @@ def test_completion_cost_prices_base_model_under_its_own_provider( ) info: Final = litellm.get_model_info("claude-sonnet-4-5", custom_llm_provider="anthropic") - assert cost == pytest.approx( - 1000 * info["input_cost_per_token"] + 1000 * info["output_cost_per_token"] - ) + assert cost == pytest.approx(1000 * info["input_cost_per_token"] + 1000 * info["output_cost_per_token"]) def test_completion_cost_strips_dated_azure_snapshot_model(_local_model_cost_map: None) -> None: @@ -1615,9 +1640,7 @@ def test_per_query_priced_rerank_deployment_completion_cost_is_nonzero(): "vertex_ai/semantic-ranker-fast-004", ), ) -def test_vertex_semantic_ranker_rerank_completion_cost_is_nonzero( - model: str, _local_model_cost_map: None -) -> None: +def test_vertex_semantic_ranker_rerank_completion_cost_is_nonzero(model: str, _local_model_cost_map: None) -> None: response: Final = RerankResponse( id="vertex_ai_rerank_pricing_test", results=[], @@ -3373,9 +3396,7 @@ def test_gemini_cache_tokens_details_no_negative_values(): assert usage.prompt_tokens_details.text_tokens == 9402 assert usage.prompt_tokens_details.image_tokens == 258 assert usage.prompt_tokens_details.cached_tokens == 9651 - assert usage.prompt_tokens_details.cached_tokens_details == CachedTokensDetails( - text_tokens=9393, image_tokens=258 - ) + assert usage.prompt_tokens_details.cached_tokens_details == CachedTokensDetails(text_tokens=9393, image_tokens=258) assert prompt_details["text_tokens"] == 9 assert prompt_details["image_tokens"] == 0 assert prompt_details["text_tokens"] >= 0 @@ -6876,23 +6897,17 @@ def test_cost_ft_gpt_35(): usage=Usage(prompt_tokens=21, completion_tokens=17, total_tokens=38), ) - cost = litellm.completion_cost( - completion_response=resp, custom_llm_provider="openai" - ) + cost = litellm.completion_cost(completion_response=resp, custom_llm_provider="openai") print("\n Calculated Cost for ft:gpt-3.5", cost) input_cost = model_cost["ft:gpt-3.5-turbo"]["input_cost_per_token"] output_cost = model_cost["ft:gpt-3.5-turbo"]["output_cost_per_token"] print(input_cost, output_cost) - expected_cost = (input_cost * resp.usage.prompt_tokens) + ( - output_cost * resp.usage.completion_tokens - ) + expected_cost = (input_cost * resp.usage.prompt_tokens) + (output_cost * resp.usage.completion_tokens) print("\n Excpected cost", expected_cost) assert cost == expected_cost except Exception as e: print(f"Error: {e}") - pytest.fail( - f"Cost Calc failed for ft:gpt-3.5. Expected {expected_cost}, Calculated cost {cost}" - ) + pytest.fail(f"Cost Calc failed for ft:gpt-3.5. Expected {expected_cost}, Calculated cost {cost}") def test_cost_azure_gpt_35(): @@ -6918,28 +6933,22 @@ def test_cost_azure_gpt_35(): usage=Usage(prompt_tokens=21, completion_tokens=17, total_tokens=38), ) - cost = litellm.completion_cost( - completion_response=resp, model="azure/chatgpt-deployment-2" - ) + cost = litellm.completion_cost(completion_response=resp, model="azure/chatgpt-deployment-2") print("\n Calculated Cost for azure/gpt-3.5-turbo", cost) input_cost = model_cost["azure/gpt-35-turbo"]["input_cost_per_token"] output_cost = model_cost["azure/gpt-35-turbo"]["output_cost_per_token"] - expected_cost = (input_cost * resp.usage.prompt_tokens) + ( - output_cost * resp.usage.completion_tokens - ) + expected_cost = (input_cost * resp.usage.prompt_tokens) + (output_cost * resp.usage.completion_tokens) print("\n Excpected cost", expected_cost) assert cost == expected_cost except Exception as e: - pytest.fail(f"Cost Calc failed for azure/gpt-3.5-turbo. {str(e)}") + pytest.fail(f"Cost Calc failed for azure/gpt-3.5-turbo. {e!s}") def test_cost_bedrock_pricing_actual_calls(): litellm.set_verbose = True model = "anthropic.claude-3-5-sonnet-20240620-v1:0" messages = [{"role": "user", "content": "Hey, how's it going?"}] - response = litellm.completion( - model=model, messages=messages, mock_response="hello cool one" - ) + response = litellm.completion(model=model, messages=messages, mock_response="hello cool one") print("response", response) cost = litellm.completion_cost( @@ -6970,8 +6979,7 @@ def test_whisper_openai(): print(f"cost: {cost}") print(f"whisper dict: {litellm.model_cost['whisper-1']}") expected_cost = round( - litellm.model_cost["whisper-1"]["output_cost_per_second"] - * _total_time_in_seconds, + litellm.model_cost["whisper-1"]["output_cost_per_second"] * _total_time_in_seconds, 5, ) assert round(cost, 5) == round(expected_cost, 5) @@ -6991,15 +6999,12 @@ def test_whisper_azure(): _total_time_in_seconds = 3 setattr(transcription, "duration", _total_time_in_seconds) - cost = litellm.completion_cost( - model="azure/azure-whisper", completion_response=transcription - ) + cost = litellm.completion_cost(model="azure/azure-whisper", completion_response=transcription) print(f"cost: {cost}") print(f"whisper dict: {litellm.model_cost['whisper-1']}") expected_cost = round( - litellm.model_cost["whisper-1"]["output_cost_per_second"] - * _total_time_in_seconds, + litellm.model_cost["whisper-1"]["output_cost_per_second"] * _total_time_in_seconds, 5, ) assert round(cost, 5) == round(expected_cost, 5) @@ -7046,9 +7051,7 @@ def test_transcription_cost_falls_back_to_provider_billed_seconds( response_type="audio_transcription", ) - cost: Final = completion_cost( - completion_response=transcription, model=requested_model, call_type="atranscription" - ) + cost: Final = completion_cost(completion_response=transcription, model=requested_model, call_type="atranscription") assert cost == pytest.approx(0.01 * expected_seconds) @@ -7140,9 +7143,7 @@ def test_replicate_llama3_cost_tracking(): model="replicate/meta/meta-llama-3-8b-instruct", object="chat.completion", system_fingerprint=None, - usage=litellm.utils.Usage( - prompt_tokens=48, completion_tokens=31, total_tokens=79 - ), + usage=litellm.utils.Usage(prompt_tokens=48, completion_tokens=31, total_tokens=79), ) cost = litellm.completion_cost( completion_response=response, @@ -7152,20 +7153,14 @@ def test_replicate_llama3_cost_tracking(): print(f"cost: {cost}") cost = round(cost, 5) expected_cost = round( - litellm.model_cost["replicate/meta/meta-llama-3-8b-instruct"][ - "input_cost_per_token" - ] - * 48 - + litellm.model_cost["replicate/meta/meta-llama-3-8b-instruct"][ - "output_cost_per_token" - ] - * 31, + litellm.model_cost["replicate/meta/meta-llama-3-8b-instruct"]["input_cost_per_token"] * 48 + + litellm.model_cost["replicate/meta/meta-llama-3-8b-instruct"]["output_cost_per_token"] * 31, 5, ) assert cost == expected_cost -@pytest.mark.parametrize("is_streaming", [True, False]) # +@pytest.mark.parametrize("is_streaming", [True, False]) def test_groq_response_cost_tracking(is_streaming): from litellm.utils import ( CallTypes, @@ -7295,7 +7290,7 @@ def test_vertex_ai_completion_cost(): model_info = litellm.get_model_info(model="gemini-3.8-flash") - print("\nExpected model info:\n{}\n\n".format(model_info)) + print(f"\nExpected model info:\n{model_info}\n\n") expected_input_cost = prompt_tokens * model_info["input_cost_per_token"] @@ -7308,8 +7303,8 @@ def test_vertex_ai_completion_cost(): ) assert round(expected_input_cost, 6) == round(calculated_input_cost, 6) - print("expected_input_cost: {}".format(expected_input_cost)) - print("calculated_input_cost: {}".format(calculated_input_cost)) + print(f"expected_input_cost: {expected_input_cost}") + print(f"calculated_input_cost: {calculated_input_cost}") def test_vertex_ai_medlm_completion_cost(): @@ -7320,9 +7315,7 @@ def test_vertex_ai_medlm_completion_cost(): model = "vertex_ai/medlm-medium" messages = [{"role": "user", "content": "Test MedLM completion cost."}] - predictive_cost = completion_cost( - model=model, messages=messages, custom_llm_provider="vertex_ai" - ) + predictive_cost = completion_cost(model=model, messages=messages, custom_llm_provider="vertex_ai") assert predictive_cost > 0 model = "vertex_ai/medlm-large" @@ -7339,13 +7332,11 @@ def test_vertex_ai_embedding_completion_cost(caplog): litellm.model_cost = litellm.get_model_cost_map(url="") text = "The quick brown fox jumps over the lazy dog." - input_tokens = litellm.token_counter( - model="vertex_ai/text-embedding-004", text=text - ) + input_tokens = litellm.token_counter(model="vertex_ai/text-embedding-004", text=text) model_info = litellm.get_model_info(model="vertex_ai/text-embedding-004") - print("\nExpected model info:\n{}\n\n".format(model_info)) + print(f"\nExpected model info:\n{model_info}\n\n") expected_input_cost = input_tokens * model_info["input_cost_per_token"] @@ -7358,16 +7349,13 @@ def test_vertex_ai_embedding_completion_cost(caplog): ) assert round(expected_input_cost, 6) == round(calculated_input_cost, 6) - print("expected_input_cost: {}".format(expected_input_cost)) - print("calculated_input_cost: {}".format(calculated_input_cost)) + print(f"expected_input_cost: {expected_input_cost}") + print(f"calculated_input_cost: {calculated_input_cost}") captured_logs = [rec.message for rec in caplog.records] for item in captured_logs: - print("\nitem:{}\n".format(item)) - if ( - "litellm.litellm_core_utils.llm_cost_calc.google.cost_per_character(): Exception occured " - in item - ): + print(f"\nitem:{item}\n") + if "litellm.litellm_core_utils.llm_cost_calc.google.cost_per_character(): Exception occured " in item: raise Exception("Error log raised for calculating embedding cost") @@ -7409,9 +7397,7 @@ def test_vertex_ai_mistral_predict_cost(usage): else: from openai.types.completion_usage import CompletionUsage - response_usage = CompletionUsage( - prompt_tokens=32, completion_tokens=55, total_tokens=87 - ) + response_usage = CompletionUsage(prompt_tokens=32, completion_tokens=55, total_tokens=87) response_object = ModelResponse( id="26c0ef045020429d9c5c9b078c01e564", choices=[ @@ -7445,9 +7431,7 @@ def test_vertex_ai_mistral_predict_cost(usage): assert predictive_cost > 0 -@pytest.mark.parametrize( - "model", ["openai/tts-1", "azure/tts-1", "openai/gpt-4o-mini-tts"] -) +@pytest.mark.parametrize("model", ["openai/tts-1", "azure/tts-1", "openai/gpt-4o-mini-tts"]) def test_completion_cost_tts(model): os.environ["LITELLM_LOCAL_MODEL_COST_MAP"] = "True" litellm.model_cost = litellm.get_model_cost_map(url="") @@ -7546,9 +7530,7 @@ def test_completion_cost_prompt_caching(model, custom_llm_provider): cost_1 = completion_cost(model=model, completion_response=response_1) - _model_info = litellm.get_model_info( - model=model, custom_llm_provider=custom_llm_provider - ) + _model_info = litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider) expected_cost = ( ( response_1.usage.prompt_tokens @@ -7556,12 +7538,9 @@ def test_completion_cost_prompt_caching(model, custom_llm_provider): - response_1.usage.prompt_tokens_details.cache_creation_tokens ) * _model_info["input_cost_per_token"] - + (response_1.usage.prompt_tokens_details.cached_tokens or 0) - * _model_info["cache_read_input_token_cost"] - + (response_1.usage.cache_creation_input_tokens or 0) - * _model_info["cache_creation_input_token_cost"] - + (response_1.usage.completion_tokens or 0) - * _model_info["output_cost_per_token"] + + (response_1.usage.prompt_tokens_details.cached_tokens or 0) * _model_info["cache_read_input_token_cost"] + + (response_1.usage.cache_creation_input_tokens or 0) * _model_info["cache_creation_input_token_cost"] + + (response_1.usage.completion_tokens or 0) * _model_info["output_cost_per_token"] ) # Cost of processing (non-cache hit + cache hit) + Cost of cache-writing (cache writing) assert round(expected_cost, 5) == round(cost_1, 5) @@ -7653,9 +7632,7 @@ def test_completion_cost_databricks_embedding(model, monkeypatch): sync_handler = HTTPHandler() with patch.object(HTTPHandler, "post", return_value=mock_response): - resp = litellm.embedding( - model=model, input=["hey, how's it going?"], client=sync_handler - ) + resp = litellm.embedding(model=model, input=["hey, how's it going?"], client=sync_handler) print(resp) cost = completion_cost(completion_response=resp) @@ -7745,11 +7722,9 @@ def test_cost_openai_prompt_caching(): usage = response_2.usage _expected_cost2 = ( - (usage.prompt_tokens - usage.prompt_tokens_details.cached_tokens) - * model_info["input_cost_per_token"] + (usage.prompt_tokens - usage.prompt_tokens_details.cached_tokens) * model_info["input_cost_per_token"] + usage.completion_tokens * model_info["output_cost_per_token"] - + usage.prompt_tokens_details.cached_tokens - * model_info["cache_read_input_token_cost"] + + usage.prompt_tokens_details.cached_tokens * model_info["cache_read_input_token_cost"] ) print("_expected_cost2", _expected_cost2) @@ -7790,9 +7765,7 @@ def test_completion_cost_azure_ai_rerank(model): }, ) print("response", response) - cost = completion_cost( - model=model, completion_response=response, call_type="arerank" - ) + cost = completion_cost(model=model, completion_response=response, call_type="arerank") assert cost > 0 @@ -8744,9 +8717,7 @@ async def test_test_completion_cost_gpt4o_audio_output_from_model(stream): completion_tokens=34, prompt_tokens=16, total_tokens=50, - completion_tokens_details=CompletionTokensDetailsWrapper( - audio_tokens=28, reasoning_tokens=0, text_tokens=6 - ), + completion_tokens_details=CompletionTokensDetailsWrapper(audio_tokens=28, reasoning_tokens=0, text_tokens=6), prompt_tokens_details=PromptTokensDetailsWrapper( audio_tokens=0, cached_tokens=0, text_tokens=16, image_tokens=0 ), @@ -8785,27 +8756,15 @@ async def test_test_completion_cost_gpt4o_audio_output_from_model(stream): print(f"model_info: {model_info}") ## input cost - input_audio_cost = ( - model_info["input_cost_per_audio_token"] - * usage_object.prompt_tokens_details.audio_tokens - ) - input_text_cost = ( - model_info["input_cost_per_token"] - * usage_object.prompt_tokens_details.text_tokens - ) + input_audio_cost = model_info["input_cost_per_audio_token"] * usage_object.prompt_tokens_details.audio_tokens + input_text_cost = model_info["input_cost_per_token"] * usage_object.prompt_tokens_details.text_tokens total_input_cost = input_audio_cost + input_text_cost ## output cost - output_audio_cost = ( - model_info["output_cost_per_audio_token"] - * usage_object.completion_tokens_details.audio_tokens - ) - output_text_cost = ( - model_info["output_cost_per_token"] - * usage_object.completion_tokens_details.text_tokens - ) + output_audio_cost = model_info["output_cost_per_audio_token"] * usage_object.completion_tokens_details.audio_tokens + output_text_cost = model_info["output_cost_per_token"] * usage_object.completion_tokens_details.text_tokens total_output_cost = output_audio_cost + output_text_cost @@ -8868,7 +8827,7 @@ def test_completion_cost_model_response_cost(response_model, custom_llm_provider def test_select_model_name_for_cost_calc(): from litellm.cost_calculator import select_model_name_for_cost_calc - from litellm.types.utils import ModelResponse, Choices, Usage, Message + from litellm.types.utils import Choices, Message, ModelResponse, Usage args = { "model": "Mistral-large-nmefg", @@ -8930,9 +8889,7 @@ def test_cost_calculator_with_base_model_with_router(base_model_arg): } if base_model_arg == "litellm_param": - model_item["litellm_params"][ - "base_model" - ] = "bedrock/anthropic.claude-sonnet-5" + model_item["litellm_params"]["base_model"] = "bedrock/anthropic.claude-sonnet-5" elif base_model_arg == "model_info": model_item["model_info"] = { "base_model": "bedrock/anthropic.claude-sonnet-5", @@ -8998,9 +8955,7 @@ def isolated_pricing_registry(local_model_cost_map: None, monkeypatch: pytest.Mo _CUSTOM_PRICING_MODEL: Final = "migration-custom-pricing-chat" -_CUSTOM_PRICING_RATES: Final = MappingProxyType( - {"input_cost_per_token": 0.0000008, "output_cost_per_token": 0.0000032} -) +_CUSTOM_PRICING_RATES: Final = MappingProxyType({"input_cost_per_token": 0.0000008, "output_cost_per_token": 0.0000032}) _CUSTOM_PRICING_EXPECTED_COST: Final = 7 * 0.0000008 + 4 * 0.0000032 @@ -9021,7 +8976,13 @@ def _custom_pricing_upstream(request: httpx.Request) -> httpx.Response: body: Final = "".join( ( _custom_pricing_chunk( - [{"index": 0, "delta": {"role": "assistant", "content": "I am ready to help."}, "finish_reason": None}] + [ + { + "index": 0, + "delta": {"role": "assistant", "content": "I am ready to help."}, + "finish_reason": None, + } + ] ), _custom_pricing_chunk([{"index": 0, "delta": {}, "finish_reason": "stop"}]), _custom_pricing_chunk([], {"prompt_tokens": 7, "completion_tokens": 4, "total_tokens": 11}), @@ -9037,7 +8998,11 @@ def _custom_pricing_upstream(request: httpx.Request) -> httpx.Response: "created": 1, "model": _CUSTOM_PRICING_MODEL, "choices": [ - {"index": 0, "message": {"role": "assistant", "content": "I am ready to help."}, "finish_reason": "stop"} + { + "index": 0, + "message": {"role": "assistant", "content": "I am ready to help."}, + "finish_reason": "stop", + } ], "usage": {"prompt_tokens": 7, "completion_tokens": 4, "total_tokens": 11}, }, @@ -9151,7 +9116,12 @@ def test_fireworks_completion_cost_uses_registered_rate(monkeypatch: pytest.Monk monkeypatch.setitem( litellm.model_cost, model, - {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002, "litellm_provider": "fireworks_ai", "mode": "chat"}, + { + "input_cost_per_token": 0.001, + "output_cost_per_token": 0.002, + "litellm_provider": "fireworks_ai", + "mode": "chat", + }, ) respx.post("https://api.fireworks.ai/inference/v1/chat/completions").mock( return_value=httpx.Response( @@ -9285,16 +9255,13 @@ def test_speech_cost_with_custom_pricing_bills_the_reported_usage() -> None: ) assert cost == pytest.approx( - usage.prompt_tokens * rates["input_cost_per_token"] - + usage.completion_tokens * rates["output_cost_per_token"] + usage.prompt_tokens * rates["input_cost_per_token"] + usage.completion_tokens * rates["output_cost_per_token"] ) def test_moderation_cost_uses_test_owned_token_rate() -> None: prompt: Final = "moderation input" - rates: Final = MappingProxyType( - {"input_cost_per_token": 0.001, "output_cost_per_token": 0.0} - ) + rates: Final = MappingProxyType({"input_cost_per_token": 0.001, "output_cost_per_token": 0.0}) cost: Final = completion_cost( completion_response=None, @@ -9312,9 +9279,7 @@ def test_moderation_cost_uses_test_owned_token_rate() -> None: def test_vertex_predictive_cost_uses_test_owned_token_rate() -> None: model: Final = "vertex_ai/migration-llama" messages: Final = [{"role": "user", "content": "price control"}] - rates: Final = MappingProxyType( - {"input_cost_per_token": 0.001, "output_cost_per_token": 0.002} - ) + rates: Final = MappingProxyType({"input_cost_per_token": 0.001, "output_cost_per_token": 0.002}) cost: Final = completion_cost( model=model, @@ -9322,9 +9287,7 @@ def test_vertex_predictive_cost_uses_test_owned_token_rate() -> None: custom_llm_provider="vertex_ai", custom_cost_per_token=rates, ) - expected_cost: Final = ( - litellm.token_counter(model=model, messages=messages) * 0.001 - ) + expected_cost: Final = litellm.token_counter(model=model, messages=messages) * 0.001 assert cost == pytest.approx(expected_cost) @@ -9333,9 +9296,7 @@ def test_vertex_predictive_cost_uses_test_owned_token_rate() -> None: @pytest.mark.asyncio @pytest.mark.usefixtures("isolated_pricing_registry") @respx.mock -async def test_custom_pricing_cost_reaches_success_callback( - async_mode: bool, monkeypatch: pytest.MonkeyPatch -) -> None: +async def test_custom_pricing_cost_reaches_success_callback(async_mode: bool, monkeypatch: pytest.MonkeyPatch) -> None: model: Final = "migration-callback-pricing-model" litellm.register_model( { @@ -9377,9 +9338,7 @@ async def test_custom_pricing_cost_reaches_success_callback( "input_cost_per_token": 0.001, "output_cost_per_token": 0.002, } - response: Final = ( - await litellm.acompletion(**request) if async_mode else litellm.completion(**request) - ) + response: Final = await litellm.acompletion(**request) if async_mode else litellm.completion(**request) await GLOBAL_LOGGING_WORKER.flush() assert await asyncio.to_thread(callback.success_event.wait, 10.0) @@ -9475,7 +9434,7 @@ def test_json_valid_model_cost_map(): json_str = json.dumps(model_cost) json.loads(json_str) except json.JSONDecodeError as e: - pytest.fail(f"Invalid JSON format: {str(e)}") + pytest.fail(f"Invalid JSON format: {e!s}") def test_batch_cost_calculator(): @@ -9780,9 +9739,7 @@ def test_xai_realtime_bills_text_messages(_xai_voice_test_model: None) -> None: assert cost == pytest.approx(30 * 0.0013333333333333333 + 0.004) -def test_xai_realtime_missing_rates_bills_nothing( - _local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch -) -> None: +def test_xai_realtime_missing_rates_bills_nothing(_local_model_cost_map: None, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setitem( litellm.model_cost, "grok-voice-unpriced", @@ -9877,9 +9834,8 @@ def test_live_session_deduplicates_delegated_responses_and_charges_voice(_local_ ) live_rate: Final = cast(float, litellm.model_cost["gpt-live-1"]["input_cost_per_second"]) backend_model: Final = litellm.model_cost["gpt-4o-mini"] - expected_backend_cost: Final = ( - 10 * cast(float, backend_model["input_cost_per_token"]) - + 5 * cast(float, backend_model["output_cost_per_token"]) + expected_backend_cost: Final = 10 * cast(float, backend_model["input_cost_per_token"]) + 5 * cast( + float, backend_model["output_cost_per_token"] ) cost: Final = handle_realtime_stream_cost_calculation( @@ -10559,12 +10515,18 @@ def _reported_cost_stream_chunk(content: str | None, finish_reason: str | None, model="reported-cost-fixture", created=1, object="chat.completion.chunk", - choices=[litellm.types.utils.StreamingChoices(index=0, delta=litellm.types.utils.Delta(content=content), finish_reason=finish_reason)], + choices=[ + litellm.types.utils.StreamingChoices( + index=0, delta=litellm.types.utils.Delta(content=content), finish_reason=finish_reason + ) + ], usage=usage, ) -def _bill_reported_cost(stream: bool, litellm_params: Mapping[str, object], hidden_params: Mapping[str, object]) -> float | None: +def _bill_reported_cost( + stream: bool, litellm_params: Mapping[str, object], hidden_params: Mapping[str, object] +) -> float | None: """The spend the proxy logs for one upstream reply carrying usage.cost, streamed or not.""" logging_obj: Final = _reported_cost_logging(stream, litellm_params) if stream: @@ -10604,7 +10566,9 @@ def test_provider_reported_usage_cost_is_billed_on_stream_and_non_stream( @pytest.mark.parametrize("stream", [False, True]) -def test_deployment_pricing_beats_the_provider_reported_usage_cost(monkeypatch: pytest.MonkeyPatch, stream: bool) -> None: +def test_deployment_pricing_beats_the_provider_reported_usage_cost( + monkeypatch: pytest.MonkeyPatch, stream: bool +) -> None: """Streams used to bill usage.cost even when the operator priced the deployment; both paths bill the operator's rates.""" deployment_id: Final = "reported-cost-deployment-priced-by-the-operator" monkeypatch.setitem( @@ -10624,7 +10588,9 @@ def test_deployment_pricing_beats_the_provider_reported_usage_cost(monkeypatch: @pytest.mark.parametrize("stream", [False, True]) -def test_base_model_pricing_beats_the_provider_reported_usage_cost(monkeypatch: pytest.MonkeyPatch, stream: bool) -> None: +def test_base_model_pricing_beats_the_provider_reported_usage_cost( + monkeypatch: pytest.MonkeyPatch, stream: bool +) -> None: base_model: Final = "openai/reported-cost-base-model-fixture" monkeypatch.setitem( litellm.model_cost, @@ -10643,34 +10609,21 @@ def test_provider_reported_usage_cost_takes_discount_then_margin(monkeypatch: py assert _bill_reported_cost(stream, {}, {}) == pytest.approx(0.037 * 0.8 * 1.5 + 0.001) -def test_a_usage_cost_litellm_stamped_on_a_stream_is_not_billed_again_as_a_provider_report( +def test_the_stream_builder_bills_the_operator_rates_and_leaves_usage_cost_as_the_upstream_reported_it( monkeypatch: pytest.MonkeyPatch, ) -> None: - monkeypatch.setitem(litellm.model_cost, _REPORTED_COST_FIXTURE_MODEL, _REPORTED_COST_FIXTURE_RATES) - monkeypatch.setattr(litellm, "cost_margin_config", {"global": {"fixed_amount": 0.002}}) - logging_obj: Final = _reported_cost_logging(True, {}) - token_math_plus_margin: Final = 100 * 1e-6 + 20 * 3e-6 + 0.002 - - response: Final[object] = litellm.stream_chunk_builder( - [ - _reported_cost_stream_chunk("hi", None, None), - _reported_cost_stream_chunk(None, "stop", None), - _reported_cost_stream_chunk(None, None, Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120)), - ], - messages=[{"role": "user", "content": "hi"}], - logging_obj=logging_obj, + """The bill rides in hidden response_cost. usage.cost stays the upstream's figure, so pricing the assembled + stream again (an MCP round, a logging copy) reads a provider report, never litellm's own bill.""" + deployment_id: Final = "reported-cost-deployment-priced-by-the-operator" + monkeypatch.setitem( + litellm.model_cost, + deployment_id, + {"litellm_provider": "openai", "mode": "chat", "input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6}, ) - - assert isinstance(response, ModelResponse) - assert response.usage is not None and response.usage.cost == pytest.approx(token_math_plus_margin) - assert logging_obj.response_cost_calculator(result=response, ignore_stamped_cost=True) == pytest.approx( - token_math_plus_margin + logging_obj: Final = _reported_cost_logging( + True, {"input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6, "model_info": {"id": deployment_id}} ) - - -def test_a_zero_bill_is_stamped_over_the_provider_reported_usage_cost_on_a_stream(monkeypatch: pytest.MonkeyPatch) -> None: - monkeypatch.setattr(litellm, "cost_discount_config", {"openai": 1.0}) - logging_obj: Final = _reported_cost_logging(True, {}) + operator_bill: Final = 100 * 2e-6 + 20 * 6e-6 response: Final[object] = litellm.stream_chunk_builder( [ @@ -10683,195 +10636,11 @@ def test_a_zero_bill_is_stamped_over_the_provider_reported_usage_cost_on_a_strea ) assert isinstance(response, ModelResponse) - assert response.usage is not None and response.usage.cost == 0.0 - assert logging_obj.response_cost_calculator(result=response) == 0.0 - - -_PRICED_DEPLOYMENT_ID: Final = "reported-cost-deployment-priced-by-the-operator" -_PRICED_LITELLM_PARAMS: Final[Mapping[str, object]] = MappingProxyType( - { - "input_cost_per_token": 2e-6, - "output_cost_per_token": 6e-6, - "metadata": {"model_info": {"id": _PRICED_DEPLOYMENT_ID}}, - } -) - - -def _reported_cost_usage() -> Usage: - return Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120, cost=0.037) - - -def _priced_logging(stream: bool, call_type: str) -> Logging: - """The logging object the proxy hands a priced deployment's call, the deployment under metadata.model_info.""" - logging_obj: Final = _reported_cost_logging(stream, _PRICED_LITELLM_PARAMS) - logging_obj.call_type = call_type - return logging_obj - - -def _reported_cost_responses_response() -> ResponsesAPIResponse: - return ResponsesAPIResponse( - id="resp_reported", - created_at=1, - status="completed", - model="reported-cost-fixture", - object="response", - output=[], - usage=ResponseAPIUsage(input_tokens=100, output_tokens=20, total_tokens=120, cost=0.037), + response._hidden_params.update({"custom_llm_provider": "openai", "model_id": deployment_id}) + assert logging_obj.response_cost_calculator(result=response, ignore_stamped_cost=True) == pytest.approx( + operator_bill ) - - -def _client_visible_reply(route: str) -> tuple[object, Logging]: - """The object a client of the given route reads usage.cost from, after litellm priced it.""" - match route: - case "chat": - logging_obj: Logging = _priced_logging(False, "completion") - response: object = ModelResponse( - id="chatcmpl-reported", - model="reported-cost-fixture", - choices=[Choices(index=0, message=Message(role="assistant", content="hi"), finish_reason="stop")], - usage=_reported_cost_usage(), - ) - update_response_metadata( - result=response, - logging_obj=logging_obj, - model=_REPORTED_COST_FIXTURE_MODEL, - kwargs={}, - start_time=datetime.datetime(2026, 1, 1), - end_time=datetime.datetime(2026, 1, 1, 0, 0, 1), - ) - case "chat-stream": - logging_obj = _priced_logging(True, "completion") - response = litellm.stream_chunk_builder( - [ - _reported_cost_stream_chunk("hi", None, None), - _reported_cost_stream_chunk(None, "stop", None), - _reported_cost_stream_chunk(None, None, _reported_cost_usage()), - ], - messages=[{"role": "user", "content": "hi"}], - logging_obj=logging_obj, - ) - case "responses": - logging_obj = _priced_logging(False, "aresponses") - response = _reported_cost_responses_response() - update_response_metadata( - result=response, - logging_obj=logging_obj, - model=_REPORTED_COST_FIXTURE_MODEL, - kwargs={}, - start_time=datetime.datetime(2026, 1, 1), - end_time=datetime.datetime(2026, 1, 1, 0, 0, 1), - ) - case _: - logging_obj = _priced_logging(True, "aresponses") - response = _reported_cost_responses_response() - stamp_responses_usage_cost(response, logging_obj) - return response, logging_obj - - -@pytest.mark.parametrize("route", ["chat", "chat-stream", "responses", "responses-stream"]) -def test_the_client_reads_what_litellm_billed_from_usage_cost_on_every_route( - monkeypatch: pytest.MonkeyPatch, route: str -) -> None: - """A priced deployment showed its bill on chat streams only; the other three routes still showed the upstream's number.""" - monkeypatch.setitem( - litellm.model_cost, - _PRICED_DEPLOYMENT_ID, - {"litellm_provider": "openai", "mode": "chat", "input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6}, - ) - bill: Final = 100 * 2e-6 + 20 * 6e-6 - - response, logging_obj = _client_visible_reply(route) - - usage: Final = getattr(response, "usage", None) - assert usage is not None and usage.cost == pytest.approx(bill) - assert logging_obj.response_cost_calculator(result=response, ignore_stamped_cost=True) == pytest.approx(bill) - - -def test_a_plain_reply_without_an_upstream_usage_cost_keeps_none_after_pricing(monkeypatch: pytest.MonkeyPatch) -> None: - """Clients comparing the usage dict to the upstream's byte for byte must not find a cost litellm added.""" - monkeypatch.setitem( - litellm.model_cost, - _PRICED_DEPLOYMENT_ID, - {"litellm_provider": "openai", "mode": "chat", "input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6}, - ) - response: Final = ModelResponse( - id="chatcmpl-unreported", - model="reported-cost-fixture", - choices=[Choices(index=0, message=Message(role="assistant", content="hi"), finish_reason="stop")], - usage=Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120), - ) - - update_response_metadata( - result=response, - logging_obj=_priced_logging(False, "completion"), - model=_REPORTED_COST_FIXTURE_MODEL, - kwargs={}, - start_time=datetime.datetime(2026, 1, 1), - end_time=datetime.datetime(2026, 1, 1, 0, 0, 1), - ) - - assert response.usage is not None and getattr(response.usage, "cost", None) is None - assert response._hidden_params["response_cost"] == pytest.approx(100 * 2e-6 + 20 * 6e-6) - - -def test_include_cost_in_streaming_usage_puts_the_bill_on_a_plain_reply_without_an_upstream_usage_cost( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """The switch that injects the bill into streamed usage gives plain replies the same key.""" - monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", True) - monkeypatch.setitem( - litellm.model_cost, - _PRICED_DEPLOYMENT_ID, - {"litellm_provider": "openai", "mode": "chat", "input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6}, - ) - response: Final = ModelResponse( - id="chatcmpl-unreported", - model="reported-cost-fixture", - choices=[Choices(index=0, message=Message(role="assistant", content="hi"), finish_reason="stop")], - usage=Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120), - ) - logging_obj: Final = _priced_logging(False, "completion") - - update_response_metadata( - result=response, - logging_obj=logging_obj, - model=_REPORTED_COST_FIXTURE_MODEL, - kwargs={}, - start_time=datetime.datetime(2026, 1, 1), - end_time=datetime.datetime(2026, 1, 1, 0, 0, 1), - ) - - bill: Final = 100 * 2e-6 + 20 * 6e-6 - assert response.usage is not None and response.usage.cost == pytest.approx(bill) - assert logging_obj.response_cost_calculator(result=response, ignore_stamped_cost=True) == pytest.approx(bill) - - -def test_a_bill_stamped_on_a_plain_reply_survives_the_metadata_pass_for_a_second_pricing( - monkeypatch: pytest.MonkeyPatch, -) -> None: - """The metadata pass installs a fresh hidden-params dict on a reply that arrived with none, which dropped the - provider's figure the stamp keeps and let a second pricing pass apply the margin twice.""" - monkeypatch.setattr(litellm, "cost_margin_config", {"global": {"percentage": 0.5}}) - monkeypatch.delitem(litellm.model_cost, _REPORTED_COST_FIXTURE_MODEL, raising=False) - logging_obj: Final = _reported_cost_logging(False, {}) - response: Final = ModelResponse( - id="chatcmpl-reported", - model="reported-cost-fixture", - choices=[Choices(index=0, message=Message(role="assistant", content="hi"), finish_reason="stop")], - usage=_reported_cost_usage(), - ) - - update_response_metadata( - result=response, - logging_obj=logging_obj, - model=_REPORTED_COST_FIXTURE_MODEL, - kwargs={}, - start_time=datetime.datetime(2026, 1, 1), - end_time=datetime.datetime(2026, 1, 1, 0, 0, 1), - ) - - assert response.usage is not None and response.usage.cost == pytest.approx(0.037 * 1.5) - assert logging_obj.response_cost_calculator(result=response, ignore_stamped_cost=True) == pytest.approx(0.037 * 1.5) + assert response.usage is not None and response.usage.cost == _REPORTED_USAGE["cost"] def test_provider_reported_usage_cost_splits_input_and_output_by_the_token_math_ratio(