AI Agents · 강의

요청 제한 및 재시도 로직

지수 백오프, 429 처리, API를 예의 있게 사용하는 방법을 다룹니다.

요청 제한 및 재시도 로직은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 4번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

요청 제한이란 무엇인가요

요청 제한은 API가 과도한 요청으로 압도되지 않도록 스스로를 보호하는 방식입니다. 에이전트가 너무 짧은 시간에 요청을 많이 보내면 API는 429 Too Many Requests를 반환합니다. 일반적인 제한으로는 초당, 분당 또는 일일 요청 수 제한이 있습니다.

요청 제한을 무시하면 에이전트가 차단되고, API 키가 폐기되며, 추가 비용이 발생할 수 있습니다.

import requests

response = requests.get(
    'https://api.example.com/data',
    headers={'Authorization': 'Bearer YOUR_KEY'}
)

if response.status_code == 429:
    print('Rate limit exceeded!')
    # Check headers for limit details
    limit = response.headers.get('X-RateLimit-Limit')
    remaining = response.headers.get('X-RateLimit-Remaining')
    reset = response.headers.get('X-RateLimit-Reset')
    print(f'Limit: {limit}, Remaining: {remaining}, Reset: {reset}')

재시도 대기 시간 헤더

API가 429를 반환할 때는 재시도하기 전에 몇 초를 기다려야 하는지 정확히 알려 주는 Retry-After 헤더가 포함되는 경우가 많습니다. 이 헤더를 항상 준수하세요. 무시하고 즉시 재시도하면 또다시 429를 받게 됩니다.

import requests
import time

def request_with_retry_after(url, headers):
    response = requests.get(url, headers=headers)

    if response.status_code == 429:
        retry_after = int(response.headers.get('Retry-After', 60))
        print(f'Rate limited. Waiting {retry_after} seconds...')
        time.sleep(retry_after)

        # Retry once after waiting
        response = requests.get(url, headers=headers)

    response.raise_for_status()
    return response.json()

지수 백오프

지수 백오프는 실패할 때마다 더 오래 기다리는 표준 재시도 전략입니다. 첫 번째 시도에서 2초, 두 번째 시도에서 4초, 세 번째 시도에서 8초를 기다리는 식입니다. 이렇게 하면 서버 부하를 점진적으로 줄이고 서버가 복구할 시간을 확보할 수 있습니다.

공식: wait = 2 ** attempt

import requests
import time

def get_with_exponential_backoff(url, headers, max_retries=5):
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers, timeout=(5, 30))

        if response.status_code == 200:
            return response.json()

        if response.status_code in (429, 500, 502, 503):
            wait = 2 ** attempt  # 1, 2, 4, 8, 16 seconds
            print(f'Attempt {attempt+1} failed ({response.status_code}). '
                  f'Waiting {wait}s before retry...')
            time.sleep(wait)
        else:
            response.raise_for_status()  # non-retryable error

    raise Exception(f'Failed after {max_retries} retries')

백오프에 무작위 지연 추가하기

많은 에이전트가 동시에 재시도하면(짧은 장애 후 흔히 발생하는 상황) 모두 동시에 깨어나 대규모 동시 재시도가 발생하고, 서버가 다시 즉시 요청 제한을 적용하게 됩니다. 무작위 지연을 추가하면 재시도 시점이 분산되어 서버 부하를 줄일 수 있습니다.

import requests
import time
import random

def get_with_jittered_backoff(url, headers, max_retries=5):
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers, timeout=(5, 30))

        if response.status_code == 200:
            return response.json()

        if response.status_code in (429, 500, 502, 503):
            base_wait = 2 ** attempt
            # Add random jitter: actual wait is 50%-100% of base
            jitter = random.uniform(0.5, 1.0)
            wait = base_wait * jitter
            print(f'Waiting {wait:.1f}s (attempt {attempt+1})')
            time.sleep(wait)
        else:
            response.raise_for_status()

    raise Exception(f'Failed after {max_retries} retries')

테너시티 라이브러리

테너시티는 재시도 로직에 가장 널리 사용되는 Python 라이브러리입니다. 깔끔한 데코레이터 문법으로 지수 백오프, 무작위 지연, 최대 재시도 횟수, 사용자 지정 중단 조건을 처리합니다. 직접 작성한 재시도 반복문보다 훨씬 안정적입니다.

from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_exception_type, before_sleep_log
)
import requests
import logging

logger = logging.getLogger(__name__)

