오늘 다룰 내용은 아래 페이지에 있습니다.

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

How to stream completions

기본적으로 OpenAI Completions api 콜을 하면 Request된 Prompt에 대해 작업을 완료 한 다음 하나의 response로 답을 보내 옵니다.

davinci 모델의 경우 아주 긴 completion을 생성하게 되면 응담까지 몇초가 걸릴 수 있습니다. 2022년 8월 현재 text-davinci-002 모델의 응답시간은 일반적으로 100 completion token당 1초에서 2초 정도가 걸립니다.

이 정도도 기다리지 않고 더 빨리 응답을 받고 싶으면 Stream을 사용하실 수 있습니다.

이렇게 되면 Request된 Prompt에 대한 답을 완료하기 전에 일부분에 대한 답을 받아서 인쇄를 시작하거나 Completion 의 시작 부분을 처리할 수 있습니다.

Completions를 stream 하려면 API를 호출 할 때 stream=True를 사용하시면 됩니다.  이렇게 하면 data-only 서버가 보낸 이벤트와 같이 텍스트를 streams back 하는 object를 return 하게 됩니다. (답변의 일부분이라도 완료 되면 그 일 부분들을 계속 해서 stream으로 응답을 보내 온 다는 얘기 입니다.)

Downsides

주의 할 점은 프로덕션 어플리케이션에서 stream=True를 사용하면 completions 의 내용을 조정하기가 더 어려워 집니다. 

(streaming 되는 동안 답변의 일부분만 가지고 있기 때문에 전체 답변에 대한 보정이나 처리 작업을 하는데는 한계가 있다는 얘기 입니다.)

이 Streaming response의 다른 작은 단점은 Response에 사용된 토큰 수를 알려 주는 정보가 없다는 겁니다. 

해당 토큰수를 알려면 모든 응답을 결합한 후 toktoken을 사용해서 직접 계산 해야 합니다.

Example code

아래에 이 streaming completions를 어떻게 사용하는지에 대한 파이썬 예제가 있습니다.

# imports
import openai  # for OpenAI API calls
import time  # for measuring time savings

우선 openai를 import 합니다. 그리고 응답 시간을 알아보기 위해 time 모듈도 import 합니다.

# Example of an OpenAI Completion request
# https://beta.openai.com/docs/api-reference/completions/create

# record the time before the request is sent
start_time = time.time()

# send a Completion request to count to 100
response = openai.Completion.create(
    model='text-davinci-002',
    prompt='1,2,3,',
    max_tokens=193,
    temperature=0,
)

# calculate the time it took to receive the response
response_time = time.time() - start_time

# extract the text from the response
completion_text = response['choices'][0]['text']

# print the time delay and text received
print(f"Full response received {response_time:.2f} seconds after request")
print(f"Full text received: {completion_text}")

이 방법은 stream을 사용하지 않은 일반적인 방법입니다.

openai.Completion.create() 파라미터를 보면 stream=True 부분이 없습니다. 즉 응답받는데 streaming을 사용하지 않는 겁니다.

openai.COmpletion.create() api를 호출하기 전 시간을 start_time에 담고 호출 한 후의 시간을 response_time에 담았습니다.

response 받은 데이터 중 choices의 첫번째 아이템에 있는 text 부분만 떼어내서 completion_text 에 담습니다.

그런 다음 전체 소요 시간을 print 하고 두번째 print에서는 completion_text를 print 합니다.

저는 위 소스코드를 조금 수정해서 openaiapikey를 text 파일에서 읽어와서 인증 받는 부분을 추가했고 Completion api를 호출할 때 davinci보다 저렴한 ada 모델을 사용하는 것으로 했습니다.

응답은 2.86초 걸렸고 그 내용은 Full text received 에 있습니다.

A streaming completion request

# Example of an OpenAI Completion request, using the stream=True option
# https://beta.openai.com/docs/api-reference/completions/create

# record the time before the request is sent
start_time = time.time()

