AI API 속도 제한: 작동 원리와 대응 방법

모든 AI API에는 속도 제한이 있습니다. 개발 중에 걸리면 짜증나고, 운영 중에 걸리면 사용자에게 오류가 발생합니다. 공급자별 속도 제한 작동 방식을 이해하고 적절한 재시도 로직을 구축하는 것이 데모와 실제 운영 애플리케이션의 차이입니다.

공급자별 제한 방식

OpenAI

OpenAI는 계정 사용 이력과 결제 수준에 따라 단계별 속도 제한을 적용합니다.

단계	RPM (요청/분)	TPM (토큰/분)	도달 조건
무료	3	40,000	신규 계정
1단계	500	200,000	5달러 결제
2단계	5,000	2,000,000	50달러 결제
3단계	5,000	10,000,000	100달러 결제
4단계	10,000	50,000,000	250달러 결제
5단계	10,000	300,000,000	1,000달러 결제

제한은 모델별로 적용됩니다. GPT-4.1을 사용해도 GPT-4.1-mini 할당량은 소모되지 않습니다.

Anthropic

Anthropic도 RPM과 TPM 제한이 있는 유사한 단계 시스템을 사용하며, 낮은 단계에서는 일일 토큰 제한도 적용합니다.

Google (Gemini)

Google AI Studio는 모델별 제한이 있습니다. 무료 단계는 일일 요청 수는 관대하지만 분당 요청 수는 엄격합니다(예: Gemini 2.5 Flash 무료 단계 15 RPM).

통합 플랫폼

LemonData, OpenRouter 같은 통합 플랫폼은 상위 공급자 제한 위에 자체 속도 제한 계층을 추가합니다. LemonData는 역할 기반 제한을 사용합니다:

역할	RPM
사용자	1,000
파트너	3,000
VIP	10,000

장점: 통합 플랫폼 제한은 개별 공급자 무료 단계보다 보통 높으며, 다중 채널 라우팅 덕분에 한 상위 채널이 제한되면 요청이 다른 채널로 전환됩니다.

속도 제한 헤더 읽기

주요 공급자들은 응답 헤더에 속도 제한 정보를 제공합니다:

x-ratelimit-limit-requests: 500
x-ratelimit-remaining-requests: 499
x-ratelimit-reset-requests: 60s
x-ratelimit-limit-tokens: 200000
x-ratelimit-remaining-tokens: 199500

이 헤더를 적극적으로 활용하세요. 429 오류가 발생할 때까지 기다리지 말고 미리 속도를 조절하세요.

재시도 로직 구축

잘못된 방법

# 이렇게 하지 마세요
import time

def call_api(messages):
    while True:
        try:
            return client.chat.completions.create(
                model="gpt-4.1",
                messages=messages
            )
        except Exception:
            time.sleep(1)  # 고정 지연, 백오프 없음, 모든 예외 포착

문제점: 지수 백오프 없음, 재시도 불가 오류도 포착, 최대 재시도 제한 없음, 지터 없음.

올바른 방법

import time
import random
from openai import RateLimitError, APIError, APIConnectionError

def call_with_retry(messages, model="gpt-4.1", max_retries=3):
    """지수 백오프와 지터를 적용한 재시도"""
    for attempt in range(max_retries + 1):
        try:
            return client.chat.completions.create(
                model=model,
                messages=messages
            )
        except RateLimitError as e:
            if attempt == max_retries:
                raise
            # 가능하면 응답의 retry_after 사용
            wait = getattr(e, 'retry_after', None)
            if wait is None:
                wait = (2 ** attempt) + random.uniform(0, 1)
            print(f"속도 제한 발생. {wait:.1f}초 대기 (시도 {attempt + 1})")
            time.sleep(wait)
        except APIConnectionError:
            if attempt == max_retries:
                raise
            wait = (2 ** attempt) + random.uniform(0, 1)
            time.sleep(wait)
        except APIError as e:
            # 클라이언트 오류(400, 401, 403)는 재시도하지 않음
            if e.status_code and 400 <= e.status_code < 500:
                raise
            if attempt == max_retries:
                raise
            time.sleep((2 ** attempt) + random.uniform(0, 1))

핵심 원칙:

