diff --git a/litellm/cost_calculator.py b/litellm/cost_calculator.py index 86b22069a39..d2fdec8688e 100644 --- a/litellm/cost_calculator.py +++ b/litellm/cost_calculator.py @@ -5,7 +5,7 @@ import math import time from collections.abc import Mapping, Sequence from datetime import datetime -from functools import lru_cache +from functools import lru_cache, partial from types import MappingProxyType from typing import TYPE_CHECKING, Any, Final, Literal, TypeAlias, cast @@ -22,7 +22,7 @@ from litellm.constants import ( OPENAI_LIVE_TERMINAL_RESPONSE_EVENT_TYPES, REALTIME_METERED_USAGE_KEY, ) -from litellm.litellm_core_utils.core_helpers import as_str_mapping +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 @@ -408,13 +408,114 @@ def _per_second_pricing_cost( return resolved_cost_per_second * seconds, 0.0 +_PROVIDERS_WITH_OWN_REPORTED_COST_READER: Final = frozenset( + {LlmProviders.XAI.value, LlmProviders.PERPLEXITY.value, LlmProviders.OPENCODE_GO.value} +) + + +def _billable_reported_cost(usage_object: Usage | None, custom_llm_provider: str | None) -> float | None: + if custom_llm_provider in _PROVIDERS_WITH_OWN_REPORTED_COST_READER: + return None + return _billable_header_cost(provider_reported_usage_cost(usage_object, custom_llm_provider)) + + +def _split_reported_cost(reported_cost: float, priced: tuple[float, float] | None) -> tuple[float, float]: + """Providers report one total, so attribute it to input and output in the ratio the token math gives.""" + priced_total: Final = (priced[0] + priced[1]) if priced is not None else 0.0 + if priced is None or priced_total <= 0: + return 0.0, reported_cost + input_cost: Final = reported_cost * priced[0] / priced_total + return input_cost, reported_cost - input_cost + + def cost_per_token( model: str = "", prompt_tokens: int = 0, completion_tokens: int = 0, response_time_ms: float | None = 0.0, custom_llm_provider: str | None = None, - region_name=None, + region_name: str | None = None, + ### CHARACTER PRICING ### + prompt_characters: int | None = None, + completion_characters: int | None = None, + ### PROMPT CACHING PRICING ### - used for anthropic + cache_creation_input_tokens: int | None = 0, + cache_read_input_tokens: int | None = 0, + ### CUSTOM PRICING ### + custom_cost_per_token: CostPerToken | None = None, + custom_cost_per_second: float | None = None, + ### NUMBER OF QUERIES ### + number_of_queries: int | None = None, + ### USAGE OBJECT ### + usage_object: Usage | None = None, # just read the usage object if provided + ### BILLED UNITS ### + rerank_billed_units: RerankBilledUnits | None = None, + ### CALL TYPE ### + call_type: CallTypesLiteral = "completion", + audio_transcription_file_duration: float = 0.0, # for audio transcription calls - the file time in seconds + ### SERVICE TIER ### + service_tier: str | None = None, # for OpenAI service tier pricing + ### DATA RESIDENCY ### + data_residency: str | None = None, # for OpenAI regional-processing uplift (e.g. "eu", "us") + ### VERTEX LOCATION ### + vertex_location: str | None = None, # for Vertex AI regional-endpoint uplift (e.g. "us-east5", "global") + response: object | None = None, + ### REQUEST MODEL ### + request_model: str | None = None, # original request model for router detection + custom_model_info: OCRPricing | None = None, + bedrock_model_info: ModelInfo | None = None, +) -> tuple[float, float]: + """Cost in USD for prompt and completion tokens. + + A numeric ``usage.cost`` the provider reported is the bill, split by the ratio of our own token math, + unless the caller priced the call itself with ``custom_cost_per_token`` or ``custom_cost_per_second``. + Without a reported cost the token math is the bill. + """ + priced_by_caller: Final = custom_cost_per_token is not None or custom_cost_per_second is not None + reported_cost: Final = None if priced_by_caller else _billable_reported_cost(usage_object, custom_llm_provider) + rated: Final = partial( + _cost_per_token_from_rates, + model=model, + prompt_tokens=prompt_tokens, + completion_tokens=completion_tokens, + response_time_ms=response_time_ms, + custom_llm_provider=custom_llm_provider, + region_name=region_name, + prompt_characters=prompt_characters, + completion_characters=completion_characters, + cache_creation_input_tokens=cache_creation_input_tokens, + cache_read_input_tokens=cache_read_input_tokens, + custom_cost_per_token=custom_cost_per_token, + custom_cost_per_second=custom_cost_per_second, + number_of_queries=number_of_queries, + usage_object=usage_object, + rerank_billed_units=rerank_billed_units, + call_type=call_type, + audio_transcription_file_duration=audio_transcription_file_duration, + service_tier=service_tier, + data_residency=data_residency, + vertex_location=vertex_location, + response=response, + request_model=request_model, + custom_model_info=custom_model_info, + bedrock_model_info=bedrock_model_info, + ) + if reported_cost is None: + return rated() + try: + return _split_reported_cost(reported_cost, rated()) + except Exception: # noqa: BLE001 # an unmapped model has no rates to split by, the reported total still bills + verbose_logger.debug("No rates for model=%s, billing the provider reported cost %s", model, reported_cost) + return _split_reported_cost(reported_cost, None) + + +def _cost_per_token_from_rates( + model: str = "", + prompt_tokens: int = 0, + completion_tokens: int = 0, + response_time_ms: float | None = 0.0, + custom_llm_provider: str | None = None, + region_name: str | None = None, ### CHARACTER PRICING ### prompt_characters: int | None = None, completion_characters: int | None = None, @@ -1515,7 +1616,7 @@ def completion_cost( call_type: CallTypesLiteral | None = None, ### REGION ### custom_llm_provider: str | None = None, - region_name=None, # used for bedrock pricing + region_name: str | None = None, # used for bedrock pricing ### IMAGE GEN ### size: str | None = None, quality: str | None = None, @@ -1610,7 +1711,9 @@ def completion_cost( audio_transcription_file_duration: float = 0.0 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 else provider_usage_object + _without_provider_stated_cost(provider_usage_object) + if custom_pricing or base_model is not None + else provider_usage_object ) rerank_billed_units: RerankBilledUnits | None = None diff --git a/litellm/litellm_core_utils/hidden_params.py b/litellm/litellm_core_utils/hidden_params.py index a74337d2549..202226fa9ad 100644 --- a/litellm/litellm_core_utils/hidden_params.py +++ b/litellm/litellm_core_utils/hidden_params.py @@ -63,7 +63,8 @@ def stored_attribute(obj: object, name: str) -> object: def _get_hidden_params_storage(obj: object) -> object | None: if isinstance(obj, dict): return obj.get(_HIDDEN_PARAMS_ATTR) - if isinstance(obj, BaseModel) and _HIDDEN_PARAMS_ATTR in type(obj).__private_attributes__: + model_type: Final = type(obj) + if issubclass(model_type, BaseModel) and _HIDDEN_PARAMS_ATTR in model_type.__private_attributes__: return getattr(obj, _HIDDEN_PARAMS_ATTR, None) return stored_attribute(obj, _HIDDEN_PARAMS_ATTR) diff --git a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py index 8d243e572fa..daa816179a1 100644 --- a/litellm/litellm_core_utils/streaming_chunk_builder_utils.py +++ b/litellm/litellm_core_utils/streaming_chunk_builder_utils.py @@ -1351,9 +1351,6 @@ def stream_chunk_builder_text_completion( return TextCompletionResponse(**response) -_CALCULATOR_PRICED_REPORTED_COST_PROVIDERS: Final = frozenset({LlmProviders.XAI.value}) - - def _logged_provider(logging_obj: Optional["Logging"]) -> object: if logging_obj is None: return None @@ -1361,10 +1358,6 @@ def _logged_provider(logging_obj: Optional["Logging"]) -> object: return provider -def _reported_cost_is_priced_by_calculator(logging_obj: Optional["Logging"]) -> bool: - return _logged_provider(logging_obj) in _CALCULATOR_PRICED_REPORTED_COST_PROVIDERS - - def _stream_builder_response_cost( response: ModelResponse, logging_obj: Optional["Logging"], chunks: Sequence[object] ) -> float | None: @@ -1375,9 +1368,6 @@ def _stream_builder_response_cost( for chunk in reversed(chunks) ) return next((cost for cost in reported if cost is not None), None) - usage_cost: Final = getattr(getattr(response, "usage", None), "cost", None) - if isinstance(usage_cost, (int, float)) and not _reported_cost_is_priced_by_calculator(logging_obj): - return float(usage_cost) if logging_obj is not None: return None provider_hint: Final = response._hidden_params.get( # pyright: ignore[reportPrivateUsage] # no public accessor @@ -1421,15 +1411,12 @@ 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 - if isinstance(getattr(usage, "cost", None), (int, float)): - return computed_cost: Final = logging_obj.response_cost_calculator(result=response) - if not isinstance(computed_cost, (int, float)) or computed_cost <= 0: + if isinstance(computed_cost, bool) or not isinstance(computed_cost, (int, float)) or computed_cost < 0: return - setattr(usage, "cost", computed_cost) _set_hidden_response_cost(response, float(computed_cost)) @@ -1600,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, @@ -1788,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/litellm_core_utils/streaming_handler.py b/litellm/litellm_core_utils/streaming_handler.py index 3fb25de43cc..1dc624706c3 100644 --- a/litellm/litellm_core_utils/streaming_handler.py +++ b/litellm/litellm_core_utils/streaming_handler.py @@ -92,6 +92,17 @@ def _next_sync_or_exhausted(it: Iterator[object]) -> object: return _SYNC_ITER_EXHAUSTED +def _stamp_hidden_response_cost( + usage_chunk: ModelResponse | TextCompletionResponse | ModelResponseStream, + complete_response: ModelResponse | TextCompletionResponse | ModelResponseStream, +) -> None: + """The bill the builder priced for the spend row rides the usage chunk, so the proxy shows the same number.""" + bill: Final = (get_hidden_params(complete_response) or {}).get("response_cost") + target: Final = get_hidden_params(usage_chunk) + if target is not None and isinstance(bill, (int, float)) and not isinstance(bill, bool): + target["response_cost"] = float(bill) + + def _stamp_served_service_tier(response: ModelResponseStream, complete_streaming_response: ModelResponse) -> None: served_tier: Final = complete_streaming_response.model_dump().get("service_tier") if isinstance(served_tier, str) and served_tier: @@ -2124,6 +2135,7 @@ class CustomStreamWrapper: getattr(complete_streaming_response, "usage"), ) _stamp_served_service_tier(response, complete_streaming_response) + _stamp_hidden_response_cost(response, complete_streaming_response) try: _cache_copy = complete_streaming_response.model_copy(deep=True) _log_copy = complete_streaming_response.model_copy(deep=True) @@ -2345,6 +2357,7 @@ class CustomStreamWrapper: getattr(complete_streaming_response, "usage"), ) _stamp_served_service_tier(response, complete_streaming_response) + _stamp_hidden_response_cost(response, complete_streaming_response) try: _copy = complete_streaming_response.model_copy(deep=True) except RuntimeError: diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 64fb2791913..854470f54d5 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -930,6 +930,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 with_client_stream_usage_cost from litellm.proxy.utils import ( # noqa: F401, RUF100 # legacy module exports PrismaClient, ProxyLogging, @@ -10563,7 +10564,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)) 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..ac29c1cef09 --- /dev/null +++ b/litellm/proxy/usage_cost_egress.py @@ -0,0 +1,63 @@ +"""What a client reads in a streamed ``usage.cost``: the bill litellm already priced for the spend row, copied once on +the way out of the proxy onto a copy of the event. Nothing is priced here. Inside the pipeline ``usage.cost`` stays +what the upstream reported.""" + +from __future__ import annotations + +import math +from typing import Final, TypeVar + +from pydantic import BaseModel + +from litellm.litellm_core_utils.hidden_params import get_hidden_params +from litellm.types.llms.openai import ResponseAPIUsage, ResponsesAPIResponse +from litellm.types.utils import ModelResponseStream, Usage + +_RESPONSES_USAGE_EVENTS: Final = frozenset({"response.completed", "response.incomplete"}) +_ReplyT: Final = 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) -> float | None: + """The bill, where the upstream put a number in usage.cost (a $0 bill included) or litellm priced the + stream above zero. None leaves the usage as the upstream sent it.""" + billable: Final = _billable(bill) + if billable is None: + return None + reported: Final = isinstance(upstream_cost, (int, float)) and not isinstance(upstream_cost, bool) + return billable if reported or billable > 0 else None + + +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 _hidden_bill(reply: object) -> object: + hidden: Final = get_hidden_params(reply) + return hidden.get("response_cost") if hidden is not None else None + + +def with_client_stream_usage_cost(chunk: BaseModel) -> BaseModel: + """A streamed event as the client gets it. Only the usage-bearing events change: the chat chunk carrying + ``usage`` and the Responses ``response.completed`` / ``response.incomplete`` events, each read against the + hidden ``response_cost`` the pipeline stamped on it.""" + 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), _hidden_bill(chunk)) + 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), _hidden_bill(response)) + 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 219944de521..9c490430594 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, @@ -1481,7 +1481,7 @@ class LiteLLMCompletionStreamingIterator(ResponsesAPIStreamingIterator): } ) ) - stamp_responses_usage_cost(partial, self.litellm_logging_obj) + normalize_responses_usage(partial, self.litellm_logging_obj) 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 4e418a91156..6a4cbc90b41 100644 --- a/litellm/responses/streaming_iterator.py +++ b/litellm/responses/streaming_iterator.py @@ -41,6 +41,7 @@ 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 get_hidden_params from litellm.litellm_core_utils.internal_call_metadata import ( is_background_response, requests_background, @@ -595,7 +596,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, self.logging_obj) if _chunk_type == openai_types.ResponsesAPIStreamEvents.RESPONSE_FAILED: self._handle_logging_failed_response() @@ -1123,7 +1124,7 @@ class BaseResponsesAPIStreamingIterator: "output": _partial_output_items(self._generated_content), } ) - stamp_responses_usage_cost(partial, self.logging_obj) + normalize_responses_usage(partial, self.logging_obj) return ResponseIncompleteEvent(type=ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, response=partial) def _billed_without_disconnect(self) -> bool: @@ -1822,23 +1823,24 @@ 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, logging_obj: LiteLLMLoggingObj | None) -> None: + """A terminal response built off an untyped dict carries its usage as a dict, logging and the proxy read a model. + The bill the logging object prices for it rides its hidden ``response_cost``, so the proxy shows the same number.""" + 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 - if isinstance(getattr(usage_obj, "cost", None), (int, float)): + response_obj.usage = usage_obj # rebind-ok: the typed usage has to ride on the response the client receives + if logging_obj is None: return try: - cost: Final[float | None] = logging_obj.response_cost_calculator(result=response_obj) - except Exception: + bill: Final = logging_obj.response_cost_calculator(result=response_obj) + except Exception: # noqa: BLE001 # a pricing failure leaves the client usage as the upstream sent it return - if isinstance(cost, (int, float)) and cost > 0: - setattr(usage_obj, "cost", cost) + hidden: Final = get_hidden_params(response_obj) + if hidden is not None and isinstance(bill, (int, float)) and not isinstance(bill, bool): + hidden["response_cost"] = float(bill) def opening_item_payload(payload: Mapping[str, object]) -> Mapping[str, object]: @@ -1872,7 +1874,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, logging_obj) events: Final[list[ResponsesAPIStreamingResponse]] = [ _build_response_status_event(openai_types.ResponsesAPIStreamEvents.RESPONSE_CREATED, transformed), diff --git a/tests/integration/cost_calculation/cases/openai.py b/tests/integration/cost_calculation/cases/openai.py index 53fd5c1cec2..923ce6136aa 100644 --- a/tests/integration/cost_calculation/cases/openai.py +++ b/tests/integration/cost_calculation/cases/openai.py @@ -2,6 +2,7 @@ PARITY pairs the streamed case with its plain twin where both bill the same row.""" +from collections.abc import Mapping from typing import Final, cast from integration.cost_calculation.cost_tracking_case import ( @@ -11878,7 +11879,215 @@ GPT_5_6_TERRA_MESSAGES_BRIDGED_STREAM: Final = GPT_5_6_TERRA_MESSAGES_BRIDGED_JS } ) + +# An OpenAI-compatible gateway that reports what it charged under usage.cost (#47533). The reported total is the bill, +# split across input and output by the ratio of the cost map rates (100 * 1e-6 : 20 * 3e-6), on plain and streamed +# replies alike. A deployment the operator priced keeps the operator's rates on both. +_GATEWAY_REPORTED_USAGE: Final = {"prompt_tokens": 100, "completion_tokens": 20, "total_tokens": 120, "cost": 0.037} +_GATEWAY_REPORTED_EXPECTED: Final = ExactExpected( + spend=0.037, input_cost=0.023125, output_cost=0.013875, prompt_tokens=100, completion_tokens=20 +) +_GATEWAY_OPERATOR_PRICED: Final = Deployment(input_cost_per_token=2e-06, output_cost_per_token=6e-06) +_GATEWAY_OPERATOR_PRICED_EXPECTED: Final = ExactExpected( + spend=0.00032, input_cost=0.0002, output_cost=0.00012, prompt_tokens=100, completion_tokens=20 +) +_GATEWAY_CHUNK: Final = { + "id": "chatcmpl-$REQUEST_ID", + "object": "chat.completion.chunk", + "created": 1, + "model": "gateway-reported-cost-fixture", +} + +GATEWAY_REPORTED_COST: Final = CostTrackingTestCase( + name="gateway-reported-cost", + covers="quota_management.spend_tracking.cost_matrix.logs_cost", + model="gateway-reported-cost-fixture", + endpoint="/v1/chat/completions", + request={"model": "$MODEL", "messages": [{"role": "user", "content": "reported cost"}], "stream": False}, + response=JsonResponse( + content_type="application/json", + body={ + "id": "chatcmpl-$REQUEST_ID", + "object": "chat.completion", + "created": 1, + "model": "gateway-reported-cost-fixture", + "choices": [{"index": 0, "message": {"role": "assistant", "content": "ok"}, "finish_reason": "stop"}], + "usage": _GATEWAY_REPORTED_USAGE, + }, + ), + expected=_GATEWAY_REPORTED_EXPECTED, +) + +GATEWAY_REPORTED_COST_STREAM: Final = CostTrackingTestCase( + name="gateway-reported-cost-stream", + covers="quota_management.spend_tracking.cost_matrix.logs_cost", + model="gateway-reported-cost-fixture", + endpoint="/v1/chat/completions", + request={ + "model": "$MODEL", + "messages": [{"role": "user", "content": "reported cost"}], + "stream": True, + "stream_options": {"include_usage": True}, + }, + response=SseResponse( + content_type="text/event-stream", + frames=sse_frames( + {**_GATEWAY_CHUNK, "choices": [{"index": 0, "delta": {"role": "assistant", "content": "ok"}}]}, + {**_GATEWAY_CHUNK, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}, + {**_GATEWAY_CHUNK, "choices": [], "usage": _GATEWAY_REPORTED_USAGE}, + done=True, + ), + ), + expected=_GATEWAY_REPORTED_EXPECTED, +) + +GATEWAY_REPORTED_COST_OPERATOR_PRICED: Final = GATEWAY_REPORTED_COST.model_copy( + update={ + "name": "gateway-reported-cost-operator_priced", + "deployment": _GATEWAY_OPERATOR_PRICED, + "expected": _GATEWAY_OPERATOR_PRICED_EXPECTED, + } +) + +GATEWAY_REPORTED_COST_OPERATOR_PRICED_STREAM: Final = GATEWAY_REPORTED_COST_STREAM.model_copy( + update={ + "name": "gateway-reported-cost-operator_priced-stream", + "deployment": _GATEWAY_OPERATOR_PRICED, + "expected": _GATEWAY_OPERATOR_PRICED_EXPECTED, + } +) + +_GATEWAY_RESPONSES_USAGE: Final[Mapping[str, JsonValue]] = {"input_tokens": 100, "output_tokens": 20, "total_tokens": 120, "cost": 0.037} +_GATEWAY_RESPONSES_ITEM: Final[Mapping[str, JsonValue]] = { + "id": "msg_gateway", + "type": "message", + "role": "assistant", + "status": "completed", + "content": [{"type": "output_text", "text": "ok", "annotations": []}], +} +_GATEWAY_RESPONSES_BODY: Final[Mapping[str, JsonValue]] = { + "id": "resp_$REQUEST_ID", + "object": "response", + "created_at": 1, + "status": "completed", + "model": "gateway-reported-cost-fixture", + "output": [_GATEWAY_RESPONSES_ITEM], + "usage": _GATEWAY_RESPONSES_USAGE, +} + +GATEWAY_REPORTED_COST_RESPONSES: Final = CostTrackingTestCase( + name="gateway-reported-cost-responses", + covers="quota_management.spend_tracking.cost_matrix.logs_cost", + model="gateway-reported-cost-fixture", + endpoint="/v1/responses", + request={"model": "$MODEL", "input": "reported cost", "cache": {"no-cache": True}}, + response=JsonResponse(content_type="application/json", body=_GATEWAY_RESPONSES_BODY), + expected=_GATEWAY_REPORTED_EXPECTED, +) + +GATEWAY_REPORTED_COST_RESPONSES_STREAM: Final = GATEWAY_REPORTED_COST_RESPONSES.model_copy( + update={ + "name": "gateway-reported-cost-responses-stream", + "request": {**GATEWAY_REPORTED_COST_RESPONSES.request, "stream": True}, + "response": SseResponse( + content_type="text/event-stream", + frames=sse_frames( + { + "type": "response.created", + "sequence_number": 0, + "response": {**_GATEWAY_RESPONSES_BODY, "status": "in_progress", "output": [], "usage": None}, + }, + { + "type": "response.output_text.delta", + "sequence_number": 1, + "item_id": "msg_gateway", + "output_index": 0, + "content_index": 0, + "delta": "ok", + }, + {"type": "response.completed", "sequence_number": 2, "response": _GATEWAY_RESPONSES_BODY}, + ), + ), + } +) + +GATEWAY_REPORTED_COST_RESPONSES_OPERATOR_PRICED: Final = GATEWAY_REPORTED_COST_RESPONSES.model_copy( + update={ + "name": "gateway-reported-cost-responses-operator_priced", + "deployment": _GATEWAY_OPERATOR_PRICED, + "expected": _GATEWAY_OPERATOR_PRICED_EXPECTED, + } +) + +GATEWAY_REPORTED_COST_RESPONSES_OPERATOR_PRICED_STREAM: Final = GATEWAY_REPORTED_COST_RESPONSES_STREAM.model_copy( + update={ + "name": "gateway-reported-cost-responses-operator_priced-stream", + "deployment": _GATEWAY_OPERATOR_PRICED, + "expected": _GATEWAY_OPERATOR_PRICED_EXPECTED, + } +) + + +_GATEWAY_ZERO_USAGE: Final = {**_GATEWAY_REPORTED_USAGE, "cost": 0} + +GATEWAY_REPORTED_ZERO_OPERATOR_PRICED: Final = GATEWAY_REPORTED_COST_OPERATOR_PRICED.model_copy( + update={ + "name": "gateway-reported-zero-operator_priced", + "response": JsonResponse( + content_type="application/json", + body={**GATEWAY_REPORTED_COST.response.body, "usage": _GATEWAY_ZERO_USAGE}, + ), + } +) + +GATEWAY_REPORTED_ZERO_OPERATOR_PRICED_STREAM: Final = GATEWAY_REPORTED_COST_OPERATOR_PRICED_STREAM.model_copy( + update={ + "name": "gateway-reported-zero-operator_priced-stream", + "response": SseResponse( + content_type="text/event-stream", + frames=sse_frames( + {**_GATEWAY_CHUNK, "choices": [{"index": 0, "delta": {"role": "assistant", "content": "ok"}}]}, + {**_GATEWAY_CHUNK, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}]}, + {**_GATEWAY_CHUNK, "choices": [], "usage": _GATEWAY_ZERO_USAGE}, + done=True, + ), + ), + } +) + +_GATEWAY_OPENROUTER_OPERATOR_PRICED: Final = _GATEWAY_OPERATOR_PRICED.model_copy( + update={"model": "openrouter/vendor/gateway-reported-cost-fixture"} +) + +GATEWAY_REPORTED_COST_OPENROUTER_OPERATOR_PRICED: Final = GATEWAY_REPORTED_COST_OPERATOR_PRICED.model_copy( + update={ + "name": "gateway-reported-cost-openrouter-operator_priced", + "deployment": _GATEWAY_OPENROUTER_OPERATOR_PRICED, + } +) + +GATEWAY_REPORTED_COST_OPENROUTER_OPERATOR_PRICED_STREAM: Final = ( + GATEWAY_REPORTED_COST_OPERATOR_PRICED_STREAM.model_copy( + update={ + "name": "gateway-reported-cost-openrouter-operator_priced-stream", + "deployment": _GATEWAY_OPENROUTER_OPERATOR_PRICED, + } + ) +) + CASES: Final[tuple[CostTrackingTestCase, ...]] = ( + GATEWAY_REPORTED_COST, + GATEWAY_REPORTED_COST_STREAM, + GATEWAY_REPORTED_COST_OPERATOR_PRICED, + GATEWAY_REPORTED_COST_OPERATOR_PRICED_STREAM, + GATEWAY_REPORTED_COST_RESPONSES, + GATEWAY_REPORTED_COST_RESPONSES_STREAM, + GATEWAY_REPORTED_COST_RESPONSES_OPERATOR_PRICED, + GATEWAY_REPORTED_COST_RESPONSES_OPERATOR_PRICED_STREAM, + GATEWAY_REPORTED_ZERO_OPERATOR_PRICED, + GATEWAY_REPORTED_ZERO_OPERATOR_PRICED_STREAM, + GATEWAY_REPORTED_COST_OPENROUTER_OPERATOR_PRICED, + GATEWAY_REPORTED_COST_OPENROUTER_OPERATOR_PRICED_STREAM, GPT_5_6_SLASH_ALIAS, GPT_5_6_SLASH_ALIAS_STREAM, GPT_6_SOL_RESPONSES_DEPLOYMENT_JSON, @@ -12120,6 +12329,22 @@ CASES: Final[tuple[CostTrackingTestCase, ...]] = ( ) PARITY: Final[tuple[StreamParityTestCase, ...]] = ( + StreamParityTestCase(plain=GATEWAY_REPORTED_COST, streamed=GATEWAY_REPORTED_COST_STREAM), + StreamParityTestCase( + plain=GATEWAY_REPORTED_COST_OPERATOR_PRICED, streamed=GATEWAY_REPORTED_COST_OPERATOR_PRICED_STREAM + ), + StreamParityTestCase( + plain=GATEWAY_REPORTED_ZERO_OPERATOR_PRICED, streamed=GATEWAY_REPORTED_ZERO_OPERATOR_PRICED_STREAM + ), + StreamParityTestCase( + plain=GATEWAY_REPORTED_COST_OPENROUTER_OPERATOR_PRICED, + streamed=GATEWAY_REPORTED_COST_OPENROUTER_OPERATOR_PRICED_STREAM, + ), + StreamParityTestCase(plain=GATEWAY_REPORTED_COST_RESPONSES, streamed=GATEWAY_REPORTED_COST_RESPONSES_STREAM), + StreamParityTestCase( + plain=GATEWAY_REPORTED_COST_RESPONSES_OPERATOR_PRICED, + streamed=GATEWAY_REPORTED_COST_RESPONSES_OPERATOR_PRICED_STREAM, + ), StreamParityTestCase(plain=GPT_5_3_CODEX_INPUT_TEXT, streamed=GPT_5_3_CODEX_STREAM), StreamParityTestCase(plain=GPT_5_3_CODEX_INPUT_TEXT, streamed=GPT_5_3_CODEX_STREAM_INCOMPLETE), StreamParityTestCase(plain=GPT_5_3_CODEX_INPUT_TEXT, streamed=GPT_5_3_CODEX_STREAM_UNVALIDATED), diff --git a/tests/integration/cost_calculation/cost_map.py b/tests/integration/cost_calculation/cost_map.py index 32722b37c9b..da4c2b6e859 100644 --- a/tests/integration/cost_calculation/cost_map.py +++ b/tests/integration/cost_calculation/cost_map.py @@ -26,6 +26,12 @@ COST_MAP_ENTRIES: Final[dict[str, dict[str, JsonValue]]] = { "input_cost_per_token_above_128k_tokens": 2.2e-07, "output_cost_per_token_above_128k_tokens": 8.8e-07, }, + "gateway-reported-cost-fixture": { + "litellm_provider": "openai", + "mode": "chat", + "input_cost_per_token": 1e-06, + "output_cost_per_token": 3e-06, + }, "openrouter/anthropic/claude-sonnet-5": { "litellm_provider": "openrouter", "mode": "chat", diff --git a/tests/unit/litellm_core_utils/test_streaming_handler.py b/tests/unit/litellm_core_utils/test_streaming_handler.py index a1b58521d7e..853a52276af 100644 --- a/tests/unit/litellm_core_utils/test_streaming_handler.py +++ b/tests/unit/litellm_core_utils/test_streaming_handler.py @@ -9,7 +9,7 @@ import time from collections.abc import Mapping, Sequence from dataclasses import dataclass from types import CodeType, FrameType, MappingProxyType -from typing import AsyncIterator, Callable, Final, Iterator, Literal, NoReturn, Optional, cast +from typing import AsyncIterator, Callable, Final, Iterator, Literal, NoReturn, Optional, Tuple, cast from unittest.mock import AsyncMock, MagicMock, Mock, patch import httpx @@ -17,8 +17,9 @@ import pytest from openai import AsyncOpenAI, AsyncStream, OpenAI, Stream from openai.types.chat import ChatCompletionChunk from pydantic import BaseModel, TypeAdapter + import litellm -from litellm import verbose_logger +from litellm import completion, verbose_logger from litellm._logging import session_id_var, trace_id_var from litellm.exceptions import MidStreamFallbackError from litellm.integrations.custom_logger import CustomLogger @@ -29,7 +30,6 @@ from litellm.litellm_core_utils.streaming_handler import ( _ProviderChunkEarlyReturn, _ProviderChunkParsed, ) -from litellm.llms.custom_httpx.http_handler import HTTPHandler from litellm.llms.openai.chat.gpt_transformation import OpenAIChatCompletionStreamingHandler from litellm.responses.mcp.chat_completions_handler import _add_mcp_metadata_to_response from litellm.types.utils import ( @@ -38,13 +38,11 @@ from litellm.types.utils import ( ModelResponse, ModelResponseStream, PromptTokensDetailsWrapper, - StreamingChoices, StandardLoggingPayload, + StreamingChoices, Usage, ) from litellm.utils import ModelResponseListIterator -from litellm import completion -from typing import Tuple @pytest.mark.parametrize("async_stream", [False, True]) @@ -2547,14 +2545,14 @@ def test_perplexity_streaming_dict_cost_bills_through_its_own_calculator(): ) == pytest.approx(0.00503) -def test_openai_compatible_streaming_cost_is_priced_from_the_cost_map(): +def test_openai_compatible_streaming_cost_is_the_reported_cost(): import litellm from litellm.cost_calculator import ( get_response_cost_from_hidden_params, response_cost_calculator, ) - model = "openai/streams-cost-in-nanodollars" + model = "openai/streams-with-a-reported-cost" litellm.register_model( { model: { @@ -2569,7 +2567,7 @@ def test_openai_compatible_streaming_cost_is_priced_from_the_cost_map(): id="chatcmpl-openai-compatible", model=model, choices=[], - usage=Usage(completion_tokens=5, prompt_tokens=10, total_tokens=15, cost=3_144_000), + usage=Usage(completion_tokens=5, prompt_tokens=10, total_tokens=15, cost=0.037), ) CustomStreamWrapper._propagate_usage_cost_to_hidden_params(complete_response, "openai") @@ -2581,7 +2579,7 @@ def test_openai_compatible_streaming_cost_is_priced_from_the_cost_map(): custom_llm_provider="openai", call_type="completion", optional_params={}, - ) == pytest.approx(2e-5) + ) == 0.037 def test_xai_streaming_reported_cost_still_takes_the_margin(monkeypatch): @@ -6204,6 +6202,25 @@ def test_chunk_creator_preserves_hidden_provider_specific_fields_from_parsed_chu assert assembled._hidden_params["provider_specific_fields"] == {"traffic_type": "ON_DEMAND_FLEX"} +def test_the_usage_chunk_carries_the_bill_the_builder_priced_for_the_spend_row() -> None: + from litellm.litellm_core_utils.hidden_params import get_hidden_params + from litellm.litellm_core_utils.streaming_handler import ( + _stamp_hidden_response_cost, # pyright: ignore[reportPrivateUsage] # the helper under test + ) + + usage: Final = Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120) + complete: Final = ModelResponse(model="omni-model", usage=usage) + complete.hidden_params = {"response_cost": 0.00032} + usage_chunk: Final = ModelResponseStream(id="chunk-1", model="omni-model", choices=[], usage=usage) + usage_chunk.hidden_params = {"response_cost": None} + + _stamp_hidden_response_cost(usage_chunk, complete) + + stamped: Final = get_hidden_params(usage_chunk) + assert stamped is not None and stamped["response_cost"] == 0.00032 + assert getattr(usage, "cost", None) is None + + def test_chunk_creator_keeps_provider_model_private_across_stream(): from litellm.router_utils.add_retry_fallback_headers import ( get_hidden_params_dict, diff --git a/tests/unit/llms/azure_ai/test_azure_ai_cost_calculator.py b/tests/unit/llms/azure_ai/test_azure_ai_cost_calculator.py index 4f8ffa25abb..cef64f6284f 100644 --- a/tests/unit/llms/azure_ai/test_azure_ai_cost_calculator.py +++ b/tests/unit/llms/azure_ai/test_azure_ai_cost_calculator.py @@ -14,7 +14,6 @@ import litellm from litellm.cost_calculator import completion_cost from litellm.litellm_core_utils.litellm_logging import Logging from litellm.llms.azure_ai.cost_calculator import ( - calculate_azure_model_router_flat_cost, cost_per_token, is_azure_model_router, ) @@ -263,7 +262,14 @@ class TestAzureModelRouterCostBreakdown: assert route.call_count == 1 assert json.loads(route.calls.last.request.content)["model"] == "model-router" assert (final_usage.prompt_tokens, final_usage.completion_tokens) == (5000, 2000) - assert final_usage.cost == pytest.approx(routed_prompt_cost + routed_completion_cost + ROUTED_FEE, rel=1e-9) + assert getattr(final_usage, "cost", None) is None, "the bill reaches a client through the proxy egress only" + assembled: Final = litellm.stream_chunk_builder( + chunks=list(chunks), messages=messages, logging_obj=stream.logging_obj + ) + assert assembled is not None + assert assembled._hidden_params["response_cost"] == pytest.approx( + routed_prompt_cost + routed_completion_cost + ROUTED_FEE, rel=1e-9 + ) @pytest.mark.respx(assert_all_called=True) def test_model_router_completion_reports_routed_cost_plus_flat_fee(self, respx_mock: respx.MockRouter) -> None: 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..62c32cd3ebc --- /dev/null +++ b/tests/unit/proxy/test_usage_cost_egress.py @@ -0,0 +1,126 @@ +from __future__ import annotations + +from typing import Final + +import pytest + +from litellm.litellm_core_utils.hidden_params import get_hidden_params +from litellm.proxy.usage_cost_egress import with_client_stream_usage_cost +from litellm.types.llms.openai import ( + ResponseAPIUsage, + ResponseCompletedEvent, + ResponseIncompleteEvent, + ResponsesAPIResponse, + ResponsesAPIStreamEvents, +) +from litellm.types.utils import Delta, ModelResponseStream, StreamingChoices, Usage + +UPSTREAM_COST: Final = 0.037 +OPERATOR_BILL: Final = 100 * 2e-6 + 20 * 6e-6 + + +def _responses_reply(upstream_cost: float | None, bill: object) -> ResponsesAPIResponse: + reply: Final = 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), + ) + hidden: Final = get_hidden_params(reply) + assert hidden is not None + hidden["response_cost"] = bill + return reply + + +def _usage_chunk(upstream_cost: float | None, bill: object) -> ModelResponseStream: + chunk: Final = ModelResponseStream( + id="chatcmpl-egress", + model="omni-model", + choices=[], + usage=Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120, cost=upstream_cost), + ) + chunk.hidden_params = {"response_cost": bill} + return chunk + + +def test_the_streamed_usage_chunk_tells_the_client_what_litellm_billed() -> None: + chunk: Final = _usage_chunk(UPSTREAM_COST, OPERATOR_BILL) + + billed: Final = with_client_stream_usage_cost(chunk) + + assert isinstance(billed, ModelResponseStream) and billed is not chunk + assert billed.usage is not None and billed.usage.cost == OPERATOR_BILL + assert chunk.usage is not None and chunk.usage.cost == UPSTREAM_COST + + +def test_a_zero_bill_replaces_the_upstream_cost() -> None: + billed: Final = with_client_stream_usage_cost(_usage_chunk(UPSTREAM_COST, 0.0)) + + assert isinstance(billed, ModelResponseStream) + assert billed.usage is not None and billed.usage.cost == 0.0 + + +@pytest.mark.parametrize("bill", [None, -0.001, float("nan"), float("inf"), True]) +def test_an_unpriced_stream_keeps_the_upstream_cost(bill: object) -> None: + chunk: Final = _usage_chunk(UPSTREAM_COST, bill) + + assert with_client_stream_usage_cost(chunk) is chunk + assert chunk.usage is not None and chunk.usage.cost == UPSTREAM_COST + + +def test_a_priced_stream_without_an_upstream_cost_still_tells_the_client_the_bill() -> None: + billed: Final = with_client_stream_usage_cost(_usage_chunk(None, OPERATOR_BILL)) + + assert isinstance(billed, ModelResponseStream) + assert billed.usage is not None and billed.usage.cost == OPERATOR_BILL + + +@pytest.mark.parametrize("bill", [None, 0.0]) +def test_an_unpriced_stream_without_an_upstream_cost_goes_out_as_sent(bill: float | None) -> None: + chunk: Final = _usage_chunk(None, bill) + + assert with_client_stream_usage_cost(chunk) is chunk + + +def test_a_content_chunk_is_not_touched() -> None: + chunk: Final = ModelResponseStream( + id="chatcmpl-egress", + model="omni-model", + choices=[StreamingChoices(index=0, delta=Delta(content="hi"), finish_reason=None)], + ) + chunk.hidden_params = {"response_cost": OPERATOR_BILL} + + assert with_client_stream_usage_cost(chunk) is chunk + + +@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( + upstream_cost: float | None, event_type: ResponsesAPIStreamEvents +) -> None: + reply: Final = _responses_reply(upstream_cost, OPERATOR_BILL) + event: Final = ( + ResponseCompletedEvent(type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED, response=reply) + if event_type is ResponsesAPIStreamEvents.RESPONSE_COMPLETED + else ResponseIncompleteEvent(type=ResponsesAPIStreamEvents.RESPONSE_INCOMPLETE, response=reply) + ) + + billed: Final = with_client_stream_usage_cost(event) + + assert billed is not event + assert isinstance(billed, (ResponseCompletedEvent, ResponseIncompleteEvent)) + assert billed.response.usage is not None and billed.response.usage.cost == OPERATOR_BILL + assert reply.usage is not None and reply.usage.cost == upstream_cost + + +def test_an_unpriced_terminal_responses_event_goes_out_as_sent() -> None: + event: Final = ResponseCompletedEvent( + type=ResponsesAPIStreamEvents.RESPONSE_COMPLETED, response=_responses_reply(UPSTREAM_COST, None) + ) + + assert with_client_stream_usage_cost(event) is event diff --git a/tests/unit/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py b/tests/unit/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py index ed90acbd2fa..e4784e92bc4 100644 --- a/tests/unit/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py +++ b/tests/unit/responses/litellm_completion_transformation/test_streaming_iterator_transformation.py @@ -10,6 +10,7 @@ before response.completed, and that every event of a bridged stream carries the spend tracking stores, so a follow-up previous_response_id still finds the conversation. """ +import itertools import json from collections.abc import Callable from typing import Final @@ -23,6 +24,7 @@ from litellm.responses.litellm_completion_transformation.streaming_iterator impo from litellm.responses.utils import ResponsesAPIRequestUtils from litellm.types.llms.openai import ( BaseLiteLLMOpenAIResponseObject, + OutputTextDeltaEvent, ResponsesAPIStreamEvents, ) from litellm.types.responses.main import build_web_search_call @@ -1558,3 +1560,29 @@ async def test_a_compaction_block_without_its_summary_streams_no_compaction_item assert [event["item"]["type"] for event in events if "item" in event] == ["message", "message"] assert [item["type"] for item in events[-1]["response"]["output"]] == ["message"] + + +def test_the_partial_response_billed_on_disconnect_carries_the_logging_objects_bill_as_hidden_response_cost() -> None: + from litellm.litellm_core_utils.hidden_params import get_hidden_params + + wrapper: Final = _FakeStreamWrapper([_chunk("Hello"), _chunk(" world", finish_reason="stop")]) + wrapper.logging_obj.response_cost_calculator.return_value = 0.00032 + iterator: Final = LiteLLMCompletionStreamingIterator( + model="claude-haiku-4-5", + litellm_custom_stream_wrapper=wrapper, # pyright: ignore[reportArgumentType] # the fake stands in for CustomStreamWrapper, as in _build_iterator + request_input="What is the weather in San Francisco?", + responses_api_request={}, + custom_llm_provider="anthropic", + litellm_metadata={}, + ) + relayed: Final = tuple( + itertools.takewhile(lambda event: not isinstance(event, OutputTextDeltaEvent), iterator) + ) + assert relayed, "the stream must relay at least one chunk before the client goes away" + + partial: Final = iterator.partial_response_on_disconnect() + + assert partial is not None + hidden: Final = get_hidden_params(partial.response) + assert hidden is not None and hidden["response_cost"] == 0.00032 + assert wrapper.logging_obj.response_cost_calculator.call_args.kwargs["result"] is partial.response diff --git a/tests/unit/responses/test_streaming_iterator.py b/tests/unit/responses/test_streaming_iterator.py index 664f4dba163..f0f9fa9738c 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 @@ -22,9 +25,11 @@ from pydantic_core import PydanticSerializationError import litellm from litellm.exceptions import MidStreamFallbackError from litellm.integrations.custom_logger import CustomLogger +from litellm.litellm_core_utils.hidden_params import get_hidden_params 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 +43,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 +82,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 +111,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: @@ -491,7 +493,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): @@ -840,32 +841,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_keeps_provider_reported_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) - - stamp_responses_usage_cost(response, logging_obj) - - assert getattr(response.usage, "cost", None) == pytest.approx(0.5) - logging_obj.response_cost_calculator.assert_not_called() - - def _unvalidated_response_with_dict_usage(usage: dict) -> ResponsesAPIResponse: return ResponsesAPIResponse.model_construct( id="resp_lit7391", @@ -879,8 +854,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, @@ -890,41 +866,52 @@ 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) - stamp_responses_usage_cost(response, logging_obj) + normalize_responses_usage(response, None) assert isinstance(response.usage, ResponseAPIUsage) assert response.usage.cost == pytest.approx(3e-05) assert response.usage.output_tokens_details.reasoning_tokens == 117 - logging_obj.response_cost_calculator.assert_not_called() -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, None) assert getattr(response.usage, "cost", None) is None +def test_normalize_responses_usage_stamps_the_logging_objects_bill_as_hidden_response_cost(): + from litellm.responses.streaming_iterator import normalize_responses_usage + + from litellm.litellm_core_utils.litellm_logging import Logging + + response: Final = _responses_api_response_with_usage() + logging_obj: Final = Logging( + model="mantle-claude", + messages=[{"role": "user", "content": "q"}], + stream=True, + call_type="aresponses", + start_time=None, + litellm_call_id="call", + function_id="fn", + ) + logging_obj.update_environment_variables( + litellm_params={"custom_llm_provider": "openai", "input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6}, + optional_params={}, + ) + logging_obj.custom_cost_per_token = {"input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6} + + normalize_responses_usage(response, logging_obj) + + hidden: Final = get_hidden_params(response) + assert hidden is not None and hidden["response_cost"] == pytest.approx(20 * 2e-6 + 60 * 6e-6) + assert getattr(response.usage, "cost", None) is None, "the client field is written at the proxy egress, not here" + + def _capture_dispatch(logged: list): """Record the object handed to the success handlers. @@ -942,7 +929,7 @@ def _capture_dispatch(logged: list): return _dispatch -def _headers_config(*, transform_hidden_params: Optional[dict] = None) -> Mock: +def _headers_config(*, transform_hidden_params: Mapping[str, object] | 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) @@ -1259,10 +1246,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", @@ -1304,8 +1291,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(): @@ -1375,6 +1361,7 @@ def _vcr_outcome_gate(request, vcr): yield record_vcr_outcome(request, vcr) + @pytest.fixture(scope="function") def setup_and_teardown(): """ @@ -1392,6 +1379,7 @@ def setup_and_teardown(): loop.close() asyncio.set_event_loop(None) + class _FakeLoggingObj: def __init__(self): self.success_calls = 0 @@ -1432,10 +1420,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, @@ -1463,6 +1453,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): @@ -1485,6 +1476,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): @@ -1500,6 +1492,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( @@ -1527,6 +1520,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( @@ -1542,6 +1536,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): @@ -1598,6 +1593,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): @@ -1638,6 +1634,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(): @@ -1667,6 +1664,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( @@ -1681,6 +1679,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() @@ -1715,8 +1714,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: pytest.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() @@ -1765,9 +1765,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() @@ -1803,6 +1804,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(): @@ -1835,6 +1837,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: @@ -1853,6 +1856,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 @@ -1900,6 +1904,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() @@ -1916,6 +1921,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(): @@ -1964,6 +1970,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() @@ -2009,6 +2016,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} @@ -2050,6 +2058,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: @@ -2079,6 +2088,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", @@ -2143,6 +2153,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 @@ -2201,6 +2212,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( @@ -2251,6 +2263,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} @@ -2296,6 +2309,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( @@ -2355,6 +2369,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} @@ -2453,7 +2468,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) @@ -2509,7 +2526,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", @@ -2643,7 +2659,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) diff --git a/tests/unit/test_cost_calculator.py b/tests/unit/test_cost_calculator.py index 84f6dab9eff..60beaccee26 100644 --- a/tests/unit/test_cost_calculator.py +++ b/tests/unit/test_cost_calculator.py @@ -1,46 +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_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): @@ -516,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: @@ -1613,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=[], @@ -3371,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 @@ -6941,23 +6964,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(): @@ -6983,28 +7000,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( @@ -7035,8 +7046,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) @@ -7056,15 +7066,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) @@ -7111,9 +7118,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) @@ -7205,9 +7210,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, @@ -7217,20 +7220,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, @@ -7360,7 +7357,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"] @@ -7373,8 +7370,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(): @@ -7385,9 +7382,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" @@ -7404,13 +7399,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"] @@ -7423,16 +7416,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") @@ -7474,9 +7464,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=[ @@ -7510,9 +7498,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="") @@ -7611,9 +7597,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 @@ -7621,12 +7605,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) @@ -7718,9 +7699,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) @@ -7810,11 +7789,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) @@ -7855,9 +7832,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 @@ -8809,9 +8784,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 ), @@ -8850,27 +8823,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 @@ -8933,7 +8894,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", @@ -8995,9 +8956,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", @@ -9063,9 +9022,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 @@ -9086,7 +9043,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}), @@ -9102,7 +9065,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}, }, @@ -9216,7 +9183,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( @@ -9350,16 +9322,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, @@ -9377,9 +9346,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, @@ -9387,9 +9354,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) @@ -9398,9 +9363,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( { @@ -9442,9 +9405,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) @@ -9540,7 +9501,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(): @@ -9845,9 +9806,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", @@ -9942,9 +9901,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( @@ -10589,3 +10547,213 @@ def test_bedrock_registered_cache_prices_preserve_dynamic_threshold_rates( assert entry is not None assert entry[1]["input_cost_per_token_above_300k_tokens"] == 0.03 assert entry[1][tier_field] == (tier_override if tier_override is not None else 0.007) + + +_REPORTED_USAGE: Final = {"prompt_tokens": 100, "completion_tokens": 20, "total_tokens": 120, "cost": 0.037} +_REPORTED_COST_FIXTURE_MODEL: Final = "openai/reported-cost-fixture" +_REPORTED_COST_FIXTURE_RATES: Final = { + "litellm_provider": "openai", + "mode": "chat", + "input_cost_per_token": 1e-6, + "output_cost_per_token": 3e-6, +} + + +def _reported_cost_logging(stream: bool, litellm_params: Mapping[str, object]) -> Logging: + logging_obj: Final = Logging( + model=_REPORTED_COST_FIXTURE_MODEL, + messages=[{"role": "user", "content": "hi"}], + stream=stream, + call_type="completion", + start_time=datetime.datetime(2026, 1, 1), + litellm_call_id="reported-cost", + function_id="reported-cost", + ) + logging_obj.update_from_kwargs( + kwargs={}, optional_params={}, litellm_params={"custom_llm_provider": "openai", **litellm_params} + ) + logging_obj.model_call_details["custom_llm_provider"] = "openai" + return logging_obj + + +def _reported_cost_stream_chunk(content: str | None, finish_reason: str | None, usage: Usage | None) -> object: + return litellm.ModelResponseStream( + id="chatcmpl-reported", + 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 + ) + ], + usage=usage, + ) + + +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: + 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(**_REPORTED_USAGE)), + ], + messages=[{"role": "user", "content": "hi"}], + logging_obj=logging_obj, + ) + else: + response = ModelResponse( + id="chatcmpl-reported", + model="reported-cost-fixture", + choices=[Choices(index=0, message=Message(role="assistant", content="hi"), finish_reason="stop")], + usage=Usage(**_REPORTED_USAGE), + ) + assert isinstance(response, ModelResponse) + response._hidden_params.update({"custom_llm_provider": "openai", **hidden_params}) + return logging_obj.response_cost_calculator(result=response) + + +@pytest.mark.parametrize("stream", [False, True]) +@pytest.mark.parametrize("mapped", [False, True]) +def test_provider_reported_usage_cost_is_billed_on_stream_and_non_stream( + monkeypatch: pytest.MonkeyPatch, stream: bool, mapped: bool +) -> None: + """Regression for #47533: a plain openai-compatible deployment billed usage.cost on streams and $0 on non-stream.""" + if mapped: + monkeypatch.setitem(litellm.model_cost, _REPORTED_COST_FIXTURE_MODEL, _REPORTED_COST_FIXTURE_RATES) + else: + monkeypatch.delitem(litellm.model_cost, _REPORTED_COST_FIXTURE_MODEL, raising=False) + + assert _bill_reported_cost(stream, {}, {}) == pytest.approx(0.037) + + +@pytest.mark.parametrize("stream", [False, True]) +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( + litellm.model_cost, + deployment_id, + {"litellm_provider": "openai", "mode": "chat", "input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6}, + ) + litellm_params: Final = { + "input_cost_per_token": 2e-6, + "output_cost_per_token": 6e-6, + "model_info": {"id": deployment_id}, + } + + assert _bill_reported_cost(stream, litellm_params, {"model_id": deployment_id}) == pytest.approx( + 100 * 2e-6 + 20 * 6e-6 + ) + + +@pytest.mark.parametrize("stream", [False, True]) +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, + base_model, + {"litellm_provider": "openai", "mode": "chat", "input_cost_per_token": 5e-6, "output_cost_per_token": 7e-6}, + ) + + assert _bill_reported_cost(stream, {"base_model": base_model}, {}) == pytest.approx(100 * 5e-6 + 20 * 7e-6) + + +@pytest.mark.parametrize("stream", [False, True]) +def test_provider_reported_usage_cost_takes_discount_then_margin(monkeypatch: pytest.MonkeyPatch, stream: bool) -> None: + monkeypatch.setattr(litellm, "cost_discount_config", {"openai": 0.2}) + monkeypatch.setattr(litellm, "cost_margin_config", {"global": {"percentage": 0.5, "fixed_amount": 0.001}}) + + assert _bill_reported_cost(stream, {}, {}) == pytest.approx(0.037 * 0.8 * 1.5 + 0.001) + + +def test_the_stream_builder_bills_the_operator_rates_and_leaves_usage_cost_as_the_upstream_reported_it( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """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}, + ) + logging_obj: Final = _reported_cost_logging( + True, {"input_cost_per_token": 2e-6, "output_cost_per_token": 6e-6, "model_info": {"id": deployment_id}} + ) + operator_bill: Final = 100 * 2e-6 + 20 * 6e-6 + + 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(**_REPORTED_USAGE)), + ], + messages=[{"role": "user", "content": "hi"}], + logging_obj=logging_obj, + ) + + assert isinstance(response, ModelResponse) + 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 + ) + 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( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setitem(litellm.model_cost, _REPORTED_COST_FIXTURE_MODEL, _REPORTED_COST_FIXTURE_RATES) + usage: Final = Usage(**_REPORTED_USAGE) + + prompt_cost, completion_cost = cost_per_token( + model=_REPORTED_COST_FIXTURE_MODEL, custom_llm_provider="openai", usage_object=usage + ) + + priced_prompt: Final = 100 * 1e-6 + priced_completion: Final = 20 * 3e-6 + assert prompt_cost + completion_cost == pytest.approx(0.037) + assert prompt_cost == pytest.approx(0.037 * priced_prompt / (priced_prompt + priced_completion)) + + +def test_provider_reported_usage_cost_without_rates_is_all_output(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.delitem(litellm.model_cost, _REPORTED_COST_FIXTURE_MODEL, raising=False) + + assert cost_per_token( + model=_REPORTED_COST_FIXTURE_MODEL, custom_llm_provider="openai", usage_object=Usage(**_REPORTED_USAGE) + ) == (0.0, 0.037) + + +@pytest.mark.parametrize("reported", [-0.037, float("nan"), float("inf"), True]) +def test_unusable_provider_reported_usage_cost_falls_back_to_the_token_math( + monkeypatch: pytest.MonkeyPatch, reported: object +) -> None: + monkeypatch.setitem(litellm.model_cost, _REPORTED_COST_FIXTURE_MODEL, _REPORTED_COST_FIXTURE_RATES) + usage: Final = Usage(prompt_tokens=100, completion_tokens=20, total_tokens=120) + setattr(usage, "cost", reported) + + assert cost_per_token(model=_REPORTED_COST_FIXTURE_MODEL, custom_llm_provider="openai", usage_object=usage) == ( + pytest.approx(100 * 1e-6), + pytest.approx(20 * 3e-6), + ) + + +def test_caller_supplied_custom_cost_per_token_beats_the_provider_reported_usage_cost() -> None: + assert cost_per_token( + model=_REPORTED_COST_FIXTURE_MODEL, + custom_llm_provider="openai", + prompt_tokens=100, + completion_tokens=20, + usage_object=Usage(**_REPORTED_USAGE), + custom_cost_per_token={"input_cost_per_token": 0.001, "output_cost_per_token": 0.002}, + ) == (pytest.approx(0.1), pytest.approx(0.04)) diff --git a/tests/unit/test_main.py b/tests/unit/test_main.py index f38bbcf4404..4df1907b3f0 100644 --- a/tests/unit/test_main.py +++ b/tests/unit/test_main.py @@ -20,7 +20,7 @@ from unittest.mock import AsyncMock, MagicMock, patch import httpx import pytest import respx -from openai import APITimeoutError, AsyncOpenAI, OpenAI +from openai import AsyncOpenAI, OpenAI from openai._legacy_response import LegacyAPIResponse from openai.types.chat.chat_completion import ChatCompletion from pydantic import TypeAdapter @@ -291,9 +291,7 @@ def test_completion_missing_role(openai_api_response): print(f"openai_api_response: {openai_api_response}") - with patch.object( - client, "post", MagicMock(return_value=mock_raw_response) - ) as mock_create: + with patch.object(client, "post", MagicMock(return_value=mock_raw_response)) as mock_create: litellm.completion( model="gpt-4o-mini", messages=[ @@ -447,7 +445,7 @@ async def test_url_with_format_param(model, sync_mode, monkeypatch): else: response = await acompletion(**args, client=client) print(response) - except Exception as e: + except Exception: pass mock_client.assert_called() @@ -2860,9 +2858,7 @@ def test_parallel_function_call_stream_reassembles_each_arguments_payload(): created=1751934860, model="gpt-4.1-mini", object="chat.completion.chunk", - choices=[ - StreamingChoices(index=0, delta=Delta(tool_calls=tool_calls), finish_reason=finish_reason) - ], + choices=[StreamingChoices(index=0, delta=Delta(tool_calls=tool_calls), finish_reason=finish_reason)], ) chunks: Final = [ @@ -2993,9 +2989,7 @@ def test_parallel_function_call_stream_over_responses_bridge_ignores_events_it_d }, ) route: Final = respx.post("https://api.openai.com/v1/responses").mock( - return_value=httpx.Response( - 200, headers={"content-type": "text/event-stream"}, content=_responses_sse(events) - ) + return_value=httpx.Response(200, headers={"content-type": "text/event-stream"}, content=_responses_sse(events)) ) response: Final = litellm.completion( @@ -3609,9 +3603,7 @@ async def test_retrying() -> None: ) -def test_anthropic_disable_url_suffix_env_var( - monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter -) -> None: +def test_anthropic_disable_url_suffix_env_var(monkeypatch: pytest.MonkeyPatch, respx_mock: respx.MockRouter) -> None: monkeypatch.setenv("ANTHROPIC_API_BASE", "https://api.example.com") default_route: Final = respx_mock.post("https://api.example.com/v1/messages").mock( return_value=httpx.Response( @@ -3874,9 +3866,7 @@ async def test_async_mock_streaming_request(): print(chunk) complete_response += chunk["choices"][0]["delta"]["content"] or "" - assert ( - complete_response == "LiteLLM is awesome" - ), f"Unexpected response got {complete_response}" + assert complete_response == "LiteLLM is awesome", f"Unexpected response got {complete_response}" def test_mock_request_n_greater_than_1(): @@ -4402,7 +4392,10 @@ async def test_acompletion_forwards_store_and_prompt_cache_key_to_openai(): assert request_body["prompt_cache_key"] == "test-cache-key" -_TOOL: Final[Mapping[str, object]] = {"type": "function", "function": {"name": "f", "parameters": {"type": "object", "properties": {}}}} +_TOOL: Final[Mapping[str, object]] = { + "type": "function", + "function": {"name": "f", "parameters": {"type": "object", "properties": {}}}, +} # Client settings ride in OPENAI_CHAT_COMPLETION_PARAMS but configure the HTTP client, never the body. _OPENAI_CLIENT_SETTINGS: Final = frozenset( @@ -4462,7 +4455,9 @@ _OPENAI_BODY_PARAM_CASES: Final[Mapping[str, tuple[str, Mapping[str, object]]]] def test_every_openai_body_param_has_a_forwarding_case(): - body_params: Final = frozenset(litellm.OPENAI_CHAT_COMPLETION_PARAMS) - _OPENAI_CLIENT_SETTINGS - _NOT_OPENAI_BODY_PARAMS + body_params: Final = ( + frozenset(litellm.OPENAI_CHAT_COMPLETION_PARAMS) - _OPENAI_CLIENT_SETTINGS - _NOT_OPENAI_BODY_PARAMS + ) assert frozenset(_OPENAI_BODY_PARAM_CASES) == body_params, ( "a param added to OPENAI_CHAT_COMPLETION_PARAMS needs a case in _OPENAI_BODY_PARAM_CASES proving " "completion() sends it to OpenAI; prompt_cache_retention (LIT-9758) and store/prompt_cache_key " @@ -5221,7 +5216,7 @@ def _stream_builder_logging_obj(model: str = "gpt-4o", custom_llm_provider: str return logging_obj -def test_stream_chunk_builder_stamps_streaming_usage_cost_by_default(monkeypatch: pytest.MonkeyPatch): +def test_stream_chunk_builder_bills_a_stream_without_writing_into_its_usage(monkeypatch: pytest.MonkeyPatch): monkeypatch.setattr(litellm, "include_cost_in_streaming_usage", False) chunks: Final = [ _stream_builder_text_chunk("gpt-4o", "Hello "), @@ -5233,10 +5228,28 @@ def test_stream_chunk_builder_stamps_streaming_usage_cost_by_default(monkeypatch ) assert response is not None - usage_cost: Final = getattr(response.usage, "cost", None) - assert usage_cost is not None - assert usage_cost > 0 - assert response._hidden_params["response_cost"] == pytest.approx(usage_cost) + assert getattr(response.usage, "cost", None) is None, "usage.cost is the upstream's field, the bill rides hidden" + assert response._hidden_params["response_cost"] > 0 + + +def test_stream_chunk_builder_keeps_a_zero_bill_as_hidden_response_cost() -> None: + logging_obj: Final = _stream_builder_logging_obj() + logging_obj.update_environment_variables( + litellm_params={"custom_llm_provider": "openai", "input_cost_per_token": 0.0, "output_cost_per_token": 0.0}, + optional_params={}, + ) + logging_obj.custom_cost_per_token = {"input_cost_per_token": 0.0, "output_cost_per_token": 0.0} + chunks: Final = [ + _stream_builder_text_chunk("gpt-4o", "Hello "), + _stream_builder_text_chunk("gpt-4o", "world.", finish_reason="stop"), + ] + + response: Final = litellm.stream_chunk_builder( + chunks=chunks, messages=[{"role": "user", "content": "hi"}], logging_obj=logging_obj + ) + + assert response is not None + assert response._hidden_params["response_cost"] == 0.0, "a free stream still tells the egress what was billed" def test_stream_chunk_builder_stamped_cost_is_not_re_read_as_xai_reported_cost(): @@ -5267,7 +5280,8 @@ def test_stream_chunk_builder_stamped_cost_is_not_re_read_as_xai_reported_cost() ) assert response is not None - usage_cost: Final = getattr(response.usage, "cost", None) + assert getattr(response.usage, "cost", None) is None + usage_cost: Final = response._hidden_params.get("response_cost") assert usage_cost is not None assert usage_cost > 0 @@ -5278,10 +5292,9 @@ def test_stream_chunk_builder_stamped_cost_is_not_re_read_as_xai_reported_cost() assert breakdown["output_cost"] > 0 assert breakdown["input_cost"] + breakdown["output_cost"] == pytest.approx(usage_cost) assert spend_row_cost == pytest.approx(usage_cost) - assert response._hidden_params.get("response_cost") == pytest.approx(usage_cost) -def test_stream_chunk_builder_skips_stamp_when_cost_is_unpriceable(): +def test_stream_chunk_builder_stamps_the_zero_bill_when_cost_is_unpriceable(): import time as time_module from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging @@ -5310,7 +5323,7 @@ def test_stream_chunk_builder_skips_stamp_when_cost_is_unpriceable(): assert response is not None assert getattr(response.usage, "cost", None) is None - assert response._hidden_params.get("response_cost") is None + assert response._hidden_params.get("response_cost") == 0.0, "the spend row reads 0, so the egress must too" def test_stream_chunk_builder_prices_alias_response_at_its_deployment_model(): @@ -5344,7 +5357,7 @@ def test_stream_chunk_builder_prices_alias_response_at_its_deployment_model(): entry: Final = litellm.model_cost[deployment_model] expected_cost: Final = 40 * entry["input_cost_per_token"] + 5 * entry["output_cost_per_token"] assert response is not None - assert getattr(response.usage, "cost", None) == pytest.approx(expected_cost) + assert getattr(response.usage, "cost", None) is None assert response._hidden_params.get("response_cost") == pytest.approx(expected_cost) @@ -5387,7 +5400,9 @@ def test_stream_chunk_builder_prices_alias_from_openai_sdk_usage_chunk(): assert response._hidden_params["response_cost"] == pytest.approx(0.000704) -def test_stream_chunk_builder_leaves_xai_reported_cost_to_the_calculator(monkeypatch: pytest.MonkeyPatch): +def test_stream_chunk_builder_bills_the_margin_over_the_xai_reported_cost_it_leaves_in_usage( + monkeypatch: pytest.MonkeyPatch, +): monkeypatch.setattr(litellm, "cost_margin_config", {"xai": 0.5}) usage_chunk: Final = _stream_builder_text_chunk("grok-4", "") usage_chunk.usage = Usage(prompt_tokens=5, completion_tokens=2, total_tokens=7, cost=0.42) @@ -5403,9 +5418,10 @@ def test_stream_chunk_builder_leaves_xai_reported_cost_to_the_calculator(monkeyp ) assert response is not None - assert getattr(response.usage, "cost", None) == pytest.approx(0.42) - assert response._hidden_params.get("response_cost") is None + assert getattr(response.usage, "cost", None) == pytest.approx(0.42), "the upstream's figure stays in usage.cost" + assert response._hidden_params.get("response_cost") == pytest.approx(0.63) assert logging_obj.response_cost_calculator(result=response) == pytest.approx(0.63) + assert logging_obj.response_cost_calculator(result=response, ignore_stamped_cost=True) == pytest.approx(0.63) def test_speech_mistral_dispatches_and_decodes_audio(respx_mock: respx.MockRouter, monkeypatch: pytest.MonkeyPatch): @@ -5908,17 +5924,11 @@ def test_drop_params_false_still_rejects_an_invalid_stream_chunk_size() -> None: def test_acompletion_params(): import inspect - from litellm.types.completion import CompletionRequest - acompletion_params_odict = inspect.signature(acompletion).parameters completion_params_dict = inspect.signature(completion).parameters - acompletion_params = { - name: param.annotation for name, param in acompletion_params_odict.items() - } - completion_params = { - name: param.annotation for name, param in completion_params_dict.items() - } + acompletion_params = {name: param.annotation for name, param in acompletion_params_odict.items()} + completion_params = {name: param.annotation for name, param in completion_params_dict.items()} keys_acompletion = set(acompletion_params.keys()) keys_completion = set(completion_params.keys()) @@ -5972,9 +5982,7 @@ def test_null_role_response(): import openai openai_client = openai.OpenAI() - with patch.object( - openai_client, "post", side_effect=_openai_mock_response - ) as mock_response: + with patch.object(openai_client, "post", side_effect=_openai_mock_response) as mock_response: response = litellm.completion( model="gpt-3.5-turbo", messages=[{"role": "user", "content": "Hey! how's it going?"}], @@ -6072,7 +6080,8 @@ def test_completion_perplexity_api(): openai_client = OpenAI() with patch.object( - openai_client, "post", + openai_client, + "post", side_effect=_return_pydantic_obj, ) as mock_client: # litellm.set_verbose= True @@ -6115,12 +6124,10 @@ async def test_openai_compatible_custom_api_base(provider): openai_client = OpenAI(api_key="fake-key") - with patch.object( - openai_client, "post", new=MagicMock() - ) as mock_call: + with patch.object(openai_client, "post", new=MagicMock()) as mock_call: try: completion( - model="{provider}/my-vllm-model".format(provider=provider), + model=f"{provider}/my-vllm-model", messages=messages, response_format={"type": "json_object"}, client=openai_client, @@ -6132,7 +6139,7 @@ async def test_openai_compatible_custom_api_base(provider): mock_call.assert_called_once() - print("Call KWARGS - {}".format(mock_call.call_args.kwargs)) + print(f"Call KWARGS - {mock_call.call_args.kwargs}") assert "hello" in mock_call.call_args.kwargs["options"]["extra_json"] @@ -6166,12 +6173,10 @@ async def test_openai_compatible_custom_api_video(provider): openai_client = OpenAI(api_key="fake-key") - with patch.object( - openai_client, "post", new=MagicMock() - ) as mock_call: + with patch.object(openai_client, "post", new=MagicMock()) as mock_call: try: completion( - model="{provider}/my-vllm-model".format(provider=provider), + model=f"{provider}/my-vllm-model", messages=messages, response_format={"type": "json_object"}, client=openai_client, @@ -6361,10 +6366,7 @@ async def test_completion_functions_param(): mock_client.assert_called() print(f"mock_client.call_args.kwargs: {mock_client.call_args.kwargs}") assert "tools" in mock_client.call_args.kwargs["json"] - assert ( - "litellm_param_is_function_call" - not in mock_client.call_args.kwargs["json"] - ) + assert "litellm_param_is_function_call" not in mock_client.call_args.kwargs["json"] assert response.choices[0].message.function_call is not None except Exception as e: pytest.fail(f"Error occurred: {e}") @@ -6427,7 +6429,7 @@ def test_bedrock_deepseek_custom_prompt_dict(): messages=messages, client=client, ) - except Exception as e: + except Exception: pass mock_post.assert_called_once() @@ -6440,9 +6442,7 @@ def test_bedrock_deepseek_custom_prompt_dict(): def test_bedrock_deepseek_known_tokenizer_config(monkeypatch): - model = ( - "deepseek_r1/arn:aws:bedrock:us-west-2:888602223428:imported-model/bnnr6463ejgf" - ) + model = "deepseek_r1/arn:aws:bedrock:us-west-2:888602223428:imported-model/bnnr6463ejgf" from unittest.mock import Mock import httpx @@ -6513,9 +6513,7 @@ def test_completion_anthropic_hanging(): {"role": "function", "name": "get_capital", "content": "Kokoko"}, ] - converted_messages = anthropic_messages_pt( - messages, model="claude-3-sonnet-20240229", llm_provider="anthropic" - ) + converted_messages = anthropic_messages_pt(messages, model="claude-3-sonnet-20240229", llm_provider="anthropic") assert len(converted_messages) == 3 for i, msg in enumerate(converted_messages): @@ -6822,9 +6820,7 @@ def test_completion_logprobs_parses_top_logprobs(): "top_logprobs": 3, "reasoning_effort": "none", } - assert [ - item.token for item in response.choices[0].logprobs.content[0].top_logprobs - ] == ["hello", "hi", "hey"] + assert [item.token for item in response.choices[0].logprobs.content[0].top_logprobs] == ["hello", "hi", "hey"] choice: Final = response.choices[0] assert "logprobs" in choice assert "content" in choice.logprobs @@ -6896,9 +6892,11 @@ def test_completion_logprobs_stream_parses_top_logprobs(): "stream_options": {"include_usage": True}, "stream": True, } - assert [ - item.token for item in response_chunks[0].choices[0].logprobs.content[0].top_logprobs - ] == ["hello", "hi", "hey"] + assert [item.token for item in response_chunks[0].choices[0].logprobs.content[0].top_logprobs] == [ + "hello", + "hi", + "hey", + ] @respx.mock @@ -6969,9 +6967,7 @@ async def test_bedrock_converse_parses_scripted_response( "api_key": "test-bedrock-token", "aws_region_name": "us-west-2", } - 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) assert route.calls[0].request.headers["authorization"] == "Bearer test-bedrock-token" assert json.loads(route.calls[0].request.content) == { @@ -7058,9 +7054,7 @@ def test_bedrock_converse_parses_scripted_tool_call_arguments( tool_call: Final = response.choices[0].message.tool_calls[0] assert route.call_count == 1 assert tool_call.function.name == "trade" - assert json.loads(tool_call.function.arguments) == { - "orders": [{"action": "buy", "asset": "BTC", "amount": 0.1}] - } + assert json.loads(tool_call.function.arguments) == {"orders": [{"action": "buy", "asset": "BTC", "amount": 0.1}]} @pytest.mark.parametrize( @@ -7098,9 +7092,9 @@ def test_bedrock_invoke_signs_request_and_parses_scripted_response( expected_request_body: dict[str, object], fake_provider_credentials: None, ): - route: Final = respx.post( - f"https://bedrock-runtime.us-west-2.amazonaws.com/model/{invoke_model_id}/invoke" - ).mock(return_value=httpx.Response(200, json=response_body)) + route: Final = respx.post(f"https://bedrock-runtime.us-west-2.amazonaws.com/model/{invoke_model_id}/invoke").mock( + return_value=httpx.Response(200, json=response_body) + ) response: Final = litellm.completion( model=model, @@ -7111,9 +7105,7 @@ def test_bedrock_invoke_signs_request_and_parses_scripted_response( ) request: Final = route.calls[0].request - assert request.headers["authorization"].startswith( - "AWS4-HMAC-SHA256 Credential=unit-test/" - ) + assert request.headers["authorization"].startswith("AWS4-HMAC-SHA256 Credential=unit-test/") assert "/us-west-2/bedrock/aws4_request" in request.headers["authorization"] assert request.headers["x-amz-date"].endswith("Z") assert json.loads(request.content) == expected_request_body @@ -7123,9 +7115,9 @@ def test_bedrock_invoke_signs_request_and_parses_scripted_response( @respx.mock def test_completion_azure_ad_token_is_forwarded(openai_api_response): - route: Final = respx.post( - "https://example.openai.azure.com/openai/deployments/gpt-6.1-sol/chat/completions" - ).mock(return_value=httpx.Response(200, json=openai_api_response)) + route: Final = respx.post("https://example.openai.azure.com/openai/deployments/gpt-6.1-sol/chat/completions").mock( + return_value=httpx.Response(200, json=openai_api_response) + ) litellm.completion( model="azure/gpt-6.1-sol", @@ -7140,9 +7132,9 @@ def test_completion_azure_ad_token_is_forwarded(openai_api_response): @respx.mock def test_completion_azure_extra_headers_are_forwarded(openai_api_response): - route: Final = respx.post( - "https://example.openai.azure.com/openai/deployments/gpt-6.1-sol/chat/completions" - ).mock(return_value=httpx.Response(200, json=openai_api_response)) + route: Final = respx.post("https://example.openai.azure.com/openai/deployments/gpt-6.1-sol/chat/completions").mock( + return_value=httpx.Response(200, json=openai_api_response) + ) litellm.completion( model="azure/gpt-6.1-sol", @@ -7163,9 +7155,9 @@ def test_completion_azure_extra_headers_are_forwarded(openai_api_response): @respx.mock def test_completion_azure_api_key_argument_is_forwarded(openai_api_response): - route: Final = respx.post( - "https://example.openai.azure.com/openai/deployments/gpt-6.1-sol/chat/completions" - ).mock(return_value=httpx.Response(200, json=openai_api_response)) + route: Final = respx.post("https://example.openai.azure.com/openai/deployments/gpt-6.1-sol/chat/completions").mock( + return_value=httpx.Response(200, json=openai_api_response) + ) response: Final = litellm.completion( model="azure/gpt-6.1-sol", @@ -7822,9 +7814,7 @@ def test_completion_deep_infra(drop_params): ], "usage": {"prompt_tokens": 20, "completion_tokens": 5, "total_tokens": 25}, } - route: Final = respx.post(endpoint).mock( - return_value=httpx.Response(200, json=response_body) - ) + route: Final = respx.post(endpoint).mock(return_value=httpx.Response(200, json=response_body)) if not drop_params: with pytest.raises(litellm.exceptions.UnsupportedParamsError): @@ -7888,9 +7878,7 @@ def test_completion_deep_infra_mistral(): ], "usage": {"prompt_tokens": 20, "completion_tokens": 5, "total_tokens": 25}, } - route: Final = respx.post(endpoint).mock( - return_value=httpx.Response(200, json=response_body) - ) + route: Final = respx.post(endpoint).mock(return_value=httpx.Response(200, json=response_body)) response: Final = completion( model=model_name, @@ -7939,19 +7927,13 @@ def test_unified_auth_params(provider, model, project, region_name, token): translated_optional_params = litellm.utils.get_optional_params(**data) if provider == "azure": - special_auth_params = ( - litellm.AzureOpenAIConfig().get_mapped_special_auth_params() - ) + special_auth_params = litellm.AzureOpenAIConfig().get_mapped_special_auth_params() elif provider == "bedrock": - special_auth_params = ( - litellm.AmazonBedrockGlobalConfig().get_mapped_special_auth_params() - ) + special_auth_params = litellm.AmazonBedrockGlobalConfig().get_mapped_special_auth_params() elif provider == "vertex_ai": special_auth_params = litellm.VertexAIConfig().get_mapped_special_auth_params() elif provider == "watsonx": - special_auth_params = ( - litellm.IBMWatsonXAIConfig().get_mapped_special_auth_params() - ) + special_auth_params = litellm.IBMWatsonXAIConfig().get_mapped_special_auth_params() for param, value in special_auth_params.items(): assert param in data @@ -7983,9 +7965,7 @@ async def test_dynamic_azure_params(stream, sync_mode): mock_client = AsyncMock(return_value="Hello world!") ## CHECK IF CLIENT IS USED (NO PARAM CHANGE) - with patch.object( - client.chat.completions.with_raw_response, "create", new=mock_client - ) as mock_client: + with patch.object(client.chat.completions.with_raw_response, "create", new=mock_client) as mock_client: try: # client.chat.completions.with_raw_response.create = mock_client if sync_mode: @@ -8014,9 +7994,7 @@ async def test_dynamic_azure_params(stream, sync_mode): new_mock_client = AsyncMock(return_value="Hello world!") ## CHECK IF NEW CLIENT IS USED (PARAM CHANGE) - with patch.object( - client.chat.completions.with_raw_response, "create", new=new_mock_client - ) as new_mock_client: + with patch.object(client.chat.completions.with_raw_response, "create", new=new_mock_client) as new_mock_client: try: if sync_mode: _ = completion( @@ -8227,9 +8205,7 @@ def test_completion_novita_ai(): openai_client = OpenAI(api_key="fake-key") - with patch.object( - openai_client, "post", return_value=MagicMock(spec=LegacyAPIResponse) - ) as mock_call: + with patch.object(openai_client, "post", return_value=MagicMock(spec=LegacyAPIResponse)) as mock_call: mock_call.return_value.headers = {} mock_call.return_value.parse.return_value = litellm.ModelResponse( choices=[{"message": {"role": "assistant", "content": "Hello"}}] @@ -8246,10 +8222,7 @@ def test_completion_novita_ai(): assert response.choices[0].message.content == "Hello" # Verify model is passed correctly - assert ( - mock_call.call_args.kwargs["body"]["model"] - == "meta-llama/llama-3.3-70b-instruct" - ) + assert mock_call.call_args.kwargs["body"]["model"] == "meta-llama/llama-3.3-70b-instruct" # Verify messages are passed correctly assert mock_call.call_args.kwargs["body"]["messages"] == messages @@ -8306,9 +8279,7 @@ def test_completion_openai_metadata(monkeypatch, enable_preview_features): litellm.set_verbose = True monkeypatch.setattr(litellm, "enable_preview_features", enable_preview_features) - with patch.object( - client, "post", return_value=MagicMock() - ) as mock_completion: + with patch.object(client, "post", return_value=MagicMock()) as mock_completion: try: resp = litellm.completion( model="openai/gpt-3.5-turbo", @@ -8321,9 +8292,7 @@ def test_completion_openai_metadata(monkeypatch, enable_preview_features): mock_completion.assert_called_once() if enable_preview_features: - assert mock_completion.call_args.kwargs["body"]["metadata"] == { - "my-test-key": "my-test-value" - } + assert mock_completion.call_args.kwargs["body"]["metadata"] == {"my-test-key": "my-test-value"} else: assert "metadata" not in mock_completion.call_args.kwargs["body"] @@ -8544,17 +8513,30 @@ async def test_async_responses_still_retries_provider_server_errors(monkeypatch) monkeypatch.setattr(litellm, "num_retries", None) monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) with respx.mock as upstream: - response = upstream.post("https://openai-test.invalid/v1/responses").mock(side_effect=[ - httpx.Response(500, json={"error": {"message": "temporary provider failure", "type": "server_error"}}), - httpx.Response(200, json={ - "id": "resp-retry", "object": "response", "created_at": 1, "status": "completed", - "model": "test-model", "output": [], - "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5}, - }), - ]) + response = upstream.post("https://openai-test.invalid/v1/responses").mock( + side_effect=[ + httpx.Response(500, json={"error": {"message": "temporary provider failure", "type": "server_error"}}), + httpx.Response( + 200, + json={ + "id": "resp-retry", + "object": "response", + "created_at": 1, + "status": "completed", + "model": "test-model", + "output": [], + "usage": {"input_tokens": 3, "output_tokens": 2, "total_tokens": 5}, + }, + ), + ] + ) result = await litellm.aresponses( - model="openai/test-model", input="ping", api_key="test-key", - api_base="https://openai-test.invalid/v1", num_retries=1, max_retries=0, + model="openai/test-model", + input="ping", + api_key="test-key", + api_base="https://openai-test.invalid/v1", + num_retries=1, + max_retries=0, ) assert result.status == "completed" assert response.call_count == 2 @@ -8740,9 +8722,7 @@ def test_openai_chat_stream_routes_legacy_function_calls_and_usage( "object": "chat.completion.chunk", "created": 1, "model": "gpt-4o-mini", - "choices": [ - {"index": 0, "delta": {}, "finish_reason": "function_call"} - ], + "choices": [{"index": 0, "delta": {}, "finish_reason": "function_call"}], }, ) @@ -8766,9 +8746,7 @@ def test_openai_chat_stream_routes_legacy_function_calls_and_usage( ), ) ) - route: Final = respx.post(f"{api_base}/chat/completions").mock( - side_effect=lambda _: next(scripted_responses) - ) + route: Final = respx.post(f"{api_base}/chat/completions").mock(side_effect=lambda _: next(scripted_responses)) stream: Final = litellm.completion( model="gpt-4o-mini", api_key="sk-test", @@ -8779,11 +8757,14 @@ def test_openai_chat_stream_routes_legacy_function_calls_and_usage( ) answer_chunks: Final = tuple(stream) - assert "".join( - chunk.choices[0].delta.content - for chunk in answer_chunks - if chunk.choices and chunk.choices[0].delta.content - ) == "The answer is 42." + assert ( + "".join( + chunk.choices[0].delta.content + for chunk in answer_chunks + if chunk.choices and chunk.choices[0].delta.content + ) + == "The answer is 42." + ) assert tuple( chunk.choices[0].finish_reason for chunk in answer_chunks @@ -8800,12 +8781,7 @@ def test_openai_chat_stream_routes_legacy_function_calls_and_usage( assert answer_chunks[-1].usage.completion_tokens == 2 assert answer_chunks[-1].usage.total_tokens == 6 assert stream._hidden_params["api_base"] == api_base - assert ( - answer_chunks[0]._hidden_params["additional_headers"][ - "llm_provider-x-ratelimit-remaining-requests" - ] - == "17" - ) + assert answer_chunks[0]._hidden_params["additional_headers"]["llm_provider-x-ratelimit-remaining-requests"] == "17" function_stream: Final = litellm.completion( model="gpt-4o-mini", @@ -8817,9 +8793,7 @@ def test_openai_chat_stream_routes_legacy_function_calls_and_usage( stream=True, ) function_chunks: Final = tuple(function_stream) - assembled: Final = litellm.stream_chunk_builder( - chunks=function_chunks, messages=messages - ) + assembled: Final = litellm.stream_chunk_builder(chunks=function_chunks, messages=messages) function_call: Final = assembled.choices[0].message.function_call assert function_call is not None @@ -8831,12 +8805,8 @@ def test_openai_chat_stream_routes_legacy_function_calls_and_usage( if chunk.choices and chunk.choices[0].finish_reason is not None ) == ("function_call",) assert route.call_count == 2 - assert json.loads(route.calls[0].request.content)["stream_options"] == { - "include_usage": True - } - assert json.loads(route.calls[1].request.content)["function_call"] == { - "name": "get_current_weather" - } + assert json.loads(route.calls[0].request.content)["stream_options"] == {"include_usage": True} + assert json.loads(route.calls[1].request.content)["function_call"] == {"name": "get_current_weather"} assert json.loads(route.calls[1].request.content)["functions"] == list(functions) @@ -8875,15 +8845,9 @@ async def test_async_openai_stream_options_include_usage_on_the_last_chunk( "usage": {"prompt_tokens": 3, "completion_tokens": 2, "total_tokens": 5}, }, ) - body: Final = "".join( - f"data: {json.dumps(event)}\n\n" for event in events - ) + "data: [DONE]\n\n" - route: Final = respx.post( - "https://api.openai.com/v1/chat/completions" - ).mock( - return_value=httpx.Response( - 200, text=body, headers={"content-type": "text/event-stream"} - ) + body: Final = "".join(f"data: {json.dumps(event)}\n\n" for event in events) + "data: [DONE]\n\n" + route: Final = respx.post("https://api.openai.com/v1/chat/completions").mock( + return_value=httpx.Response(200, text=body, headers={"content-type": "text/event-stream"}) ) stream: Final = await litellm.acompletion( model="gpt-4o-mini", @@ -8894,11 +8858,10 @@ async def test_async_openai_stream_options_include_usage_on_the_last_chunk( ) chunks: Final = tuple([chunk async for chunk in stream]) - assert "".join( - chunk.choices[0].delta.content - for chunk in chunks - if chunk.choices and chunk.choices[0].delta.content - ) == "async reply" + assert ( + "".join(chunk.choices[0].delta.content for chunk in chunks if chunk.choices and chunk.choices[0].delta.content) + == "async reply" + ) assert tuple(getattr(chunk, "usage", None) for chunk in chunks[:-1]) == ( None, None, @@ -8906,9 +8869,7 @@ async def test_async_openai_stream_options_include_usage_on_the_last_chunk( assert chunks[-1].usage.prompt_tokens == 3 assert chunks[-1].usage.completion_tokens == 2 assert chunks[-1].usage.total_tokens == 5 - assert json.loads(route.calls[0].request.content)["stream_options"] == { - "include_usage": True - } + assert json.loads(route.calls[0].request.content)["stream_options"] == {"include_usage": True} @respx.mock @@ -8945,16 +8906,10 @@ async def test_parallel_openai_streams_keep_prompt_responses_separate( "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}], }, ) - body: Final = "".join( - f"data: {json.dumps(event)}\n\n" for event in events - ) + "data: [DONE]\n\n" - return httpx.Response( - 200, text=body, headers={"content-type": "text/event-stream"} - ) + body: Final = "".join(f"data: {json.dumps(event)}\n\n" for event in events) + "data: [DONE]\n\n" + return httpx.Response(200, text=body, headers={"content-type": "text/event-stream"}) - route: Final = respx.post( - "https://api.openai.com/v1/chat/completions" - ).mock(side_effect=respond) + route: Final = respx.post("https://api.openai.com/v1/chat/completions").mock(side_effect=respond) async def collect(prompt: str) -> tuple[str, tuple[str, ...]]: stream = await litellm.acompletion( @@ -8964,9 +8919,7 @@ async def test_parallel_openai_streams_keep_prompt_responses_separate( stream=True, ) chunks: Final = tuple([chunk async for chunk in stream]) - response_text: Final = "".join( - chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices - ) + response_text: Final = "".join(chunk.choices[0].delta.content or "" for chunk in chunks if chunk.choices) finish_reasons: Final = tuple( chunk.choices[0].finish_reason for chunk in chunks @@ -8974,9 +8927,7 @@ async def test_parallel_openai_streams_keep_prompt_responses_separate( ) return response_text, finish_reasons - responses: Final = tuple( - await asyncio.gather(collect("request one"), collect("request two")) - ) + responses: Final = tuple(await asyncio.gather(collect("request one"), collect("request two"))) assert tuple(response for response, _ in responses) == ( "response one", @@ -9149,7 +9100,7 @@ def test_openai_streaming_tool_call_arguments_are_valid_json( @respx.mock -def test_streaming_openai_usage_includes_test_owned_cost( +def test_streaming_openai_usage_carries_no_cost_the_upstream_did_not_report( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(litellm, "disable_aiohttp_transport", True) @@ -9170,9 +9121,7 @@ def test_streaming_openai_usage_includes_test_owned_cost( "object": "chat.completion.chunk", "created": 1, "model": model_name, - "choices": [ - {"index": 0, "delta": {"role": "assistant", "content": "priced"}, "finish_reason": None} - ], + "choices": [{"index": 0, "delta": {"role": "assistant", "content": "priced"}, "finish_reason": None}], }, { "id": "chatcmpl-cost", @@ -9207,12 +9156,10 @@ def test_streaming_openai_usage_includes_test_owned_cost( ) stream_chunks: Final = tuple(stream) usage_chunks: Final = tuple( - getattr(chunk, "usage", None) - for chunk in stream_chunks - if getattr(chunk, "usage", None) is not None + getattr(chunk, "usage", None) for chunk in stream_chunks if getattr(chunk, "usage", None) is not None ) assert len(usage_chunks) == 1 - assert usage_chunks[0].cost == pytest.approx(4 * 0.001 + 2 * 0.002) + assert getattr(usage_chunks[0], "cost", None) is None, "the bill reaches a client through the proxy egress only" def test_completion_ensure_alternating_roles_sends_continue_messages(respx_mock: respx.MockRouter) -> None: @@ -9225,9 +9172,7 @@ def test_completion_ensure_alternating_roles_sends_continue_messages(respx_mock: "object": "chat.completion", "created": 0, "model": "databricks-meta-llama-3-1-70b-instruct", - "choices": [ - {"index": 0, "message": {"role": "assistant", "content": "Sure"}, "finish_reason": "stop"} - ], + "choices": [{"index": 0, "message": {"role": "assistant", "content": "Sure"}, "finish_reason": "stop"}], "usage": {"prompt_tokens": 40, "completion_tokens": 1, "total_tokens": 41}, }, ) @@ -9437,9 +9382,12 @@ def test_azure_ai_forwards_reasoning_effort_once_the_deployment_declares_levels( assert _chat_body_sent(respx_mock, "azure_ai/my-kimi", "adaptive", _KIMI_DEPLOYMENT)["reasoning_effort"] == ( "adaptive" ) - assert _chat_body_sent(respx_mock, "azure_ai/my-kimi", "none", {"reasoning_effort_levels": ["low", "high"]})[ - "reasoning_effort" - ] == "none" + assert ( + _chat_body_sent(respx_mock, "azure_ai/my-kimi", "none", {"reasoning_effort_levels": ["low", "high"]})[ + "reasoning_effort" + ] + == "none" + ) def test_a_model_map_declaration_alone_leaves_reasoning_effort_to_drop_params( @@ -9574,7 +9522,9 @@ def test_a_default_effort_beside_a_thinking_budget_sends_no_effort_on_a_declarin thinking={"type": "enabled", "budget_tokens": 2048, **display}, **default_effort, ) - adaptive_alone: Final = body_sent(respx_mock, _OPUS_DECLARING_LOW_AND_HIGH, thinking={"type": "adaptive", **display}) + adaptive_alone: Final = body_sent( + respx_mock, _OPUS_DECLARING_LOW_AND_HIGH, thinking={"type": "adaptive", **display} + ) assert "output_config" not in defaulted assert defaulted == adaptive_alone @@ -10079,7 +10029,9 @@ async def test_aspeech_answers_a_non_string_provider_hint_with_a_bad_request( hint: str, respx_mock: respx.MockRouter ) -> None: with pytest.raises(litellm.BadRequestError) as raised: - await litellm.aspeech(model="kokoro", input="hello", voice="alloy", api_key="fake_openai_api_key", **{hint: 123}) + await litellm.aspeech( + model="kokoro", input="hello", voice="alloy", api_key="fake_openai_api_key", **{hint: 123} + ) assert raised.value.status_code == 400 assert f"Invalid {hint}" in raised.value.message @@ -10269,7 +10221,9 @@ async def test_converse_fallback_prices_the_current_attempt( } if asynchronous: with pytest.raises(litellm.BadRequestError): - await litellm.acompletion(model="bedrock/converse/amazon.nova-micro-v1:0", model_id=failed_profile, **kwargs) + await litellm.acompletion( + model="bedrock/converse/amazon.nova-micro-v1:0", model_id=failed_profile, **kwargs + ) else: with pytest.raises(litellm.BadRequestError): litellm.completion(model="bedrock/converse/amazon.nova-micro-v1:0", model_id=failed_profile, **kwargs) @@ -10293,9 +10247,7 @@ async def test_converse_fallback_prices_the_current_attempt( assert result is not None prices: Final = litellm.get_model_info(profile or model, custom_llm_provider="bedrock") expected: Final = ( - 0.001 + 0.002 - if custom_price - else prices["input_cost_per_token"] + prices["output_cost_per_token"] + 0.001 + 0.002 if custom_price else prices["input_cost_per_token"] + prices["output_cost_per_token"] ) active_logging: Final = response.logging_obj if stream else logging_obj assert active_logging.response_cost_calculator(result) == pytest.approx(expected)