# send a Completion request to count to 100
response = openai.Completion.create(
    model='text-ada-001', #model='text-davinci-002',
    prompt='1,2,3,',
    max_tokens=193,
    temperature=0,
    stream=True,  # this time, we set stream=True
)

# create variables to collect the stream of events
collected_events = []
completion_text = ''
# iterate through the stream of events
for event in response:
    event_time = time.time() - start_time  # calculate the time delay of the event
    collected_events.append(event)  # save the event response
    event_text = event['choices'][0]['text']  # extract the text
    completion_text += event_text  # append the text
    print(f"Text received: {event_text} ({event_time:.2f} seconds after request)")  # print the delay and text

# print the time delay and text received
print(f"Full response received {event_time:.2f} seconds after request")
print(f"Full text received: {completion_text}")

이번에는 Streaming을 사용하는 에제입니다.

openai.Completion.create() 의 파라미터를 보면 stream=True 를 추가 하신것을 보실 수 있습니다.

이것이 streaming을 사용하는 방법 입니다.

그 이후의 코드들은 이 스트리밍을 어떻게 receive 하는지 print 해주기 위해 만든 스크립트 들 입니다.

for 문이 있는데 이 for 문에서는 response 에 발생한 이벤트만큼 실행 됩니다.

이벤트가 일어날 때마다 그 이벤트 시각에서 start_time을 뺍니다. 즉 Completion api 콜 한 시점에서 그 이벤트가 일어난 시점까지의 시간을 알기 위해서 이런 계산을 합니다. 그 값은 event_time에 저장 됩니다.

그리고 collected_events에 이 이벤트를 append 합니다.

그리고 그 이벤트에서 받은 reponse에서 choices의 첫번째 아이템에 있는 text 내용을 가져 옵니다.

completion_text에는 지금까지 받은 evnet_text에 현재 받은 event_text를 추가 합니다.

그리고 for  문 안에 있는 print 문에서는 응답 내용과 그 응답을 받기 까지 얼마나 걸렸는지를 print 해 줍니다.

for 문이 완료 되면 전체 소요 시간과 전체 응답을 print 합니다.

그러면 아래와 같은 응답을 받습니다.

.......

Stream을 사용하지 않았을 때는 전체 결과 값을 받는데까지 2.86초 걸렸습니다.

그런데 이렇게 Stream을 사용하니까 응답의 시작부분을 0.35초 후부터 받기 시작해서 전체 메세지를 받는데까지 1.52초가 걸렸습니다.

이렇게 streaming 해서 받은 값들을 모두 합한 값은 위에 stream을 사용하지 않았을 때와 동일합니다.

Time comparison

쿡북의 예에서는 Davinci 모델을 사용해서 Stream을 사용하지 않은 경우는 7.32초가 걸렸고 사용한 경우는 7.25초가 걸렸습니다.

둘 다 거의 비슷하게 걸렸지만 streaming을 사용한 경우에는 0.16초가 지나면서부터 응답의 일부를 받기 시작해서 전체 응답을 받기까지 7.25초가 걸린 겁니다.

How to count tokens with tiktoken

오늘 다를 예제는 아래 CookBook 페이지에 있는 토큰 관련 에제 입니다.

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

tiktoken 은 openAI 에서 제공하는 오픈 소스 토크나이저 입니다.

이 모듈은 문자열과 인코딩 방법을 전달하면 그 문자열을 토큰으로 나눠서 리스트에 담아서 반환합니다.

예를 들어 tiktoken is great! 이라는 문자열을 gpt2라는 인코딩을 사용하면 어떻게 토큰이 나누어 지는지 질문하면 아래처럼 response를 줍니다.

["t", "ik", "token", " is", " great", "!"]

이 문자열은 총 6개의 토큰으로 구성 돼 있습니다.

대개 1개의 단어가 1개의 토큰이 되지만 꼭 그렇지만은 않습니다.

그리고 인코딩 방법에 따라서 토큰을 나누는게 조금 다릅니다.

