본문 바로가기
C.W.K.
Stream
Lesson 05 of 07 · published

Cost & Rate Limit Management — per-tenant ledger

~22 min · cost, rate-limits, budgets

Level 0Tokenizer
0 XP0/54 lessons0/10 achievements
0/120 XP to next level120 XP to go0% complete

운영을 시작하면 곧 '비용이 왜 이렇게 늘었지?'와 '어느 customer가 만들었지?'라는 질문을 받게 돼. 답하려면 tenant별 ledger가 필요해. 호출마다 tenant_id, model, prompt_tokens, completion_tokens, reasoning_tokens, timestamp를 저장하고 일별·tenant별로 합산해.

전체 비용만으로는 원인을 알 수 없어

총합만 보면 특정 customer가 비용의 80%를 썼는지 알 수 없어. tenant별 breakdown이 없으면 청구액 원인을 추측하게 돼. 호출 시점에 기록하는 건 저렴하지만 지나간 세부값을 나중에 되살릴 수는 없어.

rate limit header로 동시 요청 수를 조절해

응답마다 x-ratelimit-remaining-requests를 읽어. 20% 아래면 동시 요청 수를 절반으로 줄이고 80% 위로 회복하면 늘려. 429에 부딪힌 뒤가 아니라 남은 양을 보고 먼저 움직여.

tenant별 예산 상한을 둬

ledger가 있으면 customer별 월 예산을 적용해 한도를 넘는 호출을 거부하거나 더 저렴한 모델로 내릴 수 있어. 상한이 없으면 한 customer의 버그가 전체 비용을 키울 수 있어.

Code

Per-tenant budget tracker·python
from dataclasses import dataclass

PRICING_PER_1M = {
    "gpt-5.4": {"prompt": 2.50, "completion": 15.00},
    "gpt-4.1": {"prompt": 2.00, "completion": 8.00},
    "gpt-4o-mini": {"prompt": 0.15, "completion": 0.60},
}

@dataclass
class CostTracker:
    total_cost_usd: float = 0.0
    total_prompt_tokens: int = 0
    total_completion_tokens: int = 0
    request_count: int = 0

    def record(self, model: str, usage: dict) -> float:
        p = usage.get("prompt_tokens", 0)
        c = usage.get("completion_tokens", 0)
        pricing = PRICING_PER_1M.get(model, {"prompt": 0, "completion": 0})
        cost = p * pricing["prompt"] / 1_000_000 + c * pricing["completion"] / 1_000_000
        self.total_cost_usd += cost
        self.request_count += 1
        return cost
Header 로 adaptive concurrency·python
import asyncio, random

async def retry_with_backoff(func, max_retries=6, initial_delay=1.0):
    delay = initial_delay
    for attempt in range(max_retries + 1):
        try:
            return await func()
        except httpx.HTTPStatusError as e:
            if e.response.status_code not in (429, 500, 502, 503, 504):
                raise
            if attempt == max_retries: raise
            # Honor Retry-After header
            retry_after = e.response.headers.get("Retry-After")
            sleep_time = float(retry_after) if retry_after else delay * (0.5 + random.random())
            await asyncio.sleep(min(sleep_time, 60.0))
            delay = min(delay * 2, 60.0)

External links

Exercise

API 호출마다 tenant_id, model, in_tokens, out_tokens, usd를 기록하는 CostLedger를 만들어. 일별 tenant report를 생성하고 tenant별 상한을 적용할 위치에 guard를 추가해.

Progress

Progress is local-only — sign in to sync across devices.
이 페이지에서 버그를 발견하셨거나 피드백이 있으세요?문제 신고

댓글 0

🔔 답글 알림 (로그인 필요)
로그인댓글을 남기려면 로그인해 주세요.

아직 댓글이 없어요. 첫 댓글을 남겨보세요.