@retry(
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=2, max=60),
    retry=retry_if_exception_type(requests.exceptions.HTTPError),
    before_sleep=before_sleep_log(logger, logging.WARNING)
)
def fetch_data(url, headers):
    response = requests.get(url, headers=headers, timeout=(5, 30))
    if response.status_code == 429:
        response.raise_for_status()  # triggers retry
    response.raise_for_status()
    return response.json()

사용자 지정 재시도 조건을 적용한 테너시티

테너시티가 특정 상태 코드(예: 429 및 5xx)에서만 재시도하고, 재시도해도 도움이 되지 않는 클라이언트 오류(4xx)에서는 즉시 중단하도록 설정할 수 있습니다. 응답을 검사하려면 retry_if_result 또는 사용자 지정 호출 가능 함수를 사용하세요.

from tenacity import (
    retry, stop_after_attempt, wait_exponential,
    retry_if_result
)
import requests

def is_retryable_response(response):
    return response.status_code in (429, 500, 502, 503, 504)

@retry(
    stop=stop_after_attempt(4),
    wait=wait_exponential(multiplier=2, min=2, max=30),
    retry=retry_if_result(is_retryable_response)
)
def resilient_get(url, headers):
    response = requests.get(url, headers=headers, timeout=(5, 30))
    return response  # retry logic inspects the response object

# Usage
response = resilient_get(
    'https://api.example.com/data',
    {'Authorization': 'Bearer YOUR_KEY'}
)
data = response.json()

요청 제한 사전 관리

가장 좋은 전략은 처음부터 요청 제한에 도달하지 않는 것입니다. 모든 응답에서 요청 제한 헤더를 확인하고, 제한에 가까워지면 속도를 낮추세요. 많은 API가 X-RateLimit-Remaining 및 X-RateLimit-Reset 헤더를 제공합니다.

import requests
import time

class RateLimitAwareClient:
    def __init__(self, base_url, api_key):
        self.base_url = base_url
        self.headers = {'Authorization': f'Bearer {api_key}'}
        self.remaining = 1000  # assume generous limit

    def get(self, path):
        # Proactively slow down if nearly exhausted
        if self.remaining < 10:
            print('Rate limit nearly exhausted, sleeping 5s...')
            time.sleep(5)

        response = requests.get(
            f'{self.base_url}{path}', headers=self.headers
        )

        # Update remaining from response headers
        remaining_str = response.headers.get('X-RateLimit-Remaining')
        if remaining_str:
            self.remaining = int(remaining_str)

        response.raise_for_status()
        return response.json()

최대 재시도 횟수와 포기하기

재시도 로직에는 항상 제한이 있어야 합니다. 무한히 재시도하면 모든 에이전트가 재시도 반복문에 갇히는 연쇄 장애가 발생할 수 있습니다. max_retries에 도달하면 무엇이 실패했는지에 대한 맥락을 담은 최종 예외를 발생시키세요. 그러면 에이전트가 이를 기록하고 다른 작업으로 넘어갈 수 있습니다.

import requests
import time

class MaxRetriesExceeded(Exception):
    def __init__(self, url, attempts, last_status):
        self.url = url
        self.attempts = attempts
        self.last_status = last_status
        super().__init__(
            f'Failed {url} after {attempts} attempts '
            f'(last status: {last_status})'
        )

def fetch_with_limit(url, headers, max_retries=3):
    last_response = None
    for attempt in range(max_retries):
        last_response = requests.get(url, headers=headers)
        if last_response.status_code == 200:
            return last_response.json()
        time.sleep(2 ** attempt)
    raise MaxRetriesExceeded(url, max_retries, last_response.status_code)

회로 차단기 패턴

회로 차단기 패턴은 에이전트가 실패하는 서비스에 계속 요청을 보내는 것을 막습니다. 실패 횟수가 임계값에 도달하면 회로가 "열리고", 네트워크에 연결하지 않은 채 모든 요청이 즉시 실패합니다. 냉각 기간이 지나면 요청을 한 번 시도합니다. 성공하면 회로가 닫히고 정상 작업이 재개됩니다.

import time

class CircuitBreaker:
    CLOSED, OPEN, HALF_OPEN = 'closed', 'open', 'half_open'

    def __init__(self, failure_threshold=5, recovery_timeout=60):
        self.state = self.CLOSED
        self.failures = 0
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.opened_at = None

    def call(self, func, *args, **kwargs):
        if self.state == self.OPEN:
            if time.time() - self.opened_at > self.recovery_timeout:
                self.state = self.HALF_OPEN
            else:
                raise Exception('Circuit OPEN — service unavailable')
        try:
            result = func(*args, **kwargs)
            self.failures = 0
            self.state = self.CLOSED
            return result
        except Exception as e:
            self.failures += 1
            if self.failures >= self.failure_threshold:
                self.state = self.OPEN
                self.opened_at = time.time()
                print(f'Circuit OPENED after {self.failures} failures')
            raise

