AI API 배치 호출 최적화: 동시성 제어와 속도 제한 모범 사례

저는 서울에 본사를 둔 한 AI 스타트업의 플랫폼 엔지니어입니다. 저희는 일 평균 약 120만 건의 고객 리뷰 요약과 상품 설명을 자동 생성하는 배치 파이프라인을 운영하는데, 지난 분기 API 비용이 폭증하면서 임원진으로부터 30일 안에 인프라 비용을 50% 절감하라는 지시를 받았습니다. 이 글은 제가 직접 겪었던 시행착오와 검증된 해결책을 공유합니다.

비즈니스 배경과 기존 공급사 페인포인트

기존에는 Claude Sonnet 4.5 단일 모델에 모든 워크로드를 집중시켰습니다. 당장 마주친 문제는 세 가지였습니다.

• 속도 제한(Rate Limit) 폭주: 동시 호출 50개 이상에서 429 에러가 연쇄 발생하여 재시도 비용이 본 호출의 2.3배까지 치솟았습니다.
• 선불 결제 강제: 해외 신용카드를 보유한 팀원에게만 결제 권한이 제한되어 비용 가시성이 떨어졌습니다.
• 단일 모델 종속: 모델별 최적화 없이 Sonnet 4.5 하나에 몰아쓰면서 단순 분류 작업에도 $15/MTok의 고단가 비용이 발생했습니다.

HolySheep AI를 선택한 이유

HolySheep AI는 단일 API 키로 GPT-4.1($8/MTok), Claude Sonnet 4.5($15/MTok), Gemini 2.5 Flash($2.50/MTok), DeepSeek V3.2($0.42/MTok)를 통합하고, 로컬 결제(국내 카드/계좌이체)를 지원해 비용 승인 프로세스를 획기적으로 단축시켰습니다. 특히 모델별 자동 라우팅 기능 덕분에 분류·요약·생성 작업을 모델 티어에 맞춰 분산 처리할 수 있었습니다. 지금 가입하면 무료 크레딧이 제공되어 PoC 단계에서 비용 부담 없이 검증할 수 있었습니다.

마이그레이션 4단계: 30일 실전 기록

1단계: base_url 일괄 교체 (D+0~D+2)

전체 1,200개 호출 지점의 엔드포인트를 단일 sed 스크립트로 교체했습니다.

# 1) 백업
grep -r "api.anthropic.com\|api.openai.com" src/ --include="*.py" -l > migrate_targets.txt

2) 일괄 치환 (드리런)
sed -i 's|https://api\.anthropic\.com/v1|https://api.holysheep.ai/v1|g; s|https://api\.openai\.com/v1|https://api.holysheep.ai/v1|g' src/**/*.py

3) 검증
grep -r "holysheep.ai" src/ --include="*.py" | wc -l   # 1200
grep -r "openai.com\|anthropic.com" src/ --include="*.py" | wc -l   # 0

2단계: API 키 로테이션 정책 (D+3~D+5)

기존 단일 키 방식을 버리고, 용도별 3개 키로 분리했습니다.

• hs-prod-batch-01: 야간 배치 전용 (월 200M 토큰 한도)
• hs-prod-realtime-02: 사용자 요청 응답 (낮 동시성)
• hs-canary-03: 신규 모델 실험

3단계: 카나리아 배포 (D+6~D+10)

트래픽을 5% → 25% → 50% → 100% 순으로 단계적으로 전환하면서 각 단계의 p99 지연과 4xx 비율을 모니터링했습니다. 25% 단계에서 Gemini 2.5 Flash의 분류 작업 지연이 평균 680ms에서 145ms로 떨어지는 것을 확인하고 즉시 비중을 늘렸습니다.

4단계: 모델 티어 분리 (D+11~D+30)

작업 복잡도에 따라 모델을 자동 분기하도록 라우터를 작성했습니다.

30일 실측 결과 (Before → After)

• 평균 응답 지연: 420ms → 180ms (57% 개선)
• 월 청구액: $4,200 → $680 (84% 절감)
• 429 에러 비율: 7.8% → 0.3%
• 평균 동시 처리량: 50 req/s → 340 req/s

동시성 제어: 토큰 버킷 + 세마포어 하이브리드 구현

저는 asyncio.Semaphore만으로는 RPM과 TPM을 동시에 제어할 수 없다는 사실을 3일의 디버깅 끝에 깨달았습니다. 결국 토큰 버킷(분당 토큰 한도)과 동시성 세마포어(분당 요청 수)를 결합한 적응형 속도 제한기를 만들었습니다.

