diff --git a/litellm/cost_calculator.py b/litellm/cost_calculator.py index 76cced0c1c9..4a72e88ad3f 100644 --- a/litellm/cost_calculator.py +++ b/litellm/cost_calculator.py @@ -2368,15 +2368,15 @@ def completion_cost( def get_response_cost_from_hidden_params( - hidden_params: dict | BaseModel, + hidden_params: Mapping[str, object] | BaseModel, ) -> float | None: if isinstance(hidden_params, BaseModel): _hidden_params_dict = hidden_params.model_dump() else: _hidden_params_dict = hidden_params - additional_headers: Final = _hidden_params_dict.get("additional_headers", {}) - if additional_headers and "llm_provider-x-litellm-response-cost" in additional_headers: + additional_headers: Final = as_str_mapping(_hidden_params_dict.get("additional_headers")) + if additional_headers is not None and "llm_provider-x-litellm-response-cost" in additional_headers: return _billable_header_cost(additional_headers["llm_provider-x-litellm-response-cost"]) return None diff --git a/litellm/litellm_core_utils/litellm_logging.py b/litellm/litellm_core_utils/litellm_logging.py index e22ca540864..c8f34995d06 100644 --- a/litellm/litellm_core_utils/litellm_logging.py +++ b/litellm/litellm_core_utils/litellm_logging.py @@ -61,6 +61,7 @@ from litellm.constants import ( from litellm.cost_calculator import ( RealtimeAPITokenUsageProcessor, ResponsesWebSocketTokenUsageProcessor, + get_response_cost_from_hidden_params, get_usage_object, pricing_entry_for_cost_calc, select_model_name_for_cost_calc, @@ -3076,7 +3077,9 @@ class Logging(LiteLLMLoggingBaseClass): return TypeAdapter(dict[str, JsonValue]).validate_python( { "response_ms": getattr(result, "_response_ms", None), - "provider_reported_cost": self._provider_reported_cost, + "provider_reported_cost": self._provider_reported_cost + if self._provider_reported_cost is not None + else get_response_cost_from_hidden_params(dict(get_hidden_params(result) or {})), "provider_usage_cost": billed_usage_cost(result, self.custom_llm_provider), "current_billing_time": self.billing_time_for_cost().isoformat(), "custom_pricing": self.custom_pricing_for(result), diff --git a/litellm/proxy/spend_tracking/cost_compiler.py b/litellm/proxy/spend_tracking/cost_compiler.py index 37516add6d4..a0cca9c3757 100644 --- a/litellm/proxy/spend_tracking/cost_compiler.py +++ b/litellm/proxy/spend_tracking/cost_compiler.py @@ -9,7 +9,16 @@ from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter from litellm._internal_context import pinned_billing_time from litellm._logging import verbose_proxy_logger -from litellm.accounting import CostEvent, DeliveryTarget, DimensionName, OutboundUsageEvent, PricingSnapshot, Unpriced +from litellm.accounting import ( + CostEvent, + DeliveryTarget, + DimensionName, + OutboundUsageEvent, + Outcome, + PricingSnapshot, + Unpriced, +) +from litellm.cost_calculator import batch_cost_calculator from litellm.litellm_core_utils.litellm_logging import Logging from litellm.litellm_core_utils.llm_cost_calc.finalization import finalize_cost from litellm.litellm_core_utils.llm_cost_calc.pricing_context import RetainedPricing, retained_pricing @@ -18,6 +27,7 @@ from litellm.proxy.spend_tracking.usage_meter import usage_delivery_targets from litellm.types.llms.openai import ResponsesAPIResponse from litellm.types.utils import ( CallTypes, + CostBreakdown, CostPerToken, LiteLLMRealtimeStreamLoggingObject, ModelResponse, @@ -71,6 +81,13 @@ def price_usage(envelope: str) -> PricingSnapshot | Unpriced: ) provider: Final = next((d.value for d in event.dimensions if d.name is DimensionName.PROVIDER), None) logged: Final = _fields(pricing.get("logging")) + if ( + fixed_call + and event.outcome is not Outcome.SUCCEEDED + and captured.usage is None + and not logged.get("tool_response_cost") + ): + return Unpriced(usage_event_id=event.event_id, reason="usage_unknown") reported: Final = ( logged.get("provider_reported_cost") if logged.get("provider_reported_cost") is not None @@ -180,7 +197,9 @@ def price_usage(envelope: str) -> PricingSnapshot | Unpriced: model=model, messages=[], stream=False, - call_type=event.call_type.value, + call_type=CallTypes.aresponses.value + if event.call_type in {CallTypes.get_responses, CallTypes.aget_responses} + else event.call_type.value, start_time=event.started_at, litellm_call_id=event.event_id, function_id=event.attempt_id, @@ -226,7 +245,26 @@ def price_usage(envelope: str) -> PricingSnapshot | Unpriced: ) logger.cost_billing_time = billed_at with retained_pricing(context), pinned_billing_time(billed_at): - amount: Final = logger.response_cost_calculator(response, ignore_stamped_cost=True) + batch_costs: Final = ( + batch_cost_calculator( + replay_usage, + model, + provider, + data_residency=TypeAdapter[str | None](str | None).validate_python(deployment.get("data_residency")), + vertex_location=TypeAdapter[str | None](str | None).validate_python(deployment.get("vertex_location")), + ) + if event.call_type in {CallTypes.retrieve_batch, CallTypes.aretrieve_batch} + else None + ) + amount: Final = ( + sum(batch_costs) + if batch_costs is not None + else logger.response_cost_calculator(response, ignore_stamped_cost=True) + ) + if batch_costs is not None: + logger.cost_breakdown = CostBreakdown( + input_cost=batch_costs[0], output_cost=batch_costs[1], total_cost=sum(batch_costs) + ) if amount is None: return Unpriced(usage_event_id=event.event_id, reason="calculation_failed") retained_final: Final = captured.finalized_cost or {} @@ -255,9 +293,9 @@ def price_usage(envelope: str) -> PricingSnapshot | Unpriced: "pricing_source": "upstream_header" if upstream else "per_request" - if request + if any(is_server_derived_pricing_key(name) and value is not None for name, value in request.items()) else "deployment" - if deployment + if any(is_server_derived_pricing_key(name) and value is not None for name, value in deployment.items()) else "cost_map", "priced_model_key": key, "total_cost": Decimal(str(finalized.amount)), diff --git a/litellm/proxy/spend_tracking/usage_capture.py b/litellm/proxy/spend_tracking/usage_capture.py index c1d87d752d8..674862713dc 100644 --- a/litellm/proxy/spend_tracking/usage_capture.py +++ b/litellm/proxy/spend_tracking/usage_capture.py @@ -307,7 +307,8 @@ def capture_context(payload: SpendLogsPayload, kwargs: Mapping[str, object], res } deferred: Final = billed_id.endswith(("_batch_cost", "_background_cost")) and terminal management: Final = ( - (billed_id.endswith(("_batch_cost", "_background_cost")) and not terminal) + payload["call_type"] in {CallTypes.create_batch.value, CallTypes.acreate_batch.value} + or (billed_id.endswith(("_batch_cost", "_background_cost")) and not terminal) or (is_background_response(response) and not terminal) or is_unbilled_non_inference_call(payload["call_type"], request_metadata, response) ) @@ -836,8 +837,8 @@ def capture_logging_usage( existing_pricing: Final = details.get("_accounting_pricing_snapshot") pricing: Final = ( existing_pricing - if isinstance(existing_pricing, PricingCapture) and not internal - else _capture_pricing(logging_obj, normalized, details) + if isinstance(existing_pricing, PricingCapture) and not internal and not _batch_rows(normalized) + else _capture_pricing(logging_obj, normalized_result, details) ) if freeze_only or _batch_rows(normalized): if not internal: diff --git a/tests/integration/database/test_accounting_ledger.py b/tests/integration/database/test_accounting_ledger.py index f783a128bc7..efff0d9b16a 100644 --- a/tests/integration/database/test_accounting_ledger.py +++ b/tests/integration/database/test_accounting_ledger.py @@ -889,8 +889,9 @@ async def test_restarted_compiler_recovers_transient_failures_without_revisiting @pytest.mark.asyncio @pytest.mark.parametrize("sync_ack", (False, True)) +@pytest.mark.parametrize("provider_header", (False, True)) async def test_proxy_capture_and_background_compiler_converge_on_final_header_charge( - writer: _Writer, ledger_database: str, monkeypatch: pytest.MonkeyPatch, sync_ack: bool + writer: _Writer, ledger_database: str, monkeypatch: pytest.MonkeyPatch, sync_ack: bool, provider_header: bool ) -> None: import litellm from litellm.litellm_core_utils.litellm_logging import Logging @@ -929,12 +930,17 @@ async def test_proxy_capture_and_background_compiler_converge_on_final_header_ch logger.update_environment_variables( model="capture-priced-model", custom_llm_provider="openai", litellm_params={}, optional_params={} ) - logger.set_request_pricing( - "capture-priced-model", "openai", {"input_cost_per_token": 0.01, "output_cost_per_token": 0.02} - ) + if not provider_header: + logger.set_request_pricing( + "capture-priced-model", "openai", {"input_cost_per_token": 0.01, "output_cost_per_token": 0.02} + ) response: Final = ModelResponse( id="gateway-provider", model="capture-priced-model", usage=Usage(prompt_tokens=10, completion_tokens=4) ) + if provider_header: + from litellm.litellm_core_utils.core_helpers import set_response_cost_in_hidden_params + + set_response_cost_in_hidden_params(response, 0.18) try: finalized: Final = await finalize_response_cost(logger, response, None, _AT, guardrail_cost=0.03) assert finalized is not None diff --git a/tests/unit/proxy/spend_tracking/test_cost_compiler.py b/tests/unit/proxy/spend_tracking/test_cost_compiler.py index ea769816b59..fe4e8ca7791 100644 --- a/tests/unit/proxy/spend_tracking/test_cost_compiler.py +++ b/tests/unit/proxy/spend_tracking/test_cost_compiler.py @@ -910,3 +910,179 @@ async def test_stream_replay_applies_adjustments_once_to_legacy_stamped_usage( assert float(priced.original_cost or 0) == pytest.approx(0.1 + completion_tokens * 0.02) finally: await meter.stop(1) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("amount", [0.0, 0.123]) +async def test_capture_retains_provider_header_price_without_catalog(amount: float) -> None: + from litellm.proxy.spend_tracking.usage_capture import capture_logging_usage + from litellm.proxy.spend_tracking.usage_meter import DurableUsageMeter, UsageCaptureSettings + from litellm.types.utils import ModelResponse, Usage + from tests.unit.proxy.spend_tracking.test_usage_capture import StoredUsage, _logger + from tests.unit.proxy.spend_tracking.test_usage_meter import _manual_flush + + response: Final = ModelResponse( + model="capture-model", + usage=Usage(prompt_tokens=10, completion_tokens=4), + ) + from litellm.litellm_core_utils.core_helpers import set_response_cost_in_hidden_params + + set_response_cost_in_hidden_params(response, amount) + database: Final = StoredUsage() + meter: Final = DurableUsageMeter(database.insert, UsageCaptureSettings(), wait_for_batch=_manual_flush) + try: + capture_logging_usage(meter, _logger(), response, _AT, _AT) + await meter.close() + result: Final = price_usage(database.rows[-1].envelope) + assert isinstance(result, PricingSnapshot) + assert result.total_cost == Decimal(str(amount)) + assert result.pricing_source == "upstream_header" + finally: + await meter.stop(1) + + +@pytest.mark.parametrize("metadata", [{"input_cost_per_token": None}, {"service_tier": "default"}]) +def test_catalog_provenance_ignores_metadata_without_price_overrides(metadata: dict[str, object]) -> None: + result: Final = price_usage(json.dumps(_retained_envelope(deployment=metadata, request=metadata))) + assert isinstance(result, PricingSnapshot) + assert result.pricing_source == "cost_map" + + +@pytest.mark.parametrize("call_type", ["get_responses", "aget_responses"]) +def test_terminal_response_replay_prices_usage_without_changing_event_identity(call_type: str) -> None: + from pydantic import TypeAdapter + + original: Final = _retained_envelope() + event: Final = TypeAdapter(dict[str, object]).validate_python(original["event"]) + result: Final = price_usage(json.dumps({**original, "event": {**event, "call_type": call_type}})) + assert isinstance(result, PricingSnapshot) + assert float(result.total_cost) == pytest.approx(10 * 0.01 + 4 * 0.02) + + +@pytest.mark.asyncio +async def test_batch_terminal_capture_refreshes_row_prices_after_creation(monkeypatch: pytest.MonkeyPatch) -> None: + import litellm + from litellm.proxy.spend_tracking.usage_capture import capture_batch_callback, capture_logging_usage + from litellm.proxy.spend_tracking.usage_meter import DurableUsageMeter, UsageCaptureSettings + from tests.unit.proxy.spend_tracking.test_usage_capture import StoredUsage, _logger + from tests.unit.proxy.spend_tracking.test_usage_meter import _manual_flush + + rates: Final = { + "mode": "chat", + "litellm_provider": "openai", + "input_cost_per_token": 0.01, + "output_cost_per_token": 0.02, + "input_cost_per_token_batches": 0.003, + "output_cost_per_token_batches": 0.004, + } + monkeypatch.setitem(litellm.model_cost, "capture-model", rates) + monkeypatch.setattr(litellm, "cost_margin_config", {"global": {"fixed_amount": 0.1}}) + monkeypatch.setitem(litellm.model_cost, "second-model", {**rates, "input_cost_per_token_batches": 0.005}) + logger: Final = _logger("aretrieve_batch") + database: Final = StoredUsage() + meter: Final = DurableUsageMeter(database.insert, UsageCaptureSettings(), wait_for_batch=_manual_flush) + usage: Final = {"prompt_tokens": 10, "completion_tokens": 4, "total_tokens": 14} + response: Final = { + "id": "batch", + "status": "completed", + "usage": { + **usage, + "batch_usage_by_model": {"capture-model": usage, "second-model": usage}, + "batch_usage_by_request": [ + {"model": "capture-model", "usage": usage}, + {"model": "capture-model", "usage": usage}, + {"model": "second-model", "usage": usage}, + ], + }, + } + try: + capture_logging_usage(meter, logger, {"id": "batch", "status": "in_progress"}, _AT, _AT, freeze_only=True) + capture_logging_usage(meter, logger, response, _AT, _AT) + await capture_batch_callback(meter, database.load, logger.model_call_details, response, _AT, _AT) + first_rows: Final = database.rows + monkeypatch.setitem(litellm.model_cost, "capture-model", {**rates, "input_cost_per_token_batches": 99.0}) + capture_logging_usage(meter, logger, response, _AT, _AT) + await capture_batch_callback(meter, database.load, logger.model_call_details, response, _AT, _AT) + await meter.close() + assert database.rows == first_rows + results: Final = tuple(price_usage(row.envelope) for row in database.rows) + priced: Final = tuple(result for result in results if isinstance(result, PricingSnapshot)) + assert len(priced) == 3 + assert sorted(float(price.total_cost) for price in priced) == pytest.approx([0.046, 0.046, 0.066]) + finally: + await meter.stop(1) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("call_type", ["get_responses", "aget_responses"]) +async def test_background_capture_compiles_terminal_usage_once(call_type: str) -> None: + from litellm.constants import INTERNAL_CALL_ORIGIN_METADATA_KEY + from litellm.proxy.spend_tracking.usage_capture import capture_logging_usage + from litellm.proxy.spend_tracking.usage_meter import DurableUsageMeter, UsageCaptureSettings + from litellm.types.llms.openai import ResponsesAPIResponse + from litellm.types.utils import BACKGROUND_RESPONSE_COST_POLL_CALL_ORIGIN + from tests.unit.proxy.spend_tracking.test_usage_capture import StoredUsage + from tests.unit.proxy.spend_tracking.test_usage_meter import _manual_flush + + logger: Final = _capture_logger(call_type) + logger.update_environment_variables( + model="capture-model", + custom_llm_provider="openai", + optional_params={}, + litellm_params={"metadata": {INTERNAL_CALL_ORIGIN_METADATA_KEY: BACKGROUND_RESPONSE_COST_POLL_CALL_ORIGIN}}, + ) + response: Final = ResponsesAPIResponse.model_validate( + { + "id": "background-job", + "object": "response", + "created_at": int(_AT.timestamp()), + "status": "completed", + "background": True, + "model": "capture-model", + "output": [], + "usage": {"input_tokens": 10, "output_tokens": 4, "total_tokens": 14}, + } + ) + database: Final = StoredUsage() + meter: Final = DurableUsageMeter(database.insert, UsageCaptureSettings(), wait_for_batch=_manual_flush) + try: + capture_logging_usage(meter, logger, response, _AT, _AT) + await meter.close() + assert len(database.rows) == 1 + assert database.rows[0].event.call_type.value == call_type + assert database.rows[0].event.event_id == "background-job_background_cost" + result: Final = price_usage(database.rows[0].envelope) + assert isinstance(result, PricingSnapshot) + assert float(result.total_cost) == pytest.approx(10 * 0.01 + 4 * 0.02) + finally: + await meter.stop(1) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("call_type", ["call_mcp_tool", "asend_message"]) +async def test_failed_fixed_call_without_usage_does_not_invent_a_free_charge(call_type: str) -> None: + from litellm.proxy.spend_tracking.usage_capture import capture_callback + from litellm.proxy.spend_tracking.usage_meter import DurableUsageMeter, UsageCaptureSettings + from tests.unit.proxy.spend_tracking.test_usage_capture import StoredUsage, _logger + from tests.unit.proxy.spend_tracking.test_usage_meter import _manual_flush + + logger: Final = _logger(call_type) + logger.model_call_details["response_cost"] = 0.0 + database: Final = StoredUsage() + meter: Final = DurableUsageMeter(database.insert, UsageCaptureSettings(), wait_for_batch=_manual_flush) + try: + capture_callback( + meter, + {**logger.model_call_details, "litellm_logging_obj": logger, "log_event_type": "failed_api_call"}, + RuntimeError("Request rejected before dispatch"), + _AT, + _AT, + ) + await meter.close() + outbound: Final = database.rows[-1] + assert outbound.event.outcome is Outcome.FAILED + assert price_usage(outbound.envelope) == Unpriced( + usage_event_id=outbound.event.event_id, reason="usage_unknown" + ) + finally: + await meter.stop(1) diff --git a/tests/unit/proxy/spend_tracking/test_usage_capture.py b/tests/unit/proxy/spend_tracking/test_usage_capture.py index 24bb3480560..1ef6e6d2d4b 100644 --- a/tests/unit/proxy/spend_tracking/test_usage_capture.py +++ b/tests/unit/proxy/spend_tracking/test_usage_capture.py @@ -1754,3 +1754,34 @@ async def test_sync_ack_deadline_bounds_compilation_after_usage_commit(monkeypat finally: compiler.release.set() await runtime.stop(1) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("call_type", ["create_batch", "acreate_batch"]) +async def test_batch_creation_is_not_a_charge_owner(call_type: str) -> None: + from litellm.proxy.spend_tracking.cost_compiler import price_usage + from litellm.proxy.spend_tracking.usage_capture import capture_callback + from litellm.accounting import Unpriced + + database: Final = StoredUsage() + meter: Final = DurableUsageMeter(database.insert, UsageCaptureSettings(), wait_for_batch=_manual_flush) + try: + captured: Final = capture_callback( + meter, + { + "call_type": call_type, + "litellm_call_id": "create", + "model": "capture-model", + "litellm_params": {}, + }, + {"id": "batch", "status": "validating"}, + NOW, + NOW, + ) + await meter.close() + outbound: Final = captured.writes[-1] + assert price_usage(outbound.envelope) == Unpriced( + usage_event_id=outbound.event.event_id, reason="not_charge_owner" + ) + finally: + await meter.stop(1)