feat(proxy): split Azure PTU deployments across teams by share

A PTU deployment's model_info takes ptu_shares (team_id -> whole PTUs adding up to
ptu_count) in place of a single team_id. With LITELLM_ENABLE_PTU_COST_ATTRIBUTION on,
routing hides the deployment from teams holding no share, the v3 limiter enforces each
share as a per-team TPM ceiling from Azure's per-model TPM-per-PTU sizing, the daily
flat-cost rollup splits the hourly cost by share, and /team/daily/activity plus the Team
Usage page report PTU-hours next to tokens.
This commit is contained in:
mateo-berri 2026-09-24 13:37:08 -07:00
parent f39a56b004
commit 2d49656a2a
30 changed files with 1479 additions and 48 deletions

View file

@ -0,0 +1,104 @@
"""How much throughput one Azure OpenAI provisioned throughput unit (PTU) serves per model.
Azure sizes a provisioned deployment in normalized tokens per minute:
``input TPM x (1 - cache hit rate) + output-to-input ratio x output TPM``, divided by the
model's "Input TPM per PTU" to get the PTUs required. The same two numbers turn a team's
PTU share into a per-minute token ceiling and a request's usage back into PTU-hours.
Table read from
https://learn.microsoft.com/en-us/azure/foundry/openai/how-to/provisioned-throughput-sizing#deployment-parameters-and-throughput-values-by-model
on 2026-09-24 (page dated 2026-09-23). Azure deducts cached input tokens in full for every
model except the GPT-6 family, where a cached input token costs a tenth of an uncached one.
"""
import re
from collections.abc import Mapping
from dataclasses import dataclass
from types import MappingProxyType
from typing import Final
@dataclass(frozen=True, slots=True)
class PTUCapacity:
input_tpm_per_ptu: int
output_to_input_ratio: float
cached_input_ratio: float = 0.0
@property
def normalized_tokens_per_ptu_hour(self) -> int:
return self.input_tpm_per_ptu * 60
AZURE_PTU_CAPACITY: Final[Mapping[str, PTUCapacity]] = MappingProxyType(
{
"gpt-6-sol": PTUCapacity(3_000, 5.0, cached_input_ratio=0.1),
"gpt-6-astra": PTUCapacity(600, 5.0, cached_input_ratio=0.1),
"gpt-5.6-luna": PTUCapacity(30_000, 6.0),
"gpt-5.6-terra": PTUCapacity(3_000, 6.0),
"gpt-5.6-sol": PTUCapacity(1_200, 6.0),
"gpt-5.5": PTUCapacity(1_200, 6.0),
"gpt-5.4": PTUCapacity(2_400, 6.0),
"gpt-5.4-mini": PTUCapacity(7_900, 6.0),
"gpt-5.3-codex": PTUCapacity(3_400, 8.0),
"gpt-5.2": PTUCapacity(3_400, 8.0),
"gpt-5.2-codex": PTUCapacity(3_400, 8.0),
"gpt-5.1": PTUCapacity(4_750, 8.0),
"gpt-5.1-codex": PTUCapacity(4_750, 8.0),
"gpt-5": PTUCapacity(4_750, 8.0),
"gpt-5-mini": PTUCapacity(23_750, 8.0),
"gpt-4.1": PTUCapacity(3_000, 4.0),
"gpt-4.1-mini": PTUCapacity(14_900, 4.0),
"gpt-4.1-nano": PTUCapacity(59_400, 4.0),
"o3": PTUCapacity(3_000, 4.0),
"o4-mini": PTUCapacity(5_400, 4.0),
"gpt-4o": PTUCapacity(2_500, 4.0),
"gpt-4o-mini": PTUCapacity(37_000, 4.0),
"o3-mini": PTUCapacity(2_500, 4.0),
"o1": PTUCapacity(230, 4.0),
"llama-3.3-70b-instruct": PTUCapacity(8_450, 4.0),
}
)
_VERSION_SUFFIX: Final = re.compile(r"-\d{4}-\d{2}-\d{2}$")
def azure_ptu_capacity(model: str) -> PTUCapacity | None:
"""The sizing row for ``model``, read as its last path segment with a dated version dropped.
``azure/gpt-4.1-2025-04-14`` and ``gpt-4.1`` both resolve to the ``gpt-4.1`` row; a
deployment name that is not a model name resolves to nothing, which is why callers
prefer ``model_info.base_model``.
"""
name: Final = model.rsplit("/", 1)[-1].strip().lower()
return AZURE_PTU_CAPACITY.get(name) or AZURE_PTU_CAPACITY.get(_VERSION_SUFFIX.sub("", name))
def deployment_ptu_capacity(deployment: Mapping[str, object]) -> PTUCapacity | None:
"""The sizing row a deployment resolves to: ``model_info.base_model`` first, since an
Azure deployment name is arbitrary, then ``litellm_params.model``."""
model_info: Final = deployment.get("model_info")
litellm_params: Final = deployment.get("litellm_params")
candidates: Final = tuple(
value
for value in (
model_info.get("base_model") if isinstance(model_info, Mapping) else None,
litellm_params.get("model") if isinstance(litellm_params, Mapping) else None,
)
if isinstance(value, str) and value
)
return next((capacity for capacity in map(azure_ptu_capacity, candidates) if capacity is not None), None)
def normalized_tokens(
capacity: PTUCapacity, *, prompt_tokens: int, completion_tokens: int, cache_read_tokens: int = 0
) -> float:
"""Azure's normalized token count for one request: uncached input in full, cached input
at the model's cached ratio, output weighted by the output-to-input ratio."""
cached: Final = min(max(cache_read_tokens, 0), max(prompt_tokens, 0))
uncached: Final = max(prompt_tokens, 0) - cached
return uncached + capacity.cached_input_ratio * cached + capacity.output_to_input_ratio * max(completion_tokens, 0)
def ptu_hours(capacity: PTUCapacity, normalized: float) -> float:
"""PTU-hours ``normalized`` tokens amount to: one PTU serves its input TPM for sixty minutes."""
return normalized / capacity.normalized_tokens_per_ptu_hour

View file

