Розробка білінг-системи для 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 і підвищують довіру до платформи.