I am finding the Batch API very useful. It allows me to apply the magic of LLMs to a range of use cases that were not cost effective in the past. It means that I can divide the tasks that I want to done by an LLM into those that I need a rapid response to (chat) and those tasks that I can wait an hour or more for (batch). Here is some code I am using.
I have created three functions. The first is to create a .jsonl file from a plain text file. To allow for a mult-line prompt, you put a [begin-prompt] tag at the front of each prompt and an [end-prompt] flag at the end of each prompt.
# Create a jsonl file from a txt file containing prompts (requests)
# To support multi-line prompts, each prompt requires as [begin-prompt] and [end-prompt] tag on a separate line
ctr = 0
with open(input_file_name, 'r') as file:
# Initialize variables
request = "" # To store the concatenated string
capturing = False # Flag to check if we are between the tags
# Initialize a list to hold all JSON records
json_records = []
# Read the file line by line
for line in file:
# Check for the beginning tag
if '[begin-prompt]' in line:
capturing = True
ctr=ctr+1
continue # Skip the line with [begin-prompt]
# Check for the ending tag
elif '[end-prompt]' in line:
capturing = False
# Process the collected request here if needed, for example:
# Create a JSON record using the schema
json_record = {
"custom_id":"Prompt-"+str(ctr),
"method": "POST",
"url": "/v1/chat/completions",
"body": {
"model": model,
"messages": [
{"role": "system", "content": content},
{"role": "user", "content": request}
],
"max_tokens": max_tokens
}
}
# Append the JSON record to the list
json_records.append(json_record)
# Reset the request for the next capture
request = ""
continue # Skip the line with [end-prompt]
# If we are between the tags, add the line to request
if capturing:
request += line
# Write the list of JSON records to the output file
with open(output_file_name, 'w') as output_file:
for record in json_records:
json.dump(record, output_file)
output_file.write('\n')
The second function takes the .jsonl file from the first step and creates and submits a batch request – see below.
def create_batch(client, input_file_name):
# Upload the JSONL file
batch_input_file = client.files.create(
file=open(input_file_name, "rb"),
purpose="batch"
)
batch_input_file_id = batch_input_file.id
# Create the batch using the file id of the uploaded jsonl file
batch_object = client.batches.create(
input_file_id=batch_input_file_id,
endpoint="/v1/chat/completions",
completion_window="24h",
metadata={
"description": "Test Batch - Student Experience"
}
)
# Retrieve batch ID
batch_id = batch_object.id
# Retrieve the batch object to check the status
batch_object = client.batches.retrieve(batch_id)
# Batch statuses that indicate the job is still running
running_statuses = ["validating", "in_progress", "finalizing", "cancelling"]
# Loop until batch processing is finished
while batch_object.status in running_statuses:
# Check the current status
print(f"Current status: {batch_object.status}")
# Wait for a short period before checking again (e.g., 10 seconds)
time.sleep(10)
# Retrieve the updated batch object
batch_object = client.batches.retrieve(batch_id)
# Once the loop exits, check the final status
print(f"Final status: {batch_object.status}")
return batch_object
Once the batch processing completes, this function creates a plain text file with the prompt id, and response from the LLM.
def show_batch_output(batch_object, content_file_name):
# Get the content from the batch object's output file
output_file_response = client.files.content(batch_object.output_file_id)
print(output_file_response)
json_data = output_file_response.content.decode('utf-8')
print('\n json_data', json_data)
# Open the specified file in write mode
with open(content_file_name, 'w') as file:
# Split the data into individual JSON records by line
for line in json_data.splitlines():
# Parse the JSON record (line) to validate it
json_record = json.loads(line)
# Extract and print the custom_id
custom_id = json_record.get("custom_id")
file.write(f"\n Prompt ID: {custom_id}\n")
# Navigate to the 'choices' key within the 'response' -> 'body'
choices = json_record.get("response", {}).get("body", {}).get("choices", [])
# Loop through the choices to find messages with the 'assistant' role
for choice in choices:
message = choice.get("message", {})
if message.get("role") == "assistant":
assistant_content = message.get("content")
file.write(f"\n {assistant_content}\n")
print(f"\n Finished processing of batch output file. JSON records have been saved to {content_file_name}")
Here is an example of the usage where I translated the text from a Jules Verne novel from English to French. I added [begin-prompt] and [end-prompt] tags to each chapter. I also added the prompt request “Convert the following text from English to French”. I had assumed when I wrote the program that I would have a mixture prompt requests in a single batch file.
input_file_name='julesverne.txt'
output_file_name = 'julesverne.jsonl'
content_file_name = 'julesverne_translated.txt'
max_tokens = 10000
model = "gpt-4o-2024-08-06"
content = "You are an expert in English to French translation"
print('\n Creating a jsonl file...')
# Step 1 create a jsonl file from a plain text file
create_jsonl_from_text(input_file_name, output_file_name, max_tokens, model, content)
print('\n Creating a batch ...')
# Step 2 Create and process a batch by first uploading the jsonl file and creating the batch
batch_object = create_batch(client,output_file_name)
# Step 3 Parse the output and get the content of the response
show_batch_output(batch_object,content_file_name) ype or paste code here