Openai cookbook - API usage - How to handle rate limits

지난 글까지 해서 openai cookbook의 embedding 코너는 모두 마쳤습니다.

OpenAI API 를 공부하면서 어떻게 하다가 CookBook 의 Embedding 부터 먼저 공부 했는데요.

이 CookBook은 원래 API usage부터 시작합니다.

 

오늘부터 이 CookBook 순서대로 공부해 볼까 합니다.

 

https://github.com/openai/openai-cookbook/blob/main/examples/How_to_handle_rate_limits.ipynb

GitHub - openai/openai-cookbook: Examples and guides for using the OpenAI API

How to handle rate limits

OpenAI API를 사용하는데는 일일 제한량이 있습니다. 

API 호출이 이 제한량을 넘을 경우 API는 해당 Request에 대한 Response를 하지 않고 429: 'Too Many Requests 나 RateLimitError 메세지를 보냅니다.

 

이번 예제에서는 이 API Request 제한량을 초과해서 나타나는 오류를 방지하거나 그 오류를 처리하는데 대한 팁을 보여 줍니다.

 

이 제한량 초과 에러를 피하기 위해 병렬 요청 (Parallel Requests)를 조절하는 예제를 보려면 아래 내용을 참고 하세요.

 

https://github.com/openai/openai-cookbook/blob/2f5e350bbe66a418184899b0e12f182dbb46a156/examples/api_request_parallel_processor.py

GitHub - openai/openai-cookbook: Examples and guides for using the OpenAI API

Why rate limits exist

이 request 수 제한은 API 관리하는 일반적인 관행입니다. 이것을 관리하는 이유는 몇가지 있습니다.

* 첫번째로는 API 남용과 오용으로부터 시스템을 보호하는데 도움을 줍니다. 예를 들어 악의적인 의도를 가지고 API에 과부하를 일으키거나 서비스를 중단시키려는 의도로 API 요청을 다량 발생 시킬 수 있습니다. 이런 악의적인 공격으로 부터 OpenAI API 서비스를 보호하기 위해 Request 수 제한 정책이 필요합니다.

 

* 둘째로는 이 OpenAI API 자원을 다양한 사람이 공평하게 사용할 수 있도록 하는 목적이 있습니다. 한 개인이나 조직이 과도한 수의 요청을 하면 다른 소비자들의 API 서비스가 중단 될 수 있습니다. 단일 사용자가 만들 수 있는 요청 수를 제한 함으로서 이 API 서비스를 더 많은 사람들이 속도 저하 없이 API를 사용할 수 있는 기회를 더 보장할 수 있습니다.

 

* 세번째 마지막으로 이 요청량 제한은 OpenAI 가 인프라의 총 부하를 관리하는데 도움이 될 수 있습니다. API에 대한 요청이 급격하게 증가하면 서버에 부담을 주고 성능에 문제를 일으킬 수 있습니다. 이 요청량 베한을 설정함으로서 OpenAI는 모든 사용자에게 원활하고 일관된 경험을 유지하는데 도움을 줄 수 있습니다.

 

Default rate limits

2023년 1월 현재 요청량 제한은 아래와 같습니다.

참고로 1000개의 토큰은 한 페이지 분량의 텍스트 입니다.

 

Other rate limit resources

이 요청량 제한에 대해 좀 더 많은 정보를 얻고 싶으면 아래 자료들을 참조하세요.

• Guide: Rate limits
• Help Center: Is API usage subject to any rate limits?
• Help Center: How can I solve 429: 'Too Many Requests' errors?

Requesting a rate limit increase

제한된 요청량 이상으로 API 리소스를 사용하시려면 아래 요청서를 작성하세요.

• OpenAI Rate Limit Increase Request form

Example rate limit error

API 요청들이 너무 빨리 보내지면 아래 요청량 제한 에러가 발생합니다. OpenAI 파이썬 라이브러리를 사용한다면 그 에러 메세지는 아래와 같을 겁니다.

아래 코드는 요청량 제한을 초과하는 한 예를 보여 줍니다.

 

import openai  # for making OpenAI API requests

# request a bunch of completions in a loop
for _ in range(100):
    openai.Completion.create(
        model="code-cushman-001",
        prompt="def magic_function():\n\t",
        max_tokens=10,
    )

 

