Asyncio Queues를 사용하여 AI API Rate Limit 제어하기

지난달, 저는 API Rate Limit (속도 제한) 문제 때문에 사흘 동안 머리를 싸매고 고생했습니다.

저는 수백 개의 텍스트 입력을 AI API를 통해 처리해야 하는 작은 웹 앱을 만들고 있었습니다. 감성 분석(Sentiment Analysis)이나 요약(Summarization)처럼 프로토타입을 마법처럼 보이게 만드는 작업들이었죠. 문제는 무엇이었을까요? 테스트 데이터를 확장할 때마다 제 앱은 마치 퍼레이드의 꽃가루처럼 429 에러를 쏟아내기 시작했습니다.

익숙한 상황인가요?

제가 정확히 무엇을 시도했는지, 무엇이 실패했는지, 그리고 마침내 성공한 접근 방식이 무엇인지 단계별로 설명해 드리겠습니다. 군더더기나 "이 기묘한 비법 하나면 됩니다" 같은 소리는 없습니다. 오직 정직한 코드와 트레이드오프(Trade-offs)만 다룹니다.

진짜 문제

저에게는 다음과 같은 간단한 파이프라인(Pipeline)이 있었습니다:

1. CSV에서 텍스트 목록 읽기
2. 각 텍스트에 대해 AI API 엔드포인트(Endpoint) 호출
3. 결과를 데이터베이스에 저장

API는 요청당 약 500ms로 빨랐습니다. 하지만 초당 10개의 요청이라는 Rate Limit (속도 제한)이 있었습니다. 저는 asyncio.gather()를 사용하여 요청을 병렬로 날리면 제한 범위 내에 머물 수 있을 것이라고 순진하게 가정했습니다. 스포일러를 하자면, 그렇지 못했습니다.

스크립트를 실행하자마자 첫 10개의 요청 이후로 429 에러가 폭발적으로 발생했습니다. 저의 재시도 로직(Retry logic)은 실패 시 단순히 time.sleep(1)을 사용하는 원시적인 방식이었고, 이로 인해 전체 프로세스가 매우 느려졌습니다. 백오프(Backoff)가 불안정하고 예측 불가능했다는 점은 말할 것도 없었죠.

제가 시도했던 것들 (그리고 실패한 이유)

시도 1: 요청 사이의 고정 지연 시간 (Fixed Delay)

# 나쁜 아이디어: 비동기 루프 내에서 블로킹(Blocking) sleep 사용
async def process(texts):
    for text in texts:
...

이 방식은 API 지연 시간(Latency)이 급증하거나 실수로 두 개의 인스턴스를 실행할 때까지는 작동합니다. 게다가 낭비적입니다. API가 더 많은 트래픽을 처리할 수 있는 상황에서도 유휴 상태(Idle)로 머물게 됩니다.

시도 2: asyncio.Semaphore

# 더 나은 방법이지만 여전히 취약함
sem = asyncio.Semaphore(5)

...

이것은 동시성(Concurrency)을 제한하지만, '초당' Rate Limit을 준수하지는 않습니다. 만약 각 요청이 200ms가 걸린다면, 5개의 동시 작업자(Workers)를 통해 1초에 쉽게 25개의 요청을 몰아서 보낼 수 있습니다. 여전히 Rate Limit에 걸리게 됩니다.

시도 3: 재시도를 포함한 단순 지수 백오프 (Simple Exponential Backoff)

async def call_with_retry(text, max_retries=5):
    for attempt in range(max_retries):
        try:
...

이 방식은 가끔 발생하는 429 에러에는 효과적이지만, 모든 요청에서 제한(limit)에 걸리는 상황에서는 대부분의 시간을 대기(sleep)하며 보내게 됩니다. 이는 좋지 않은 방식입니다.

최종 해결책: 토큰 버킷 (Token Bucket)을 사용한 Async Queue

저에게는 두 가지가 필요했습니다:

