Разработка биллинг-системы для AI-as-a-Service (учёт токенов, GPU-часов, запросов)

Разработка биллинг-системы для AI-as-a-Service (токены, GPU-часы, запросы)

Биллинг-система для AI SaaS должна точно учитывать несколько разнородных ресурсов: LLM токены, GPU-время, API запросы, хранилище. Каждый ресурс имеет свою единицу измерения, цену и логику подсчёта.

Модели тарификации и их комбинирование

from dataclasses import dataclass
from enum import Enum
from decimal import Decimal

class BillingUnit(Enum):
    TOKEN = "token"           # 1M токенов
    GPU_SECOND = "gpu_second" # GPU-время
    REQUEST = "request"       # Запрос
    GB_MONTH = "gb_month"     # Хранилище

@dataclass
class PricingRule:
    resource: BillingUnit
    unit_price: Decimal
    tier_breaks: list  # [(volume_threshold, discounted_price)]
    minimum_charge: Decimal = Decimal('0')

PRICING = {
    "llm_input_token": PricingRule(
        resource=BillingUnit.TOKEN,
        unit_price=Decimal('0.000005'),  # $5 per 1M tokens
        tier_breaks=[
            (10_000_000, Decimal('0.000004')),   # >10M → $4/1M
            (100_000_000, Decimal('0.000003')),  # >100M → $3/1M
        ]
    ),
    "gpu_a100_second": PricingRule(
        resource=BillingUnit.GPU_SECOND,
        unit_price=Decimal('0.00089'),   # ~$3.20/hour per A100
        tier_breaks=[],
        minimum_charge=Decimal('0.01')   # Минимум 10 секунд биллинга
    ),
}

Real-time Usage Metering

class UsageMeter:
    def __init__(self, redis_client, kafka_producer):
        self.redis = redis_client
        self.kafka = kafka_producer

    async def record_llm_usage(self, customer_id: str, model_id: str,
                                input_tokens: int, output_tokens: int,
                                request_id: str):
        # 1. Real-time счётчики в Redis (для rate limiting и balance checks)
        pipe = self.redis.pipeline()
        month_key = f"usage:{customer_id}:{self.current_month()}"
        pipe.hincrby(month_key, f"{model_id}:input_tokens", input_tokens)
        pipe.hincrby(month_key, f"{model_id}:output_tokens", output_tokens)
        pipe.expire(month_key, 60 * 60 * 24 * 40)  # 40 дней
        await pipe.execute()

        # 2. Детальные события в Kafka для аудита и биллинга
        await self.kafka.send("usage_events", {
            "event_type": "llm_inference",
            "customer_id": customer_id,
            "model_id": model_id,
            "request_id": request_id,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "timestamp": datetime.utcnow().isoformat()
        })

    async def record_gpu_job(self, customer_id: str, job_id: str,
                              gpu_type: str, duration_seconds: float):
        # GPU биллинг с округлением вверх до секунды
        billable_seconds = max(ceil(duration_seconds), 10)  # Минимум 10 сек

        await self.kafka.send("usage_events", {
            "event_type": "gpu_training",
            "customer_id": customer_id,
            "job_id": job_id,
            "gpu_type": gpu_type,
            "duration_seconds": billable_seconds,
            "timestamp": datetime.utcnow().isoformat()
        })

Invoice генерация

class InvoiceGenerator:
    async def generate_monthly_invoice(self, customer_id: str,
                                        billing_period: str) -> Invoice:
        # Агрегация usage событий из ClickHouse
        usage = await self.clickhouse.query("""
            SELECT
                model_id,
                sum(input_tokens) as total_input,
                sum(output_tokens) as total_output,
                sum(gpu_seconds) as total_gpu,
                count() as total_requests
            FROM usage_events
            WHERE customer_id = %(customer_id)s
              AND toYYYYMM(timestamp) = %(period)s
            GROUP BY model_id
        """, {"customer_id": customer_id, "period": billing_period})

        line_items = []
        total = Decimal('0')

        for row in usage:
            input_cost = self.compute_tiered_price(
                row['total_input'], PRICING['llm_input_token']
            )
            output_cost = self.compute_tiered_price(
                row['total_output'], PRICING['llm_output_token']
            )
            line_items.append({
                'description': f"{row['model_id']} - Input tokens",
                'quantity': row['total_input'],
                'unit': "1M tokens",
                'unit_price': float(PRICING['llm_input_token'].unit_price * 1_000_000),
                'amount': float(input_cost)
            })
            total += input_cost + output_cost

        # Применение кредитов и скидок
        credits = await self.get_customer_credits(customer_id)
        final_amount = max(total - credits, Decimal('0'))

        invoice = Invoice(
            customer_id=customer_id,
            period=billing_period,
            line_items=line_items,
            subtotal=float(total),
            credits_applied=float(min(credits, total)),
            total=float(final_amount)
        )

        # Отправка через Stripe
        if final_amount > 0:
            await self.stripe.create_invoice(customer_id, invoice)

        return invoice

Алерты и контроль бюджета

async def check_budget_alerts(customer_id: str):
    customer = await db.get_customer(customer_id)
    current_spend = await usage_meter.get_current_month_spend(customer_id)

    thresholds = [0.5, 0.8, 0.9, 1.0]  # % от месячного бюджета
    for threshold in thresholds:
        alert_amount = customer.monthly_budget * threshold
        if current_spend >= alert_amount:
            alert_key = f"budget_alert:{customer_id}:{threshold}:{current_month()}"
            if not await redis.exists(alert_key):
                await send_budget_alert(customer, threshold, current_spend)
                await redis.setex(alert_key, 86400 * 31, "1")

Правильная биллинг-система — это не только техническая задача, но и бизнес-процесс: прозрачность для клиентов, точность учёта и простота оспаривания счетов снижают churn и повышают доверие к платформе.