scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-15 23:31:29 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 165

fix(dashscope): bill tiered pricing at the request-size tier instead of graduated slices

Browse source
This commit is contained in:
Devin AI 2026-07-27 15:44:43 +00:00
parent 24123269cc
commit d40df843bc
3 changed files with 131 additions and 190 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

76
litellm/litellm_core_utils/llm_cost_calc/tiered_pricing.py
View file

@ -1,5 +1,5 @@
"""
Provider-neutral graduated tiered pricing calculation.
Provider-neutral request-size (step) tiered pricing selection.
Shared by provider cost calculators (e.g. Dashscope) and the proxy budget
reservation logic so neither has to depend on the other.
@ -25,80 +25,6 @@ def _coerce_cost_per_token(value: Union[float, int, str, None]) -> float:
return float(value)
def calculate_tiered_cost(
tokens: int,
tiered_pricing: List[dict],
cost_key: str,
fallback_cost_key: Optional[str] = None,
) -> float:
"""
Calculate cost for a given number of tokens based on a true tiered pricing structure.
This function iterates through sorted pricing tiers, calculates the cost for the
number of tokens that fall into each tier's range, and sums them up to get the total cost.
Args:
tokens (int): The total number of tokens to calculate the cost for.
tiered_pricing (List[dict]): A list of dictionaries, where each dictionary
represents a pricing tier.
cost_key (str): The key in the tier dictionary that holds the per-token cost
(e.g., 'input_cost_per_token').
fallback_cost_key (Optional[str], optional): A fallback key to use if the
primary `cost_key` is not found in a tier. Defaults to None.
Returns:
float: The total calculated cost for the given tokens.
Example:
>>> tiered_pricing = [
... {"range": [0, 100000], "input_cost_per_token": 0.0001},
... {"range": [100000, 500000], "input_cost_per_token": 0.00005},
... ]
Calculating cost for 150,000 tokens:
(100,000 * 0.0001) + (50,000 * 0.00005) = $12.5
"""
if not tiered_pricing or tokens <= 0:
return 0.0
total_cost = 0.0
tokens_processed = 0
sorted_tiers = sorted(tiered_pricing, key=lambda x: x.get("range", [0, 0])[0])
for tier in sorted_tiers:
if tokens_processed >= tokens:
break
tier_range = tier.get("range", [])
if len(tier_range) != 2:
continue
range_start, range_end = tier_range
if tokens <= range_start:
continue
tier_start = max(range_start, tokens_processed)
tier_end = min(range_end, tokens)
if tier_end > tier_start:
tokens_in_tier = tier_end - tier_start
cost_per_token = tier.get(cost_key) or tier.get(fallback_cost_key, 0)
total_cost += tokens_in_tier * _coerce_cost_per_token(cost_per_token)
tokens_processed = tier_end
# After loop, check if any tokens remain (i.e., tokens > highest tier's end range)
# and charge them at the last tier's rate.
if tokens_processed < tokens and sorted_tiers:
last_tier = sorted_tiers[-1]
remaining_tokens = tokens - tokens_processed
cost_per_token = last_tier.get(cost_key) or last_tier.get(fallback_cost_key, 0)
total_cost += remaining_tokens * _coerce_cost_per_token(cost_per_token)
return total_cost
def select_tier_for_input(
tiered_pricing: List[dict],
input_tokens: int,

109
litellm/llms/dashscope/cost_calculator.py
View file

@ -2,12 +2,16 @@
Cost calculator for Dashscope Chat models.
Handles tiered pricing and prompt caching scenarios.
Alibaba Model Studio tiered pricing is step pricing: the total input token count of a
request selects one tier, and every token of that request (plain input, cached input,
completion, reasoning) is billed at that tier's rates.
"""
from dataclasses import dataclass
from typing import List, Optional, Tuple
from litellm.litellm_core_utils.llm_cost_calc.tiered_pricing import calculate_tiered_cost
from litellm.litellm_core_utils.llm_cost_calc.tiered_pricing import select_tier_for_input, tier_rate
from litellm.types.utils import ModelInfo, Usage
from litellm.utils import get_model_info
@ -43,68 +47,49 @@ def _extract_token_breakdown(usage: Usage) -> TokenBreakdown:
return TokenBreakdown(text_tokens, cached_tokens, completion_tokens, reasoning_tokens)
def _calculate_prompt_cost(
breakdown: TokenBreakdown,
model_info: ModelInfo,
tiered_pricing: Optional[List[dict]],
) -> float:
"""Calculate total prompt cost including cached tokens."""
if tiered_pricing:
text_cost = calculate_tiered_cost(
tokens=breakdown.text_tokens,
tiered_pricing=tiered_pricing,
cost_key="input_cost_per_token",
)
cache_cost = calculate_tiered_cost(
tokens=breakdown.cached_tokens,
tiered_pricing=tiered_pricing,
cost_key="cache_read_input_token_cost",
fallback_cost_key="input_cost_per_token",
)
return text_cost + cache_cost
@dataclass(frozen=True, slots=True)
class TokenRates:
"""Per-token rates applied to a single request."""
input: float
cached_input: float
output: float
reasoning_output: float
def _rates_from_tier(tier: dict) -> TokenRates:
return TokenRates(
input=tier_rate(tier, "input_cost_per_token"),
cached_input=tier_rate(tier, "cache_read_input_token_cost", "input_cost_per_token"),
output=tier_rate(tier, "output_cost_per_token"),
reasoning_output=tier_rate(tier, "output_cost_per_reasoning_token", "output_cost_per_token"),
)
def _rates_from_flat_pricing(model_info: ModelInfo) -> TokenRates:
input_cost = float(model_info.get("input_cost_per_token") or 0.0)
# For cache_cost, first try the specific key, then fall back to input_cost.
cache_cost_val = model_info.get("cache_read_input_token_cost")
if cache_cost_val is None:
cache_cost = input_cost
else:
cache_cost = float(cache_cost_val)
return (breakdown.text_tokens * input_cost) + (breakdown.cached_tokens * cache_cost)
def _calculate_completion_cost(
breakdown: TokenBreakdown,
model_info: ModelInfo,
tiered_pricing: Optional[List[dict]],
) -> float:
"""Calculate total completion cost including reasoning tokens."""
if tiered_pricing:
completion_cost = calculate_tiered_cost(
tokens=breakdown.completion_tokens,
tiered_pricing=tiered_pricing,
cost_key="output_cost_per_token",
)
reasoning_cost = calculate_tiered_cost(
tokens=breakdown.reasoning_tokens,
tiered_pricing=tiered_pricing,
cost_key="output_cost_per_reasoning_token",
fallback_cost_key="output_cost_per_token",
)
return completion_cost + reasoning_cost
output_cost = float(model_info.get("output_cost_per_token") or 0.0)
# For reasoning_cost, first try the specific key, then fall back to output_cost.
cache_cost_val = model_info.get("cache_read_input_token_cost")
reasoning_cost_val = model_info.get("output_cost_per_reasoning_token")
if reasoning_cost_val is None:
reasoning_cost = output_cost
else:
reasoning_cost = float(reasoning_cost_val)
return (breakdown.completion_tokens * output_cost) + (breakdown.reasoning_tokens * reasoning_cost)
return TokenRates(
input=input_cost,
cached_input=input_cost if cache_cost_val is None else float(cache_cost_val),
output=output_cost,
reasoning_output=output_cost if reasoning_cost_val is None else float(reasoning_cost_val),
)
def _select_rates(model_info: ModelInfo, input_tokens: int) -> TokenRates:
tiered_pricing: Optional[List[dict]] = (
model_info.get("tiered_pricing") if isinstance(model_info.get("tiered_pricing"), list) else None
)
if tiered_pricing:
tier = select_tier_for_input(tiered_pricing=tiered_pricing, input_tokens=input_tokens)
if tier is not None:
return _rates_from_tier(tier)
return _rates_from_flat_pricing(model_info)
def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
@ -122,11 +107,11 @@ def cost_per_token(model: str, usage: Usage) -> Tuple[float, float]:
"""
model_info = get_model_info(model=model, custom_llm_provider="dashscope")
breakdown = _extract_token_breakdown(usage)
tiered_pricing = model_info.get("tiered_pricing") if isinstance(model_info.get("tiered_pricing"), list) else None
rates = _select_rates(model_info=model_info, input_tokens=usage.prompt_tokens or 0)
prompt_cost = _calculate_prompt_cost(breakdown=breakdown, model_info=model_info, tiered_pricing=tiered_pricing)
completion_cost = _calculate_completion_cost(
breakdown=breakdown, model_info=model_info, tiered_pricing=tiered_pricing
prompt_cost = (breakdown.text_tokens * rates.input) + (breakdown.cached_tokens * rates.cached_input)
completion_cost = (breakdown.completion_tokens * rates.output) + (
breakdown.reasoning_tokens * rates.reasoning_output
)
return prompt_cost, completion_cost

136
tests/test_litellm/llms/dashscope/test_dashscope_cost_calculator.py
View file

@ -2,7 +2,7 @@
Test suite for Dashscope cost calculation functionality.
Tests the cost calculation for Dashscope models including:
- Correctly calculates graduated tiered pricing.
- Selects one pricing tier from the request's total input token count (step pricing).
- Falls back to flat-rate pricing for non-tiered models.
- Handles interactions with cached tokens.
- Correctly calculates costs for token counts exceeding the highest defined tier.
@ -73,38 +73,79 @@ class TestDashscopeCostCalculator:
assert math.isclose(prompt_cost, expected_prompt_cost, rel_tol=1e-10)
assert math.isclose(completion_cost, expected_completion_cost, rel_tol=1e-10)
def test_dashscope_tiered_pricing_spanning_multiple_tiers(self):
def test_dashscope_input_above_first_tier_bills_all_tokens_at_selected_tier(self):
"""
Tests the dashscope tiered pricing with the corrected graduated calculation logic.
This is the most important test for validating the fix.
Regression for #34729: a request whose total input exceeds the first tier is
billed entirely at the selected tier's rates, with no graduated slicing, and
completion tokens follow the tier chosen by the input size.
"""
# Tiering for qwen-flash: Tier 1: [0, 256k], Tier 2: [256k, 1M]
usage = Usage(prompt_tokens=300000, completion_tokens=300000)
usage = Usage(prompt_tokens=300000, completion_tokens=2000)
prompt_cost, completion_cost = dashscope_cost_per_token(
model="qwen-flash", usage=usage
)
model_info = litellm.get_model_info("dashscope/qwen-flash")
tier_2 = model_info["tiered_pricing"][1]
expected_prompt_cost = 300000 * tier_2["input_cost_per_token"]
expected_completion_cost = 2000 * tier_2["output_cost_per_token"]
assert math.isclose(prompt_cost, expected_prompt_cost, rel_tol=1e-10)
assert math.isclose(completion_cost, expected_completion_cost, rel_tol=1e-10)
def test_dashscope_small_input_with_large_output_stays_in_first_tier(self):
"""
Regression for #34729: output tokens never select their own tier. A small
input with an output count that would land in a higher tier is still billed
at the tier picked by the input size.
"""
usage = Usage(prompt_tokens=1000, completion_tokens=300000)
prompt_cost, completion_cost = dashscope_cost_per_token(
model="qwen-flash", usage=usage
)
model_info = litellm.get_model_info("dashscope/qwen-flash")
tier_1 = model_info["tiered_pricing"][0]
assert math.isclose(
prompt_cost, 1000 * tier_1["input_cost_per_token"], rel_tol=1e-10
)
assert math.isclose(
completion_cost, 300000 * tier_1["output_cost_per_token"], rel_tol=1e-10
)
def test_dashscope_input_at_tier_boundary_stays_in_lower_tier(self):
"""
A request of exactly the first tier's upper bound stays in that tier, matching
the official ``0 < Token <= 256K`` phrasing.
"""
model_info = litellm.get_model_info("dashscope/qwen-flash")
tier_1 = model_info["tiered_pricing"][0]
tier_2 = model_info["tiered_pricing"][1]
boundary = tier_1["range"][1]
# Expected prompt cost: (256,000 tokens * tier_1_price) + (44,000 tokens * tier_2_price)
expected_prompt_cost = (256000 * tier_1["input_cost_per_token"]) + (
44000 * tier_2["input_cost_per_token"]
at_boundary, _ = dashscope_cost_per_token(
model="qwen-flash",
usage=Usage(prompt_tokens=boundary, completion_tokens=0),
)
just_above, _ = dashscope_cost_per_token(
model="qwen-flash",
usage=Usage(prompt_tokens=boundary + 1, completion_tokens=0),
)
# Expected completion cost: (256,000 tokens * tier_1_price) + (44,000 tokens * tier_2_price)
expected_completion_cost = (256000 * tier_1["output_cost_per_token"]) + (
44000 * tier_2["output_cost_per_token"]
assert math.isclose(
at_boundary, boundary * tier_1["input_cost_per_token"], rel_tol=1e-10
)
assert math.isclose(
just_above, (boundary + 1) * tier_2["input_cost_per_token"], rel_tol=1e-10
)
assert math.isclose(prompt_cost, expected_prompt_cost, rel_tol=1e-10)
assert math.isclose(completion_cost, expected_completion_cost, rel_tol=1e-10)
def test_dashscope_tiered_pricing_with_caching(self):
"""
Tests tiered pricing with cached tokens. This replaces the old, incorrect test.
Uses qwen3-coder-plus, which has cache-specific pricing defined.
Cached and plain input tokens are both billed at the tier selected by the
request's total input size, using the tier's cache-specific rate for the
cached portion. Uses qwen3-coder-plus, which has cache-specific pricing.
"""
usage = Usage(
prompt_tokens=50000, # 10k cached + 40k new
@ -116,18 +157,16 @@ class TestDashscopeCostCalculator:
prompt_cost, _ = dashscope_cost_per_token(model="qwen3-coder-plus", usage=usage)
model_info = litellm.get_model_info("dashscope/qwen3-coder-plus")
tier_1 = model_info["tiered_pricing"][0]
tier_2 = model_info["tiered_pricing"][1]
# 10k cached tokens are all in the first tier
expected_cache_cost = 10000 * tier_1["cache_read_input_token_cost"]
# 40k new tokens: 32k in tier 1, and the remaining 8k in tier 2
expected_text_cost = (32000 * tier_1["input_cost_per_token"]) + (
8000 * tier_2["input_cost_per_token"]
# 50k total input selects the tier containing 50k (first tier ends at 32k)
tier = next(
t
for t in model_info["tiered_pricing"]
if t["range"][0] < 50000 <= t["range"][1]
)
expected_total_prompt_cost = expected_cache_cost + expected_text_cost
expected_total_prompt_cost = (10000 * tier["cache_read_input_token_cost"]) + (
40000 * tier["input_cost_per_token"]
)
assert math.isclose(prompt_cost, expected_total_prompt_cost, rel_tol=1e-10)
@ -162,8 +201,8 @@ class TestDashscopeCostCalculator:
model="qwen-str-tier-test", usage=usage
)
expected_prompt_cost = 500 * float("4e-07")
expected_completion_cost = 200 * float("1.6e-06")
expected_prompt_cost = 500 * 4e-07
expected_completion_cost = 200 * 1.6e-06
assert prompt_cost > 0
assert completion_cost > 0
@ -182,14 +221,9 @@ class TestDashscopeCostCalculator:
model="qwen-str-tier-test", usage=usage
)
# prompt: 1000 @ tier1 + 1000 @ tier2 + 500 remaining @ tier2 rate
expected_prompt_cost = (
(1000 * float("4e-07")) + (1000 * float("8e-07")) + (500 * float("8e-07"))
)
# completion: 1000 @ tier1 + 1000 @ tier2 + 1000 remaining @ tier2 rate
expected_completion_cost = (
(1000 * float("1.6e-06")) + (1000 * float("3.2e-06")) + (1000 * float("3.2e-06"))
)
# 2500 input tokens exceed the highest range, so the last tier's rates apply
expected_prompt_cost = 2500 * 8e-07
expected_completion_cost = 3000 * 3.2e-06
assert prompt_cost > 0
assert completion_cost > 0
@ -198,27 +232,23 @@ class TestDashscopeCostCalculator:
def test_dashscope_tiered_pricing_exceeding_highest_tier(self):
"""
Tests tiered pricing when token count exceeds the highest defined tier range.
This replaces the old, incorrect test and validates the new fallback logic.
Input beyond the highest declared range falls back to the last (most
expensive) tier for every token in the request.
"""
usage = Usage(
prompt_tokens=1200000, completion_tokens=1000
) # Max defined range for qwen-flash is 1M
prompt_cost, _ = dashscope_cost_per_token(model="qwen-flash", usage=usage)
model_info = litellm.get_model_info("dashscope/qwen-flash")
tier_1 = model_info["tiered_pricing"][0]
tier_2 = model_info["tiered_pricing"][1]
# Expected cost: (tier_1_tokens * tier_1_price) + (tokens_up_to_max_range_in_tier_2 * tier_2_price) + (remaining_tokens * tier_2_price)
tokens_in_tier_2_range = 1000000 - 256000
remaining_tokens_over_max = 1200000 - 1000000
expected_prompt_cost = (
(256000 * tier_1["input_cost_per_token"])
+ (tokens_in_tier_2_range * tier_2["input_cost_per_token"])
+ (remaining_tokens_over_max * tier_2["input_cost_per_token"])
prompt_cost, completion_cost = dashscope_cost_per_token(
model="qwen-flash", usage=usage
)
assert math.isclose(prompt_cost, expected_prompt_cost, rel_tol=1e-10)
model_info = litellm.get_model_info("dashscope/qwen-flash")
last_tier = model_info["tiered_pricing"][-1]
assert math.isclose(
prompt_cost, 1200000 * last_tier["input_cost_per_token"], rel_tol=1e-10
)
assert math.isclose(
completion_cost, 1000 * last_tier["output_cost_per_token"], rel_tol=1e-10
)