GPT-3 같은 모델은 텍스트를 토큰들의 집합으로 봅니다. 그렇기 때문에 문자열을 토큰으로 분할하게 되면 유용하게 사용할 수 있습니다. 

이 모델이 사용하기에 너무 많은 토큰의 갯수는 아닌지를 알아 볼 수도 있고 이 토큰 갯수에 따라서 API 요청에 대한 과금이 매겨지기 때문에 이 요청을 하면 돈이 얼마가 들지 미리 알 수 있습니다.

GPT-3의 모델들은 하나의 인코딩만을 사용하는 것이 아니고 모델별로 다른 인코딩을 사용하는 경우가 있습니다.

이 tiktoken은 OpenAI 모델들이 사용하는 아래 세가지 인코딩 방법을 제공합니다.

p50k_base는 gpt2 와 많은 부분 겹치는 인코딩 입니다. 코드가 아닌 어플리케이션인 경우 일반적으로 동일하게 토큰을 나눕니다.

Tokenizer libraries and languages

gpt2 인코딩은 다음과 같은 많은 프로그래밍 언어에서 사용할 수 있는 tokenizer가 제공 됩니다.

• Python: tiktoken (or alternatively GPT2TokenizerFast)
• JavaScript: gpt-3-encoder
• .NET / C#: GPT Tokenizer
• Java: gpt2-tokenizer-java
• PHP: GPT-3-Encoder-PHP

(OpenAI는 3rd party 라이브러리에 대해서는 그 성능과 리스크에 대한 보장을 하지 않습니다.)

p50k_base와 cl100k_base 인코딩의 경우 2023년 1월 현재 tiktoken에서만 tokenizer가 가능합니다.

• Python: tiktoken

How strings are typically tokenized

영어로 된 텍스트인 경우 토큰의 길이는 일반적으로 문자 하나이거나 단어 하나가 한개의 토큰이 됩니다. (e.g., "t" or " great")

다른 언어에서는 한문자보다 짧거나 한단어보다 길 수 있습니다.

공백은 일반적으로 단어의 시작으로 그룹화 됩니다. (e.g., " is" instead of "is " or " "+"is")

OpenAI Tokenizer를 사용해서 문자열이 어떻게 tokenized 되는지 빠르게 체크할 수 있습니다.

0. Install tiktoken

이 tiktoken을 사용하려면 먼저 이 모듈을 인스톨 해야 합니다.

pip install tiktoken

1. Import tiktoken

그러면 이 tiktoken을 import 할 수 있습니다.

import tiktoken

2. Load an encoding

tiktoken.get_encoding()을 사용해서 인코딩을 로딩 합니다. 이 때 인코딩 이름을 파라미터로 전달합니다.

처음 이 작업을 할 때는 인터넷에 연결 돼 있어야 합니다.

한번 로딩한 다음에는 인터넷이 연결 돼 있지 않아도 사용할 수 있습니다.

encoding = tiktoken.get_encoding("gpt2")

3. Turn text into tokens with encoding.encode()

.encode() 메소드는 문자열을 token integer들로 변환합니다.

위에서 tiktoken is great! 이 아래와 같은 토큰들로 나누어 진다고 했습니다.

["t", "ik", "token", " is", " great", "!"]

그렇다면 t는 83이고 ik 는 1134 이고 token 은 30001 ..... 마지막으로 ! 는 0 이 됩니다.

아래 함수는 토큰 갯수를 계산해서 반환해 주는 함수 입니다.

def num_tokens_from_string(string: str, encoding_name: str) -> int:
    """Returns the number of tokens in a text string."""
    encoding = tiktoken.get_encoding(encoding_name)
    num_tokens = len(encoding.encode(string))
    return num_tokens

함수 이름은 num_tokens_from_string() 이고 파라미터로는 String 과 인코딩 이름이 전달 됩니다.

그리고 반환 타입은 integer 입니다.

get_encoding()을 사용해서 사용할 encoding을 정의 합니다.

그리고 len() 함수를 사용해서 인코드해서 얻은 리스트의 아이템 갯수를 num_tokens 변수에 넣습니다.