# --- demo ---
def flaky():
    raise ValueError('upstream 500')

def works():
    return 'ok'

cb = CircuitBreaker(failure_threshold=3, recovery_timeout=60)
for i in range(3):
    try:
        cb.call(flaky)
    except Exception as e:
        print(f'call {i+1} failed: {e}')
print(f'Breaker state after 3 failures: {cb.state}')
try:
    cb.call(flaky)
except Exception as e:
    print(f'Rejected without calling flaky(): {e}')

제한을 지키기 위한 요청 대기열 사용하기

한 번에 많은 호출을 하는 에이전트에서는 제한을 지키기 위해 토큰 버킷이나 간단한 sleep 기반 속도 제한을 사용하세요. API의 요청 제한에 따라 호출 사이의 안전한 간격을 계산하면 됩니다(예: 분당 60회 = 초당 1회).

import requests
import time

def batch_requests(urls, headers, calls_per_minute=60):
    interval = 60.0 / calls_per_minute  # seconds between calls
    results = []

    for i, url in enumerate(urls):
        start = time.time()

        response = requests.get(url, headers=headers, timeout=(5, 30))
        response.raise_for_status()
        results.append(response.json())

        print(f'Processed {i+1}/{len(urls)}')

        # Sleep for remaining time in the interval
        elapsed = time.time() - start
        sleep_time = interval - elapsed
        if sleep_time > 0:
            time.sleep(sleep_time)

    return results

재시도 로직과 백오프 헤더 결합하기

가장 견고한 패턴은 서버가 지정한 대기 시간(Retry-After)과 대체 수단으로 사용하는 지수 백오프를 결합하는 것입니다. 서버의 안내를 제공받을 수 있다면 항상 우선하세요. 서버가 언제 다시 재시도할 수 있는지 가장 정확하게 알고 있기 때문입니다.

import requests
import time
import random

def smart_retry(url, headers, max_retries=5):
    for attempt in range(max_retries):
        response = requests.get(url, headers=headers, timeout=(5, 30))

        if response.status_code == 200:
            return response.json()

        if response.status_code == 429:
            # Use Retry-After if provided, else exponential backoff
            retry_after = response.headers.get('Retry-After')
            if retry_after:
                wait = int(retry_after)
            else:
                wait = (2 ** attempt) + random.uniform(0, 1)
            print(f'429 rate limit. Waiting {wait:.1f}s...')
            time.sleep(wait)

        elif response.status_code >= 500:
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f'Server error {response.status_code}. Waiting {wait:.1f}s...')
            time.sleep(wait)

        else:
            response.raise_for_status()  # non-retryable

    raise Exception(f'Gave up after {max_retries} attempts')

빠른 확인: 지수 백오프

재시도 전략에 대한 이해도를 확인해 보세요.

요청 제한 및 재시도 복습

이제 에이전트가 요청 제한을 안정적으로 처리할 수 있습니다.

429 요청이 너무 많음 — Retry-After 헤더를 준수하고 재시도하기 전에 기다리세요
지수 백오프 — wait = 2^attempt는 재시도할 때마다 대기 시간을 두 배로 늘립니다
무작위 지연 — 여러 에이전트 인스턴스에 재시도를 분산하도록 무작위성을 추가합니다
테너시티 — 데코레이터와 깔끔한 설정으로 모든 재시도 로직을 처리합니다
회로 차단기 — 임계값에 도달하면 실패하는 서비스에 계속 요청을 보내는 것을 중단합니다
사전 속도 제한 — X-RateLimit-Remaining을 확인하고 제한에 도달하기 전에 속도를 낮춥니다

자주 묻는 질문

“요청 제한 및 재시도 로직” 강의는 무료인가요?

네 — “요청 제한 및 재시도 로직” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

“요청 제한 및 재시도 로직”에서 뭘 배우나요?

지수 백오프, 429 처리, API를 예의 있게 사용하는 방법을 다룹니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

AI Agents을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 4번째 강의입니다.

“요청 제한 및 재시도 로직” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

이 강의의 모든 강의

← AI Agents(으)로 돌아가기