비동기 호출

• 이전 api 호출을 통한 AI응답을 받는 코드를 작성했었다.
  그러나, 한 번의 응답을 받기 위해서는 짧지 않은 시간이 소요된다는걸 알게되었을 것이다.
  하나의 질문에 대해서만 응답이 필요하다면 상관 없지만 여러 질문을 하거나 서비스로 제공하여 사용자가 많아진다면 질문의 수 만큼 대기시간이 늘어나게 된다.
• 이러한 문제를 개선하기위해 비동기로 호출할 수 있도록 코드를 재 작성하자.

import asyncio
import os
from dotenv import load_dotenv

from openai import AsyncOpenAI

load_dotenv()

openai_client = AsyncOpenAI(api_key=os.environ.get("OPEN_API_KEY"))

async def call_async_openai(prompt:str, model:str="gpt-5-mini") -> str :
    response = await openai_client.chat.completions.create(
        model=model,
        messages=[{"role":"user", "content":prompt}]
    )
    
    return response.choices[0].message.content


async def main():
    print("동시에 2번 API  호출하기")
    prompt = "비동기 프로그래밍에 대해 두세 문장으로 설명해주세요."
    
    # 비동기 함수 호출시 코루틴 객체 반환(실행은 안됨)
    openai_task = call_async_openai(prompt)
    openai_task2 = call_async_openai(prompt)
    
    # 두 API 호출 병려렬로 실행하고 완료될 때까지 대기.    
    openai_response, openai_response2 = await asyncio.gather(openai_task, openai_task2)
    print(f"OpenAI 응답 1 : {openai_response}")
    print(f"OpenAI 응답 2 : {openai_response2}")
    

if __name__ == "__main__" :
    asyncio.run(main()) # 비동기 메인 함수를 이벤트 루프에서 실행

• 간단하게 asyncio를 통한 비동기 호출을 만들었다.
  실행해보면 응답시간이 질문한 개수만큼 늘어나지 않는다는걸 알 수 있다.

오류 처리

• 당연히 API호출이 모두 성공할 수 없다. 그렇다면 실패했을 때의 로직도 필요하다.
• 모듈 설치
  pip install tenacity
• 코드 작성 (위에서 작성했던 코드에 추가된 버전)

import asyncio
import os, logging, random
from dotenv import load_dotenv
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
from openai import AsyncOpenAI

# logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

load_dotenv()

openai_client = AsyncOpenAI(api_key=os.environ.get("OPEN_API_KEY"))

@retry(
    stop=stop_after_attempt(3), # 최대 3번시도
    wait=wait_exponential(multiplier=1, min=2, max=10), # 지수 백오프 : 2초, 4초, 8초....
    retry=retry_if_exception_type(), # 모든 예외에 대해 재시도
    before_sleep=lambda retry_state : logger.warning(
        f"API 호출 실패: {retry_state.outcome.exception()}, {retry_state.attempt_number} 번쨰 시도중..."
    )
    
    
)
async def call_async_openai(prompt:str, model:str="gpt-5-mini") -> str :
    
    
    logger.info(f"OpenAI API 호출 시작 : {model}")
    
    await simulate_random_failure()
    
    response = await openai_client.chat.completions.create(
        model=model,
        messages=[{"role":"user", "content":prompt}]
    )
    
    logger.info(f"OpenAI API 호출  성공 ")
    
    return response.choices[0].message.content

# 인위적 실패 생성 함수
async def simulate_random_failure() :
    if random.random() < 0.5 :
        logger.warning("인위적 API 호출 실패 생성")
        raise ConnectionError("인위적 연결 오류 발생")
    
    await asyncio.sleep(random.uniform(0.1, 0.5))

async def main():
    try :
        print("동시에 2번 API  호출하기")
        prompt = "비동기 프로그래밍에 대해 두세 문장으로 설명해주세요."
        
        # 비동기 함수 호출시 코루틴 객체 반환(실행은 안됨)
        openai_task = call_async_openai(prompt)
        openai_task2 = call_async_openai(prompt)
        
        # 두 API 호출 병려렬로 실행하고 완료될 때까지 대기.    
        openai_response, openai_response2 = await asyncio.gather(openai_task, openai_task2)
        print(f"OpenAI 응답 1 : {openai_response}")
        print(f"OpenAI 응답 2 : {openai_response2}")
    except Exception as e:
        logger.error(f"처리되지 않은 오류 발생 {e} ")
        
    

if __name__ == "__main__" :
    asyncio.run(main()) # 비동기 메인 함수를 이벤트 루프에서 실행

• 실행시켜보면 실패 되었을때, 바로 종료되지 않고 retry 어노테이션에 설정된 값과 같이 재시도를 진행하는걸 확인할 수 있다.