이 아이템들이 각각의 토큰들 입니다.

이 함수를 사용해서 토큰의 갯수를 아래와 같이 얻을 수 있습니다.

4. Turn tokens into text with encoding.decode()

인코딩이 있으면 디코딩도 있겠죠.

.decode()를 사용하면 숫자로 된 토큰 정보를 문자로 바꿀 수 있습니다.

Warning : .decode()는 single tokens에 적용될 수 있지만 utf-8 boundaries가 아닌 경우 토큰의 손실이 발생 할 수 있습니다.

single tokens의경우 decode_single_token_bytes() 는 단일 정수 토큰을 그것이 나타내는 바이트로 안전하게 변환합니다. 

여기서 각 string 앞에 있는 b가 가리키는 것은 이 String들이 byte string들이라는 의미 입니다.

5. Comparing encodings

그렇다면 각 인코딩 별로 어떻게 입력값을 분할 해서 토큰으로 나누는지를 알아 보겠습니다.

def compare_encodings(example_string: str) -> None:
    """Prints a comparison of three string encodings."""
    # print the example string
    print(f'\nExample string: "{example_string}"')
    # for each encoding, print the # of tokens, the token integers, and the token bytes
    for encoding_name in ["gpt2", "p50k_base", "cl100k_base"]:
        encoding = tiktoken.get_encoding(encoding_name)
        token_integers = encoding.encode(example_string)
        num_tokens = len(token_integers)
        token_bytes = [encoding.decode_single_token_bytes(token) for token in token_integers]
        print()
        print(f"{encoding_name}: {num_tokens} tokens")
        print(f"token integers: {token_integers}")
        print(f"token bytes: {token_bytes}")

이 compare_encodings() 함수는 string을 입력 파라미터로 받습니다. return 값은 없고 그냥 이 함수 안에서 계산해서 그 정보를 print 해 줍니다.

첫번째 있는 print 문은 Example string : 입력 받은 문자열 을 출력합니다.

그 다음 for 문이 있는데 각 인코딩 별로 이 for 문안에 있는 작업을 해 줄 겁니다.

우선 get_encoding() 을 사용해서 사용할 encoding을 세팅해 줍니다.

그리고 encode()를 사용해서 토큰으로 나눠 줍니다. (결과값을 리스트 형식으로 반환 합니다.)

그리고 len()을 이용해서 이 리스트안에 있는 아이템 갯수들을 num_tokens에 담습니다.

그리고 decode_single_token_bytes()를 사용해서 각 바이트별 텍스트를 token_bytes에 담습니다.

그 다음 인코딩 이름과 토큰 리스트의 인티저 값 그리고 각 바이트 별 문자를 표시해 줍니다.

이렇게 한 단어인데 긴것을 넣어 봤습니다. 이것은 반체제주의라는 의미라고 합니다.

gpt2와 p50k_base는 5개의 토큰으로 나눴고 cl100k_base는 6개의 토큰으로 나눴습니다.

한번 OpenAI의 토크나이저 페이지에 가서 같은 문자열을 넣고 확인해 보겠습니다.

gpt2와 p50k_base 인코딩과 동일한 결과를 볼 수 있습니다.

이 OpenAI 토크나이저 페이지는 아래에 가시면 볼 수 있습니다.

https://platform.openai.com/tokenizer

그 다음 계산식도 넣어보고 일본어도 넣어 봤습니다.

그리고 한글도 넣어 봤습니다.

그리고 긴 영어 문장도 한번 넣어 봤구요.

이 예제들에서는 gpt2와 p50k_base가 동일하게 나왔습니다.

cl100k_base는 어떤 때는 다른 두개보다 더 많은 토큰으로 나누었고 또 어떤 때는 토큰수가 더 적을 때도 있었습니다.

인코딩 방법에 따라 이렇게 다른 결과들이 나옵니다.

지난 글까지 해서 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

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

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

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

@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

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

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

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

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

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

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

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