[tenacity] 데코레이터로 클린하게 재시도 기능 구현하기

LLM 호출이 과도하게 몰리는 등 LLM 호출에 있어서 에러가 나는 경우가 발생할 수 있다.

이런 경우, 재시도하는 코드를 작성하는것은 어렵지 않지만 while문이나 for 문을 사용하다 보니 코드가 지저분하게 된다.

이를 tenacity라는 데코레이터로 쉽게 해결할 수 있다.

 

import asyncio
from openai import AsyncOpenAI
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
    before_sleep_log,
    after_log
)
import logging

from typing import List

# 로깅 설정
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)

# OpenAI 클라이언트 초기화
client = AsyncOpenAI(api_key="your-api-key-here")


# tenacity 데코레이터로 재시도 로직 구성
@retry(
    # API 오류 발생 시에만 재시도
    retry=retry_if_exception_type((Exception,)),
    # 최대 5번까지 시도
    stop=stop_after_attempt(5),
    # 지수 백오프: 2^x초 대기 (최소 1초, 최대 60초)
    wait=wait_exponential(multiplier=1, min=1, max=60),
    # 재시도 전 로그 출력
    before_sleep=before_sleep_log(logger, logging.WARNING),
    # 재시도 후 로그 출력
    after=after_log(logger, logging.INFO)
)
async def call_openai_with_retry(prompt: str, model: str = "gpt-3.5-turbo") -> str:
    """
    tenacity를 사용한 OpenAI API 호출
    네트워크 오류나 일시적 장애 시 자동으로 재시도합니다.
    """
    logger.info(f"API 호출 시도: {prompt[:50]}...")
    
    try:
        response = await client.chat.completions.create(
            model=model,
            messages=[
                {"role": "system", "content": "당신은 도움이 되는 AI 어시스턴트입니다."},
                {"role": "user", "content": prompt}
            ],
            max_tokens=150,
            temperature=0.7
        )
        
        result = response.choices[0].message.content
        logger.info(f"API 호출 성공")
        return result
        
    except Exception as e:
        logger.error(f"API 호출 실패: {str(e)}")
        raise


async def process_multiple_prompts(prompts: List[str]) -> List[str]:
    """
    여러 프롬프트를 동시에 처리하는 함수
    """
    tasks = [call_openai_with_retry(prompt) for prompt in prompts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # 결과 처리
    processed_results = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            logger.error(f"프롬프트 {i+1} 처리 실패: {str(result)}")
            processed_results.append(f"Error: {str(result)}")
        else:
            processed_results.append(result)
    
    return processed_results


# 커스텀 재시도 조건 예제
@retry(
    # Rate Limit 에러만 재시도
    retry=retry_if_exception_type((Exception,)),
    stop=stop_after_attempt(3),
    # 고정 대기 시간 5초
    wait=wait_exponential(multiplier=5, min=5, max=30),
)
async def call_openai_rate_limit_aware(prompt: str) -> str:
    """
    Rate Limit을 고려한 OpenAI API 호출
    """
    response = await client.chat.completions.create(
        model="gpt-3.5-turbo",
        messages=[{"role": "user", "content": prompt}],
        max_tokens=100
    )
    return response.choices[0].message.content


async def main():
    """
    메인 실행 함수
    """
    print("=" * 60)
    print("OpenAI API + Tenacity 비동기 재시도 예제")
    print("=" * 60)
    
    # 단일 프롬프트 테스트
    print("\n[1] 단일 API 호출 테스트")
    try:
        result = await call_openai_with_retry(
            "Python에서 비동기 프로그래밍이란 무엇인가요?"
        )
        print(f"결과: {result}\n")
    except Exception as e:
        print(f"최종 실패: {e}\n")
    
    # 다중 프롬프트 동시 처리
    print("[2] 다중 API 호출 테스트")
    prompts = [
        "tenacity 라이브러리의 주요 기능은?",
        "asyncio와 동기 코드의 차이점은?",
        "API 재시도 전략의 중요성은?",
    ]
    
    results = await process_multiple_prompts(prompts)
    
    for i, (prompt, result) in enumerate(zip(prompts, results), 1):
        print(f"\n프롬프트 {i}: {prompt}")
        print(f"응답: {result[:100]}...")
    
    print("\n" + "=" * 60)
    print("모든 테스트 완료!")


if __name__ == "__main__":
    # 비동기 메인 함수 실행
    asyncio.run(main())