Batch

Initialize a Batch

You can initialize a batch by passing the path to a JSONL file. File should be in OpenAI format.

from langbatch import OpenAIChatCompletionBatch

batch = OpenAIChatCompletionBatch("data.jsonl")
print(batch.id)

You can also pass a list of requests to the create method to create a batch.

batch = OpenAIChatCompletionBatch.create([
    {"messages": [
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is the capital of the moon?"}
    ]},
    {
        "messages": [
            {"role": "system", "content": "You are a helpful assistant."},
            {"role": "user", "content": "What is the capital of the moon?"}
        ]
    }
])

Info

When you initialize a batch, it is not started automatically. You need to call the start method to start the batch job.

Start a Batch

You can start a batch by calling the start method.

batch.start()

Get Batch Status

You can get the status of a batch by calling the get_status method.

batch.get_status()

Retry Batch

Incase of any failure with batch job due to rate limits, exceeded 24h wait time or other issues, you can retry the same batch by calling the retry method.

batch.retry()

Info

A batch instance in LangBatch can be retried multiple times. It will internally create multiple batch jobs until the batch is successful.

Get Batch Results

You can get the results of a completed batch by calling the get_results method.

successful_results, unsuccessful_results = batch.get_results()

for result in successful_results:
    print(result)

Get Batch Results File

You can get the results file of a completed batch by calling the get_results_file method. This file will be in OpenAI Batch Result JSONL format.

file_path = batch.get_results_file()

with open(file_path, "r") as file:
    results = file.readlines()

for result in results:
    print(result)

Get Unsuccessful Requests

You can get the unsuccessful requests of a batch by calling the get_unsuccessful_requests method. This will be useful to retry failed requests or to debug the issues with the requests.

unsuccessful_requests = batch.get_unsuccessful_requests()

for request in unsuccessful_requests:
    print(request)

Get Requests by Custom IDs

You can get the requests of a batch by passing the custom ids to the get_requests_by_custom_ids method. This will be useful to get the requests to debug the issues with the requests.

custom_ids = ["custom_id_1", "custom_id_2"]
requests = batch.get_requests_by_custom_ids(custom_ids)

for request in requests:
    print(request)

Save Batch

You can save a batch by calling the save method. This will be useful to keep track of created batches and resume the batch job later.

# save batch
batch.save()

# Save Batch to Custom Storage
storage = CustomStorage()
batch = OpenAIChatCompletionBatch("data.jsonl")
batch.save(storage=storage)

Info

By default, File based storage will be used to save the batch. You can also pass a custom storage instance when initializing the batch. For example, you can save the batch to a cloud storage.

Load Batch

You can load a batch by calling the load method. This will be useful to resume the batch job later.

# load batch
batch = OpenAIChatCompletionBatch.load("batch_id")

# Load Batch from Custom Storage
storage = CustomStorage()
batch = OpenAIChatCompletionBatch.load("batch_id", storage=storage)

You can pass additional kwargs to the batch when loading the batch.

from openai import OpenAI
client = OpenAI(api_key="your-api-key", base_url="your-base-url")

# Load batch
batch = OpenAIChatCompletionBatch.load("batch_id", batch_kwargs={"client": client})

Info

This is needed only for provider classes where you usually pass client objects when initializing the batch. Usually for OpenAI API supported providers like Azure OpenAI.