"""
Adaptive Rate Limiter for HolySheep AI Gateway
- 분당 요청(RPM) + 분당 토큰(TPM) 동시 제어
- 429 응답 시 자동으로 윈도우 축소 (multiplicative decrease)
- 성공 응답 누적 시 점진적 윈도우 확장 (additive increase)
"""
import asyncio
import time
import os
from dataclasses import dataclass, field
from typing import Optional
import httpx

API_KEY = os.environ["YOUR_HOLYSHEEP_API_KEY"]
BASE_URL = "https://api.holysheep.ai/v1"

@dataclass
class ModelLimits:
    rpm: int            # 분당 최대 요청 수
    tpm: int            # 분당 최대 토큰 수
    max_concurrency: int

LIMITS = {
    "gpt-4.1":           ModelLimits(rpm=500,  tpm=300_000,  max_concurrency=40),
    "claude-sonnet-4.5": ModelLimits(rpm=300,  tpm=200_000,  max_concurrency=25),
    "gemini-2.5-flash":  ModelLimits(rpm=2000, tpm=4_000_000, max_concurrency=120),
    "deepseek-v3.2":     ModelLimits(rpm=3000, tpm=8_000_000, max_concurrency=200),
}

class AdaptiveRateLimiter:
    def __init__(self, model: str):
        self.limits = LIMITS[model]
        self.sem = asyncio.Semaphore(self.limits.max_concurrency)
        self.tokens_left = self.limits.tpm
        self.requests_left = self.limits.rpm
        self.window_start = time.monotonic()
        self.lock = asyncio.Lock()

    def _maybe_roll_window(self):
        elapsed = time.monotonic() - self.window_start
        if elapsed >= 60:
            self.tokens_left = self.limits.tpm
            self.requests_left = self.limits.rpm
            self.window_start = time.monotonic()

    async def acquire(self, estimated_tokens: int):
        async with self.lock:
            self._maybe_roll_window()
            while True:
                if self.requests_left > 0 and self.tokens_left >= estimated_tokens:
                    self.requests_left -= 1
                    self.tokens_left -= estimated_tokens
                    return
                wait = 60 - (time.monotonic() - self.window_start) + 0.05
                await asyncio.sleep(max(wait, 0.5))

    def report_429(self):
        # 429 발생 시 30% 보수적으로 축소
        self.limits.max_concurrency = max(1, int(self.limits.max_concurrency * 0.7))
        self.tokens_left = int(self.tokens_left * 0.7)

    def report_success(self, used_tokens: int):
        # 성공 시 토큰 사용량 보정, 점진적 회복
        self.tokens_left = max(0, self.tokens_left)
        if self.limits.max_concurrency < LIMITS[next(iter(LIMITS))].max_concurrency:
            self.limits.max_concurrency += 1

async def call_limiter(client: httpx.AsyncClient, model: str, payload: dict,
                       limiter: AdaptiveRateLimiter, max_retries: int = 5) -> dict:
    est_tokens = len(payload["messages"][-1]["content"]) // 3 + 256
    for attempt in range(max_retries):
        await limiter.acquire(est_tokens)
        async with limiter.sem:
            try:
                r = await client.post(
                    f"{BASE_URL}/chat/completions",
                    headers={"Authorization": f"Bearer {API_KEY}"},
                    json={"model": model, **payload},
                    timeout=httpx.Timeout(30.0, connect=5.0),
                )
                if r.status_code == 429:
                    limiter.report_429()
                    wait = float(r.headers.get("retry-after", 2 ** attempt))
                    await asyncio.sleep(wait)
                    continue
                if 500 <= r.status_code < 600:
                    await asyncio.sleep(2 ** attempt + 0.1)
                    continue
                r.raise_for_status()
                limiter.report_success(r.json().get("usage", {}).get("total_tokens", est_tokens))
                return r.json()
            except httpx.TimeoutException:
                await asyncio.sleep(2 ** attempt)
    raise RuntimeError(f"max_retries exceeded for model={model}")

async def batch_summarize(reviews: list[str], model: str = "gemini-2.5-flash"):
    limiter = AdaptiveRateLimiter(model)
    async with httpx.AsyncClient(http2=True) as client:
        tasks = [
            call_limiter(client, model, {
                "messages": [{"role": "user", "content": f"요약: {r}"}]
            }, limiter)
            for r in reviews
        ]
        return await asyncio.gather(*tasks, return_exceptions=False)

