Spaces:
Running
Running
| """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, | |
| } | |