Molayo

© 2026 Molayo

Dev.to헤드라인2026. 05. 27. 06:09

AI 애플리케이션을 위한 비동기 Python: 부하 상황에서도 무너지지 않는 패턴

요약

AI 애플리케이션 개발 시 발생하는 비동기 Python의 속도 제한 및 연결 오류를 해결하기 위한 견고한 패턴을 소개합니다. 세마포어를 통한 동시성 제어, 지수 백오프를 적용한 재시도 전략 등 실무적인 최적화 방법을 다룹니다.

핵심 포인트

  • asyncio.Semaphore를 사용하여 API 속도 제한 및 연결 풀 고갈 방지
  • 지수 백오프와 지터(Jitter)를 결합하여 재시도 시 발생하는 부하 분산
  • Rate limit 및 일시적 서버 오류에 대한 선별적 재시도 전략
  • 제한 없는 asyncio.gather 사용 시 발생하는 오류 전파 문제 해결

대부분의 Python 개발자들이 처음 작성하는 비동기 AI 애플리케이션은 다음과 같은 모습입니다:

import asyncio
from anthropic import AsyncAnthropic

...

작동은 합니다. 하지만 500개의 문서를 대상으로 실행하면 속도 제한(Rate limit) 오류, 연결 시간 초과(Connection timeout), 그리고 부분적인 결과가 뒤섞여 나타납니다. 어떤 작업은 완료되지만, 어떤 작업은 조용히 실패하며, 여러분은 어떤 것이 실패했는지조차 알 수 없게 됩니다.

이 포스트에서는 실제 부하 상황에서도 견고하게 버티는 비동기 패턴인 제한된 동시성 (Bounded concurrency), 지수 백오프를 이용한 재시도 (Retry with backoff), 오류 격리를 포함한 결과 수집 (Result collection with error isolation), 그리고 취소 (Cancellation)를 다룹니다.

제한 없는 gather의 문제점

asyncio.gather(*[summarize(doc) for doc in docs])는 모든 작업을 동시에 실행합니다. 500개의 문서가 있다면, 이는 500개의 동시 API 연결을 의미합니다. 이때 세 가지 현상이 발생합니다:

  1. 속도 제한 (Rate limit) 오류. 대부분의 AI API는 분당 토큰 제한을 가지고 있습니다. 500개의 동시 요청은 즉시 이 제한에 걸리게 됩니다.
  2. 연결 풀 고갈 (Connection pool exhaustion). Anthropic SDK의 기반이 되는 기본 httpx 연결 풀은 기본적으로 100개의 연결 제한을 가집니다. 이를 초과하면 요청이 대기열에 쌓이거나 실패합니다.
  3. 오류 전파 (Error propagation). asyncio.gather는 기본적으로 첫 번째 예외(Exception)를 발생시키고 나머지 작업을 취소합니다. 문서 하나가 잘못되면 전체 배치(Batch) 작업이 중단됩니다.

해결책은 세마포어 (Semaphore)입니다.

asyncio.Semaphore를 이용한 제한된 동시성

import asyncio
from anthropic import AsyncAnthropic

...

세마포어는 큐 (Queue) 역할을 합니다. 10개의 작업만 동시에 실행되고 나머지는 대기합니다. 이를 통해 속도 제한 오류가 현저히 줄어들고, 연결 풀(Connection pool) 상태도 건강하게 유지됩니다.

세마포어 값 조정하기: 개발 단계에서는 5로 시작하세요. 운영 환경에서는 floor(분당_속도_제한 / 호출당_평균_초 / 60)로 조정하십시오. 만약 API가 분당 60,000 토큰을 허용하고, 각 호출이 약 1,000 토큰을 사용하며 약 1초가 걸린다면, 이론적으로 60개의 동시 호출이 안전합니다. 여유 공간을 두기 위해 40-50 정도를 사용하세요.

지수 백오프를 이용한 재시도 (Retry with exponential backoff)

제한된 동시성을 사용하더라도 속도 제한 오류는 여전히 발생할 수 있습니다. 다른 프로세스와 할당량(Quota)을 공유해야 하고, 시간대에 따라 할당량이 변하며, API가 가끔 529 오류를 반환하기 때문입니다. 재시도 데코레이터 (Retry decorator):

import asyncio
import random
import logging
...

핵심 세부 사항:

  • 지터 (Jitter, random.uniform(0, 1))를 포함한 지수 백오프 (Exponential backoff, base_delay * 2^attempt)는 천둥 치는 들소 (Thundering herd) 현상을 방지합니다. 즉, 모든 재시도 작업이 동시에 실행되지 않도록 합니다.
  • 속도 제한 (Rate limits, 429) 및 일시적인 서버 오류 (500, 502, 503, 529)에 대해서만 재시도합니다.
  • 400번대 오류는 재시도하지 마세요. 잘못된 요청 (Bad requests)은 재시도해도 성공하지 않으며 할당량 (Quota)만 낭비하게 됩니다.

배치 처리에서의 오류 격리 (Error isolation)

