Batch

Using the Batch endpoint allows you to submit a collection of requests at one time and come back later when all the request have completed.

The overall process is:

1. create a JSONL file with the requests that you would like to make
2. upload the JSONL file using the Files API while setting purpose to batch
3. Use the Batch API to submit a batch job that references the file_id returned from the previous step
4. Query the Batch API with the batch id returned from the previous step to see if the task has completed
5. Fetch the Batch output and parse the output

Limitations

• The Batch API is limited for use with any of the locally deployed models. See the model list for which models are deployed locally.
• The current limit is 4 parallel requests at a time per batch submission
• There is currently not a limit on the max number of jobs per batch submission
• Batch submissions are limited to using the following endpoints:
  • Chat Completions API - /v1/chat/completions
  • Embeddings - /v1/embeddings
  • Responses API - /v1/responses
• Batch submissions are limited to using a single endpoint for the entire request

Creating the JSONL file

The JSONL file consists of multiple json objects. Each json object represents a request to send to an endpoint where the body field is the same as the paramaters that would normally be sent to that endpoint. See the example below for using the Chat Completion API:

{"custom_id": "task-0", "method": "POST", "url": "/v1/chat/completions", "body": {"messages": [{"role": "system", "content": "You are an AI assistant that helps people find information."}, {"role": "user", "content": "When was Microsoft founded?"}]}}
{"custom_id": "task-1", "method": "POST", "url": "/v1/chat/completions", "body": {"messages": [{"role": "system", "content": "You are an AI assistant that helps people find information."}, {"role": "user", "content": "When was the first XBOX released?"}]}}
{"custom_id": "task-2", "method": "POST", "url": "/v1/chat/completions", "body": {"messages": [{"role": "system", "content": "You are an AI assistant that helps people find information."}, {"role": "user", "content": "Why is the sky blue?"}]}}

and the following for using the Requests API:

{"custom_id": "task-0", "method": "POST", "url": "/v1/responses", "body": {"input": [{"role": "system", "content": "You are an AI assistant that helps people find information."}, {"role": "user", "content": "When was Microsoft founded?"}]}}
{"custom_id": "task-1", "method": "POST", "url": "/v1/responses", "body": {"input": [{"role": "system", "content": "You are an AI assistant that helps people find information."}, {"role": "user", "content": "When was the first XBOX released?"}]}}
{"custom_id": "task-2", "method": "POST", "url": "/v1/responses", "body": {"input": [{"role": "system", "content": "You are an AI assistant that helps people find information."}, {"role": "user", "content": "Why is the sky blue?"}]}}

Examples

Chat Completions API

The following example shows how to write a python script that:

• uploads a jsonl file
• calls the Batch API setting the endpoint to /v1/chat/completions
• waits for the batch submission to complete checking on it's status
• parses the response output

This is the jsonl file that will be uploaded:

{"custom_id": "task-0", "method": "POST", "url": "/v1/chat/completions", "body": {"messages": [{"role": "system", "content": "You are an AI assistant that helps people find information."}, {"role": "user", "content": "When was Microsoft founded?"}]}}
{"custom_id": "task-1", "method": "POST", "url": "/v1/chat/completions", "body": {"messages": [{"role": "system", "content": "You are an AI assistant that helps people find information."}, {"role": "user", "content": "When was the first XBOX released?"}]}}
{"custom_id": "task-3", "method": "POST", "url": "/v1/chat/completions", "body": {"messages": [{"role": "system", "content": "You are an AI assistant that helps people find information."}, {"role": "user", "content": "Why is the sky blue?"}]}}

from openai import OpenAI
from time import sleep
import json

api_key = "sk-XXXXXXXX"
base_url = "https://api.navigator.ai.ufl.edu/v1/"

client = OpenAI(base_url=base_url, api_key=api_key)

file = "batch-chat.jsonl" # path to the batch file
model = "gpt-oss-20b" # or any other local model

loop_count = 25 # number of times to check the status of the batch
wait_time = 3 # amount of time to wait inbetween each check

file_id = None

try:
    # make the request to upload the jsonl file
    batch_input_file = client.files.create(
        file=open(file, "rb"),
        purpose="batch",
        extra_headers={
            "x-litellm-model": model
        }
    )
    file_id = batch_input_file.id
    print(f"File id is:\n{file_id}")
except Exception as e:
    print(f"Exception occurored during file creation: {e}")
    exit(1)

try:
    # create the batch
    print("Creating the batch....")
    batch = client.batches.create( # re use client from above
        input_file_id=file_id,
        endpoint="/v1/chat/completions",
        completion_window="24h",    )
    print(f"Batch created: {batch}")
except Exception as e:
    print(f"Exception occurored during batch creation: {e}")
    exit(1)

