From 0d47347ad781b78142bba7f99a3ab1ac995041c0 Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 24 Sep 2026 21:03:59 -0700 Subject: [PATCH] fix(cost): apply a deployment's pricing override to realtime sessions (#43114) * fix(cost): apply a deployment's pricing override to realtime sessions Pass the resolved custom pricing model into the realtime and transcription cost paths so model_info rates and base_model on a realtime deployment are honoured instead of the model the session reported. Adds an integration test that bills a realtime turn at the deployment's configured rates Carries the fix from #36958 Co-authored-by: Marty Sullivan Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(cost): honour audio-only and base_model realtime pricing overrides Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(cost): keep flat per-unit prices from claiming the deployment pricing key Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(cost): keep base_model out of realtime transcription rate overrides Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * test(cost): type the realtime pricing test parameters Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> * fix(cost): try a realtime deployment's base_model ahead of the session model Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --------- Co-authored-by: kerry Co-authored-by: Marty Sullivan Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> --- litellm/cost_calculator.py | 103 ++- .../pricing/test_configured_prices.py | 93 +++ .../test_realtime_cached_audio_pricing.py | 4 +- tests/test_litellm/test_cost_calculator.py | 622 ++++++++++++++++++ 4 files changed, 796 insertions(+), 26 deletions(-) diff --git a/litellm/cost_calculator.py b/litellm/cost_calculator.py index 6cc0d9444cd..a279b9f0903 100644 --- a/litellm/cost_calculator.py +++ b/litellm/cost_calculator.py @@ -790,6 +790,16 @@ def _get_hidden_str_for_cost_calc(hidden_params: object, key: str) -> str | None return value if isinstance(value, str) and value else None +_NON_TOKEN_RATE_FIELDS: Final = frozenset({"input_cost_per_second", "input_cost_per_query", "tiered_pricing"}) + + +def _cost_map_entry_prices_anything(entry: Mapping[str, object]) -> bool: + return any( + value is not None and (field in _NON_TOKEN_RATE_FIELDS or ("cost_per" in field and "token" in field)) + for field, value in entry.items() + ) + + def _select_model_name_for_cost_calc( model: str | None, completion_response: object | None, @@ -828,12 +838,7 @@ def _select_model_name_for_cost_calc( if custom_pricing is True: if router_model_id is not None and router_model_id in litellm.model_cost: entry: Final = litellm.model_cost[router_model_id] - if ( - entry.get("input_cost_per_token") is not None - or entry.get("input_cost_per_second") is not None - or entry.get("input_cost_per_query") is not None - or entry.get("tiered_pricing") is not None - ): + if _cost_map_entry_prices_anything(entry): return_model = router_model_id else: return_model = model @@ -1699,6 +1704,8 @@ def completion_cost( litellm_model_name=model, data_residency=data_residency, litellm_logging_obj=litellm_logging_obj, + custom_pricing_model=selected_model if custom_pricing else None, + base_pricing_model=(selected_model if base_model is not None and not custom_pricing else None), ) elif call_type == _MCP_CALL_TYPE: from litellm.proxy._experimental.mcp_server.cost_calculator import ( @@ -2870,14 +2877,20 @@ def _candidate_realtime_token_costs( def _cost_map_entry_declares_pricing(model_name: str, custom_llm_provider: str) -> bool: + """Whether the entry behind ``model_name`` sets any rate of its own, even a zero one. + + The name is resolved the way ``get_model_info`` resolves it before the raw entry is read, + because a deployment-scoped name arrives here already carrying its provider prefix. Two raw + lookups cannot strip that prefix, so a zero-rated override read as declaring nothing, and a + session that should bill nothing fell through to the public rates instead. + """ + resolved: Final = _get_model_info_or_none(model_name, custom_llm_provider) entries: Final = ( + litellm.model_cost.get(resolved.get("key")) if resolved is not None else None, litellm.model_cost.get(model_name), litellm.model_cost.get(f"{custom_llm_provider}/{model_name}"), ) - return any( - entry is not None and any("cost_per" in field and value is not None for field, value in entry.items()) - for entry in entries - ) + return any(entry is not None and _cost_map_entry_prices_anything(entry) for entry in entries) def _first_priced_realtime_token_costs( @@ -2917,6 +2930,8 @@ def handle_realtime_stream_cost_calculation( litellm_model_name: str, data_residency: str | None = None, litellm_logging_obj: LitellmLoggingObject | None = None, + custom_pricing_model: str | None = None, + base_pricing_model: str | None = None, ) -> float: """ Handles the cost calculation for realtime stream responses. @@ -2925,9 +2940,13 @@ def handle_realtime_stream_cost_calculation( Args: results: A list of OpenAIRealtimeStreamBaseObject objects + custom_pricing_model: deployment-scoped pricing key from the deployment's + custom rates, tried ahead of the session-reported model + base_pricing_model: the deployment's resolved base_model, tried ahead of the + session-reported model but after custom rates """ received_model = None - potential_model_names: Final = [] + potential_model_names: Final = [custom_pricing_model, base_pricing_model] for result in results: if result["type"] == "session.created": received_model = cast(OpenAIRealtimeStreamSessionEvents, result)["session"].get("model", None) @@ -2945,6 +2964,7 @@ def handle_realtime_stream_cost_calculation( results=results, custom_llm_provider=custom_llm_provider, litellm_model_name=litellm_model_name, + custom_pricing_model=custom_pricing_model, ) if any(r.get("type") == _TRANSCRIPTION_COMPLETED_EVENT_TYPE for r in results) else 0.0 @@ -2968,6 +2988,7 @@ def handle_realtime_transcription_cost_calculation( results: OpenAIRealtimeStreamList, custom_llm_provider: str, litellm_model_name: str, + custom_pricing_model: str | None = None, ) -> float: """ Cost for realtime transcription sessions (e.g. gpt-realtime-whisper). @@ -2985,15 +3006,15 @@ def handle_realtime_transcription_cost_calculation( return 0.0 model_name: Final = _get_transcription_model_name_from_results(results) or litellm_model_name - try: - model_info = litellm.get_model_info(model=model_name, custom_llm_provider=custom_llm_provider) - except Exception: - model_info = None + model_info: Final = _get_model_info_or_none(model_name, custom_llm_provider) + override_info: Final = ( + _get_model_info_or_none(custom_pricing_model, custom_llm_provider) if custom_pricing_model is not None else None + ) total_cost = 0.0 for event in completed_events: usage = event.get("usage") or {} - total_cost += _transcription_usage_cost(usage, model_info) + total_cost += _transcription_usage_cost(usage, model_info, override_info) return total_cost @@ -3018,23 +3039,57 @@ def _get_transcription_model_name_from_results( return None -def _transcription_usage_cost(usage: dict, model_info: ModelInfo | None) -> float: - if model_info is None: +def _get_model_info_or_none(model: str, custom_llm_provider: str) -> ModelInfo | None: + try: + return litellm.get_model_info(model=model, custom_llm_provider=custom_llm_provider) + except Exception: + return None + + +def _declared_transcription_rate(info: ModelInfo | None, keys: tuple[str, ...]) -> float | None: + """First of ``keys`` this entry prices, read off the raw ``litellm.model_cost`` entry + because ``get_model_info`` synthesizes zero token rates for entries that omit them.""" + if info is None: + return None + declared: Final = litellm.model_cost.get(info.get("key")) + if declared is None: + return None + return next( + (float(value) for key in keys if declared.get(key) is not None and (value := info.get(key)) is not None), + None, + ) + + +def _transcription_rate(keys: tuple[str, ...], override: ModelInfo | None, base: ModelInfo | None) -> float: + rates: Final = (_declared_transcription_rate(info, keys) for info in (override, base)) + return next((rate for rate in rates if rate is not None), 0.0) + + +def _transcription_usage_cost( + usage: dict, + model_info: ModelInfo | None, + override_info: ModelInfo | None = None, +) -> float: + if model_info is None and override_info is None: return 0.0 + usage_type: Final = usage.get("type") if usage_type == "duration": seconds: Final = usage.get("seconds") or 0.0 - per_second: Final = model_info.get("input_cost_per_second") or 0.0 - return float(seconds) * float(per_second) + return float(seconds) * _transcription_rate(("input_cost_per_second",), override_info, model_info) if usage_type == "tokens": input_token_details: Final = usage.get("input_token_details") or {} audio_tokens: Final = input_token_details.get("audio_tokens") or 0 text_tokens: Final = input_token_details.get("text_tokens") or 0 output_tokens: Final = usage.get("output_tokens") or 0 - audio_cost: Final = float(audio_tokens) * float( - model_info.get("input_cost_per_audio_token") or model_info.get("input_cost_per_token") or 0.0 + audio_cost: Final = float(audio_tokens) * _transcription_rate( + ("input_cost_per_audio_token", "input_cost_per_token"), override_info, model_info + ) + text_cost: Final = float(text_tokens) * _transcription_rate( + ("input_cost_per_token",), override_info, model_info + ) + output_cost: Final = float(output_tokens) * _transcription_rate( + ("output_cost_per_token",), override_info, model_info ) - text_cost: Final = float(text_tokens) * float(model_info.get("input_cost_per_token") or 0.0) - output_cost: Final = float(output_tokens) * float(model_info.get("output_cost_per_token") or 0.0) return audio_cost + text_cost + output_cost return 0.0 diff --git a/tests/integration/pricing/test_configured_prices.py b/tests/integration/pricing/test_configured_prices.py index 0e4efea3a15..93290a404fc 100644 --- a/tests/integration/pricing/test_configured_prices.py +++ b/tests/integration/pricing/test_configured_prices.py @@ -1,6 +1,9 @@ +import asyncio import json +import os import uuid from collections.abc import Iterator, Mapping +from hashlib import sha256 from pathlib import Path from typing import Final @@ -12,6 +15,96 @@ from litellm import get_model_info from tests.integration._support.client import Gateway, eventually, object_value, string_value from tests.integration._support.database import read_rows from tests.integration._support.process import owned_proxy +from tests.integration._support.upstream import delete_scenario, register_scenario +from tests.integration.cost_calculation.cost_tracking_case import RealtimeResponse +from tests.integration.pricing.test_realtime_cached_audio_pricing import one_realtime_turn + +REALTIME_MODEL: Final = "gpt-realtime-2" +REALTIME_INPUT_TEXT_TOKENS: Final = 10 +REALTIME_INPUT_AUDIO_TOKENS: Final = 20 +REALTIME_OUTPUT_TEXT_TOKENS: Final = 5 +REALTIME_OUTPUT_AUDIO_TOKENS: Final = 7 + + +def _realtime_response_done() -> RealtimeResponse: + return RealtimeResponse( + content_type="application/x-realtime", + events=( + { + "type": "response.done", + "event_id": "evt_$REQUEST_ID", + "response": { + "id": "resp_$REQUEST_ID", + "object": "realtime.response", + "status": "completed", + "output": [], + "usage": { + "total_tokens": REALTIME_INPUT_TEXT_TOKENS + + REALTIME_INPUT_AUDIO_TOKENS + + REALTIME_OUTPUT_TEXT_TOKENS + + REALTIME_OUTPUT_AUDIO_TOKENS, + "input_tokens": REALTIME_INPUT_TEXT_TOKENS + REALTIME_INPUT_AUDIO_TOKENS, + "output_tokens": REALTIME_OUTPUT_TEXT_TOKENS + REALTIME_OUTPUT_AUDIO_TOKENS, + "input_token_details": { + "text_tokens": REALTIME_INPUT_TEXT_TOKENS, + "audio_tokens": REALTIME_INPUT_AUDIO_TOKENS, + "cached_tokens": 0, + }, + "output_token_details": { + "text_tokens": REALTIME_OUTPUT_TEXT_TOKENS, + "audio_tokens": REALTIME_OUTPUT_AUDIO_TOKENS, + }, + }, + }, + }, + ), + ) + + +@pytest.mark.parametrize( + ("input_text_rate", "input_audio_rate", "output_text_rate", "output_audio_rate"), + ((0.001, 0.002, 0.003, 0.004), (0.0, 0.0, 0.0, 0.0)), + ids=("custom_rates", "zero_rated"), +) +def test_realtime_session_is_charged_at_the_deployment_configured_rates( + gateway: Gateway, + input_text_rate: float, + input_audio_rate: float, + output_text_rate: float, + output_audio_rate: float, +) -> None: + with gateway.scenario() as scenario: + scenario_id: Final = f"realtime-configured-price-{uuid.uuid4().hex[:12]}" + handle: Final = register_scenario(scenario_id, _realtime_response_done()) + scenario.cleanups.callback(delete_scenario, handle) + key: Final = scenario.key() + model: Final = scenario.model( + model=f"openai/{REALTIME_MODEL}", + api_key=scenario_id, + api_base=gateway.upstream_url.rstrip("/"), + input_cost_per_token=input_text_rate, + input_cost_per_audio_token=input_audio_rate, + output_cost_per_token=output_text_rate, + output_cost_per_audio_token=output_audio_rate, + ) + session: Final = asyncio.run(one_realtime_turn(os.environ["INTEGRATION_PROXY_URL"].rstrip("/"), key, model)) + assert session.get("type") == "session.created", session + rows: Final = eventually( + lambda: read_rows( + 'SELECT spend, call_type FROM "LiteLLM_SpendLogs" WHERE api_key = %s', + (sha256(key.encode()).hexdigest(),), + ), + lambda values: len(values) == 1, + seconds=70, + ) + assert rows[0]["call_type"] == "_arealtime", rows + assert float(str(rows[0]["spend"])) == pytest.approx( + REALTIME_INPUT_TEXT_TOKENS * input_text_rate + + REALTIME_INPUT_AUDIO_TOKENS * input_audio_rate + + REALTIME_OUTPUT_TEXT_TOKENS * output_text_rate + + REALTIME_OUTPUT_AUDIO_TOKENS * output_audio_rate, + abs=1e-9, + ), rows @pytest.mark.covers("quota_management.spend_tracking.custom_price.matches_input_rates") diff --git a/tests/integration/pricing/test_realtime_cached_audio_pricing.py b/tests/integration/pricing/test_realtime_cached_audio_pricing.py index 4a7598d0cbf..42e90cf2c3e 100644 --- a/tests/integration/pricing/test_realtime_cached_audio_pricing.py +++ b/tests/integration/pricing/test_realtime_cached_audio_pricing.py @@ -92,7 +92,7 @@ def cached_audio_response_done() -> RealtimeResponse: ) -async def _one_realtime_turn(proxy_url: str, key: str, model: str) -> dict[str, JsonValue]: +async def one_realtime_turn(proxy_url: str, key: str, model: str) -> dict[str, JsonValue]: async with websockets.connect( f"{proxy_url.replace('http://', 'ws://').replace('https://', 'wss://')}/v1/realtime?model={model}", additional_headers={"Authorization": f"Bearer {key}"}, @@ -115,7 +115,7 @@ def test_realtime_cached_audio_tokens_bill_at_audio_cache_read_rate_not_full_aud model: Final = scenario.model( model=f"openai/{MODEL}", api_key=scenario_id, api_base=gateway.upstream_url.rstrip("/") ) - session: Final = asyncio.run(_one_realtime_turn(os.environ["INTEGRATION_PROXY_URL"].rstrip("/"), key, model)) + session: Final = asyncio.run(one_realtime_turn(os.environ["INTEGRATION_PROXY_URL"].rstrip("/"), key, model)) assert session.get("type") == "session.created", session rows: Final = eventually( lambda: read_rows( diff --git a/tests/test_litellm/test_cost_calculator.py b/tests/test_litellm/test_cost_calculator.py index f7d6cfaf079..99dea6366f9 100644 --- a/tests/test_litellm/test_cost_calculator.py +++ b/tests/test_litellm/test_cost_calculator.py @@ -309,6 +309,243 @@ def test_realtime_logging_object_does_not_validate_unknown_event_types(): assert len(dumped["results"]) == len(results) +def test_realtime_transcription_honors_deployment_pricing_override(monkeypatch: pytest.MonkeyPatch) -> None: + """A deployment's pricing override must reach transcription events too. + + Transcription is billed separately from response usage inside the same realtime + session, so a deployment registered at zero rates has to zero both. Resolving + transcription against the public ASR model instead billed a zero-rated + deployment for every .completed event. + """ + monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") + monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url="")) + + deployment_id = "deployment-hash-zero-rated-asr" + litellm.register_model( + model_cost={ + deployment_id: { + "litellm_provider": "openai", + "mode": "realtime", + "input_cost_per_second": 0.0, + "input_cost_per_token": 0.0, + "output_cost_per_token": 0.0, + "input_cost_per_audio_token": 0.0, + } + } + ) + + results: OpenAIRealtimeStreamList = [ + { + "type": "session.created", + "session": { + "type": "transcription", + "audio": {"input": {"transcription": {"model": "gpt-realtime-whisper"}}}, + }, + }, + { + "type": "conversation.item.input_audio_transcription.completed", + "usage": {"type": "duration", "seconds": 120.0}, + }, + ] + + public_rate_cost = 120.0 * litellm.model_cost["gpt-realtime-whisper"]["input_cost_per_second"] + assert public_rate_cost > 0, "the public ASR rate must be non-zero for this test to mean anything" + + without_override = handle_realtime_stream_cost_calculation( + results=results, + combined_usage_object=Usage(), + custom_llm_provider="openai", + litellm_model_name="gpt-realtime-whisper", + ) + assert abs(without_override - public_rate_cost) < 1e-9 + + with_override = handle_realtime_stream_cost_calculation( + results=results, + combined_usage_object=Usage(), + custom_llm_provider="openai", + litellm_model_name="gpt-realtime-whisper", + custom_pricing_model=deployment_id, + ) + assert with_override == 0.0, "the zero-rated deployment must not be billed for transcription" + + +def test_realtime_transcription_partial_override_keeps_unset_rates(monkeypatch: pytest.MonkeyPatch) -> None: + """An override must not blank the rates it does not set. + + A deployment that prices tokens but omits input_cost_per_second would otherwise + bill duration-based transcription at nothing, because the cost helpers read + `.get(key) or 0.0`. Only the fields the operator actually set may win. + """ + monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") + monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url="")) + + deployment_id = "deployment-hash-tokens-only" + litellm.register_model( + model_cost={ + deployment_id: { + "litellm_provider": "openai", + "mode": "realtime", + "input_cost_per_token": 0.0, + "output_cost_per_token": 0.0, + } + } + ) + + results: OpenAIRealtimeStreamList = [ + { + "type": "session.created", + "session": { + "type": "transcription", + "audio": {"input": {"transcription": {"model": "gpt-realtime-whisper"}}}, + }, + }, + { + "type": "conversation.item.input_audio_transcription.completed", + "usage": {"type": "duration", "seconds": 120.0}, + }, + ] + + cost = handle_realtime_stream_cost_calculation( + results=results, + combined_usage_object=Usage(), + custom_llm_provider="openai", + litellm_model_name="gpt-realtime-whisper", + custom_pricing_model=deployment_id, + ) + + expected = 120.0 * litellm.model_cost["gpt-realtime-whisper"]["input_cost_per_second"] + assert expected > 0, "the public ASR per-second rate must be non-zero for this test to mean anything" + assert cost == pytest.approx(expected, rel=1e-9), ( + "duration must keep the ASR per-second rate the override left unset" + ) + + +@pytest.mark.parametrize( + "label,override,expected_audio_rate,expected_per_second", + [ + ("tokens only", {"input_cost_per_token": 0.0}, 0.0, 0.017 / 60), + ("audio zeroed", {"input_cost_per_audio_token": 0.0}, 0.0, 0.017 / 60), + ("per second only", {"input_cost_per_second": 0.001}, 6e-06, 0.001), + ("empty override", {}, 6e-06, 0.017 / 60), + ("no override", None, 6e-06, 0.017 / 60), + ], +) +def test_transcription_rate_precedence( + monkeypatch: pytest.MonkeyPatch, + label: str, + override: dict[str, float] | None, + expected_audio_rate: float, + expected_per_second: float, +) -> None: + """Rates resolve within one entry before moving to the next, and zero is a real value. + + An override that prices only tokens must apply its own token rate to audio rather + than reaching past itself for the public audio rate, a deliberate zero must win + instead of being treated as unset, and a rate the override never mentions must keep + the base entry's value. + """ + from litellm.cost_calculator import handle_realtime_transcription_cost_calculation + + monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") + monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url="")) + + base_model = "asr-precedence-base" + deployment_id = "asr-precedence-deployment" + litellm.register_model( + model_cost={ + base_model: { + "litellm_provider": "openai", + "mode": "audio_transcription", + "input_cost_per_audio_token": 6e-06, + "input_cost_per_token": 2.5e-06, + "input_cost_per_second": 0.017 / 60, + } + } + ) + if override is not None: + litellm.register_model( + model_cost={deployment_id: {"litellm_provider": "openai", "mode": "audio_transcription", **override}} + ) + + def cost_for(usage: dict[str, object]) -> float: + return handle_realtime_transcription_cost_calculation( + results=[ + {"type": "transcription_session.created", "session": {"model": base_model}}, + {"type": "conversation.item.input_audio_transcription.completed", "usage": usage}, + ], + custom_llm_provider="openai", + litellm_model_name=base_model, + custom_pricing_model=deployment_id if override is not None else None, + ) + + audio_cost = cost_for({"type": "tokens", "input_token_details": {"audio_tokens": 100}}) + assert audio_cost == pytest.approx(100 * expected_audio_rate, rel=1e-9), f"{label}: audio rate" + + per_second_cost = cost_for({"type": "duration", "seconds": 120.0}) + assert per_second_cost == pytest.approx(120.0 * expected_per_second, rel=1e-9), ( + f"{label}: an override must never blank a rate it does not set" + ) + + +def test_realtime_transcription_per_second_override_keeps_public_token_rates( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A per-second override must not zero the token rates ``get_model_info`` synthesizes. + + ``get_model_info`` defaults input_cost_per_token and output_cost_per_token to 0 for entries + that omit them, so a deployment priced only per second looked like it had declared token + rates of 0. Token-shaped transcription then billed nothing instead of falling through to the + public ASR rates, while the per-second rate the operator did set stayed in force. + """ + from litellm.cost_calculator import handle_realtime_transcription_cost_calculation + + monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") + monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url="")) + + asr_model = "gpt-4o-transcribe" + per_second_rate = 0.001 + deployment_id = "deployment-hash-per-second-only" + litellm.register_model( + model_cost={ + deployment_id: { + "litellm_provider": "openai", + "mode": "audio_transcription", + "input_cost_per_second": per_second_rate, + } + } + ) + + public = litellm.model_cost[asr_model] + session_event = {"type": "transcription_session.created", "session": {"model": asr_model}} + + def cost_for(usage: dict[str, object]) -> float: + return handle_realtime_transcription_cost_calculation( + results=[session_event, {"type": "conversation.item.input_audio_transcription.completed", "usage": usage}], + custom_llm_provider="openai", + litellm_model_name=asr_model, + custom_pricing_model=deployment_id, + ) + + token_cost = cost_for( + { + "type": "tokens", + "input_token_details": {"audio_tokens": 400, "text_tokens": 12}, + "output_tokens": 30, + } + ) + expected_token_cost = ( + 400 * public["input_cost_per_audio_token"] + + 12 * public["input_cost_per_token"] + + 30 * public["output_cost_per_token"] + ) + assert expected_token_cost > 0, "the public ASR token rates must be non-zero for this test to mean anything" + assert token_cost == pytest.approx(expected_token_cost, rel=1e-9), ( + "an override that prices only seconds must leave the public token rates in place" + ) + + assert cost_for({"type": "duration", "seconds": 120.0}) == pytest.approx(120.0 * per_second_rate, rel=1e-9) + + def test_realtime_transcription_no_completed_events_is_zero(monkeypatch): """A realtime stream without transcription completed events adds no extra cost.""" monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") @@ -4635,6 +4872,391 @@ def test_gemini_live_native_audio_limits_and_capabilities_match_vendor_model_car assert info["supports_pdf_input"] is False +def test_realtime_honours_deployment_custom_pricing(monkeypatch: pytest.MonkeyPatch) -> None: + """Regression: a deployment's pricing override never reached realtime costing. + + `model_info` overrides are registered under the deployment's own model_id, and + only `_select_model_name_for_cost_calc` knows to look there. The realtime branch + discarded that result and priced by the model the session reported, so a config + that zeroes a realtime deployment was billed at the public rate anyway. Audio is + the bulk of a voice call, so the gap was most of the cost. + """ + from litellm.types.utils import CompletionTokensDetailsWrapper + + model = "gemini-3.1-flash-live-preview" + deployment_key = "deployment-id-for-a-zero-rated-realtime-group" + paid = litellm.model_cost[model] + monkeypatch.setitem( + litellm.model_cost, + deployment_key, + { + **paid, + "input_cost_per_token": 0.0, + "output_cost_per_token": 0.0, + "input_cost_per_audio_token": 0.0, + "output_cost_per_audio_token": 0.0, + "cache_read_input_token_cost": 0.0, + }, + ) + + results: OpenAIRealtimeStreamList = [ + {"type": "session.created", "session": {"model": model}}, + { + "type": "response.done", + "response": {"usage": {"input_tokens": 10, "output_tokens": 200, "total_tokens": 210}}, + }, + ] + usage = Usage( + prompt_tokens=10, + completion_tokens=200, + total_tokens=210, + prompt_tokens_details=PromptTokensDetailsWrapper(text_tokens=10, cached_tokens=0), + completion_tokens_details=CompletionTokensDetailsWrapper(text_tokens=20, audio_tokens=180), + ) + + paid_cost = handle_realtime_stream_cost_calculation( + results=results, + combined_usage_object=usage, + custom_llm_provider="gemini", + litellm_model_name=model, + ) + expected_paid = ( + 10 * paid["input_cost_per_token"] + + 20 * paid["output_cost_per_token"] + + 180 * paid["output_cost_per_audio_token"] + ) + assert paid_cost == pytest.approx(expected_paid, rel=1e-9) + assert paid_cost > 0 + + zero_rated_cost = handle_realtime_stream_cost_calculation( + results=results, + combined_usage_object=usage, + custom_llm_provider="gemini", + litellm_model_name=model, + custom_pricing_model=deployment_key, + ) + assert zero_rated_cost == 0.0 + + +def test_realtime_honours_a_provider_prefixed_zero_rated_deployment(monkeypatch: pytest.MonkeyPatch) -> None: + """Regression: the override arrived provider-prefixed and was read as pricing nothing. + + `_select_model_name_for_cost_calc` hands back `/`, so the name reaching + the pricing guard carries a prefix the raw cost-map lookups cannot strip. The rates resolved + correctly through `get_model_info`, then the guard rejected them as undeclared and the session + billed the public rates. A zero-rated deployment must stay at zero however its name arrives. + """ + from litellm.types.utils import CompletionTokensDetailsWrapper + + model = "gemini-live-2.5-flash-native-audio" + deployment_key = "deployment-id-for-a-prefixed-zero-rated-realtime-group" + paid = litellm.model_cost[model] + monkeypatch.setitem( + litellm.model_cost, + deployment_key, + { + **paid, + "input_cost_per_token": 0.0, + "output_cost_per_token": 0.0, + "input_cost_per_audio_token": 0.0, + "output_cost_per_audio_token": 0.0, + }, + ) + + results: OpenAIRealtimeStreamList = [ + {"type": "session.created", "session": {"model": model}}, + { + "type": "response.done", + "response": {"usage": {"input_tokens": 219, "output_tokens": 81, "total_tokens": 300}}, + }, + ] + usage = Usage( + prompt_tokens=219, + completion_tokens=81, + total_tokens=300, + prompt_tokens_details=PromptTokensDetailsWrapper(text_tokens=16, audio_tokens=203), + completion_tokens_details=CompletionTokensDetailsWrapper(text_tokens=23, audio_tokens=58), + ) + + paid_cost = handle_realtime_stream_cost_calculation( + results=results, + combined_usage_object=usage, + custom_llm_provider="vertex_ai", + litellm_model_name=model, + ) + assert paid_cost > 0 + + zero_rated_cost = handle_realtime_stream_cost_calculation( + results=results, + combined_usage_object=usage, + custom_llm_provider="vertex_ai", + litellm_model_name=model, + custom_pricing_model=f"vertex_ai/{deployment_key}", + ) + assert zero_rated_cost == 0.0 + + +def test_unpriced_deployment_entry_still_falls_through_to_the_session_model( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The guard's own purpose must survive: an entry that prices nothing is not an override. + + Deployments are auto-registered under their model_id with no rates at all, and those must + keep billing at the session model's public rates rather than silently costing nothing. + """ + from litellm.types.utils import CompletionTokensDetailsWrapper + + model = "gemini-live-2.5-flash-native-audio" + deployment_key = "deployment-id-with-no-declared-rates" + monkeypatch.setitem( + litellm.model_cost, + deployment_key, + {key: value for key, value in litellm.model_cost[model].items() if "cost_per" not in key}, + ) + + results: OpenAIRealtimeStreamList = [ + {"type": "session.created", "session": {"model": model}}, + { + "type": "response.done", + "response": {"usage": {"input_tokens": 219, "output_tokens": 81, "total_tokens": 300}}, + }, + ] + usage = Usage( + prompt_tokens=219, + completion_tokens=81, + total_tokens=300, + prompt_tokens_details=PromptTokensDetailsWrapper(text_tokens=16, audio_tokens=203), + completion_tokens_details=CompletionTokensDetailsWrapper(text_tokens=23, audio_tokens=58), + ) + + with_unpriced_override = handle_realtime_stream_cost_calculation( + results=results, + combined_usage_object=usage, + custom_llm_provider="vertex_ai", + litellm_model_name=model, + custom_pricing_model=f"vertex_ai/{deployment_key}", + ) + without_override = handle_realtime_stream_cost_calculation( + results=results, + combined_usage_object=usage, + custom_llm_provider="vertex_ai", + litellm_model_name=model, + ) + assert with_unpriced_override == pytest.approx(without_override, rel=1e-9) + assert with_unpriced_override > 0 + + +def test_realtime_audio_only_override_bills_audio_at_the_deployment_rate( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Regression: an audio-only pricing override was never selected as the pricing key. + + The deployment-selection guard recognised only text, per-second, per-query and + tiered rates, so a deployment that priced just the audio meters was passed over + and the session kept billing the public rates for the exact tokens it priced. + """ + from litellm.types.utils import CompletionTokensDetailsWrapper + + model = "gemini-live-2.5-flash-native-audio" + deployment_key = "deployment-id-for-an-audio-only-realtime-group" + monkeypatch.setitem( + litellm.model_cost, + deployment_key, + { + "litellm_provider": "vertex_ai", + "mode": "realtime", + "input_cost_per_audio_token": 0.0, + "output_cost_per_audio_token": 0.0, + }, + ) + + logging_object = LiteLLMRealtimeStreamLoggingObject( + usage=Usage( + prompt_tokens=203, + completion_tokens=58, + total_tokens=261, + prompt_tokens_details=PromptTokensDetailsWrapper(audio_tokens=203), + completion_tokens_details=CompletionTokensDetailsWrapper(audio_tokens=58), + ), + results=[ + {"type": "session.created", "session": {"model": model}}, + { + "type": "response.done", + "response": {"usage": {"input_tokens": 203, "output_tokens": 58, "total_tokens": 261}}, + }, + ], + ) + + public_cost = completion_cost( + completion_response=logging_object, + model=model, + call_type=CallTypes.arealtime.value, + custom_llm_provider="vertex_ai", + ) + assert public_cost > 0 + + overridden_cost = completion_cost( + completion_response=logging_object, + model=model, + call_type=CallTypes.arealtime.value, + custom_llm_provider="vertex_ai", + custom_pricing=True, + router_model_id=deployment_key, + ) + assert overridden_cost == pytest.approx(0.0) + + +def test_realtime_session_falls_back_to_base_model_pricing(monkeypatch: pytest.MonkeyPatch) -> None: + """Regression: a priced base_model was discarded for realtime sessions. + + The resolved base model only reached the realtime cost path when custom pricing + was on, so a session reporting an alias unmapped in the cost map recorded zero + instead of the base model's published price. + """ + from litellm.types.utils import CompletionTokensDetailsWrapper + + base_model = "gemini-live-2.5-flash-native-audio" + logging_object = LiteLLMRealtimeStreamLoggingObject( + usage=Usage( + prompt_tokens=219, + completion_tokens=81, + total_tokens=300, + prompt_tokens_details=PromptTokensDetailsWrapper(text_tokens=16, audio_tokens=203), + completion_tokens_details=CompletionTokensDetailsWrapper(text_tokens=23, audio_tokens=58), + ), + results=[ + { + "type": "session.created", + "session": {"model": "my-voice-alias"}, + }, + { + "type": "response.done", + "response": {"usage": {"input_tokens": 219, "output_tokens": 81, "total_tokens": 300}}, + }, + ], + ) + + aliased_cost = completion_cost( + completion_response=logging_object, + model="my-voice-alias", + call_type=CallTypes.arealtime.value, + custom_llm_provider="vertex_ai", + base_model=base_model, + ) + base_cost = completion_cost( + completion_response=logging_object, + model=base_model, + call_type=CallTypes.arealtime.value, + custom_llm_provider="vertex_ai", + ) + assert aliased_cost == pytest.approx(base_cost, rel=1e-9) + assert aliased_cost > 0 + + +def test_base_model_does_not_override_transcription_rates(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") + monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url="")) + + base_model = "gpt-realtime-2" + asr_model = "gpt-4o-transcribe" + logging_object = LiteLLMRealtimeStreamLoggingObject( + usage=Usage(), + results=[ + { + "type": "session.created", + "session": { + "model": "my-voice-alias", + "audio": {"input": {"transcription": {"model": asr_model}}}, + }, + }, + { + "type": "conversation.item.input_audio_transcription.completed", + "usage": { + "type": "tokens", + "input_token_details": {"audio_tokens": 400, "text_tokens": 12}, + "output_tokens": 30, + }, + }, + ], + ) + + with_base_model = completion_cost( + completion_response=logging_object, + model="my-voice-alias", + call_type=CallTypes.arealtime.value, + custom_llm_provider="openai", + base_model=base_model, + ) + asr_priced = completion_cost( + completion_response=logging_object, + model="my-voice-alias", + call_type=CallTypes.arealtime.value, + custom_llm_provider="openai", + ) + realtime_card = litellm.model_cost[base_model] + billed_at_realtime = ( + 400 * realtime_card["input_cost_per_audio_token"] + + 12 * realtime_card["input_cost_per_token"] + + 30 * realtime_card["output_cost_per_audio_token"] + ) + assert billed_at_realtime != pytest.approx(asr_priced, rel=1e-9) + assert with_base_model == pytest.approx(asr_priced, rel=1e-9) + assert with_base_model > 0 + + +def test_realtime_base_model_outranks_the_session_reported_model(monkeypatch: pytest.MonkeyPatch) -> None: + monkeypatch.setenv("LITELLM_LOCAL_MODEL_COST_MAP", "True") + monkeypatch.setattr(litellm, "model_cost", litellm.get_model_cost_map(url="")) + + from litellm.types.utils import CompletionTokensDetailsWrapper + + session_model = "gpt-realtime-mini" + base_model = "gpt-realtime-2" + + def logging_object_for(session: str) -> LiteLLMRealtimeStreamLoggingObject: + return LiteLLMRealtimeStreamLoggingObject( + usage=Usage( + prompt_tokens=120, + completion_tokens=60, + total_tokens=180, + prompt_tokens_details=PromptTokensDetailsWrapper(text_tokens=20, audio_tokens=100), + completion_tokens_details=CompletionTokensDetailsWrapper(text_tokens=10, audio_tokens=50), + ), + results=[ + { + "type": "session.created", + "session": {"model": session}, + }, + { + "type": "response.done", + "response": {"usage": {"input_tokens": 120, "output_tokens": 60, "total_tokens": 180}}, + }, + ], + ) + + with_base_model = completion_cost( + completion_response=logging_object_for(session_model), + model=session_model, + call_type=CallTypes.arealtime.value, + custom_llm_provider="openai", + base_model=base_model, + ) + base_priced = completion_cost( + completion_response=logging_object_for(base_model), + model=base_model, + call_type=CallTypes.arealtime.value, + custom_llm_provider="openai", + ) + session_priced = completion_cost( + completion_response=logging_object_for(session_model), + model=session_model, + call_type=CallTypes.arealtime.value, + custom_llm_provider="openai", + ) + assert base_priced != pytest.approx(session_priced, rel=1e-9) + assert with_base_model == pytest.approx(base_priced, rel=1e-9) + + def test_baseten_glm_5_3_fast_is_priced_from_registry(_local_model_cost_map: None) -> None: model: Final = "baseten/zai-org/GLM-5.3-Fast" prompt_tokens: Final = 1000