asyncio.gather(*tasks)는 첫 번째 예외 (Exception)를 전파합니다. 부분적인 성공이 허용되는 배치 처리 (Batch processing)의 경우, return_exceptions=True를 사용하세요:

from dataclasses import dataclass
from typing import Any

...

return_exceptions=True를 사용하면 예외가 발생(raise)되는 대신 결과 리스트의 값으로 반환됩니다. 실패한 작업에 대해 로그를 남기고 계속 진행할지, 재시도를 위해 다시 큐에 넣을지, 데드 레터 큐 (Dead-letter queue)에 기록할지, 아니면 예외를 발생시킬지 직접 결정할 수 있습니다.

긴 배치 작업을 위한 진행 상황 추적 (Progress tracking)

수 분이 소요되는 배치 작업의 경우, 차단 (Blocking) 없이 진행 상황 업데이트를 확인하고 싶을 것입니다:

import asyncio
from tqdm.asyncio import tqdm

...

tqdm.asyncio.tqdm.as_completed는 asyncio.as_completed를 진행률 표시줄 (Progress bar)으로 감쌉니다. 결과는 완료된 순서대로 도착하므로, 원래의 순서가 필요한 경우 마지막에 index를 기준으로 정렬하세요.

타임아웃 (Timeouts) 및 취소 (Cancellation)

AI API 호출은 응답 없이 멈춰 있을 수 있습니다. AsyncAnthropic 클라이언트는 기본 타임아웃을 가지고 있지만, 더 엄격한 제어가 필요할 수 있습니다:

async def summarize_with_timeout(text: str, timeout: float = 30.0) -> str:
    try:
        async with asyncio.timeout(timeout):
...

asyncio.timeout (Python 3.11+)은 asyncio.wait_for보다 깔끔합니다. 이 방식은 TimeoutError를 발생시키고 기반이 되는 태스크 (Task)를 적절히 취소합니다. 타임아웃은 단순히 딱 떨어지는 숫자가 아니라, max_tokens와 예상되는 모델 지연 시간 (Latency)을 기반으로 설정하세요.

Python 3.11 미만 버전의 경우 asyncio.wait_for를 사용하세요:

result = await asyncio.wait_for(
    client.messages.create(...),
    timeout=30.0
...

종합: 프로덕션용 배치 프로세서

import asyncio
import logging
from dataclasses import dataclass
...

비동기 대신 스레드(threads)를 사용해야 하는 경우

비동기(Async)는 병목 현상이 I/O 대기(I/O wait) — 네트워크 호출, 파일 읽기, 데이터베이스 쿼리 — 에 있을 때 효과적입니다. 병목 현상이 CPU에 있을 때는 도움이 되지 않습니다.

AI 출력물에 대해 무거운 후처리(parsing, classification, 대규모 텍스트에 대한 regex 등)를 수행하는 경우, asyncio.to_thread를 사용하여 이벤트 루프(event loop)를 차단하지 않고 스레드 풀(thread pool)에서 해당 작업을 실행하십시오:

import asyncio
import re

...

asyncio.to_thread는 동기 함수를 기본 ThreadPoolExecutor에서 실행하고 결과를 기다립니다(await). 이벤트 루프는 차단되지 않은 상태로 유지됩니다. 진정으로 무거운 CPU 작업(모델 추론, 대규모 numpy 연산)의 경우에는 대신 ProcessPoolExecutor를 사용하십시오.

비동기 AI 애플리케이션을 위한 체크리스트

  • 모든 배치(batch)에 asyncio.Semaphore 적용 — 외부 API에 대해 제한 없는 gather를 절대 수행하지 마십시오.
  • 속도 제한(rate limit) 및 5xx 에러 발생 시 지터(jitter)를 포함한 지수 백오프(Exponential backoff) 적용.
  • 배치 처리를 위한 gather 호출 시 return_exceptions=True 설정 — 실패를 예외(exception)로 발생시키지 말고 값(value)으로 처리하십시오.
  • 모든 API 호출에 타임아웃(Timeout) 설정 — 외부 지연 시간(latency)이 제한적일 것이라고 절대 신뢰하지 마십시오.
  • CPU 집약적인 후처리를 위한 asyncio.to_thread 사용 — 이벤트 루프를 깨끗하게 유지하십시오.

AI Dev Toolkit의 비동기 패턴에는 asyncio 배치 프로세서를 생성하기 위한 프롬프트 템플릿, 재시도 데코레이터(retry decorators), 그리고 함수 시그니처로부터 진행 상황이 추적되는 파이프라인이 포함되어 있습니다. 따라서 새로운 AI 워크플로우를 구축할 때마다 보일러플레이트(boilerplate) 코드를 처음부터 작성할 필요가 없습니다.

추가 읽을거리

AI 자동 생성 콘텐츠

본 콘텐츠는 Dev.to AI tag의 원문을 AI가 자동으로 요약·번역·분석한 것입니다. 원 저작권은 원저작자에게 있으며, 정확한 내용은 반드시 원문을 확인해 주세요.

원문 바로가기
0

댓글

0