try:
    loopTotal = loop_count
    sleepInterval = wait_time
    i = 1

    # Use a loop to check on the status of the Batch
    while i <= loopTotal:
        print(f"Entering loop {i} of {loopTotal}")
        
        batch_status = client.batches.retrieve(batch.id)

        if batch_status.status == "completed":
            print(f"Batch has completed with a status of: {batch_status.status}")
            print()
            break
        else:
            print(f"Batch has not completed yet. Status is: {batch_status.status}")

        print(f"Sleeping for {sleepInterval} seconds")
        sleep(sleepInterval)
        print(f"Exiting loop {i} of {loopTotal}")
        print()
        i = i + 1
except Exception as e:
    print(f"Exception occurored while retrieving batch request: {e}")
    exit(1)

try:
    file_response = client.files.content(batch_status.output_file_id)
    file_contents = file_response.content.decode("utf-8")

    for line in file_contents.strip().split("\n"):
        if line:
            result = json.loads(line)
            custom_id = result["custom_id"]
            
            # Check for errors in the individual request
            if "error" in result and result["error"]:
                print(f"Request {custom_id} failed: {result['error']}")
                continue
                
            # Extract response payload
            response_body = result["response"]["body"]
            choices = response_body["choices"]
            for i in choices:
                if i["finish_reason"] != "stop":
                    print(f"{custom_id} did not have a stop reason: {i["finish_reason"]}")
                    continue

                index = i["index"]
                message = i["message"]["content"]
                print(f"Result for {custom_id}/{index}: {message}\n")

    exit(0)
except Exception as e:
    print(f"Exception occurored while processing batch request: {e}")
    exit(1)

Responses API

The following example shows how to write a python script that:

• uploads a jsonl file
• calls the Batch API setting the endpoint to /v1/responses
• waits for the batch submission to complete checking on it's status
• parses the response output

This is the jsonl file that will be uploaded:

{"custom_id": "task-0", "method": "POST", "url": "/v1/responses", "body": {"input": [{"role": "system", "content": "You are an AI assistant that helps people find information."}, {"role": "user", "content": "When was Microsoft founded?"}]}}
{"custom_id": "task-1", "method": "POST", "url": "/v1/responses", "body": {"input": [{"role": "system", "content": "You are an AI assistant that helps people find information."}, {"role": "user", "content": "When was the first XBOX released?"}]}}

from openai import OpenAI
from time import sleep
import json

api_key = "sk-xxxxxxx"
base_url = "https://api.navigator.ai.ufl.edu/v1/"

client = OpenAI(base_url=base_url, api_key=api_key)

file = "batch-responses.jsonl"
model = "gpt-oss-20b" # or any other local model

loop_count = 25 # number of times to check the status of the batch
wait_time = 3 # amount of time to wait inbetween each check

file_id = None

try:
    # make the request to upload the jsonl file
    batch_input_file = client.files.create(
        file=open(file, "rb"),
        purpose="batch",
        extra_headers={
            "x-litellm-model": model
        }
    )
    file_id = batch_input_file.id
    print(f"File id is:\n{file_id}")
except Exception as e:
    print(f"Exception occurored during file creation: {e}")
    exit(1)

try:
    print("Creating the batch....")
    batch = client.batches.create( # re use client from above
        input_file_id=file_id,
        endpoint="/v1/responses",
        completion_window="24h",
    )
    print(f"Batch created: {batch}")
except Exception as e:
    print(f"Exception occurored during batch creation: {e}")
    exit(1)

try:
    loopTotal = loop_count
    sleepInterval = wait_time
    i = 1

    while i <= loopTotal:
        print(f"Entering loop {i} of {loopTotal}")
        
        batch_status = client.batches.retrieve(batch.id)

        if batch_status.status == "completed":
            print(f"Batch has completed with a status of: {batch_status.status}")
            print()
            break
        else:
            print(f"Batch has not completed yet. Status is: {batch_status.status}")

        print(f"Sleeping for {sleepInterval} seconds")
        sleep(sleepInterval)
        print(f"Exiting loop {i} of {loopTotal}")
        print()
        i = i + 1
except Exception as e:
    print(f"Exception occurored while retrieving batch request: {e}")
    exit(1)

try:
    file_response = client.files.content(batch_status.output_file_id)
    file_contents = file_response.content.decode("utf-8")

    for line in file_contents.strip().split("\n"):
        if line:
            result = json.loads(line)
            custom_id = result["custom_id"]

            # This is some extra debug output if you want to see what the result output shape is
            # print(f"DEBUG:\nResult is:\n=========\n{result}\n=========\n")
            
            # Check for errors in the individual request
            if "error" in result and result["error"]:
                print(f"Request {custom_id} failed: {result['error']}")
                continue
                
            # Extract response payload
            response_body = result["response"]["body"]
            output = response_body["output"]
            for i in output:
                if i["type"] != "message":
                    # If the type is not message then it is another type of response like the reasoning output.
                    # Let's skip that
                    print(f"Skipping because type is: {i["type"]}")
                    continue

                message = i["content"][0]["text"]
                print(f"Result for {custom_id}: {message}\n")

    exit(0)
except Exception as e:
    print(f"Exception occurored while processing batch request: {e}")
    exit(1)