저는 굳이 이 코드는 실행해 보지 않겠습니다.

코드는 그냥 간단합니다. max_tokens를 10으로 설정한 다음에 Completion.create() api를 100번 요청하는 for 문 입니다.

 

How to avoid rate limit errors

Retrying with exponential backoff

이 요청량 제한 오류를 방지하는 쉬운 방법 중 하나는 random exponential backoff로 요청을 자동적으로 재 시도 하는 겁니다.

exponential backoff로 retry를 한다는 의미는 요청량 제한 에러에 도달했을 때 잠시 쉬었다가 실패한 요청을 다시 요청한다는 겁니다. 만약 그 요청이 또다시 실패 한다면 잠시 쉬는 시간이 좀 더 늘어나고 그 다음에 요청하는 과정을 반복하는 겁니다. 이 과정은 요청이 성공할 때까지 이루어지게 할 수도 있고 특정 시도 횟수를 지정해서 그 횟수 만큼만 실행하게 할 수도 있습니다.

 

이 접근법은 다음과 같은 이점이 있습니다.

 

* 자동재시도는 crash나 데이터 손실 없이 요청량 제한 에러를 극복할 수 있다는 의미입니다.

* Exponential backoff는 첫번째 시도를 빠르게 시도할 수 있음을 의미하며 처음 몇번의 재시도가 실패할 경우 점점 더 재시도까지의 쉬는 시간이 점점 더 길어진다는 겁니다. 그렇기 때문에 초기에 성공하면 좀 더 빠른 시간안에 에러를 복구 할 수 있습니다.

* Random jitter를 추가하면 동시에 모든 hitting으로부터 재시도를 하는데 도움을 줍니다.

Note. 실패한 요청은 여러분의 per-minute limit에 포함 되므로 계속 재시도를 해도 작동하지 않을 수 있습니다.

 

아래에 이 방법을 이용하는 몇가지 방법을 소개합니다.

 

Example #1: Using the Tenacity library

Tenacity 는 파이썬으로 작성된 아파치 2.0 라이센스 범용 retrying 라이브러리 입니다. 이것을 이용하면 거의 모든 상황에 재시도 동작을 추가하는 작업을 단순화 할 수 있습니다.

 

요청에 exponential backoff를 추가하려면 tenacity.retry decorator를 사용하면 됩니다.

tenacity.wait_random_exponential  함수를 사용하여 random exponential backoff를 요청에 추가하는 방법을 아래 예제에서 보여 줍니다.

 

Note : Tenacity 라이브러리는 별도로 개발한 회사가 있고 그 회사가 배포한 라이브러리 입니다. OpenAI에서 그 안정성이나 보안을 보장하지는 않습니다.

 

import openai  # for OpenAI API calls
from tenacity import (
    retry,
    stop_after_attempt,
    wait_random_exponential,
)  # for exponential backoff


@retry(wait=wait_random_exponential(min=1, max=60), stop=stop_after_attempt(6))
def completion_with_backoff(**kwargs):
    return openai.Completion.create(**kwargs)


completion_with_backoff(model="text-davinci-002", prompt="Once upon a time,")

@retry() 를 사용하며 wait과 stop을 지정합니다. 

재시도 사이 대기시간이 1초에서 60초 사이가 되고 재시도는 6번 시도하라는 겁니다.

이것은 그 아래 함수인 Completion_with_backoff() 함수에 적용 됩니다.

이 함수에서는 openai.Completion.create() api를 호출합니다.

 

그리고 그 함수 밖에서 이 completion_with_backoff()를 호출합니다.

 

Example #2: Using the backoff library

다른 라이브러리로는 backoff 가 있습니다.

Tenacity와 마찬가지로 Backoff 라이브러리는 third-party tool 입니다. OpenAI에서 그 안정성이나 보안성을 보장하지는 않습니다.

 

import backoff  # for exponential backoff
import openai  # for OpenAI API calls


@backoff.on_exception(backoff.expo, openai.error.RateLimitError)
def completions_with_backoff(**kwargs):
    return openai.Completion.create(**kwargs)


completions_with_backoff(model="text-davinci-002", prompt="Once upon a time,")

이 라이브러리는 @backoff.on_exception()이라고 지정합니다.

 

