134 lines
5 KiB
Python
134 lines
5 KiB
Python
"""AI usage cost verification reports.
|
|
|
|
The admin API already owns collection. This module turns that existing usage
|
|
shape into a deterministic model/provider cost report for ops evidence without
|
|
adding any enforcement policy.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any, Mapping
|
|
|
|
|
|
REPORT_SCHEMA = "vignette.ai_usage_model_cost_report.v1"
|
|
|
|
|
|
def _number(value: Any, default: float = 0.0) -> float:
|
|
try:
|
|
return float(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def _integer(value: Any, default: int = 0) -> int:
|
|
try:
|
|
return int(value)
|
|
except (TypeError, ValueError):
|
|
return default
|
|
|
|
|
|
def _ratio(part: float, whole: float) -> float:
|
|
if whole <= 0:
|
|
return 0.0
|
|
return round(part / whole, 6)
|
|
|
|
|
|
def _cost_per_1k_tokens(cost_usd: float, tokens: int) -> float | None:
|
|
if tokens <= 0:
|
|
return None
|
|
return round(cost_usd / tokens * 1000.0, 6)
|
|
|
|
|
|
def _cost_per_turn(cost_usd: float, turns: int) -> float | None:
|
|
if turns <= 0:
|
|
return None
|
|
return round(cost_usd / turns, 6)
|
|
|
|
|
|
def build_model_cost_report(usage: Mapping[str, Any]) -> dict[str, Any]:
|
|
"""Build an ops report from an AdminUsageResponse-like mapping."""
|
|
total_cost = round(_number(usage.get("cost_usd")), 6)
|
|
total_turns = _integer(usage.get("total_turns"))
|
|
metered_turns = _integer(usage.get("metered_turns"))
|
|
tokens_in = _integer(usage.get("tokens_in"))
|
|
tokens_out = _integer(usage.get("tokens_out"))
|
|
total_tokens = tokens_in + tokens_out
|
|
by_provider = list(usage.get("by_provider") or [])
|
|
budget = dict(usage.get("budget") or {})
|
|
evaluator_cache = dict(usage.get("evaluator_cache") or {})
|
|
|
|
models: list[dict[str, Any]] = []
|
|
for item in by_provider:
|
|
row = dict(item or {})
|
|
turns = _integer(row.get("turns"))
|
|
row_tokens_in = _integer(row.get("tokens_in"))
|
|
row_tokens_out = _integer(row.get("tokens_out"))
|
|
row_tokens = row_tokens_in + row_tokens_out
|
|
row_cost = round(_number(row.get("cost_usd")), 6)
|
|
models.append(
|
|
{
|
|
"provider": str(row.get("provider") or "unknown"),
|
|
"model": str(row.get("model") or "unknown"),
|
|
"turns": turns,
|
|
"tokens_in": row_tokens_in,
|
|
"tokens_out": row_tokens_out,
|
|
"tokens_total": row_tokens,
|
|
"cost_usd": row_cost,
|
|
"cost_share": _ratio(row_cost, total_cost),
|
|
"token_share": _ratio(float(row_tokens), float(total_tokens)),
|
|
"cost_per_turn_usd": _cost_per_turn(row_cost, turns),
|
|
"cost_per_1k_tokens_usd": _cost_per_1k_tokens(row_cost, row_tokens),
|
|
}
|
|
)
|
|
models.sort(key=lambda item: (-float(item["cost_usd"]), item["provider"], item["model"]))
|
|
|
|
warnings: list[str] = []
|
|
if total_turns > 0 and metered_turns < total_turns:
|
|
warnings.append("partial_metering")
|
|
if total_cost == 0 and metered_turns > 0:
|
|
warnings.append("zero_cost_metered_usage")
|
|
if str(budget.get("status") or "") in {"warn", "exceeded"}:
|
|
warnings.append(f"budget_{budget.get('status')}")
|
|
cache_hit_rate = _number(evaluator_cache.get("hit_rate"))
|
|
if bool(evaluator_cache.get("enabled")) and _integer(evaluator_cache.get("requests")) > 0:
|
|
if cache_hit_rate < 0.25:
|
|
warnings.append("low_evaluator_cache_hit_rate")
|
|
if models and models[0]["cost_share"] >= 0.8:
|
|
warnings.append("dominant_model_cost")
|
|
|
|
return {
|
|
"schema": REPORT_SCHEMA,
|
|
"source": str(usage.get("source") or "unknown"),
|
|
"durable": bool(usage.get("durable")),
|
|
"window_days": _integer(usage.get("window_days")),
|
|
"summary": {
|
|
"total_turns": total_turns,
|
|
"metered_turns": metered_turns,
|
|
"metered_coverage": _ratio(float(metered_turns), float(total_turns)),
|
|
"tokens_in": tokens_in,
|
|
"tokens_out": tokens_out,
|
|
"tokens_total": total_tokens,
|
|
"cost_usd": total_cost,
|
|
"cost_per_turn_usd": _cost_per_turn(total_cost, metered_turns),
|
|
"cost_per_1k_tokens_usd": _cost_per_1k_tokens(total_cost, total_tokens),
|
|
},
|
|
"budget": {
|
|
"status": str(budget.get("status") or "disabled"),
|
|
"limit_usd": round(_number(budget.get("limit_usd")), 6),
|
|
"used_ratio": round(_number(budget.get("used_ratio")), 6),
|
|
"remaining_usd": round(_number(budget.get("remaining_usd")), 6),
|
|
},
|
|
"evaluator_cache": {
|
|
"enabled": bool(evaluator_cache.get("enabled")),
|
|
"requests": _integer(evaluator_cache.get("requests")),
|
|
"hits": _integer(evaluator_cache.get("hits")),
|
|
"misses": _integer(evaluator_cache.get("misses")),
|
|
"hit_rate": round(cache_hit_rate, 6),
|
|
},
|
|
"models": models,
|
|
"top_cost_model": models[0] if models else None,
|
|
"warnings": warnings,
|
|
}
|
|
|
|
|
|
__all__ = ["REPORT_SCHEMA", "build_model_cost_report"]
|