scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-11-13 08:52:02 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 174

fix(accounting): preserve terminal usage pricing and charge ownership

Browse source
This commit is contained in:
Joshua Valluru 2026-10-31 11:19:01 -07:00
parent a2b911d97b
commit da08f7c269
7 changed files with 271 additions and 16 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
litellm/cost_calculator.py
View file

@ -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

5
litellm/litellm_core_utils/litellm_logging.py
View file

@ -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),

48
litellm/proxy/spend_tracking/cost_compiler.py
View file

@ -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)),

7
litellm/proxy/spend_tracking/usage_capture.py
View file

@ -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:

14
tests/integration/database/test_accounting_ledger.py
View file

@ -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

176
tests/unit/proxy/spend_tracking/test_cost_compiler.py
View file

@ -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)

31
tests/unit/proxy/spend_tracking/test_usage_capture.py
View file

@ -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)