파라미터로는 backoff.expo와 에러메세지인 openai.error.RateLimitError를 전달합니다.

그 아래 함수와 이 함수 호출 부분은 위 예제와 동일합니다.

 

Example 3: Manual backoff implementation

이런 3rd-party 툴을 사용하지 않고 직접 backoff 로직을 구현할 수도 있습니다.

 

# imports
import random
import time

import openai

# define a retry decorator
def retry_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 specified 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


@retry_with_exponential_backoff
def completions_with_backoff(**kwargs):
    return openai.Completion.create(**kwargs)


completions_with_backoff(model="text-davinci-002", prompt="Once upon a time,")

이 에제를 보면 위에서 설명했던 동작인 요청하고 error가 발생하면 재시도를 정해진 횟수만큼 하고 그 재시도 사이의 지연 시간은 점차 증가해 나가는 동작을 하도록 직접 wrapper() 함수 안에 구현을 해 놨습니다.

그리고 최대 재시도 횟수와 지연시간 관련 값들은 retry_with_exponential_backoff() 함수에서 세팅을 했습니다.

그리고 위에 설명한 wrapper() 함수는 이 retry_with_exponential_backoff() 함수에 속해 있습니다.

 

이렇게 설정을 해놓고 이것을 이용하는 방법은 @ decorator를 사용하는 겁니다.

https://builtin.com/software-engineering-perspectives/python-symbol

What Is the @ Symbol in Python and How Do I Use It?

https://dojang.io/mod/page/view.php?id=2427 

파이썬 코딩 도장: 42.1 데코레이터 만들기

@retry_with_exponential_backoff -> 이런식으로 사용합니다.

 

How to maximize throughput of batch processing given rate limits

만약 여러분의 애플리케이션이 고객의 실시간 요청을 처리하는 서비스를 제공한다면 backoff and retry는 요청량 제한 에러를 피하면서 latency를 minimize할 수 있는 아주 좋은 전략입니다.

 

그런데 이렇게 서비스 제공 시간이 중요한게 아니라 얼마나 많은 양을 처리하느냐가 더 중요한 애플리케이션이 있을 수 있습니다. 대량의 batch data를 처리하는 것이 더 중요한 경우이죠. 이런 경우 backoff and retry 말고 여러분이 사용할 수 있는 몇가지 다른 방법이 있습니다.

 

Proactively adding delay between requests

만약 여러분의 앱이 요청량 제한에 걸리고 back off 하고 재시도하고 하는 상황이 계속 반복 되면서 금방 요청량 제한에 걸린다면 남아 있는 시간은 제한량 초과 메세지만 받으면서 요청을 계속 할 수 있습니다.  (예를 들어 1분에 20회가 제한량인데 이 20회가 10초만에 도달하면 나머지 50초는 제한량 초과 메세지만 받게 될 것입니다.) 그러면 이 50초 동안의 요청은 낭비 되는 것이죠. 그것을 하기 위해 사용한 내 시스템의 리소스가 낭비 되는 것입니다. 

 

이 경우 잠재적 해결책이 될 수 있는 것은 여러분의 요청량 제한을 계산하고 그 reciprocal에 맞게 지연 시간을 만드는 것입니다.

reciprocal(예: 만약 여러분의 요청량 제한이 분당 20이라면 각 request마다 지연시간을 3~6초씩 추가 해 주는 것)

 

이렇게 하면 요청량 제한의 상한선에 도달하지 않고 낭비되는 요청도 발생 시키지 않는 상황에 가깝게 만들 수 있습니다.

 

Example of adding delay to a request

# imports
import time
import openai

# Define a function that adds a delay to a Completion API call
def delayed_completion(delay_in_seconds: float = 1, **kwargs):
    """Delay a completion by a specified amount of time."""

    # Sleep for the delay
    time.sleep(delay_in_seconds)

    # Call the Completion API and return the result
    return openai.Completion.create(**kwargs)


# Calculate the delay based on your rate limit
rate_limit_per_minute = 20
delay = 60.0 / rate_limit_per_minute

delayed_completion(
    delay_in_seconds=delay,
    model="text-davinci-002",
    prompt="Once upon a time,"
)

이 소스 코드는 위에 예시로 들었던 상황에 맞게 만든 겁니다.

