Spaces:
Running
Running
File size: 1,666 Bytes
bd44418 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 | """Pricing post-processor computing costs per model and totals."""
from config import settings
from ..context import RunContext
class PricingProcessor:
name = "pricing"
def process(self, ctx: RunContext) -> None:
currency = getattr(settings, "currency", "EUR")
pricing_cfg = getattr(settings, "pricing", {})
by_model = {}
total_cost = 0.0
for model_id, usage in ctx.usage_by_model.items():
cfg = pricing_cfg.get(model_id) or {}
# Prefer per-million token pricing; fallback to per-1k if provided
in_rate_1m = cfg.get("input_per_1m")
out_rate_1m = cfg.get("output_per_1m")
if in_rate_1m is None and cfg.get("input_per_1k") is not None:
in_rate_1m = float(cfg.get("input_per_1k")) * 1000.0
if out_rate_1m is None and cfg.get("output_per_1k") is not None:
out_rate_1m = float(cfg.get("output_per_1k")) * 1000.0
in_rate_1m = float(in_rate_1m or 0)
out_rate_1m = float(out_rate_1m or 0)
cost_in = (usage.get("input_tokens", 0) / 1_000_000.0) * in_rate_1m
cost_out = (usage.get("output_tokens", 0) / 1_000_000.0) * out_rate_1m
model_total = cost_in + cost_out
by_model[model_id] = {
"input": round(cost_in, 6),
"output": round(cost_out, 6),
"total": round(model_total, 6),
}
total_cost += model_total
ctx.metadata_out["pricing"] = {
"currency": currency,
"total_cost": round(total_cost, 6),
"by_model": by_model,
}
|