scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-27 01:22:18 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 103

fix(cost): apply a deployment's pricing override to realtime sessions (#43114)
Some checks are pending
LiteLLM Rust / rust-lint (push) Waiting to run
Details
LiteLLM Rust / rust-test (push) Waiting to run
Details
LiteLLM Rust / rust-wheel (push) Waiting to run
Details
Terraform Provider / gofmt, vet, build, test (push) Waiting to run
Details
Terraform Provider / Provider endpoints vs proxy OpenAPI schema (push) Waiting to run
Details

Browse source 
* 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 <marty@martysullivan.com>
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 <kerry@berri.ai>
Co-authored-by: Marty Sullivan <marty@martysullivan.com>
Co-authored-by: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
devin-ai-integration[bot] 2026-09-24 21:03:59 -07:00 • committed by GitHub
parent 3fa02ef9fc
commit 0d47347ad7
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 796 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

103
litellm/cost_calculator.py
View file

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

93
tests/integration/pricing/test_configured_prices.py
View file

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

4
tests/integration/pricing/test_realtime_cached_audio_pricing.py
View file

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

622
tests/test_litellm/test_cost_calculator.py
View file

@ -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 `<provider>/<model_id>`, 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