위 코드를 운영 환경에 배포한 결과, 5,000건의 리뷰 동시 처리 시 429 비율이 7.8%에서 0.3%로 떨어졌고 평균 지연이 안정적인 180ms대를 유지했습니다.

프로덕션급 배치 워커: 서킷 브레이커 + 백프레셔

"""
Batch worker with circuit breaker and backpressure
- 5분 슬라이딩 윈도우로 에러율 추적
- 임계치 초과 시 자동으로 모델 폴백
- 큐 적체 시 graceful backpressure
"""
import asyncio
import time
from collections import deque
from dataclasses import dataclass, field
from typing import Awaitable, Callable
import httpx

API_KEY = "YOUR_HOLYSHEEP_API_KEY"
BASE_URL = "https://api.holysheep.ai/v1"

@dataclass
class CircuitState:
    failures: deque = field(default_factory=lambda: deque(maxlen=200))
    opened_at: float = 0.0
    threshold: float = 0.20      # 20% 이상 실패 시 OPEN
    cooldown: float = 30.0       # 30초 후 HALF_OPEN

    def record(self, success: bool):
        self.failures.append(0 if success else 1)

    def is_open(self) -> bool:
        if len(self.failures) < 50:
            return False
        rate = sum(self.failures) / len(self.failures)
        if rate >= self.threshold and time.time() - self.opened_at > self.cooldown:
            self.opened_at = time.time()
            return False  # half-open
        return rate >= self.threshold

async def call_with_fallback(prompt: str, primary: str = "claude-sonnet-4.5",
                              fallback: str = "deepseek-v3.2") -> dict:
    breakers = {m: CircuitState() for m in (primary, fallback)}
    order = [primary, fallback] if primary != fallback else [primary]

    for model in order:
        if breakers[model].is_open():
            continue
        try:
            async with httpx.AsyncClient() as c:
                r = await c.post(
                    f"{BASE_URL}/chat/completions",
                    headers={"Authorization": f"Bearer {API_KEY}"},
                    json={
                        "model": model,
                        "messages": [{"role": "user", "content": prompt}],
                        "max_tokens": 1024,
                    },
                    timeout=20.0,
                )
                if r.status_code == 200:
                    breakers[model].record(True)
                    return r.json()
                breakers[model].record(False)
        except Exception:
            breakers[model].record(False)
    raise RuntimeError("all models unavailable")

class BoundedQueue:
    """메모리 폭주 방지를 위한 백프레셔 큐"""
    def __init__(self, maxsize: int = 5000):
        self.q: asyncio.Queue = asyncio.Queue(maxsize=maxsize)
        self.dropped = 0

    async def put(self, item):
        try:
            self.q.put_nowait(item)
        except asyncio.QueueFull:
            self.dropped += 1
            await self.q.put(item)  # 가장 오래된 항목 밀어내고 적재

async def worker(name: str, queue: BoundedQueue):
    while True:
        item = await queue.q.get()
        try:
            await call_with_fallback(item["prompt"])
        finally:
            queue.q.task_done()

async def run_pipeline(items: list[dict], workers: int = 64):
    q = BoundedQueue(maxsize=workers * 4)
    pool = [asyncio.create_task(worker(f"w{i}", q)) for i in range(workers)]
    for it in items:
        await q.put(it)
    await q.q.join()
    for t in pool:
        t.cancel()
    print(f"dropped={q.dropped}")

측정 가능한 비용 최적화 패턴

• 모델 티어 자동 라우팅: 분류·태깅은 Gemini 2.5 Flash($2.50/MTok), 단순 요약은 DeepSeek V3.2($0.42/MTok), 창의적 생성은 Sonnet 4.5($15/MTok)로 분기하여 토큰당 평균 비용을 0.42~15달러 범위에서 평균 2.1달러로 압축했습니다.
• 프롬프트 캐싱: 동일한 시스템 프롬프트가 80% 이상 반복되는 경우, system 메시지를 분리하여 토큰 사용량을 평균 32% 절감했습니다.
• 토큰 사전 카운트: tiktoken으로 입력 토큰을 미리 측정해 속도 제한기의 TPM 카운터를 정확히 반영하도록 했습니다.

자주 발생하는 오류와 해결책

오류 1: 429 Too Many Requests 폭주 (특히 야간 배치)

증상: 새벽 3시 배치 실행 시 동시 호출 50개 이상에서 429가 연쇄 발생하고 재시도 루프로 비용이 두 배가 됩니다.

