OpenAI API 429 폭탄 대응 실전 가이드 지수 백오프 큐잉 토큰 버짓으로 비용과 지연을 함께 줄이기

import time
from openai import OpenAI

client = OpenAI()

def call(prompt: str) -> str:
    for _ in range(10):
        try:
            r = client.responses.create(
                model="gpt-4.1-mini",
                input=prompt,
            )
            return r.output_text
        except Exception:
            time.sleep(0.1)
    raise RuntimeError("failed")

import random
import time
from typing import Optional

from openai import OpenAI
from openai import RateLimitError, APIError, APITimeoutError

client = OpenAI()


def _sleep_seconds(attempt: int, retry_after: Optional[float] = None) -> float:
    # 서버가 Retry-After를 줬다면 우선
    if retry_after is not None:
        return max(0.0, retry_after)

    # Full jitter exponential backoff
    cap = 10.0
    base = 0.5
    exp = min(cap, base * (2 ** attempt))
    return random.uniform(0, exp)


def call_with_backoff(prompt: str, *, max_wait_s: float = 20.0) -> str:
    start = time.time()
    attempt = 0

    while True:
        try:
            r = client.responses.create(
                model="gpt-4.1-mini",
                input=prompt,
                # 출력 상한은 비용/TPM에도 직결
                max_output_tokens=300,
            )
            return r.output_text

        except RateLimitError as e:
            # 일부 SDK/환경에서는 headers 접근이 다를 수 있음
            retry_after = None
            try:
                retry_after = float(e.response.headers.get("retry-after"))
            except Exception:
                pass

            sleep_s = _sleep_seconds(attempt, retry_after)

        except (APITimeoutError, APIError) as e:
            # 5xx/timeout은 짧게 재시도하되 지터 적용
            sleep_s = _sleep_seconds(attempt)

        except Exception:
            # 영구 오류(인증/파라미터 등) 가능성이 높으니 즉시 실패
            raise

        elapsed = time.time() - start
        if elapsed + sleep_s > max_wait_s:
            raise TimeoutError(f"retry budget exceeded after {elapsed:.2f}s")

        time.sleep(sleep_s)
        attempt += 1

import asyncio
from dataclasses import dataclass
from openai import OpenAI

client = OpenAI()

@dataclass
class Job:
    job_id: str
    prompt: str
    max_output_tokens: int

async def worker(name: str, q: asyncio.Queue, sem: asyncio.Semaphore):
    while True:
        job = await q.get()
        try:
            async with sem:
                # 동시 호출 수 제한
                r = await asyncio.to_thread(
                    client.responses.create,
                    model="gpt-4.1-mini",
                    input=job.prompt,
                    max_output_tokens=job.max_output_tokens,
                )
                text = r.output_text
                # TODO: 결과 저장/콜백
        finally:
            q.task_done()

async def main():
    q: asyncio.Queue[Job] = asyncio.Queue(maxsize=1000)
    sem = asyncio.Semaphore(5)  # 동시성 상한

    # worker 5개
    for i in range(5):
        asyncio.create_task(worker(f"w{i}", q, sem))

    # producer 예시
    for n in range(100):
        await q.put(Job(str(n), f"summarize: doc {n}", 200))

    await q.join()

asyncio.run(main())

import time
import threading

class TokenBucket:
    def __init__(self, tokens_per_minute: int):
        self.capacity = tokens_per_minute
        self.tokens = tokens_per_minute
        self.refill_rate_per_sec = tokens_per_minute / 60.0
        self.last = time.time()
        self.lock = threading.Lock()

    def _refill(self):
        now = time.time()
        delta = now - self.last
        self.tokens = min(self.capacity, self.tokens + delta * self.refill_rate_per_sec)
        self.last = now

    def acquire(self, cost: int):
        # cost만큼 토큰이 찰 때까지 대기
        while True:
            with self.lock:
                self._refill()
                if self.tokens >= cost:
                    self.tokens -= cost
                    return
                need = cost - self.tokens
                wait_s = need / self.refill_rate_per_sec
            time.sleep(min(wait_s, 1.0))


def estimate_tokens(text: str) -> int:
    # 매우 단순한 휴리스틱(운영에서는 토크나이저 권장)
    return max(1, len(text) // 4)


def guarded_call(prompt: str, max_output_tokens: int, bucket: TokenBucket) -> str:
    in_tok = estimate_tokens(prompt)
    cost = in_tok + max_output_tokens
    bucket.acquire(cost)

    from openai import OpenAI
    client = OpenAI()
    r = client.responses.create(
        model="gpt-4.1-mini",
        input=prompt,
        max_output_tokens=max_output_tokens,
    )
    return r.output_text