요청량 제한이 분당 20이라면 delay는 60/20 즉 3초가 됩니다.

 

delay_completion() 함수 안데 time.sleep() 을 이 3초로 했기 때문에 이 함수를 호출할 때 마다 3초씩 기다렸다가 openai.Completion.create() api 를 사용하게 되는 겁니다.

 

그러면 이 1분이라는 시간에 할 수 있는 요청량을 최대한으로 사용할 수 있게 됩니다.

단 10초만에 요청량 제한에 걸려서 나머지 50초는 그냥 손 놓고 있는 상황을 피할 수 있게 되는 거죠.

 

Batching requests

OpenAI api에는 분당 요청량 제한 이외에 분당 요청 토큰에 대한 별도의 제한이 있습니다.

만약 여러분이 분당 요청량 제한에 도달했지만 분당 토큰에 여유가 있는 경우 각 요청에 여러 작업을 batch 함으로서 throughput(처리량)을 증가시킬 수 있습니다.

 

프롬프트들의 batch를 보내는 방법은 일반적인 API 콜과 동일하게 작동합니다.

단지 프롬프트 파라미터가 Single string이 아니라 String 의 List라는 것만 다릅니다.

 

Warning: Response 는 그 prompt 의 순서대로 반환을 하지 않을 수 있습니다. 그렇기 때문에 index 필드를 사용하여 그 response를 prompt 의 파라미터 순서와 맞게 일치시키는 작업을 반드시 해야 합니다.

 

Example without batching

import openai  # for making OpenAI API requests


num_stories = 10
prompt = "Once upon a time,"

# serial example, with one story completion per request
for _ in range(num_stories):
    response = openai.Completion.create(
        model="curie",
        prompt=prompt,
        max_tokens=20,
    )

    # print story
    print(prompt + response.choices[0].text)

 

Example with batching

import openai  # for making OpenAI API requests


num_stories = 10
prompts = ["Once upon a time,"] * num_stories

# batched example, with 10 stories completions per request
response = openai.Completion.create(
    model="curie",
    prompt=prompts,
    max_tokens=20,
)

# match completions to prompts by index
stories = [""] * len(prompts)
for choice in response.choices:
    stories[choice.index] = prompts[choice.index] + choice.text

# print stories
for story in stories:
    print(story)

위의 예제는 prompt 'once upon a time' 단일 파라미터로 하는 openai.Completion.create() 호출을 for 문을 통해서 10번 일으켰습니다.

즉 10번 요청을 한 것입니다.

 

그런데 두번째 예제는 prompt 자체를 Once upon a time이 10번 들어가 있는 리스트로 만든 다음 이 리스트를 파라미터로 전달해서 openai.Completion.create() api를 호출 했습니다.

즉 1번 요청한 것입니다.

 

이렇게 함으로서 요청량을 줄일 수 있게 되는 것입니다.

 

두번째 예제의 첫번째 for 문에서는 response를 request의 index 와 맞게 매치 시키는 작업을 하고 있습니다.

그래서 질문 + 응답 이런 식으로 stories[] 에 저장되게 했습니다.

 

이제 이 stories[]를 두번째 for 문처럼 print 하기만 하면 해당 질문과 그에 대한 답변 식으로 출력이 됩니다.

 

Example parallel processing script

대량의 API 요청을 parallel (병렬) 처리하기 위한 예제 스크립트는 아래에 있습니다.

https://github.com/openai/openai-cookbook/blob/2f5e350bbe66a418184899b0e12f182dbb46a156/examples/api_request_parallel_processor.py

GitHub - openai/openai-cookbook: Examples and guides for using the OpenAI API

 

이 스크립트에는 다음과 같은 몇가지 편리한 기능들이 결합돼 있습니다.

* 대용량 작업의 메모리 부족을 방지하기 위해 파일에서 request를 스트림하기

* 요청을 동시에 해서 throughput (처리량)을 극대화 하기

* 요청량 제한을 넘지 않기 위해 request 와 token 사용을 모두 제한하기

* 데이터 누락을 피하기 위해 실패한 요청을 재시도 하기

* 오류를 기록해서 그 요청에 대한 문제를 진단할 수 있게 하기

 

아 코드를 그대로 사용하시거나 필요에 맞게 수정해서 사용하시면 됩니다.