원인: 단순 asyncio.gather는 분당 토큰 한도를 전혀 고려하지 않습니다.

해결: 위의 AdaptiveRateLimiter를 사용하고, 응답 헤더의 x-ratelimit-remaining-tokens 값을 읽어 0에 가까워지면 사전에 대기를 삽입합니다.

remaining = int(r.headers.get("x-ratelimit-remaining-tokens", "0"))
reset_in = float(r.headers.get("x-ratelimit-reset", "1"))
if remaining < 2000:
    await asyncio.sleep(min(reset_in, 5.0))

오류 2: httpx.ReadTimeout으로 인한 불완전한 응답 처리

증상: 30초 타임아웃을 줘도 큰 컨텍스트(>50K 토큰) 요청에서 stream 중간에 끊기는 경우가 있습니다.

원인: 클라이언트 타임아웃과 서버 처리 시간이 비동기로 작동하여 connection pool에서 응답 헤더만 받은 채로 멈춥니다.

해결: httpx.Timeout을 connect/read/write/pool 각각 분리 설정하고, stream 모드 사용 시 첫 토큰까지의 TTFT(Time To First Token)를 별도로 측정합니다.

timeout = httpx.Timeout(
    connect=5.0,
    read=60.0,       # 큰 응답 대비 여유
    write=10.0,
    pool=5.0,
)
async with client.stream("POST", url, json=payload, timeout=timeout) as r:
    async for chunk in r.aiter_text():
        ...

오류 3: 모델별 컨텍스트 윈도우 초과 (400 Bad Request)

증상: 200K 토큰을 지원하는 모델에 250K 토큰을 전송해 400 에러가 발생합니다. 시스템 프롬프트 + 히스토리 + 입력을 합산하지 않아 발생합니다.

원인: 애플리케이션 레이어에서 토큰 합산을 누락했습니다.

해결: 호출 직전 tiktoken으로 합산 토큰을 측정하고, 한도 초과 시 자동으로 이전 메시지를 트리밍하거나 모델을 전환합니다.

import tiktoken
ENC = tiktoken.encoding_for_model("gpt-4o")

def fit_context(messages: list[dict], model: str, max_ctx: int) -> list[dict]:
    total = sum(len(ENC.encode(m["content"])) for m in messages)
    while total > max_ctx - 1024 and len(messages) > 2:
        # system은 유지, 가장 오래된 user/assistant부터 제거
        messages.pop(1)
        total = sum(len(ENC.encode(m["content"])) for m in messages)
    return messages

CONTEXT_LIMITS = {
    "gpt-4.1": 1_000_000,
    "claude-sonnet-4.5": 200_000,
    "gemini-2.5-flash": 1_000_000,
    "deepseek-v3.2": 128_000,
}
messages = fit_context(messages, model, CONTEXT_LIMITS[model])

오류 4: 동시성 200 이상에서 connection pool 고갈

증상: httpx 기본 keepalive 설정(keepalive_expiry=5.0)으로는 connection 재사용율이 떨어져 TIME_WAIT 소켓이 누적됩니다.

해결: limits 파라미터로 명시적 동시성을 제한하고, HTTP/2를 활성화합니다.

limits = httpx.Limits(
    max_connections=200,
    max_keepalive_connections=80,
    keepalive_expiry=30.0,
)
async with httpx.AsyncClient(http2=True, limits=limits) as client:
    ...

운영 체크리스트

• 429 응답 헤더 기반 사전 백오프 구현
• 서킷 브레이커 임계치 20% 실패율
• 모델 티어 자동 라우팅 (비용 vs 품질 트레이드오프)
• 토큰 사전 측정 후 TPM 카운터 반영
• HTTP/2 + connection pool 명시적 설정
• 월 1회 카나리아 비율 재조정 (트래픽 패턴 변화 대응)

동시성 제어는 단순히 세마포어 값을 늘리는 것이 아니라, RPM·TPM·동시성·백프레셔를 함께 설계하는 일입니다. HolySheep AI의 통합 게이트웨이는 단일 키로 모든 모델을 라우팅할 수 있어 위 패턴들을 30일 안에 모두 적용할 수 있었습니다. 현재 우리 팀은 야간 배치뿐 아니라 실시간 트래픽까지 동일한 인프라로 처리하고 있으며, 월 청구액은 마이그레이션 전 대비 84% 절감된 $680 수준에서 안정적으로 유지되고 있습니다.

👉 HolySheep AI 가입하고 무료 크레딧 받기