@ -56,9 +56,13 @@ PTU_ZEROED_PRICING: Final[Mapping[str, float | tuple[()] | Mapping[str, float]]]
@dataclass(frozen=True, slots=True)
class PTUTerms:
"""The reservation a deployment declares, once every field has been validated."""
"""The reservation a deployment declares, once every field has been validated.
team_id: str
``shares`` maps every team the capacity is attributed to onto its PTUs and adds up to
``ptu_count``: a deployment declaring a single ``team_id`` holds the whole count under it.
"""
shares: Mapping[str, int]
ptu_count: int
cost_per_ptu_per_hour: float
effective_from: datetime
@ -127,7 +131,44 @@ def ptu_identity_error(
return None
PTU_MODEL_INFO_FIELDS: Final = ("ptu_count", "cost_per_ptu_per_hour", "ptu_effective_from", "ptu_effective_to")
PTU_MODEL_INFO_FIELDS: Final = (
"ptu_count",
"cost_per_ptu_per_hour",
"ptu_effective_from",
"ptu_effective_to",
"ptu_shares",
)
def parsed_ptu_shares(raw: object) -> Mapping[str, int] | None:
"""``ptu_shares`` as team id -> whole PTUs, else None when empty or any entry is unusable.
A share is a count of reserved units, so it has to be a positive integer; ``bool`` is
excluded because it is an ``int`` subclass and ``True`` would read as one PTU.
"""
if not isinstance(raw, Mapping) or not raw:
return None
entries: Final = tuple((str(team_id), share) for team_id, share in raw.items())
if any(
not team_id or isinstance(share, bool) or not isinstance(share, int) or share <= 0 for team_id, share in entries
):
return None
return MappingProxyType(dict(entries))
def _declared_shares(model_info: Mapping[str, object], ptu_count: int) -> Mapping[str, int] | None:
"""Who holds the capacity: the single ``team_id`` holding all of it, or the ``ptu_shares``
that add up to it, else None."""
team_id: Final = model_info.get("team_id")
raw_shares: Final = model_info.get("ptu_shares")
if team_id and raw_shares is None:
return MappingProxyType({str(team_id): ptu_count})
if team_id or raw_shares is None:
return None
shares: Final = parsed_ptu_shares(raw_shares)
if shares is None or sum(shares.values()) != ptu_count:
return None
return shares
def declares_ptu(model_info: Mapping[str, object]) -> bool:
@ -164,8 +205,37 @@ def ptu_config_error(model_info: Mapping[str, object], *, model_name: str | None
"today could be billed for days it did not exist",
model_name,
)
if not model_info.get("team_id"):
return _named("team_id is required when PTU fields are set (one model maps to one team)", model_name)
return _ptu_holder_error(model_info, model_name)
def _ptu_holder_error(model_info: Mapping[str, object], model_name: str | None) -> str | None:
"""Why the teams this capacity is attributed to cannot be read, else None.
The shares have to add up to the count exactly: a shortfall would leave PTU-hours the
provider bills attributed to nobody, and a surplus would attribute capacity that was
never reserved.
"""
team_id: Final = model_info.get("team_id")
raw_shares: Final = model_info.get("ptu_shares")
if team_id and raw_shares is not None:
return _named(
"team_id and ptu_shares cannot both be set; ptu_shares lists every team the capacity is split across",
model_name,
)
if not team_id and raw_shares is None:
return _named("team_id or ptu_shares is required when PTU fields are set", model_name)
if raw_shares is None:
return None
shares: Final = parsed_ptu_shares(raw_shares)
if shares is None:
return _named("ptu_shares must map at least one team_id to a positive whole number of PTUs", model_name)
try:
ptu_count: Final = int(str(model_info.get("ptu_count")))
except ValueError:
return None
allocated: Final = sum(shares.values())
if allocated != ptu_count:
return _named(f"ptu_shares must add up to ptu_count ({allocated} of {ptu_count} allocated)", model_name)
return None
@ -178,8 +248,7 @@ def ptu_terms(model_info: Mapping[str, object]) -> PTUTerms | None:
"""
ptu_count: Final = model_info.get("ptu_count")
cost_per_hour: Final = model_info.get("cost_per_ptu_per_hour")
team_id: Final = model_info.get("team_id")
if ptu_count is None or cost_per_hour is None or not team_id:
if ptu_count is None or cost_per_hour is None:
return None
try:
ptu_count_int: Final = int(ptu_count)
@ -199,8 +268,11 @@ def ptu_terms(model_info: Mapping[str, object]) -> PTUTerms | None:
return None
if effective_to is not None and effective_to <= effective_from:
return None
shares: Final = _declared_shares(model_info, ptu_count_int)
if shares is None:
return None
return PTUTerms(
team_id=str(team_id),
shares=shares,
ptu_count=ptu_count_int,
cost_per_ptu_per_hour=cost_per_hour_float,
effective_from=effective_from,

View file

@ -3177,6 +3177,11 @@
"title": "Total Prompt Tokens",
"type": "integer"
},
"total_ptu_hours": {
"default": 0.0,
"title": "Total Ptu Hours",
"type": "number"
},
"total_response_time_ms": {
"default": 0,
"title": "Total Response Time Ms",
@ -3720,6 +3725,11 @@
"title": "Prompt Tokens",
"type": "integer"
},
"ptu_hours": {
"default": 0.0,
"title": "Ptu Hours",
"type": "number"
},
"spend": {
"default": 0.0,
"title": "Spend",

View file

@ -36,6 +36,7 @@ from litellm.integrations.custom_logger import CustomLogger
from litellm.litellm_core_utils.prompt_templates.common_utils import (
get_str_from_messages,
)
from litellm.litellm_core_utils.ptu_pricing import is_ptu_cost_attribution_enabled
from litellm.litellm_core_utils.token_counter import offload_token_count
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.auth.auth_utils import (
@ -64,6 +65,7 @@ from litellm.router_utils.add_retry_fallback_headers import (
response_has_hidden_params,
)
from litellm.router_utils.common_utils import resolve_model_group_alias
from litellm.router_utils.ptu_shares import PTUTeamCeiling, team_ptu_ceiling
from litellm.types.caching import RedisPipelineIncrementOperation
from litellm.types.llms.openai import BaseLiteLLMOpenAIResponseObject, ResponseAPIUsage
from litellm.types.utils import (
@ -112,6 +114,14 @@ def _resolve_model_group_alias_via_proxy_router(model: str) -> str | None:
return resolve_model_group_alias(llm_router.model_group_alias, model)
def _resolve_ptu_team_ceiling_via_proxy_router(team_id: str, model_group: str) -> PTUTeamCeiling | None:
from litellm.proxy.proxy_server import llm_router
if llm_router is None or not is_ptu_cost_attribution_enabled():
return None
return team_ptu_ceiling(llm_router.get_model_list(model_name=model_group) or (), team_id)
def _sibling_counter_keys(window_key: str) -> tuple[str, str]:
prefix: Final = window_key.removesuffix(":window")
return f"{prefix}:requests", f"{prefix}:tokens"
@ -420,6 +430,10 @@ _AUDIO_BYTES_PER_TOKEN: Final = 1600
# on the same project+model simultaneously without colliding on cache keys.
PROJECT_ITPM_DESCRIPTOR_KEY: Final = "model_per_project_itpm"
PROJECT_OTPM_DESCRIPTOR_KEY: Final = "model_per_project_otpm"
# Descriptor "key" for a team's PTU share of a shared Azure provisioned deployment,
# counted in Azure normalized tokens (output weighted by the model's ratio) so it
# never collides with the raw-token "model_per_team" counter on the same team+model.
PTU_TEAM_DESCRIPTOR_KEY: Final = "model_per_team_ptu"
# How long an acquired slot counts toward the in-flight total before it is
# considered leaked (worker crashed without any release callback firing) and
# pruned. Also the longest request duration the gauge can track: a request
@ -648,10 +662,14 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
internal_usage_cache: InternalUsageCache,
time_provider: Callable[[], datetime] | None = None,
model_group_resolver: Callable[[str], str | None] = _resolve_model_group_alias_via_proxy_router,
ptu_team_ceiling_resolver: Callable[
[str, str], PTUTeamCeiling | None
] = _resolve_ptu_team_ceiling_via_proxy_router,
):
self.internal_usage_cache = internal_usage_cache
self._time_provider = time_provider or datetime.now
self._model_group_resolver = model_group_resolver
self._ptu_team_ceiling_resolver = ptu_team_ceiling_resolver
if self.internal_usage_cache.dual_cache.redis_cache is not None:
self.batch_rate_limiter_script = self.internal_usage_cache.dual_cache.redis_cache.async_register_script(
BATCH_RATE_LIMITER_SCRIPT
@ -2879,6 +2897,11 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
requested_model=requested_model if isinstance(requested_model, str) else None,
descriptors=descriptors,
)
self._add_team_ptu_rate_limit_descriptor(
user_api_key_dict=user_api_key_dict,
requested_model=requested_model if isinstance(requested_model, str) else None,
descriptors=descriptors,
)
# Agent-level and session-level rate limits
resolved_agent_id: Final = self._get_resolved_agent_id(user_api_key_dict, data)
@ -3017,6 +3040,28 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
)
)
def _add_team_ptu_rate_limit_descriptor(
self,
user_api_key_dict: UserAPIKeyAuth,
requested_model: str | None,
descriptors: list[RateLimitDescriptor],
) -> None:
model: Final = self._rate_limited_model(requested_model)
if model is None or not user_api_key_dict.team_id:
return
ceiling: Final = self._ptu_team_ceiling_resolver(user_api_key_dict.team_id, model.group)
if ceiling is None:
return
descriptors.append(
RateLimitDescriptor(
key=PTU_TEAM_DESCRIPTOR_KEY,
value=f"{user_api_key_dict.team_id}:{model.group}",
rate_limit=RateLimitDescriptorRateLimitObject(
requests_per_unit=None, tokens_per_unit=ceiling.tpm_limit, window_size=self.window_size
),
)
)
def _add_project_model_rate_limit_descriptor_from_metadata(
self,
user_api_key_dict: UserAPIKeyAuth,
@ -4550,9 +4595,54 @@ class _PROXY_MaxParallelRequestsHandler_v3(CustomLogger):
reserved_tokens=reserved_tokens,
)
)
pipeline_operations.extend(
self._build_team_ptu_tpm_ops(
standard_logging_metadata=standard_logging_metadata,
response_obj=response_obj,
reconcile_model=reconcile_model,
reserved_scopes=reserved_scopes,
reserved_tokens=reserved_tokens,
total_tokens=total_tokens,
)
)
return pipeline_operations
def _build_team_ptu_tpm_ops(
self,
standard_logging_metadata: Mapping[str, object],
response_obj: object,
reconcile_model: RateLimitedModel | None,
reserved_scopes: Set[tuple[str, str]],
reserved_tokens: int,
total_tokens: int,
) -> Sequence[RedisPipelineIncrementOperation]:
"""Settle the team's PTU counter in Azure normalized tokens: uncached input in full plus
output weighted by the model's output-to-input ratio, the way Azure sizes a PTU.
The pre-call reservation was raw estimated tokens, so this is the same reconcile as the
other TPM scopes with a weighted actual; when usage cannot be resolved it charges the raw
total the other scopes charge.
"""
team_id: Final = standard_logging_metadata.get("user_api_key_team_id")
if reconcile_model is None or not isinstance(team_id, str) or not team_id:
return ()
ceiling: Final = self._ptu_team_ceiling_resolver(team_id, reconcile_model.group)
if ceiling is None:
return ()
billable_input, completion_tokens, usage_resolved = self._resolve_io_token_reconcile_usage(response_obj)
normalized: Final = (
billable_input + round(ceiling.output_to_input_ratio * completion_tokens)
if usage_resolved
else total_tokens
)
return self._build_reservation_aware_tpm_ops(
targets=((PTU_TEAM_DESCRIPTOR_KEY, f"{team_id}:{reconcile_model.group}"),),
reserved_scopes=reserved_scopes,
actual_tokens=normalized,
reserved_tokens=reserved_tokens,
)
async def async_log_success_event(self, kwargs, response_obj, start_time, end_time):
"""
Update TPM usage on successful API calls by incrementing counters using pipeline

View file

@ -0,0 +1,95 @@
"""PTU-hours a team's tokens amount to, attached to a daily activity response.
Azure sizes a provisioned deployment in normalized tokens per minute per PTU, so the prompt,
cached, and completion tokens a team sent to a PTU model group convert back to the share of
a PTU-hour it consumed, reported next to the raw token counts.
"""
from collections.abc import Callable
from types import MappingProxyType
from typing import Final
from litellm.litellm_core_utils.azure_ptu_capacity import PTUCapacity, normalized_tokens, ptu_hours
from litellm.types.proxy.management_endpoints.common_daily_activity import (
DailySpendData,
MetricWithMetadata,
SpendAnalyticsPaginatedResponse,
SpendMetrics,
)
def _with_ptu_hours(metrics: SpendMetrics, capacity: PTUCapacity) -> SpendMetrics:
consumed: Final = ptu_hours(
capacity,
normalized_tokens(
capacity,
prompt_tokens=metrics.prompt_tokens,
completion_tokens=metrics.completion_tokens,
cache_read_tokens=metrics.cache_read_input_tokens,
),
)
return metrics.model_copy(update=MappingProxyType({"ptu_hours": consumed}))
def _model_group_with_ptu_hours(bucket: MetricWithMetadata, capacity: PTUCapacity) -> MetricWithMetadata:
api_key_breakdown: Final = { # mutable-ok: pydantic serializes a dict[...] field only from a plain dict
api_key: key_bucket.model_copy(
update=MappingProxyType({"metrics": _with_ptu_hours(key_bucket.metrics, capacity)})
)
for api_key, key_bucket in bucket.api_key_breakdown.items()
}
return bucket.model_copy(
update=MappingProxyType(
{"metrics": _with_ptu_hours(bucket.metrics, capacity), "api_key_breakdown": api_key_breakdown}
)
)
def _day_with_ptu_hours(
day: DailySpendData, capacity_for_model_group: Callable[[str], PTUCapacity | None]
) -> DailySpendData:
priced: Final = MappingProxyType(
{
model_group: _model_group_with_ptu_hours(bucket, capacity)
for model_group, bucket in day.breakdown.model_groups.items()
if (capacity := capacity_for_model_group(model_group)) is not None
}
)
if not priced:
return day
model_groups: Final = { # mutable-ok: pydantic serializes a dict[...] field only from a plain dict
**day.breakdown.model_groups,
**priced,
}
return day.model_copy(
update=MappingProxyType(
{
"metrics": day.metrics.model_copy(
update=MappingProxyType({"ptu_hours": sum(bucket.metrics.ptu_hours for bucket in priced.values())})
),
"breakdown": day.breakdown.model_copy(update=MappingProxyType({"model_groups": model_groups})),
}
)
)
def attach_ptu_hours(
response: SpendAnalyticsPaginatedResponse, capacity_for_model_group: Callable[[str], PTUCapacity | None]
) -> SpendAnalyticsPaginatedResponse:
"""The response with ``ptu_hours`` filled in on every PTU model group and its api keys,
on each day, and on the total, from the tokens already on the page.
A model group the resolver has no sizing row for keeps ``ptu_hours`` at zero.
"""
days: Final = tuple(_day_with_ptu_hours(day, capacity_for_model_group) for day in response.results)
results: Final = list(days) # mutable-ok: pydantic serializes a list[...] field only from a plain list
return response.model_copy(
update=MappingProxyType(
{
"results": results,
"metadata": response.metadata.model_copy(
update=MappingProxyType({"total_ptu_hours": sum(day.metrics.ptu_hours for day in days)})
),
}
)
)

View file

@ -41,6 +41,7 @@ import litellm
from litellm._logging import verbose_proxy_logger
from litellm._uuid import uuid
from litellm.integrations.prometheus import PrometheusLogger
from litellm.litellm_core_utils.ptu_pricing import is_ptu_cost_attribution_enabled
from litellm.litellm_core_utils.safe_json_dumps import safe_dumps
from litellm.proxy._types import (
UI_TEAM_ID,
@ -143,6 +144,7 @@ from litellm.proxy.management_endpoints.common_utils import (
from litellm.proxy.management_endpoints.organization_endpoints import (
add_member_to_organization,
)
from litellm.proxy.management_endpoints.ptu_consumption import attach_ptu_hours
from litellm.proxy.management_endpoints.router_weights import validate_router_settings_weights
from litellm.proxy.management_endpoints.tag_management_endpoints import (
get_daily_activity,
@ -194,6 +196,7 @@ from litellm.repositories.verification_token_repository import (
VerificationTokenRepository,
)
from litellm.router import Router
from litellm.router_utils.ptu_shares import model_group_ptu_capacity
from litellm.types.proxy.auth.auth_checks import UserNotFoundError
from litellm.types.proxy.management_endpoints.common_daily_activity import (
SpendAnalyticsPaginatedResponse,
@ -6647,6 +6650,17 @@ async def _resolve_team_daily_activity_scope(
)
def _with_ptu_consumption(
activity: SpendAnalyticsPaginatedResponse, llm_router: Router | None
) -> SpendAnalyticsPaginatedResponse:
if llm_router is None or not is_ptu_cost_attribution_enabled():
return activity
return attach_ptu_hours(
activity,
lambda model_group: model_group_ptu_capacity(llm_router.get_model_list(model_name=model_group) or ()),
)
@router.get(
"/team/daily/activity",
response_model=SpendAnalyticsPaginatedResponse,
@ -6679,6 +6693,7 @@ async def get_team_daily_activity(
SpendAnalyticsPaginatedResponse: Paginated response containing daily activity data.
"""
from litellm.proxy.proxy_server import (
llm_router,
prisma_client,
proxy_logging_obj,
user_api_key_cache,
@ -6697,7 +6712,7 @@ async def get_team_daily_activity(
proxy_logging_obj=proxy_logging_obj,
)
return await get_daily_activity(
activity: Final = await get_daily_activity(
prisma_client=prisma_client,
table_name="litellm_dailyteamspend",
entity_id_field="team_id",
@ -6711,6 +6726,7 @@ async def get_team_daily_activity(
page=page,
page_size=page_size,
)
return _with_ptu_consumption(activity, llm_router)
_MAX_AGGREGATED_RANGE_DAYS: Final = 400
@ -6767,6 +6783,7 @@ async def get_team_daily_activity_aggregated(
SpendAnalyticsPaginatedResponse: Response containing all daily activity data for the range.
"""
from litellm.proxy.proxy_server import (
llm_router,
prisma_client,
proxy_logging_obj,
user_api_key_cache,
@ -6789,7 +6806,7 @@ async def get_team_daily_activity_aggregated(
proxy_logging_obj=proxy_logging_obj,
)
return await get_daily_activity_aggregated(
activity: Final = await get_daily_activity_aggregated(
prisma_client=prisma_client,
table_name="litellm_dailyteamspend",
entity_id_field="team_id",
@ -6803,6 +6820,7 @@ async def get_team_daily_activity_aggregated(
timezone_offset_minutes=timezone,
include_entity_breakdown=True,
)
return _with_ptu_consumption(activity, llm_router)
def _team_user_spend_sql(*, team_count: int, restrict_to_user: bool) -> str:

View file

@ -82,7 +82,8 @@ class BackfillResult:
@dataclass(frozen=True, slots=True)
class PTUModel:
"""A model deployment carrying valid manual PTU config."""
"""One team's holding on a deployment carrying valid manual PTU config: the whole count
under a single team_id, or that team's share of a split deployment."""
model_id: str
model_name: str
@ -98,10 +99,10 @@ def _public_model_name(row: object, model_info: Mapping[str, object]) -> str:
Creating a team-scoped deployment rewrites model_name to a synthetic routing key
(``model_name_<team_id>_<uuid4>``) and keeps the chosen name in
``model_info.team_public_model_name``. PTU config is only accepted alongside a
team_id, so every PTU deployment carries that synthetic name; keying the sentinel
row on it would file each charge under a UUID that no usage view can resolve and
that never lines up with the same model's request rows.
``model_info.team_public_model_name``. A PTU deployment held by one team_id carries
that synthetic name; keying the sentinel row on it would file each charge under a
UUID that no usage view can resolve and that never lines up with the same model's
request rows. A deployment split into ptu_shares keeps the name it was declared with.
"""
public_name: Final = model_info.get("team_public_model_name")
if isinstance(public_name, str) and public_name:
@ -128,7 +129,7 @@ def _decode_model_info(raw: object) -> "Mapping[str, object] | None":
@dataclass(frozen=True, slots=True)
class _PTUDeployment:
"""A deployment in the shape ``_parse_ptu_model`` reads, whatever declared it.
"""A deployment in the shape ``_parse_ptu_models`` reads, whatever declared it.
A ``LiteLLM_ProxyModelTable`` row already has it. A router entry does not: its id
lives in ``model_info.id`` rather than on the entry itself.
@ -160,26 +161,30 @@ def _router_deployment(deployment: Mapping[str, object]) -> _PTUDeployment | Non
)
def _parse_ptu_model(row: object) -> PTUModel | None:
"""Return a PTUModel when the deployment carries valid manual PTU config, else None.
def _parse_ptu_models(row: object) -> tuple[PTUModel, ...]:
"""One PTUModel per team the deployment's valid PTU config attributes capacity to, else empty.
Valid means model_info has a positive ptu_count, a non-negative
cost_per_ptu_per_hour, and a team_id (1 model -> 1 team).
A deployment held by one team_id yields one entry carrying the whole count; a deployment
split into ptu_shares yields one per team carrying that team's share, so each team's row
accrues its share of the flat cost.
"""
model_info: Final = _decode_model_info(getattr(row, "model_info", None))
if model_info is None:
return None
return ()
terms: Final = ptu_terms(model_info)
if terms is None:
return None
return PTUModel(
model_id=str(getattr(row, "model_id", "") or ""),
model_name=_public_model_name(row, model_info),
team_id=terms.team_id,
ptu_count=terms.ptu_count,
cost_per_ptu_per_hour=terms.cost_per_ptu_per_hour,
effective_from=terms.effective_from,
effective_to=terms.effective_to,
return ()
return tuple(
PTUModel(
model_id=str(getattr(row, "model_id", "") or ""),
model_name=_public_model_name(row, model_info),
team_id=team_id,
ptu_count=share,
cost_per_ptu_per_hour=terms.cost_per_ptu_per_hour,
effective_from=terms.effective_from,
effective_to=terms.effective_to,
)
for team_id, share in sorted(terms.shares.items())
)
@ -376,9 +381,7 @@ async def _load_ptu_models(prisma_client: "PrismaClient", *, router: object | No
rows: Final = await _proxy_model_table(prisma_client).find_many()
db_ids: Final = frozenset(model_id for row in rows if (model_id := str(getattr(row, "model_id", "") or "")))
config_records: Final = _config_deployments(router, owned_by_db=db_ids)
models: Final = tuple(
parsed for parsed in (_parse_ptu_model(row) for row in (*rows, *config_records)) if parsed is not None
)
models: Final = tuple(parsed for row in (*rows, *config_records) for parsed in _parse_ptu_models(row))
return _LoadedDeployments(
models=models,
scanned_ids=db_ids
@ -485,11 +488,13 @@ def _lapsed_models(ptu_models: tuple[PTUModel, ...], now: datetime) -> tuple[str
charge the provider does not make for reserved capacity.
"""
return tuple(
_slack_safe(model.model_name)
for model in sorted(
(m for m in ptu_models if m.effective_to is not None and m.effective_to <= now),
key=lambda m: m.effective_to,
reverse=True,
dict.fromkeys(
_slack_safe(model.model_name)
for model in sorted(
(m for m in ptu_models if m.effective_to is not None and m.effective_to <= now),
key=lambda m: m.effective_to,
reverse=True,
)
)
)

View file

@ -238,6 +238,7 @@ from litellm.router_utils.pre_call_checks.model_rate_limit_check import (
from litellm.router_utils.pre_call_checks.prompt_caching_deployment_check import (
PromptCachingDeploymentCheck,
)
from litellm.router_utils.ptu_shares import filter_ptu_shared_deployments, ptu_capacity_warning
from litellm.router_utils.reasoning_effort_capability import (
deployment_is_catalog_mapped,
intersect_supported_reasoning_efforts,
@ -8824,6 +8825,15 @@ class Router:
access_windows_error: Final = access_windows_config_error(_model_info, model_name=_model_name)
if access_windows_error is not None:
raise ValueError(access_windows_error)
capacity_warning: Final = (
ptu_capacity_warning(
_model_name, MappingProxyType({"model_info": _model_info, "litellm_params": _litellm_params})
)
if is_ptu_cost_attribution_enabled()
else None
)
if capacity_warning is not None:
verbose_router_logger.warning(capacity_warning)
zeroed_pricing: Final = zeroed_ptu_pricing(_model_info, _litellm_params) if config_sourced else None
litellm_params: Final[LiteLLM_Params] = LiteLLM_Params(
**( # pyright: ignore[reportArgumentType] # untyped merged dict; already true for every field here
@ -12778,7 +12788,16 @@ class Router:
model=model,
llm_provider="",
)
return result.deployments
if not is_ptu_cost_attribution_enabled():
return result.deployments
shared: Final = filter_ptu_shared_deployments(result.deployments, request_team_id)
if shared.withheld and len(shared.deployments) == 0:
raise litellm.BadRequestError(
message=f"Deployment {model} is reserved for the teams holding a PTU share of it",
model=model,
llm_provider="",
)
return shared.deployments
def _filter_deployments_by_model_access_groups(
self,

View file

@ -0,0 +1,101 @@
"""Per-team PTU shares on a shared Azure provisioned deployment.
A deployment declaring ``model_info.ptu_shares`` is served only to the teams named in it,
and each team's share converts to a normalized-tokens-per-minute ceiling through the
model's Azure sizing row, so the proxy enforces the split instead of an operator hand
converting PTUs to ``model_tpm_limit``.
"""
from collections.abc import Mapping, Sequence
from dataclasses import dataclass
from typing import Final, Generic, TypeVar
from litellm.litellm_core_utils.azure_ptu_capacity import PTUCapacity, deployment_ptu_capacity
from litellm.litellm_core_utils.ptu_pricing import parsed_ptu_shares, ptu_terms
_DeploymentT = TypeVar("_DeploymentT", bound=Mapping[str, object])
@dataclass(frozen=True, slots=True)
class PTUTeamCeiling:
tpm_limit: int
output_to_input_ratio: float
@dataclass(frozen=True, slots=True)
class PTUShareFilterResult(Generic[_DeploymentT]):
deployments: tuple[_DeploymentT, ...]
withheld: bool
def _deployment_shares(deployment: Mapping[str, object]) -> Mapping[str, int] | None:
model_info: Final = deployment.get("model_info")
if not isinstance(model_info, Mapping):
return None
return parsed_ptu_shares(model_info.get("ptu_shares"))
def filter_ptu_shared_deployments(
healthy_deployments: Sequence[_DeploymentT], request_team_id: str | None
) -> PTUShareFilterResult[_DeploymentT]:
"""Drop every deployment split into PTU shares that ``request_team_id`` holds none of.
A caller with no team, the master key included, holds no share, the same way an access
window reserves a deployment away from it.
"""
checks: Final = tuple((deployment, _deployment_shares(deployment)) for deployment in healthy_deployments)
kept: Final = tuple(
deployment for deployment, shares in checks if shares is None or (request_team_id or "") in shares
)
return PTUShareFilterResult(deployments=kept, withheld=len(kept) < len(checks))
def team_ptu_ceiling(deployments: Sequence[Mapping[str, object]], team_id: str) -> PTUTeamCeiling | None:
"""The per-minute normalized-token ceiling ``team_id``'s shares across ``deployments`` add
up to, else None when the team holds no share on a deployment with a known sizing row.
Two shared deployments of different models in one group are weighted by the larger
output ratio, which over-counts output on the cheaper one rather than under-counting it
on the dearer one.
"""
priced: Final = tuple(
(shares[team_id], capacity)
for deployment in deployments
if (shares := _deployment_shares(deployment)) is not None
and team_id in shares
and (capacity := deployment_ptu_capacity(deployment)) is not None
)
if not priced:
return None
return PTUTeamCeiling(
tpm_limit=sum(share * capacity.input_tpm_per_ptu for share, capacity in priced),
output_to_input_ratio=max(capacity.output_to_input_ratio for _, capacity in priced),
)
def model_group_ptu_capacity(deployments: Sequence[Mapping[str, object]]) -> PTUCapacity | None:
"""The sizing row of the group's first reserved deployment, single-team or shared, so a
team's tokens on that group convert to PTU-hours."""
return next(
(
capacity
for deployment in deployments
if isinstance(model_info := deployment.get("model_info"), Mapping)
and ptu_terms(model_info) is not None
and (capacity := deployment_ptu_capacity(deployment)) is not None
),
None,
)
def ptu_capacity_warning(model_name: str, deployment: Mapping[str, object]) -> str | None:
"""Why this reserved deployment's tokens cannot be converted to PTUs, else None."""
model_info: Final = deployment.get("model_info")
if not isinstance(model_info, Mapping) or ptu_terms(model_info) is None:
return None
if deployment_ptu_capacity(deployment) is not None:
return None
return (
f"PTU deployment '{model_name}' has no Azure sizing row for its model, so its PTU shares set no "
"team ceiling and its usage reports no PTU hours; set model_info.base_model to the Azure model name"
)

View file

@ -19,6 +19,7 @@ class GroupByDimension(str, Enum):
class SpendMetrics(BaseModel):
spend: float = Field(default=0.0)
flat_cost: float = Field(default=0.0)
ptu_hours: float = Field(default=0.0)
prompt_tokens: int = Field(default=0)
completion_tokens: int = Field(default=0)
cache_read_input_tokens: int = Field(default=0)
@ -83,6 +84,7 @@ class DailySpendData(BaseModel):
class DailySpendMetadata(BaseModel):
total_spend: float = Field(default=0.0)
total_flat_cost: float = Field(default=0.0)
total_ptu_hours: float = Field(default=0.0)
total_prompt_tokens: int = Field(default=0)
total_completion_tokens: int = Field(default=0)
total_tokens: int = Field(default=0)

View file

@ -260,6 +260,7 @@ class ModelInfo(MirroredPricingParams):
cost_per_ptu_per_hour: float | None = None
ptu_effective_from: datetime.datetime | None = None
ptu_effective_to: datetime.datetime | None = None
ptu_shares: Mapping[str, int] | None = None
# when tag-based routing's "!" or "&" constraints eliminate every deployment
# in this model group, fall back to the default-tagged pool instead of

View file

@ -0,0 +1,88 @@
"""Tests for the Azure PTU sizing table and the conversions built on it."""
from typing import Final
import pytest
from litellm.litellm_core_utils.azure_ptu_capacity import (
AZURE_PTU_CAPACITY,
PTUCapacity,
azure_ptu_capacity,
deployment_ptu_capacity,
normalized_tokens,
ptu_hours,
)
_GPT41: Final = AZURE_PTU_CAPACITY["gpt-4.1"]
_ROW: Final = PTUCapacity(input_tpm_per_ptu=1_000, output_to_input_ratio=4.0)
_CACHED_ROW: Final = PTUCapacity(input_tpm_per_ptu=1_000, output_to_input_ratio=5.0, cached_input_ratio=0.1)
@pytest.mark.parametrize(
"model",
["gpt-4.1", "azure/gpt-4.1", "azure/gpt-4.1-2025-04-14", "GPT-4.1", "azure/eastus/gpt-4.1-2025-04-14"],
)
def test_a_model_resolves_to_its_row_through_a_provider_prefix_and_a_dated_version(model):
assert azure_ptu_capacity(model) is _GPT41
def test_a_deployment_name_that_is_not_a_model_has_no_row():
assert azure_ptu_capacity("azure/my-ptu-deployment") is None
assert azure_ptu_capacity("") is None
def test_a_dated_version_only_drops_a_full_date_suffix():
"""``gpt-4o-mini`` must not lose its ``-mini`` the way ``-2024-07-18`` is dropped."""
assert azure_ptu_capacity("azure/gpt-4o-mini-2024-07-18") is AZURE_PTU_CAPACITY["gpt-4o-mini"]
assert azure_ptu_capacity("gpt-4o-mini") is not AZURE_PTU_CAPACITY["gpt-4o"]
def test_every_row_serves_its_input_tpm_for_an_hour():
for capacity in AZURE_PTU_CAPACITY.values():
assert capacity.normalized_tokens_per_ptu_hour == capacity.input_tpm_per_ptu * 60
assert capacity.input_tpm_per_ptu > 0
assert capacity.output_to_input_ratio >= 1.0
assert 0.0 <= capacity.cached_input_ratio < 1.0
def test_a_deployment_prefers_its_declared_base_model_over_its_deployment_name():
deployment: Final = {
"model_info": {"base_model": "azure/gpt-4.1"},
"litellm_params": {"model": "azure/gpt-4o"},
}
assert deployment_ptu_capacity(deployment) is _GPT41
def test_a_deployment_falls_back_to_its_litellm_model_when_no_base_model_is_declared():
assert deployment_ptu_capacity({"litellm_params": {"model": "azure/gpt-4o"}}) is AZURE_PTU_CAPACITY["gpt-4o"]
assert deployment_ptu_capacity({"model_info": {"base_model": ""}, "litellm_params": {"model": "azure/gpt-4o"}}) is (
AZURE_PTU_CAPACITY["gpt-4o"]
)
def test_a_deployment_with_no_recognisable_model_has_no_row():
assert deployment_ptu_capacity({"litellm_params": {"model": "azure/team-a-ptu"}}) is None
assert deployment_ptu_capacity({"model_info": None, "litellm_params": None}) is None
assert deployment_ptu_capacity({}) is None
def test_output_is_weighted_by_the_models_ratio_and_uncached_input_counts_in_full():
assert normalized_tokens(_ROW, prompt_tokens=100, completion_tokens=10) == pytest.approx(140.0)
def test_cached_input_is_free_unless_the_row_prices_it():
assert normalized_tokens(_ROW, prompt_tokens=100, completion_tokens=0, cache_read_tokens=60) == pytest.approx(40.0)
assert normalized_tokens(_CACHED_ROW, prompt_tokens=100, completion_tokens=0, cache_read_tokens=60) == pytest.approx(
46.0
)
def test_cached_input_never_exceeds_the_prompt_and_negatives_count_as_zero():
assert normalized_tokens(_ROW, prompt_tokens=100, completion_tokens=0, cache_read_tokens=250) == 0.0
assert normalized_tokens(_ROW, prompt_tokens=-5, completion_tokens=-5, cache_read_tokens=-5) == 0.0
def test_one_ptu_hour_is_one_ptus_input_tpm_served_for_sixty_minutes():
assert ptu_hours(_ROW, _ROW.input_tpm_per_ptu * 60) == pytest.approx(1.0)
assert ptu_hours(_ROW, _ROW.input_tpm_per_ptu * 30) == pytest.approx(0.5)
assert ptu_hours(_ROW, 0.0) == 0.0

View file

@ -38,7 +38,7 @@ def test_a_complete_reservation_is_accepted():
terms = ptu_terms(_VALID)
assert terms is not None
assert terms.team_id == "team-alpha"
assert terms.shares == {"team-alpha": 100}
assert terms.ptu_count == 100
assert terms.effective_from == datetime(2026, 1, 1, tzinfo=timezone.utc)
assert terms.effective_to is None
@ -193,8 +193,8 @@ def test_a_deployment_with_no_ptu_fields_is_not_a_ptu_deployment():
@pytest.mark.parametrize(
"override, expected",
[
({"team_id": None}, "team_id is required when PTU fields are set (one model maps to one team)"),
({"team_id": ""}, "team_id is required when PTU fields are set (one model maps to one team)"),
({"team_id": None}, "team_id or ptu_shares is required when PTU fields are set"),
({"team_id": ""}, "team_id or ptu_shares is required when PTU fields are set"),
({"cost_per_ptu_per_hour": None}, "ptu_count and cost_per_ptu_per_hour must be set together"),
({"ptu_count": None}, "ptu_count and cost_per_ptu_per_hour must be set together"),
({"ptu_effective_to": "2025-01-01T00:00:00Z"}, "ptu_effective_to must be after ptu_effective_from"),
@ -356,3 +356,78 @@ def test_no_spillover_marker_returns_none():
is None
)
assert azure_spillover(response_headers=None, additional_headers=None) is None
# --- a reservation split into per-team shares ------------------------------------------
_SHARED = {**{k: v for k, v in _VALID.items() if k != "team_id"}, "ptu_shares": {"team-a": 60, "team-b": 40}}
def test_a_split_reservation_attributes_each_share_to_its_team():
terms = ptu_terms(_SHARED)
assert terms is not None
assert terms.shares == {"team-a": 60, "team-b": 40}
assert terms.ptu_count == 100
assert ptu_config_error(_SHARED) is None
def test_a_single_team_reservation_holds_the_whole_count_under_that_team():
terms = ptu_terms(_VALID)
assert terms is not None
assert sum(terms.shares.values()) == terms.ptu_count
@pytest.mark.parametrize(
"override, expected",
[
(
{"team_id": "team-alpha"},
"team_id and ptu_shares cannot both be set; ptu_shares lists every team the capacity is split across",
),
({"ptu_shares": {}}, "ptu_shares must map at least one team_id to a positive whole number of PTUs"),
(
{"ptu_shares": {"team-a": 0, "team-b": 100}},
"ptu_shares must map at least one team_id to a positive whole number of PTUs",
),
(
{"ptu_shares": {"team-a": 50.5, "team-b": 49.5}},
"ptu_shares must map at least one team_id to a positive whole number of PTUs",
),
({"ptu_shares": {"team-a": True}}, "ptu_shares must map at least one team_id to a positive whole number of PTUs"),
({"ptu_shares": {"": 100}}, "ptu_shares must map at least one team_id to a positive whole number of PTUs"),
({"ptu_shares": ["team-a"]}, "ptu_shares must map at least one team_id to a positive whole number of PTUs"),
({"ptu_shares": {"team-a": 60, "team-b": 30}}, "ptu_shares must add up to ptu_count (90 of 100 allocated)"),
({"ptu_shares": {"team-a": 60, "team-b": 50}}, "ptu_shares must add up to ptu_count (110 of 100 allocated)"),
],
ids=[
"team_id and shares",
"empty shares",
"zero share",
"fractional share",
"boolean share",
"blank team",
"not a mapping",
"shares short of the count",
"shares over the count",
],
)
def test_an_incoherent_split_names_its_reason_and_reserves_nothing(override, expected):
assert ptu_config_error({**_SHARED, **override}) == expected
assert ptu_terms({**_SHARED, **override}) is None
def test_the_split_is_named_after_the_deployment_when_the_caller_supplies_one():
error = ptu_config_error({**_SHARED, "ptu_shares": {"team-a": 1}}, model_name="gpt-4.1-ptu")
assert error is not None
assert "'gpt-4.1-ptu'" in error
assert "1 of 100 allocated" in error
def test_a_split_reservation_is_zeroed_like_a_single_team_one():
zeroed = _with_flag(_SHARED)
assert zeroed["input_cost_per_token"] == 0.0
assert zeroed["output_cost_per_token"] == 0.0

View file

@ -37,11 +37,13 @@ from litellm.proxy.hooks.parallel_request_limiter_v3 import (
_PROXY_MaxParallelRequestsHandler_v3 as _PROXY_MaxParallelRequestsHandler,
)
from litellm.proxy.utils import InternalUsageCache, ProxyLogging, hash_token
from litellm.router_utils.ptu_shares import PTUTeamCeiling
from litellm.types.caching import RedisPipelineIncrementOperation
from litellm.types.llms.openai import ResponsesAPIResponse
from litellm.types.utils import (
EmbeddingResponse,
ModelResponse,
PromptTokensDetailsWrapper,
TextCompletionResponse,
Usage,
)
@ -7150,3 +7152,152 @@ async def test_success_tpm_accounting_keeps_the_admission_target_after_an_alias_
charged: Final = {op["key"]: op["increment_value"] for op in ops}
assert charged[admission_bucket] == 150 - stash.reserved_tokens
assert not any(":target-b" in key for key in charged)
# --- a team's PTU share on a shared Azure provisioned deployment ---------------------------
def _ptu_ceiling_for(team_id: str, model_group: str, tpm_limit: int, ratio: float):
calls: list[tuple[str, str]] = []
def resolve(requested_team: str, requested_group: str) -> PTUTeamCeiling | None:
calls.append((requested_team, requested_group))
if (requested_team, requested_group) != (team_id, model_group):
return None
return PTUTeamCeiling(tpm_limit=tpm_limit, output_to_input_ratio=ratio)
return resolve, calls
def _ptu_request(model: str = "test-model") -> dict:
return {"model": model, "messages": [{"role": "user", "content": "hi"}], "max_tokens": 300}
@pytest.mark.asyncio
async def test_a_teams_ptu_share_is_a_hard_tpm_ceiling_on_the_shared_model():
cache = DualCache()
resolve, _ = _ptu_ceiling_for("t", "test-model", tpm_limit=500, ratio=4.0)
handler = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(cache), ptu_team_ceiling_resolver=resolve)
key = UserAPIKeyAuth(api_key=hash_token("sk-ptu"), team_id="t")
await handler.async_pre_call_hook(user_api_key_dict=key, cache=cache, data=_ptu_request(), call_type="acompletion")
with pytest.raises(HTTPException) as exc:
await handler.async_pre_call_hook(user_api_key_dict=key, cache=cache, data=_ptu_request(), call_type="acompletion")
assert exc.value.status_code == 429
assert "model_per_team_ptu" in str(exc.value.detail)
assert exc.value.headers["rate_limit_type"] == "tokens"
@pytest.mark.asyncio
async def test_a_ptu_ceiling_on_one_model_leaves_the_teams_other_models_alone():
cache = DualCache()
resolve, calls = _ptu_ceiling_for("t", "test-model", tpm_limit=500, ratio=4.0)
handler = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(cache), ptu_team_ceiling_resolver=resolve)
key = UserAPIKeyAuth(api_key=hash_token("sk-ptu"), team_id="t")
for _ in range(3):
await handler.async_pre_call_hook(
user_api_key_dict=key, cache=cache, data=_ptu_request("other-model"), call_type="acompletion"
)
assert calls == [("t", "other-model")] * 3
@pytest.mark.asyncio
async def test_a_team_without_a_share_and_a_key_without_a_team_get_no_ptu_ceiling():
cache = DualCache()
resolve, calls = _ptu_ceiling_for("t", "test-model", tpm_limit=500, ratio=4.0)
handler = _PROXY_MaxParallelRequestsHandler(internal_usage_cache=InternalUsageCache(cache), ptu_team_ceiling_resolver=resolve)
other_team = UserAPIKeyAuth(api_key=hash_token("sk-other"), team_id="u")
no_team = UserAPIKeyAuth(api_key=hash_token("sk-no-team"))
for _ in range(3):
await handler.async_pre_call_hook(user_api_key_dict=other_team, cache=cache, data=_ptu_request(), call_type="acompletion")
await handler.async_pre_call_hook(user_api_key_dict=no_team, cache=cache, data=_ptu_request(), call_type="acompletion")
assert set(calls) == {("u", "test-model")}
def _ptu_success_kwargs() -> dict:
return {
"standard_logging_object": {"metadata": {"user_api_key_hash": hash_token("sk-ptu"), "user_api_key_team_id": "t"}},
"litellm_params": {"metadata": {"model_group": "test-model", "user_api_key_metadata": {}, "user_api_key_team_metadata": {}}},
"model": "test-model",
}
def _ptu_response(usage: Usage) -> ModelResponse:
return ModelResponse(
id="ptu-share", object="chat.completion", created=int(datetime.now().timestamp()), model="test-model", usage=usage, choices=[]
)
def _ptu_increment(handler: _PROXY_MaxParallelRequestsHandler, ops: list) -> float | None:
ptu_key = handler.create_rate_limit_keys("model_per_team_ptu", "t:test-model", "tokens")
return next((op["increment_value"] for op in ops if op["key"] == ptu_key), None)
def test_success_accounting_settles_the_ptu_counter_in_azure_normalized_tokens():
"""100 uncached input tokens plus 50 output tokens at a 4:1 output ratio are 300
normalized tokens, which is what the share was sized in."""
resolve, _ = _ptu_ceiling_for("t", "test-model", tpm_limit=500, ratio=4.0)
handler = _PROXY_MaxParallelRequestsHandler(
internal_usage_cache=InternalUsageCache(DualCache()), ptu_team_ceiling_resolver=resolve
)
response = _ptu_response(Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150))
ops = handler._build_success_event_pipeline_operations(
kwargs=_ptu_success_kwargs(), response_obj=response, rate_limit_type="output"
)
assert _ptu_increment(handler, ops) == 300
def test_cached_input_is_not_charged_to_the_ptu_counter():
resolve, _ = _ptu_ceiling_for("t", "test-model", tpm_limit=500, ratio=4.0)
handler = _PROXY_MaxParallelRequestsHandler(
internal_usage_cache=InternalUsageCache(DualCache()), ptu_team_ceiling_resolver=resolve
)
response = _ptu_response(
Usage(
prompt_tokens=100,
completion_tokens=50,
total_tokens=150,
prompt_tokens_details=PromptTokensDetailsWrapper(cached_tokens=40),
)
)
ops = handler._build_success_event_pipeline_operations(
kwargs=_ptu_success_kwargs(), response_obj=response, rate_limit_type="output"
)
assert _ptu_increment(handler, ops) == 60 + 4 * 50
def test_usage_that_only_reports_a_total_charges_that_total_to_the_ptu_counter():
resolve, _ = _ptu_ceiling_for("t", "test-model", tpm_limit=500, ratio=4.0)
handler = _PROXY_MaxParallelRequestsHandler(
internal_usage_cache=InternalUsageCache(DualCache()), ptu_team_ceiling_resolver=resolve
)
response = _ptu_response(Usage(prompt_tokens=0, completion_tokens=0, total_tokens=150))
ops = handler._build_success_event_pipeline_operations(
kwargs=_ptu_success_kwargs(), response_obj=response, rate_limit_type="output"
)
assert _ptu_increment(handler, ops) == 150
def test_success_accounting_charges_no_ptu_counter_without_a_ceiling():
resolve, _ = _ptu_ceiling_for("u", "test-model", tpm_limit=500, ratio=4.0)
handler = _PROXY_MaxParallelRequestsHandler(
internal_usage_cache=InternalUsageCache(DualCache()), ptu_team_ceiling_resolver=resolve
)
response = _ptu_response(Usage(prompt_tokens=100, completion_tokens=50, total_tokens=150))
ops = handler._build_success_event_pipeline_operations(
kwargs=_ptu_success_kwargs(), response_obj=response, rate_limit_type="output"
)
assert _ptu_increment(handler, ops) is None

View file

@ -0,0 +1,130 @@
"""Tests for attaching PTU-hours to a daily activity response."""
from typing import Final
import pytest
from litellm.litellm_core_utils.azure_ptu_capacity import PTUCapacity
from litellm.proxy.management_endpoints.ptu_consumption import attach_ptu_hours
from litellm.types.proxy.management_endpoints.common_daily_activity import (
BreakdownMetrics,
DailySpendData,
DailySpendMetadata,
KeyMetricWithMetadata,
MetricWithMetadata,
SpendAnalyticsPaginatedResponse,
SpendMetrics,
)
_ROW: Final = PTUCapacity(input_tpm_per_ptu=1_000, output_to_input_ratio=4.0)
_CACHED_ROW: Final = PTUCapacity(input_tpm_per_ptu=1_000, output_to_input_ratio=4.0, cached_input_ratio=0.1)
_CAPACITY: Final = {"gpt-4.1-ptu": _ROW, "gpt-6-ptu": _CACHED_ROW}
def _metrics(prompt: int, completion: int, cached: int = 0) -> SpendMetrics:
return SpendMetrics(
prompt_tokens=prompt,
completion_tokens=completion,
cache_read_input_tokens=cached,
total_tokens=prompt + completion,
api_requests=1,
successful_requests=1,
)
def _bucket(metrics: SpendMetrics, keys: dict[str, SpendMetrics] | None = None) -> MetricWithMetadata:
return MetricWithMetadata(
metrics=metrics,
metadata={},
api_key_breakdown={key: KeyMetricWithMetadata(metrics=m, metadata={}) for key, m in (keys or {}).items()},
)
def _day(date: str, model_groups: dict[str, MetricWithMetadata]) -> DailySpendData:
total: Final = _metrics(
sum(b.metrics.prompt_tokens for b in model_groups.values()),
sum(b.metrics.completion_tokens for b in model_groups.values()),
)
return DailySpendData(date=date, metrics=total, breakdown=BreakdownMetrics(model_groups=model_groups))
def _response(*days: DailySpendData) -> SpendAnalyticsPaginatedResponse:
return SpendAnalyticsPaginatedResponse(results=list(days), metadata=DailySpendMetadata())
def test_a_ptu_model_groups_tokens_become_ptu_hours_on_the_group_the_day_and_the_total():
"""60,000 normalized tokens on a 1,000 input-TPM row is one PTU-hour."""
day: Final = _day("2026-09-23", {"gpt-4.1-ptu": _bucket(_metrics(prompt=40_000, completion=5_000))})
attached: Final = attach_ptu_hours(_response(day), _CAPACITY.get)
group: Final = attached.results[0].breakdown.model_groups["gpt-4.1-ptu"]
assert group.metrics.ptu_hours == pytest.approx(1.0)
assert attached.results[0].metrics.ptu_hours == pytest.approx(1.0)
assert attached.metadata.total_ptu_hours == pytest.approx(1.0)
def test_a_model_group_without_a_sizing_row_keeps_zero_ptu_hours_and_the_rest_of_the_day_intact():
day: Final = _day(
"2026-09-23",
{
"gpt-4.1-ptu": _bucket(_metrics(prompt=30_000, completion=0)),
"gpt-4o-mini": _bucket(_metrics(prompt=1_000_000, completion=1_000_000)),
},
)
attached: Final = attach_ptu_hours(_response(day), _CAPACITY.get)
groups: Final = attached.results[0].breakdown.model_groups
assert groups["gpt-4o-mini"].metrics.ptu_hours == 0.0
assert groups["gpt-4o-mini"].metrics.prompt_tokens == 1_000_000
assert groups["gpt-4.1-ptu"].metrics.ptu_hours == pytest.approx(0.5)
assert attached.results[0].metrics.ptu_hours == pytest.approx(0.5)
assert attached.results[0].metrics.total_tokens == day.metrics.total_tokens
def test_cached_input_is_charged_at_the_rows_cached_ratio():
day: Final = _day("2026-09-23", {"gpt-6-ptu": _bucket(_metrics(prompt=60_000, completion=0, cached=60_000))})
attached: Final = attach_ptu_hours(_response(day), _CAPACITY.get)
assert attached.results[0].breakdown.model_groups["gpt-6-ptu"].metrics.ptu_hours == pytest.approx(0.1)
def test_each_api_key_under_a_ptu_group_gets_its_own_ptu_hours():
day: Final = _day(
"2026-09-23",
{
"gpt-4.1-ptu": _bucket(
_metrics(prompt=60_000, completion=0),
keys={"key-a": _metrics(prompt=45_000, completion=0), "key-b": _metrics(prompt=15_000, completion=0)},
)
},
)
attached: Final = attach_ptu_hours(_response(day), _CAPACITY.get)
breakdown: Final = attached.results[0].breakdown.model_groups["gpt-4.1-ptu"].api_key_breakdown
assert breakdown["key-a"].metrics.ptu_hours == pytest.approx(0.75)
assert breakdown["key-b"].metrics.ptu_hours == pytest.approx(0.25)
def test_the_total_sums_every_day_on_the_page():
first: Final = _day("2026-09-22", {"gpt-4.1-ptu": _bucket(_metrics(prompt=60_000, completion=0))})
second: Final = _day("2026-09-23", {"gpt-4.1-ptu": _bucket(_metrics(prompt=0, completion=15_000))})
attached: Final = attach_ptu_hours(_response(first, second), _CAPACITY.get)
assert [day.metrics.ptu_hours for day in attached.results] == [pytest.approx(1.0), pytest.approx(1.0)]
assert attached.metadata.total_ptu_hours == pytest.approx(2.0)
def test_a_page_with_no_ptu_group_is_returned_unchanged():
day: Final = _day("2026-09-23", {"gpt-4o-mini": _bucket(_metrics(prompt=100, completion=100))})
response: Final = _response(day)
attached: Final = attach_ptu_hours(response, _CAPACITY.get)
assert attached.results[0] is day
assert attached.metadata.total_ptu_hours == 0.0
assert response.metadata.total_ptu_hours == 0.0

View file

@ -16,6 +16,7 @@ from litellm.proxy._types import (
UserAPIKeyAuth,
)
from litellm.litellm_core_utils.llm_cost_calc.utils import generic_cost_per_token
from litellm.litellm_core_utils.ptu_pricing import ptu_terms
from litellm.proxy.auth.auth_checks import _is_model_cost_zero
from litellm.llms.gemini.cost_calculator import cost_per_web_search_request
from litellm.proxy.management_endpoints.model_management_endpoints import (
@ -1284,3 +1285,48 @@ class TestPtuDeploymentsAreNotBilledPerToken:
assert "input_cost_per_token" in str(exc.value)
add_team_model_to_db.assert_not_called()
_SHARED_START = "2026-08-01T00:00:00Z"
def test_validate_helper_accepts_shares_in_place_of_a_team_id():
shared = {
"ptu_count": 5,
"cost_per_ptu_per_hour": 2.0,
"ptu_effective_from": _SHARED_START,
"ptu_shares": {"team-a": 3, "team-b": 2},
}
_validate_ptu_model_info(shared)
assert ptu_terms(shared) is not None
def test_validate_helper_refuses_shares_that_do_not_add_up_to_the_count():
with pytest.raises(HTTPException) as exc:
_validate_ptu_model_info(
{
"ptu_count": 5,
"cost_per_ptu_per_hour": 2.0,
"ptu_effective_from": _SHARED_START,
"ptu_shares": {"team-a": 3, "team-b": 1},
}
)
assert exc.value.status_code == 400
assert "4 of 5 allocated" in exc.value.detail
def test_validate_helper_refuses_a_team_id_beside_shares():
with pytest.raises(HTTPException) as exc:
_validate_ptu_model_info(
{
"team_id": "team-a",
"ptu_count": 5,
"cost_per_ptu_per_hour": 2.0,
"ptu_effective_from": _SHARED_START,
"ptu_shares": {"team-a": 5},
}
)
assert exc.value.status_code == 400
assert "cannot both be set" in exc.value.detail

View file

@ -16889,3 +16889,90 @@ def test_list_team_v2_answers_503_no_db_connection_when_the_callers_user_read_hi
assert response.status_code == 503, response.text
assert response.json() == _DB_OUTAGE_503_BODY
# --- PTU-equivalent consumption on /team/daily/activity ------------------------------------
def _ptu_activity_page():
from litellm.types.proxy.management_endpoints.common_daily_activity import (
BreakdownMetrics,
DailySpendData,
DailySpendMetadata,
MetricWithMetadata,
SpendAnalyticsPaginatedResponse,
SpendMetrics,
)
metrics = SpendMetrics(
prompt_tokens=180_000, completion_tokens=0, total_tokens=180_000, api_requests=3, successful_requests=3
)
return SpendAnalyticsPaginatedResponse(
results=[
DailySpendData(
date=datetime(2026, 9, 23).date(),
metrics=metrics,
breakdown=BreakdownMetrics(
model_groups={"gpt-4.1-ptu": MetricWithMetadata(metrics=metrics, metadata={}, api_key_breakdown={})}
),
)
],
metadata=DailySpendMetadata(total_tokens=180_000, total_api_requests=3, total_successful_requests=3),
)
def _shared_ptu_router() -> Router:
return Router(
model_list=[
{
"model_name": "gpt-4.1-ptu",
"litellm_params": {"model": "azure/gpt-4.1", "api_key": "sk-ptu", "api_base": "https://ptu.example"},
"model_info": {
"id": "shared-ptu",
"base_model": "azure/gpt-4.1",
"ptu_count": 50,
"cost_per_ptu_per_hour": 1.0,
"ptu_effective_from": "2026-01-01T00:00:00Z",
"ptu_shares": {"team-a": 30, "team-b": 20},
},
}
]
)
@pytest.mark.asyncio
@pytest.mark.parametrize("attribution_enabled", [True, False])
async def test_team_daily_activity_reports_ptu_hours_only_while_attribution_is_on(
mock_db_client, mock_admin_auth, monkeypatch, attribution_enabled
):
"""One PTU serves 3,000 input tokens per minute on gpt-4.1, so 180,000 uncached
input tokens are one PTU-hour; the figure appears beside tokens only once the
PTU flag is on, and the token totals are untouched either way."""
from litellm.proxy.management_endpoints.team_endpoints import get_team_daily_activity
monkeypatch.setenv("LITELLM_ENABLE_PTU_COST_ATTRIBUTION", "True" if attribution_enabled else "False")
mock_db_client.db.litellm_teamtable.find_many = AsyncMock(return_value=[])
page = _ptu_activity_page()
with (
patch("litellm.proxy.management_endpoints.team_endpoints.get_daily_activity", AsyncMock(return_value=page)),
patch("litellm.proxy.proxy_server.llm_router", _shared_ptu_router()),
):
result = await get_team_daily_activity(
team_ids="team-a",
start_date="2026-09-23",
end_date="2026-09-24",
model=None,
api_key=None,
page=1,
page_size=10,
exclude_team_ids=None,
user_api_key_dict=mock_admin_auth,
)
expected_ptu_hours = 1.0 if attribution_enabled else 0.0
assert result.metadata.total_ptu_hours == expected_ptu_hours
assert result.results[0].metrics.ptu_hours == expected_ptu_hours
assert result.results[0].breakdown.model_groups["gpt-4.1-ptu"].metrics.ptu_hours == expected_ptu_hours
assert result.metadata.total_tokens == 180_000
assert result.results[0].metrics.total_tokens == 180_000

View file

@ -15,7 +15,7 @@ from litellm.proxy.spend_tracking.ptu_flat_cost_rollup import (
PTUModel,
_active_hours_on_day,
_compute_daily_flat_cost,
_parse_ptu_model,
_parse_ptu_models,
run_ptu_flat_cost_backfill,
run_ptu_flat_cost_rollup,
run_scheduled_ptu_rollup,
@ -41,6 +41,14 @@ def _ptu_enabled(monkeypatch):
_VALID_PTU = {"ptu_count": 5, "cost_per_ptu_per_hour": 2.0, "team_id": "t"}
def _parse_ptu_model(row):
"""The single holding a team_id deployment parses to, else None; shared deployments
parse to one holding per team and have their own tests below."""
parsed = _parse_ptu_models(row)
assert len(parsed) <= 1
return parsed[0] if parsed else None
def _model_row(model_id="m1", model_name="gpt-4o-mini-ptu", model_info=None, with_start=True):
row = MagicMock()
row.model_id = model_id
@ -2082,3 +2090,51 @@ async def test_the_catch_up_pass_reaches_a_config_declared_deployment():
assert len(charged) == 3, charged
assert charged[-1] == yesterday
assert all(row["ptu_flat_cost"] == pytest.approx(48.0) for row in table.rows.values())
# --- a deployment split into per-team PTU shares -----------------------------------------
_SHARED_PTU = {"ptu_count": 50, "cost_per_ptu_per_hour": 1.0, "ptu_shares": {"team-b": 20, "team-a": 30}}
def test_a_shared_deployment_parses_to_one_holding_per_team_carrying_its_share():
parsed = _parse_ptu_models(_model_row(model_info=dict(_SHARED_PTU)))
assert [(m.team_id, m.ptu_count) for m in parsed] == [("team-a", 30), ("team-b", 20)]
assert {m.model_id for m in parsed} == {"m1"}
assert {m.model_name for m in parsed} == {"gpt-4o-mini-ptu"}
assert {m.cost_per_ptu_per_hour for m in parsed} == {1.0}
def test_a_shared_deployment_whose_shares_do_not_add_up_is_not_priced():
assert _parse_ptu_models(_model_row(model_info={**_SHARED_PTU, "ptu_shares": {"team-a": 30}})) == ()
@pytest.mark.asyncio
async def test_rollup_splits_a_shared_deployments_flat_cost_by_share():
"""50 PTUs at $1/hour for a day is $1,200; team-a's 30 PTUs are $720 of it and team-b's
20 are $480, each keyed on the same deployment id and public name."""
prisma, table = _prisma_with_models([_model_row(model_info=dict(_SHARED_PTU))])
result = await run_ptu_flat_cost_rollup(prisma, target_date=DAY)
assert result.rows_written == 2
created = {call.kwargs["data"]["create"]["team_id"]: call.kwargs["data"]["create"] for call in table.upsert.await_args_list}
assert created["team-a"]["ptu_flat_cost"] == pytest.approx(720.0)
assert created["team-b"]["ptu_flat_cost"] == pytest.approx(480.0)
assert sum(row["ptu_flat_cost"] for row in created.values()) == pytest.approx(50 * 1.0 * 24)
assert {row["model"] for row in created.values()} == {"m1"}
assert {row["model_group"] for row in created.values()} == {"gpt-4o-mini-ptu"}
assert {row["api_key"] for row in created.values()} == {PTU_SENTINEL_API_KEY}
@pytest.mark.asyncio
async def test_a_shared_deployment_with_a_closed_window_is_reported_lapsed_once():
closed = {**_SHARED_PTU, "ptu_effective_from": "2020-01-01T00:00:00Z", "ptu_effective_to": "2020-02-01T00:00:00Z"}
prisma, _ = _prisma_with_models([_model_row(model_id="dep-shared", model_info=closed)])
alert = AsyncMock()
result = await run_scheduled_ptu_rollup(prisma, target_date=DAY, alert=alert)
assert result.lapsed == ("gpt-4o-mini-ptu",)
alert.assert_awaited_once()

View file

@ -0,0 +1,115 @@
"""Tests for per-team PTU shares: who a shared deployment is served to and what a share is worth."""
from typing import Final
from litellm.litellm_core_utils.azure_ptu_capacity import AZURE_PTU_CAPACITY
from litellm.router_utils.ptu_shares import (
PTUTeamCeiling,
filter_ptu_shared_deployments,
model_group_ptu_capacity,
ptu_capacity_warning,
team_ptu_ceiling,
)
_GPT41: Final = AZURE_PTU_CAPACITY["gpt-4.1"]
_GPT4O: Final = AZURE_PTU_CAPACITY["gpt-4o"]
_SHARES: Final = {"team-a": 30, "team-b": 20}
def _shared(model: str = "azure/gpt-4.1", shares: object = _SHARES, deployment_id: str = "shared") -> dict:
return {
"model_name": "gpt-4.1-ptu",
"litellm_params": {"model": model},
"model_info": {
"id": deployment_id,
"ptu_count": 50,
"cost_per_ptu_per_hour": 1.0,
"ptu_effective_from": "2026-01-01T00:00:00Z",
"ptu_shares": shares,
},
}
def _single_team(model: str = "azure/gpt-4.1") -> dict:
return {
"model_name": "gpt-4.1-ptu",
"litellm_params": {"model": model},
"model_info": {
"id": "single",
"team_id": "team-a",
"ptu_count": 50,
"cost_per_ptu_per_hour": 1.0,
"ptu_effective_from": "2026-01-01T00:00:00Z",
},
}
_OPEN: Final = {"model_name": "gpt-4.1-ptu", "litellm_params": {"model": "azure/gpt-4.1"}, "model_info": {"id": "open"}}
def test_a_team_holding_a_share_keeps_the_shared_deployment():
result: Final = filter_ptu_shared_deployments([_shared(), _OPEN], "team-a")
assert [d["model_info"]["id"] for d in result.deployments] == ["shared", "open"]
assert result.withheld is False
def test_a_team_without_a_share_only_sees_the_unshared_deployments():
result: Final = filter_ptu_shared_deployments([_shared(), _OPEN], "team-c")
assert [d["model_info"]["id"] for d in result.deployments] == ["open"]
assert result.withheld is True
def test_a_caller_with_no_team_holds_no_share():
for team_id in (None, ""):
result = filter_ptu_shared_deployments([_shared()], team_id)
assert result.deployments == ()
assert result.withheld is True
def test_a_single_team_deployment_and_a_malformed_share_map_are_not_filtered_here():
"""A ``team_id`` deployment is scoped by the router's team filter, and an unusable
``ptu_shares`` is refused at registration, so neither is withheld by the share filter."""
result: Final = filter_ptu_shared_deployments([_single_team(), _shared(shares={"team-a": 0})], "team-z")
assert [d["model_info"]["id"] for d in result.deployments] == ["single", "shared"]
assert result.withheld is False
def test_a_share_converts_to_the_models_input_tpm_per_ptu():
ceiling: Final = team_ptu_ceiling([_shared()], "team-a")
assert ceiling == PTUTeamCeiling(
tpm_limit=30 * _GPT41.input_tpm_per_ptu, output_to_input_ratio=_GPT41.output_to_input_ratio
)
def test_shares_across_deployments_add_up_and_the_larger_output_ratio_wins():
gpt4o: Final = _shared(model="azure/gpt-4o", shares={"team-a": 10}, deployment_id="shared-4o")
ceiling: Final = team_ptu_ceiling([_shared(), gpt4o, _OPEN], "team-a")
assert ceiling is not None
assert ceiling.tpm_limit == 30 * _GPT41.input_tpm_per_ptu + 10 * _GPT4O.input_tpm_per_ptu
assert ceiling.output_to_input_ratio == max(_GPT41.output_to_input_ratio, _GPT4O.output_to_input_ratio)
def test_no_share_or_no_sizing_row_sets_no_ceiling():
assert team_ptu_ceiling([_shared()], "team-c") is None
assert team_ptu_ceiling([_shared(model="azure/unknown-deployment")], "team-a") is None
assert team_ptu_ceiling([_single_team(), _OPEN], "team-a") is None
def test_a_groups_capacity_comes_from_its_first_reserved_deployment_with_a_row():
assert model_group_ptu_capacity([_OPEN, _single_team()]) is _GPT41
assert model_group_ptu_capacity([_shared(model="azure/unknown"), _shared(model="azure/gpt-4o")]) is _GPT4O
assert model_group_ptu_capacity([_OPEN]) is None
assert model_group_ptu_capacity([]) is None
def test_a_reserved_deployment_without_a_sizing_row_is_warned_about_by_name():
warning: Final = ptu_capacity_warning("gpt-4.1-ptu", _shared(model="azure/my-ptu-deployment"))
assert warning is not None
assert "gpt-4.1-ptu" in warning
assert "base_model" in warning
def test_a_sized_reservation_and_an_unreserved_deployment_raise_no_warning():
assert ptu_capacity_warning("gpt-4.1-ptu", _shared()) is None
assert ptu_capacity_warning("gpt-4.1-ptu", _single_team()) is None
assert ptu_capacity_warning("gpt-4.1-ptu", {**_OPEN, "litellm_params": {"model": "azure/my-ptu-deployment"}}) is None

View file

@ -17829,3 +17829,72 @@ def test_access_windows_filter_reserved_deployments_method():
request_team_id="team-a",
)
] == ["reserved-deployment", "open-deployment"]
def _shared_ptu_model_list() -> list:
return [
{
"model_name": "gpt-4.1-ptu",
"litellm_params": {"model": "gpt-4.1", "mock_response": "shared"},
"model_info": {
"id": "shared-deployment",
"base_model": "azure/gpt-4.1",
"ptu_count": 50,
"cost_per_ptu_per_hour": 1.0,
"ptu_effective_from": "2026-01-01T00:00:00Z",
"ptu_shares": {"team-a": 30, "team-b": 20},
},
},
{
"model_name": "gpt-4.1-ptu",
"litellm_params": {"model": "gpt-4.1", "mock_response": "open"},
"model_info": {"id": "open-deployment"},
},
]
def test_ptu_shares_hide_the_shared_deployment_from_teams_holding_no_share(monkeypatch):
monkeypatch.setenv("LITELLM_ENABLE_PTU_COST_ATTRIBUTION", "True")
router = Router(model_list=_shared_ptu_model_list())
for request_kwargs in ({"metadata": {"user_api_key_team_id": "team-c"}}, {"metadata": {}}, {}):
_, deployments = router._common_checks_available_deployment(model="gpt-4.1-ptu", request_kwargs=request_kwargs)
assert [d["model_info"]["id"] for d in deployments] == ["open-deployment"]
for team_id in ("team-a", "team-b"):
_, deployments = router._common_checks_available_deployment(
model="gpt-4.1-ptu",
request_kwargs={"metadata": {"user_api_key_team_id": team_id}},
)
assert {d["model_info"]["id"] for d in deployments} == {"shared-deployment", "open-deployment"}
def test_ptu_shares_raise_when_only_shared_deployments_remain(monkeypatch):
monkeypatch.setenv("LITELLM_ENABLE_PTU_COST_ATTRIBUTION", "True")
router = Router(model_list=_shared_ptu_model_list()[:1])
with pytest.raises(litellm.BadRequestError, match="reserved for the teams holding a PTU share"):
router._common_checks_available_deployment(
model="gpt-4.1-ptu",
request_kwargs={"metadata": {"user_api_key_team_id": "team-c"}},
)
_, deployments = router._common_checks_available_deployment(
model="gpt-4.1-ptu",
request_kwargs={"metadata": {"user_api_key_team_id": "team-b"}},
)
assert [d["model_info"]["id"] for d in deployments] == ["shared-deployment"]
def test_ptu_shares_do_not_filter_while_the_feature_is_off(monkeypatch):
monkeypatch.delenv("LITELLM_ENABLE_PTU_COST_ATTRIBUTION", raising=False)
router = Router(model_list=_shared_ptu_model_list())
_, deployments = router._common_checks_available_deployment(
model="gpt-4.1-ptu",
request_kwargs={"metadata": {"user_api_key_team_id": "team-c"}},
)
assert {d["model_info"]["id"] for d in deployments} == {"shared-deployment", "open-deployment"}
def test_a_shared_ptu_deployment_whose_shares_do_not_add_up_is_refused_at_registration(monkeypatch):
monkeypatch.setenv("LITELLM_ENABLE_PTU_COST_ATTRIBUTION", "True")
model_list = _shared_ptu_model_list()[:1]
model_list[0]["model_info"]["ptu_shares"] = {"team-a": 30}
with pytest.raises(ValueError, match=r"gpt-4\.1-ptu.*30 of 50 allocated"):
Router(model_list=model_list)

View file

@ -9,7 +9,13 @@ import {
type ExtendedDailyData,
type ProviderSpendRow,
} from "./entityUsageAggregations";
import { buildCostBreakdownTiles, buildSummaryTiles, hasFlatCost, type SummaryTile } from "./entityUsageSummary";
import {
buildCostBreakdownTiles,
buildSummaryTiles,
hasFlatCost,
hasPtuHours,
type SummaryTile,
} from "./entityUsageSummary";
import { MoneyCell } from "@/components/shared/table_cells";
import { Card as ShadcnCard, CardContent, CardHeader, CardTitle } from "@/components/ui/card";
import { hasCapability, type Capability } from "@/utils/capabilities";
@ -67,6 +73,7 @@ interface EntitySpendData {
metadata: {
total_spend: number;
total_flat_cost?: number;
total_ptu_hours?: number;
total_api_requests: number;
total_successful_requests: number;
total_failed_requests: number;
@ -284,6 +291,7 @@ const EntityUsage: React.FC<EntityUsageProps> = ({
const capitalizedEntityLabel = entityType.charAt(0).toUpperCase() + entityType.slice(1);
const showFlatCost = entityType === "team" && hasFlatCost(spendData.metadata);
const showPtuHours = entityType === "team" && hasPtuHours(spendData.metadata);
const userSpendTeamIds = useMemo(
() =>
selectedTags.length > 0
@ -392,7 +400,7 @@ const EntityUsage: React.FC<EntityUsageProps> = ({
);
const breakdownTiles = showFlatCost && showCostBreakdown ? buildCostBreakdownTiles(spendData.metadata) : [];
const summaryTiles = [...buildSummaryTiles(spendData.metadata, showFlatCost), ...breakdownTiles];
const summaryTiles = [...buildSummaryTiles(spendData.metadata, showFlatCost, showPtuHours), ...breakdownTiles];
const modelViewTitle = modelViewType === "groups" ? "Top Public Model Names" : "Top Litellm Models";
@ -402,7 +410,9 @@ const EntityUsage: React.FC<EntityUsageProps> = ({
<ShadcnCard>
<CardContent>
<h3 className="text-lg font-medium text-foreground">{capitalizedEntityLabel} Spend Overview</h3>
<div className="grid grid-cols-5 gap-4 mt-4">{summaryTiles.map(renderSummaryTile)}</div>
<div className={`grid ${showPtuHours ? "grid-cols-6" : "grid-cols-5"} gap-4 mt-4`}>
{summaryTiles.map(renderSummaryTile)}
</div>
</CardContent>
</ShadcnCard>
</div>

View file

@ -1,5 +1,5 @@
import { describe, expect, it } from "vitest";
import { buildCostBreakdownTiles, buildSummaryTiles, hasFlatCost } from "./entityUsageSummary";
import { buildCostBreakdownTiles, buildSummaryTiles, hasFlatCost, hasPtuHours } from "./entityUsageSummary";
const metadata = {
total_spend: 100,
@ -10,6 +10,17 @@ const metadata = {
total_tokens: 3456,
};
describe("hasPtuHours", () => {
it("is false when no PTU hours were consumed or the backend predates the field", () => {
expect(hasPtuHours({ ...metadata, total_ptu_hours: 0 })).toBe(false);
expect(hasPtuHours(metadata)).toBe(false);
});
it("is true once a PTU model group has been used", () => {
expect(hasPtuHours({ ...metadata, total_ptu_hours: 0.25 })).toBe(true);
});
});
describe("hasFlatCost", () => {
it("is false when there is no flat cost to report", () => {
expect(hasFlatCost({ ...metadata, total_flat_cost: 0 })).toBe(false);
@ -28,6 +39,23 @@ describe("buildSummaryTiles", () => {
expect(buildSummaryTiles(metadata, true)).toHaveLength(5);
});
it("adds a sixth PTU Hours tile only when asked, after Total Tokens", () => {
const withPtu = { ...metadata, total_ptu_hours: 1234.5678 };
expect(buildSummaryTiles(withPtu, true)).toHaveLength(5);
const tiles = buildSummaryTiles(withPtu, true, true);
expect(tiles).toHaveLength(6);
const last = tiles[tiles.length - 1];
expect(last.title).toBe("PTU Hours");
expect(last.value).toBe("1,234.57");
expect(last.tooltip).toContain("PTU-hours");
expect(tiles[tiles.length - 2].title).toBe("Total Tokens");
});
it("treats missing PTU hours as zero when the tile is forced on", () => {
const tiles = buildSummaryTiles(metadata, false, true);
expect(tiles[tiles.length - 1].value).toBe("0.00");
});
it("shows request-only spend under the original title when there is no flat cost", () => {
const [first] = buildSummaryTiles(metadata, false);
expect(first.title).toBe("Total Spend");

View file

@ -11,6 +11,7 @@ export interface SummaryTile {
interface SpendSummaryMetadata {
total_spend: number;
total_flat_cost?: number;
total_ptu_hours?: number;
total_api_requests: number;
total_successful_requests: number;
total_failed_requests: number;
@ -26,10 +27,28 @@ export const REQUEST_COST_TOOLTIP =
export const FLAT_COST_TOOLTIP =
"Reserved provisioned throughput, billed per hour whether or not requests are sent. Reported here only; it does not count toward team, key, user, or organization budgets.";
export const PTU_HOURS_TOOLTIP =
"Provisioned throughput consumed, in PTU-hours: the tokens sent to PTU deployments converted through the Azure sizing table for each model. Compare it with the PTU-hours reserved to see how much of the reservation was used.";
export const hasFlatCost = (metadata: SpendSummaryMetadata): boolean => (metadata.total_flat_cost ?? 0) > 0;
export const buildSummaryTiles = (metadata: SpendSummaryMetadata, showFlatCost: boolean): SummaryTile[] => {
export const hasPtuHours = (metadata: SpendSummaryMetadata): boolean => (metadata.total_ptu_hours ?? 0) > 0;
export const buildSummaryTiles = (
metadata: SpendSummaryMetadata,
showFlatCost: boolean,
showPtuHours: boolean = false,
): SummaryTile[] => {
const flatCost = metadata.total_flat_cost ?? 0;
const ptuHoursTile: SummaryTile[] = showPtuHours
? [
{
title: "PTU Hours",
value: formatNumberWithCommas(metadata.total_ptu_hours ?? 0, 2),
tooltip: PTU_HOURS_TOOLTIP,
},
]
: [];
return [
showFlatCost
? {
@ -47,6 +66,7 @@ export const buildSummaryTiles = (metadata: SpendSummaryMetadata, showFlatCost:
},
{ title: "Failed Requests", value: metadata.total_failed_requests.toLocaleString(), className: "text-destructive" },
{ title: "Total Tokens", value: metadata.total_tokens.toLocaleString() },
...ptuHoursTile,
];
};

View file

@ -43,6 +43,7 @@ describe("sumMetadata", () => {
total_cache_read_input_tokens: 1,
total_cache_creation_input_tokens: 1,
total_flat_cost: 1,
total_ptu_hours: 1,
total_response_time_ms: 1,
total_timed_requests: 1,
};

View file

@ -30,6 +30,7 @@ const SUMMABLE_METADATA_KEYS = [
"total_cache_read_input_tokens",
"total_cache_creation_input_tokens",
"total_flat_cost",
"total_ptu_hours",
"total_response_time_ms",
"total_timed_requests",
] as const;

View file

@ -10,6 +10,7 @@ export interface EntitySpendData {
metadata: {
total_spend: number;
total_flat_cost?: number;
total_ptu_hours?: number;
total_api_requests: number;
total_successful_requests: number;
total_failed_requests: number;
@ -41,6 +42,7 @@ export interface ExportMetadata {
total_spend: number;
total_flat_cost?: number;
total_cost?: number;
total_ptu_hours?: number;
total_requests: number;
successful_requests: number;
failed_requests: number;

View file

@ -2100,6 +2100,23 @@ describe("EntityUsageExport utils", () => {
expect(result.summary.total_flat_cost).toBeUndefined();
expect(result.summary.total_cost).toBeUndefined();
});
it("should carry total_ptu_hours into the summary once a PTU model group was used", () => {
const withPtu: EntitySpendData = {
...mockSpendData,
metadata: { ...mockSpendData.metadata, total_ptu_hours: 2.5 },
};
const result = generateMetadata("team", mockDateRange, [], "daily", withPtu);
expect(result.summary.total_ptu_hours).toBeCloseTo(2.5, 4);
});
it("should omit total_ptu_hours when no PTU hours were consumed", () => {
const zeroPtu = { ...mockSpendData, metadata: { ...mockSpendData.metadata, total_ptu_hours: 0 } };
expect(generateMetadata("team", mockDateRange, [], "daily", zeroPtu).summary.total_ptu_hours).toBeUndefined();
expect(
generateMetadata("team", mockDateRange, [], "daily", mockSpendData).summary.total_ptu_hours,
).toBeUndefined();
});
});
describe("generateDailyData PTU flat cost", () => {

View file

@ -436,6 +436,9 @@ export const generateMetadata = (
summary.total_flat_cost = flatCost;
summary.total_cost = spendData.metadata.total_spend + flatCost;
}
if ((spendData.metadata.total_ptu_hours ?? 0) > 0) {
summary.total_ptu_hours = spendData.metadata.total_ptu_hours;
}
return {
export_date: new Date().toISOString(),
entity_type: entityType,

View file

@ -1,6 +1,7 @@
export interface SpendMetrics {
spend: number;
flat_cost?: number;
ptu_hours?: number;
prompt_tokens: number;
completion_tokens: number;
total_tokens: number;

View file

@ -29311,6 +29311,11 @@ export interface components {
* @default 0
*/
total_prompt_tokens: number;
/**
* Total Ptu Hours
* @default 0
*/
total_ptu_hours: number;
/**
* Total Response Time Ms
* @default 0
@ -42229,6 +42234,11 @@ export interface components {
* @default 0
*/
prompt_tokens: number;
/**
* Ptu Hours
* @default 0
*/
ptu_hours: number;
/**
* Spend
* @default 0
@ -46156,6 +46166,10 @@ export interface components {
ptu_effective_from?: string | null;
/** Ptu Effective To */
ptu_effective_to?: string | null;
/** Ptu Shares */
ptu_shares?: {
[key: string]: number;
} | null;
/** Team Id */
team_id?: string | null;
/** Team Public Model Name */