[LLM] openai 에러 완화 시키기, 안정성 확보 방법(무응답시 재요청): Error Mitigation/ Using the Tenacity library/ Using the backoff library/Manual backoff implementation
LLM
openAI 안정성 확보 방법
- LLM으로 gpt를 사용시 가끔 응답이 없는 문제가 발생해 timeout이 날 수 있다.
이러한 timeout error에 대한 안정성을 확보하는 방법으로
OpenAI 공식문서에서는 backoff, tenacity를 사용해서 구현하는 방법을 제시하고 있다.
Error Mitigation
OpenAI 공식사이트에서는 이러한 에러 완화 방법으로 Retrying with exponential backoff 와 Tenacity 를 사용하는 방법을 게시해두었다.
OpenAI Cookbook에는 속도 제한 오류를 피하는 방법을 설명하는 파이썬 노트북과
API 요청을 일괄 처리하는 동안 속도 제한을 유지하기 위한 파이썬 스크립트의 예를 제시하고 있다.
자동화된 대량 사용 및 오용으로부터 보호하기 위해 지정된 기간(매일, 매주 또는 매월) 내에 개별 사용자에 대한 사용 제한을 설정할 필요가 있고, 한도를 초과하는 사용자를 위해 하드 캡이나 수동 검토 프로세스를 구현하는 것을 고려 해야 한다고 덧붙인다.
Retrying with expoential backoff
- 속도 제한 오류를 피하는 쉬운 방법 중 하나는 무작위 지수 백오프로 요청을 자동으로 재시도하는 것이다.
지수 백오프로 다시 시도하는 것은 속도 제한 오류가 발생했을 때 짧은 수면을 수행한 다음 실패한 요청을 다시 시도하는 것을 의미한다.
요청이 여전히 실패하면, 수면 길이가 증가하고 과정이 반복된다.
해당 방법은 요청이 성공하거나 최대 재시도 횟수에 도달할 때까지 계속된다.
이러한 방법으로는
(1) Using the Tenacity library (Tenacity library 사용하기)
(2) Using the backoff library (backoff 라이브러리 사용하기)
(3) Manual backoff implementation (수동으로 backoff 구현하기)
가 있다.
방법 (1) Using the Tenacity library (Tenacity library 사용하기)
Tenacity는 Python으로 작성된 Apache 2.0 라이센스 범용 재시도 라이브러리로,
거의 모든 것에 재시도 동작을 추가하는 작업을 단순화할 수 있다고 한다.
요청에 지수 백오프를 추가하려면 tenacity.retry 데코레이터를 사용하면 된다.
openai에서 제공하고 있는 샘플 코드는 아래와 같은데,
장난하시나요.. openai? docs가 업데이트가 안된
이미 openai에서 OpenAI를 import 하는 건 옛 방법이고 저렇게 하면 안굴러간다..
gpt-3.5-turbo를 사용한다고 했을 때, message 형식을 맞춰준 다음 Tenacity library를 쓸 수 있다.
일단 나의 시나리오는 chatgpt에게 너는 이제 파이썬 엔지니어고, quick sort에 대해서 설명하라는 프롬프트를 작성한 후 답변을 기다리는 것이다.
단계 1. 필요한 라이브러리 import
먼저 필요한 라이브러리를 import 한다.
import os
import openai
from tenacity import (
retry,
stop_after_attempt,
wait_random_exponential,
)단계 2. openai key 등록
발급 받은 openai key를 환경변수에 등록한다.
나는 .env 파일에 명시해두고 불러와서 쓴다.
openai.api_key = os.environ["OPENAI_API_KEY"]단계 3. Tenacity 데코레이터를 활용해서 backoff 메소드 구현
@retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(6))
def completion_backoff(**kwargs):
return openai.ChatCompletion.create(**kwargs)
단계 4. openai chatgpt message 형식에 따라 prompt 작성
message 형식에서 role의 system과 user에 따른 prompt는 다음과 같이 할당했다.
system_prompt="""
You are a python engineer.
Please give the correct answer according to the python language \
characteristics according to the corresponding question.
"""
user_prompt="""tell me the quick sort algorithm"""단계 5. 응답 받을 최종 메소드 구현
def retryExample1(system_prompt, user_prompt):
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
]
response = completion_backoff(
model='gpt-3.5-turbo',
messages=messages,
n = 3,
)
return response.choices[0].message.contentretryExample1 메소드에 system_prompt, user_prompt를 인자로 받아서,
gpt-3.5-turbo 모델에게 해당 프롬프트를 주입하고 답변을 기다린다.
그러면 이렇게 친절햐게 구현하는 파이썬 코드 까지 포함해서 답을 준다.
The quicksort algorithm is a well-known sorting algorithm that follows the divide-and-conquer approach. It works by selecting a pivot element from the list and partitioning the other elements into two sub-arrays, according to whether they are less than or greater than the pivot. The process is then repeated recursively on the two sub-arrays until the entire list is sorted.
Here's the implementation of the quicksort algorithm in Python:
def quicksort(arr):
if len(arr) <= 1:
return arr
else:
pivot = arr[0]
smaller = [x for x in arr[1:] if x <= pivot]
greater = [x for x in arr[1:] if x > pivot]
return quicksort(smaller) + [pivot] + quicksort(greater)You can use this function by passing a list as the input, and it will return a sorted version of that list. For example:
my_list = [4, 2, 8, 5, 1, 3]
sorted_list = quicksort(my_list)
print(sorted_list) # Output: [1, 2, 3, 4, 5, 8]한국말로 대답해달라고 해야 겠다.
프롬프트를 약간 수정해봤다.
그러니까 이제 코드를 안준다. 코드도 달라고 해야 겠다.
갑자기 프롬프트를 수정하고 다른 곳으로 새버렸지만, Tenacity를 사용하는 방법이다.
방법 (2) : Using the backoff library (backoff 라이브러리 사용하기)
backoff와 retry를 위한 기능으로 데코레이터를 제공하는 또 다른 파이썬 라이브러리는 backoff(일명 백오프)이다.
openai에서 제공하는 샘플코드는 아래와 같지만 물론 그대로 사용할 수 없다.
참고해서 사용할 수 있는 코드로 변형해야 한다. (원래 그런가..? docs에 있는 샘플 코드는 가이드라인이라서 제대로된 코드가 있어야 하는거 아닌가..?)
암튼 openai.RateLimitError는 openai.error.RateLimitError로 가져와야 한다.
docs에 변경 좀 해주쇼..
위 방법과 달라진 점은 backoff 데코레이터를 사용한다는 건데, 다시 단계별로 보자면
단계 1. 필요한 라이브러리 import
먼저 필요한 라이브러리를 import 한다.
import os
import backoff
import openai
)위에서는 tenaticy를 import 했지만 이번에는 backoff를 import 한다.
단계 2. openai key 등록
발급 받은 openai key를 환경변수에 등록한다.
나는 .env 파일에 명시해두고 불러와서 쓴다.
openai.api_key = os.environ["OPENAI_API_KEY"]단계 3. backoff 데코레이터를 활용해서 backoff 메소드 구현
@backoff.on_exception(backoff.expo, openai.error.RateLimitError)
def completions_with_backoff(**kwargs):
return openai.ChatCompletion.create(**kwargs)단계 4. openai chatgpt message 형식에 따라 prompt 작성
message 형식에서 role의 system과 user에 따른 prompt는 다음과 같이 할당했다.
방법 1을 하면서 수정된 프롬프트로 진행해보자!
system_prompt="""
You are a python engineer.
Please give the correct answer according to the python language \
characteristics according to the corresponding question.
Please answer in Korean.
Please also provide sample code.
"""
user_prompt="""tell me the quick sort algorithm"""단계 5. 응답 받을 최종 메소드 구현
def retryExample2(system_prompt, user_prompt):
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
]
response = completions_with_backoff(
model='gpt-3.5-turbo',
messages=messages,
n = 3,
)
return response.choices[0].message.content
값 잘 준다
이건 위와 동일하다.
방법 (3) : Using the backoff library (backoff 라이브러리 사용하기)
위의 방법들은 각자 다른 타사 라이브러리를 사용하고 있다.
위의 라이브러리 사용없이도 자신만의 백오프 로직을 커스텀해서 구현 가능하다.
제공하고 있는 커스텀 방법 예시 코드는 바로 이것이다.
아래 openai 불러서 message 형식만 맞춰주면 이건 돌아간다 굿굿
wrapper를 사용하는 커스텀 방법이다.
아참참 중간에 retryExample3 에서 들어오는 인자의 데이터타입을 결정할 때
errors 타입만 openai.RateLimitError가 아니라 openai.error.RateLimitError로 수정해 주면 된다.
단계 1. 필요한 라이브러리 import
import time
import random
import openai단계 2. retry decorator 커스텀하기
def Retry a function with exponential backoff(
func,
initial_delay: float = 1,
exponential_base: float = 2,
jitter: bool = True,
max_retries: int = 10,
errors: tuple = (openai.error.RateLimitError,),
):
"""Retry a function with exponential backoff."""
def wrapper(*args, **kwargs):
# Initialize variables
num_retries = 0
delay = initial_delay
# Loop until a successful response or max_retries is hit or an exception is raised
while True:
try:
return func(*args, **kwargs)
# Retry on specific errors
except errors as e:
# Increment retries
num_retries += 1
# Check if max retries has been reached
if num_retries > max_retries:
raise Exception(
f"Maximum number of retries ({max_retries}) exceeded."
)
# Increment the delay
delay *= exponential_base * (1 + jitter * random.random())
# Sleep for the delay
time.sleep(delay)
# Raise exceptions for any errors not specified
except Exception as e:
raise e
return wrapper단계 3. 커스텀한 데코레이터 붙여서 backoff 메소드 구현
@retry_with_exponential_backoff
def completions_with_backoff(**kwargs):
return openai.ChatCompletion.create(**kwargs)
단계 4. 응답 받을 최종 메소드 구현
system_prompt="""
You are a python engineer.
Please give the correct answer according to the python language \
characteristics according to the corresponding question.
Please answer in Korean.
Please also provide sample code.
"""
user_prompt="""tell me the quick sort algorithm"""
def retryExample3(system_prompt, user_prompt):
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt},
]
response = completions_with_backoff(
model='gpt-3.5-turbo',
messages=messages,
n = 3,
)
return response.choices[0].message.content
print(retryExample3(system_prompt, user_prompt))굿 굿 완성
해당 샘플 코드는 git에 업로드 했당~!
참고 사이트
https://platform.openai.com/docs/guides/rate-limits/error-mitigation?context=tier-free