지수 백오프: 1초, 2초, 4초, 8초
지터: 0~1초 랜덤 추가로 동시 재시도 방지
제공 시 retry_after 헤더 존중
클라이언트 오류(잘못된 요청, 인증 실패)는 재시도하지 않음
최대 재시도 횟수 설정

비동기 버전

import asyncio
import random
from openai import AsyncOpenAI, RateLimitError

async_client = AsyncOpenAI(
    api_key="sk-lemon-xxx",
    base_url="https://api.lemondata.cc/v1"
)

async def call_with_retry_async(messages, model="gpt-4.1", max_retries=3):
    for attempt in range(max_retries + 1):
        try:
            return await async_client.chat.completions.create(
                model=model,
                messages=messages
            )
        except RateLimitError:
            if attempt == max_retries:
                raise
            wait = (2 ** attempt) + random.uniform(0, 1)
            await asyncio.sleep(wait)

고급: 토큰 버킷 속도 제한기

고속 처리 애플리케이션에서는 클라이언트 측 속도 제한을 구현해 서버 제한에 걸리지 않도록 하세요:

import time
import asyncio

class TokenBucket:
    def __init__(self, rate: float, capacity: int):
        self.rate = rate          # 초당 토큰 수
        self.capacity = capacity  # 최대 버스트 크기
        self.tokens = capacity
        self.last_refill = time.monotonic()

    async def acquire(self, tokens: int = 1):
        while True:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(
                self.capacity,
                self.tokens + elapsed * self.rate
            )
            self.last_refill = now

            if self.tokens >= tokens:
                self.tokens -= tokens
                return
            # 충분한 토큰이 생길 때까지 대기
            wait = (tokens - self.tokens) / self.rate
            await asyncio.sleep(wait)

# 분당 500 요청 = 초당 약 8.3 요청
limiter = TokenBucket(rate=8.0, capacity=20)

async def rate_limited_call(messages, model="gpt-4.1"):
    await limiter.acquire()
    return await async_client.chat.completions.create(
        model=model,
        messages=messages
    )

속도 제한 시 모델 대체

주요 모델이 속도 제한에 걸리면 대체 모델로 전환하세요:

FALLBACK_CHAIN = [
    "claude-sonnet-4-6",
    "gpt-4.1",
    "gpt-4.1-mini",
]

async def call_with_fallback(messages):
    for model in FALLBACK_CHAIN:
        try:
            return await async_client.chat.completions.create(
                model=model,
                messages=messages
            )
        except RateLimitError:
            continue
    raise Exception("모든 모델이 속도 제한에 걸림")

이 점에서 API 통합기가 빛을 발합니다. 300개 이상의 모델이 하나의 엔드포인트 뒤에 있으니 항상 대체 모델을 사용할 수 있습니다.

속도 제한 사용량 모니터링

사용량을 추적해 문제가 사용자에게 영향을 미치기 전에 감지하세요:

import logging

def log_rate_limits(response):
    headers = response.headers
    remaining = headers.get("x-ratelimit-remaining-requests")
    limit = headers.get("x-ratelimit-limit-requests")
    if remaining and int(remaining) < int(limit) * 0.1:
        logging.warning(
            f"속도 제한 경고: {remaining}/{limit} 요청 남음"
        )

남은 용량이 10% 미만일 때 알림을 설정하세요. 이렇게 하면 사용자에게 429 오류가 발생하기 전에 속도 조절을 할 시간이 확보됩니다.

요약

전략	사용 시기
지수 백오프	항상 (기본)
클라이언트 측 속도 제한기	고속 처리 앱 (>100 RPM)
모델 대체	SLA가 필요한 운영 앱
사전 모니터링	모든 운영 배포
배치 API	실시간이 아닌 작업

목표는 속도 제한을 완전히 피하는 것이 아니라, 사용자에게 전혀 티 나지 않게 우아하게 처리하는 것입니다.

탄탄한 AI 애플리케이션 구축: lemondata.cc는 상위 공급자 속도 제한을 자동으로 처리하는 다중 채널 라우팅을 제공합니다. 하나의 API 키로 300개 이상의 모델 사용 가능.

AI API 속도 제한: 작동 원리와 대처 방법