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

fix(dashscope): bill tiered pricing as all-or-nothing per official docs

Browse source 
Replaces the graduated/income-tax-bracket logic in _calculate_tiered_cost
with all-or-nothing tier selection, per Alibaba Bailian's official billing
docs: "the unit price is determined by total input tokens, and ALL tokens
are billed at that single tier's rate".
https://help.aliyun.com/zh/model-studio/billing-for-model-studio

Also corrects tier-selection input: previously _calculate_prompt_cost
passed text_tokens and cached_tokens separately to the tier selector,
which could pick the wrong tier on cache-heavy requests. Now uses
total input = text + cache to select one tier whose rates apply
uniformly to text, cache, and output.

Returns None for zero-input requests so the caller falls back to flat
pricing instead of charging completion tokens at the highest tier.

Eight tests cover within-tier, cross-tier (the regression case), cache,
overflow, boundary at range_end, output-uses-input-tier, and zero-input.
This commit is contained in:
HaiYangBG1 2026-05-16 16:51:38 +08:00
parent ec2f3aadb8
commit b62fb19f08
2 changed files with 178 additions and 139 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -1,7 +1,26 @@
"""
Cost calculator for Dashscope Chat models.
Cost calculator for Dashscope Chat models.
Handles tiered pricing and prompt caching scenarios.
Tiered pricing semantics
------------------------
Dashscope (Alibaba Bailian) uses **all-or-nothing** tiered pricing. The unit
price of a single request is determined by its total input-token count, and
*all* tokens of the request (input, cache hits, and output) are billed at that
tier's rate. This is documented in the official help center:
https://help.aliyun.com/zh/model-studio/billing-for-model-studio
"百炼部分模型实行阶梯计费。单价取决于单次请求的输入 Token 总量。
该请求的所有 Token 均按对应阶梯的单价结算。
例如，某模型设有两档计费区间：0 < Token ≤ 32K 和 32K < Token ≤ 128K。
若输入 100K Token，因数值落在第二区间（32K < 100K ≤ 128K），
所有 Token 均按第二档单价结算。"
i.e. a 100K input request priced under [0, 32K]=A, (32K, 128K]=B is billed at
B for the entire request — *not* at A for the first 32K plus B for the next
68K (which would be income-tax / graduated bracket logic).
"""
from dataclasses import dataclass
@ -46,78 +65,50 @@ def _extract_token_breakdown(usage: Usage) -> TokenBreakdown:
)
def _calculate_tiered_cost(
tokens: int,
def _select_tier_for_input(
total_input_tokens: int,
tiered_pricing: List[dict],
cost_key: str,
fallback_cost_key: Optional[str] = None,
) -> float:
) -> Optional[dict]:
"""
Calculate cost for a given number of tokens based on a true tiered pricing structure.
Return the single tier whose range contains ``total_input_tokens``.
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.
Per Dashscope rules the tier is selected by total input tokens of the
request, and that tier's rates apply uniformly to text, cached and output
tokens. If the request exceeds the highest declared range, the last tier
is used.
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
A tier matches when ``range_start < total_input_tokens <= range_end``, so
a request of exactly ``range_end`` tokens falls into the lower tier — matching
the official example ``0 < Token ≤ 32K``. A request with no input tokens
returns ``None`` so the caller can fall back to flat pricing (the tier
concept does not apply to an empty request).
"""
if not tiered_pricing or tokens <= 0:
return 0.0
total_cost = 0.0
tokens_processed = 0
if not tiered_pricing or total_input_tokens <= 0:
return None
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 range_start < total_input_tokens <= range_end:
return tier
if tokens <= range_start:
continue
return sorted_tiers[-1]
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 * 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 * cost_per_token
return total_cost
def _tier_rate(
tier: dict,
cost_key: str,
fallback_cost_key: Optional[str] = None,
) -> float:
"""Read ``cost_key`` from ``tier``, falling back to ``fallback_cost_key`` if absent."""
rate = tier.get(cost_key)
if rate is None and fallback_cost_key is not None:
rate = tier.get(fallback_cost_key)
return float(rate or 0.0)
def _calculate_prompt_cost(
@ -127,18 +118,17 @@ def _calculate_prompt_cost(
) -> 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
total_input = breakdown.text_tokens + breakdown.cached_tokens
tier = _select_tier_for_input(total_input, tiered_pricing)
if tier is not None:
input_rate = _tier_rate(tier, "input_cost_per_token")
cache_rate = _tier_rate(
tier, "cache_read_input_token_cost", "input_cost_per_token"
)
return (
breakdown.text_tokens * input_rate
+ breakdown.cached_tokens * cache_rate
)
input_cost = float(model_info.get("input_cost_per_token") or 0.0)
@ -157,20 +147,26 @@ def _calculate_completion_cost(
model_info: ModelInfo,
tiered_pricing: Optional[List[dict]],
) -> float:
"""Calculate total completion cost including reasoning tokens."""
"""
Calculate total completion cost including reasoning tokens.
Tier selection is based on *input* tokens, per Dashscope rules — output
tokens do not affect which tier is used. This is consistent with the
examples in the official documentation, which only ever reference input
Token counts when choosing a tier.
"""
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
total_input = breakdown.text_tokens + breakdown.cached_tokens
tier = _select_tier_for_input(total_input, tiered_pricing)
if tier is not None:
output_rate = _tier_rate(tier, "output_cost_per_token")
reasoning_rate = _tier_rate(
tier, "output_cost_per_reasoning_token", "output_cost_per_token"
)
return (
breakdown.completion_tokens * output_rate
+ breakdown.reasoning_tokens * reasoning_rate
)
output_cost = float(model_info.get("output_cost_per_token") or 0.0)

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

@ -2,13 +2,15 @@
Test suite for Dashscope cost calculation functionality.
Tests the cost calculation for Dashscope models including:
- Correctly calculates graduated tiered 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.
- All-or-nothing tiered pricing (single tier rate applied to the whole request).
- Flat-rate pricing fallback for models without ``tiered_pricing``.
- Cached-token interactions: cache cost is billed at the input-tier's rate
(selected by total input tokens, not by cached-token count alone).
- Output cost uses the same tier as the input.
- Boundary semantics at ``range_end``.
- Requests exceeding the highest declared tier fall back to the last tier.
"""
import json
import math
import os
import sys
@ -22,7 +24,7 @@ import litellm
from litellm.llms.dashscope.cost_calculator import (
cost_per_token as dashscope_cost_per_token,
)
from litellm.types.utils import Usage, PromptTokensDetailsWrapper
from litellm.types.utils import PromptTokensDetailsWrapper, Usage
class TestDashscopeCostCalculator:
@ -41,7 +43,6 @@ class TestDashscopeCostCalculator:
"""
usage = Usage(prompt_tokens=1000, completion_tokens=500)
# We call the specific calculator for dashscope
prompt_cost, completion_cost = dashscope_cost_per_token(
model="qwen-max", usage=usage
)
@ -55,56 +56,53 @@ class TestDashscopeCostCalculator:
def test_dashscope_tiered_pricing_within_first_tier(self):
"""
Tests the dashscope tiered pricing when token count is entirely within the first tier.
Uses 'dashscope/qwen-flash' as a real-world example.
Tokens entirely within tier 1 are billed at tier 1's rate.
Uses 'dashscope/qwen-flash' (tier 1 = [0, 256k]).
"""
# Tier 1 for qwen-flash is [0, 256,000] tokens
usage = Usage(prompt_tokens=100000, completion_tokens=50000)
prompt_cost, completion_cost = dashscope_cost_per_token(
model="qwen-flash", usage=usage
)
model_info = litellm.get_model_info("dashscope/qwen-flash")
tier_1_pricing = model_info["tiered_pricing"][0]
tier_1 = model_info["tiered_pricing"][0]
expected_prompt_cost = 100000 * tier_1_pricing["input_cost_per_token"]
expected_completion_cost = 50000 * tier_1_pricing["output_cost_per_token"]
expected_prompt_cost = 100000 * tier_1["input_cost_per_token"]
expected_completion_cost = 50000 * tier_1["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_tiered_pricing_spanning_multiple_tiers(self):
def test_dashscope_tiered_pricing_crossing_into_higher_tier(self):
"""
Tests the dashscope tiered pricing with the corrected graduated calculation logic.
This is the most important test for validating the fix.
All-or-nothing: a 300k input request on qwen-flash (tier 1 = [0, 256k],
tier 2 = [256k, 1M]) is billed entirely at tier 2's rate — *not* split
as 256k @ tier 1 + 44k @ tier 2 (which would be income-tax / graduated).
Reference: https://help.aliyun.com/zh/model-studio/billing-for-model-studio
"""
# Tiering for qwen-flash: Tier 1: [0, 256k], Tier 2: [256k, 1M]
usage = Usage(prompt_tokens=300000, 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]
tier_2 = model_info["tiered_pricing"][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"]
)
# 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"]
)
expected_prompt_cost = 300000 * tier_2["input_cost_per_token"]
expected_completion_cost = 300000 * 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_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.
Cache cost is billed at the input-tier's rate.
qwen3-coder-plus tiers: [0, 32k], (32k, 128k], (128k, 256k], (256k, 1M].
50k total input falls in tier 2 (32k, 128k], so both the text and the
cache portions are billed at tier 2 — even though the cached portion
alone (10k) would individually sit in tier 1.
"""
usage = Usage(
prompt_tokens=50000, # 10k cached + 40k new
@ -113,47 +111,92 @@ class TestDashscopeCostCalculator:
prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=10000),
)
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"]
prompt_cost, completion_cost = dashscope_cost_per_token(
model="qwen3-coder-plus", usage=usage
)
expected_total_prompt_cost = expected_cache_cost + expected_text_cost
model_info = litellm.get_model_info("dashscope/qwen3-coder-plus")
# 50k total input falls in tier 2 = index 1
tier_2 = model_info["tiered_pricing"][1]
assert math.isclose(prompt_cost, expected_total_prompt_cost, rel_tol=1e-10)
expected_text_cost = 40000 * tier_2["input_cost_per_token"]
expected_cache_cost = 10000 * tier_2["cache_read_input_token_cost"]
expected_prompt_cost = expected_text_cost + expected_cache_cost
expected_completion_cost = 1000 * 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_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.
Requests larger than the highest declared range fall back to the last
tier's rate. qwen-flash declares up to 1M; a 1.2M request is still
billed entirely at tier 2's rate.
"""
usage = Usage(
prompt_tokens=1200000, completion_tokens=1000
) # Max defined range for qwen-flash is 1M
usage = Usage(prompt_tokens=1200000, completion_tokens=1000)
prompt_cost, completion_cost = dashscope_cost_per_token(
model="qwen-flash", usage=usage
)
model_info = litellm.get_model_info("dashscope/qwen-flash")
last_tier = model_info["tiered_pricing"][-1]
expected_prompt_cost = 1200000 * last_tier["input_cost_per_token"]
expected_completion_cost = 1000 * last_tier["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_tiered_pricing_boundary_at_range_end(self):
"""
A request of exactly ``range_end`` tokens belongs to the lower tier,
matching the official phrasing ``0 < Token ≤ 32K``.
On qwen-flash (tier 1 ends at 256000) an input of exactly 256000 must
bill at tier 1, not tier 2.
"""
usage = Usage(prompt_tokens=256000, completion_tokens=10)
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"]
assert math.isclose(prompt_cost, expected_prompt_cost, rel_tol=1e-10)
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"])
def test_dashscope_tiered_pricing_output_uses_input_tier(self):
"""
Output token rate is taken from the tier selected by *input* tokens,
regardless of how many output tokens were produced. Sending 10k input
(tier 1) + 50k output on qwen-flash must bill the 50k output at tier 1.
"""
usage = Usage(prompt_tokens=10000, completion_tokens=50000)
_, 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]
expected_completion_cost = 50000 * tier_1["output_cost_per_token"]
assert math.isclose(completion_cost, expected_completion_cost, rel_tol=1e-10)
def test_dashscope_tiered_pricing_with_zero_input(self):
"""
Edge case: a request with zero input tokens has no tier to match —
``_select_tier_for_input`` returns ``None`` and the caller falls back
to flat pricing (which is 0 for purely-tiered models like qwen-flash).
This guards against the regression of charging zero-input completions
at the highest tier's rate.
"""
usage = Usage(prompt_tokens=0, completion_tokens=100)
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)
# qwen-flash has no flat input_cost_per_token / output_cost_per_token,
# so the fallback yields 0 for both — not the (incorrect) last-tier rate.
assert prompt_cost == 0.0
assert completion_cost == 0.0