1. 초당 최대 요청 수를 강제하는 Rate Limiter (속도 제한기)
2. 메인 파이프라인을 차단하지 않으면서, 지수 백오프 (Exponential Backoff)를 적용해 실패한 요청을 다시 큐에 넣는 Retry Queue (재시도 큐)

해결책은 토큰 버킷 (Token Bucket) 알고리즘을 사용하는 커스텀 RateLimiter와 재시도를 위한 asyncio.Queue를 결합하는 것이었습니다.

핵심 코드는 다음과 같습니다 (명확성을 위해 단순화되었습니다):

import asyncio
import time
from collections import deque
...

핵심 아이디어:

• **토큰 버킷 (Token Bucket)**은 많은 수의 동시 작업자 (Concurrent Workers)가 있더라도 속도 제한을 초과하지 않도록 보장합니다. acquire() 메서드는 토큰을 사용할 수 있을 때까지 차단(block)됩니다.
• **재시도 큐 (Retry Queue)**는 재시도 로직을 메인 흐름에서 분리합니다. 실패한 요청은 계산된 지연 시간(지수 백오프)과 함께 큐에 다시 삽입됩니다.
• **세마포어 (Semaphore)**는 수천 개의 태스크가 생성되지 않도록 동시성 (Concurrency)을 제한합니다.

교훈 및 트레이드오프 (Trade-offs)

좋았던 점:

• 토큰 버킷은 단순하고 예측 가능합니다. 더 이상 원인 모를 429 에러를 겪지 않아도 됩니다.
• 재시도 큐가 메인 흐름을 차단하지 않고 백그라운드 태스크로 실행됩니다.
• 코드를 테스트하기 쉽습니다. 토큰 버킷을 모킹 (Mock)하여 재시도 동작을 검증할 수 있습니다.

다음에 개선할 점:

• 토큰 버킷 구현에 락 (Lock)을 사용합니다. 매우 높은 동시성이 필요한 경우, asyncio.Event를 사용한 락 프리 (Lock-free) 방식을 고려해야 합니다.

• 재시도 워커 (Retry Worker)가 무한히 실행됩니다. 실제 환경에서는 최대 재시도 횟수를 추가하고 데드 레터 큐 (Dead Letter Queue)를 도입해야 합니다.

• 지수 백오프에 지터 (Jitter)가 없습니다. 무작위 지터를 추가하면 여러 요청이 동시에 실패할 때 발생하는

• 가끔씩만 API 호출을 수행한다면, 이러한 복잡성을 도입할 가치가 없습니다. 429 에러 이후에 간단히 time.sleep을 사용하는 것만으로도 충분합니다.

• 만약 API가 Retry-After 헤더를 제공한다면, 직접 지연 시간을 계산하는 대신 이를 사용하세요.

• 고수준 클라이언트 라이브러리(예: openai)를 사용 중이라면 이미 재시도(Retries) 로직이 포함되어 있습니다. 하지만 동시 사용(Concurrent usage)을 위해서는 여전히 속도 제한(Rate limiting)이 필요합니다.

핵심 요약 (The Takeaway)

제3자 API, 특히 토큰당 비용이 발생하는 AI 서비스를 다룰 때 속도 제한(Rate limiting)은 피할 수 없는 현실입니다. 제한 사항과 싸우기보다는 적절한 큐(Queue)와 백오프(Backoff) 전략을 통해 이를 수용하세요. 제가 공유한 코드는 프로덕션 환경에 바로 적용할 수준은 아니지만(로깅이나 429 에러 이외의 예외 처리가 없음), 견고한 시작점이 될 수 있습니다.

API 속도 제한을 처리하는 여러분만의 선호하는 전략은 무엇인가요? 프로젝트에서 토큰 버킷(Token buckets)이나 리키 버킷(Leaky buckets) 알고리즘을 사용해 본 적이 있나요? 여러분에게 효과적이었던 방법이 무엇인지 궁금합니다.