Efficient Overnight Processing of Large Document Sets Using Batch Inference
Imagine you need to classify and summarize one million support tickets stored in object storage by the following morning. Processing these documents one by one through a real ti...
Imagine you need to classify and summarize one million support tickets stored in object storage by the following morning. Processing these documents one by one through a real-time API might seem like the logical approach, but it isn't suitable. This method is time-consuming, expensive due to token costs, and prone to disruptions like network failures.
This article argues that most large language model (LLM) tasks are throughput-oriented rather than conversational. Tasks such as categorizing documents, summarizing them, tagging records, and processing backlogs are better suited for batch processing than real-time APIs, which are optimized for latency. By using batch inference—where you send bulk requests and retrieve results upon completion—you save both time and money, especially when deadlines are more flexible.
Project Overview
Consider this scenario:
- Documents: 1,000,000 plain-text documents, averaging 1,200 tokens each (around 900 words)
- Task: Classify each document into one of eight categories and provide a 3-4 sentence summary
- Deadline: Results are needed by the next morning
- Model: Utilize GPT-5 mini via a serverless inference platform, billed at $0.25 per million input tokens and $2.00 per million output tokens in real-time, with batch processing costing up to half these rates.
For cost-effectiveness, classification and summarization should be performed in a single request per document. This approach reduces the number of requests and simplifies the process, resulting in 1.5 billion input tokens and 200 million output tokens for the entire job.
Why Real-time Inference Isn't Suitable
Before implementing any solution, consider whether a real-time API could meet the deadline. With limitations on requests and tokens per minute, real-time processing would take too long and is prone to failure from transient issues. Batch inference, however, circumvents these problems by using separate quotas and retrying failed requests automatically.
Batch jobs have a 24-hour completion window, making them ideal for tasks where immediate results are unnecessary. For tasks requiring instant results, real-time APIs remain the best option.
Pre-requisites for Batch Inference
To begin, you'll need the following:
- A Tier 3 or higher account with the inference platform.
- A prepaid balance sufficient to cover estimated costs.
- A model access key for authentication.
Batch inference supports only one model per job, so different model requirements necessitate separate jobs.
Planning the Batch Job
Batch inference imposes several limits:
- Maximum 50,000 requests per input file
- Maximum 200 MB per input file
- Maximum of 10 billion tokens per model per account
Calculate the number of files needed based on these constraints. For this project, the file size limit is the main factor, requiring 40 files of 25,000 requests each.
Creating Input Files
Each line of the input file represents a single request. Ensure that each line includes a unique custom_id to link results back to the original documents. Limit the output length to avoid unnecessary token usage.
import json
SYSTEM_PROMPT = (
"You classify and summarize documents. Respond with a single JSON object: "
'{"category": "<one of: billing, bug_report, feature_request, account, '
'security, performance, documentation, other>", '
'"summary": "<3-4 sentence summary>"} '
"The category value must be one of the listed strings, lowercase."
)
CHUNK_SIZE = 25_000
def write_batch_files(documents, prefix="batch_input"):
paths, out, count, part = [], None, 0, 0
for doc_id, text in documents:
if count % CHUNK_SIZE == 0:
if out: out.close()
part += 1
path = f"{prefix}_{part:03d}.jsonl"
out = open(path, "w", encoding="utf-8")
paths.append(path)
line = {
"custom_id": doc_id,
"method": "POST",
"url": "/v1/chat/completions",
"body": {
"model": "gpt-5-mini",
"messages": [
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": text},
],
"max_completion_tokens": 500,
"reasoning_effort": "minimal",
},
}
out.write(json.dumps(line, ensure_ascii=False) + "\n")
count += 1
if out: out.close()
return paths
Submitting Jobs
Submit each file in three steps: reserve a file ID, upload the file, and create the batch job. Use a consistent request_id to avoid duplicate jobs when retrying.
import hashlib
import os
import uuid
import requests
from pydo import Client
client = Client(token=os.environ["DIGITALOCEAN_TOKEN"])
def submit_file(path):
intent = client.batches.files.create(file_name=os.path.basename(path))
file_id, upload_url = intent["file_id"], intent["upload_url"]
with open(path, "rb") as fh:
put = requests.put(
upload_url,
data=fh,
headers={"Content-Type": "application/octet-stream"},
timeout=300,
)
put.raise_for_status()
request_id = str(uuid.UUID(
hashlib.md5(f"doc-pipeline-2026-08:{path}".encode()).hexdigest()
))
batch = client.batches.create(
file_id=file_id,
provider="openai",
endpoint="/v1/chat/completions",
completion_window="24h",
request_id=request_id,
)
return batch["batch_id"]
batch_ids = {}
for path in sorted(paths):
batch_ids[path] = submit_file(path)
print(f"submitted {path} -> {batch_ids[path]}")
Monitoring Job Progress
Each job progresses through a sequence of states. Monitor these states by polling at regular intervals to determine when a job has completed.
Handling Failures
Failures can occur at various levels. Request-level errors are captured in an error file, while job-level issues can often be resolved by resubmitting only the unprocessed requests.
Retrieving and Processing Results
Once a job completes, download the results immediately. Each output line includes the custom_id and a summary of the processing, allowing for easy integration with original documents.
import json
import requests as http
CATEGORIES = {"billing", "bug_report", "feature_request", "account",
"security", "performance", "documentation", "other"}
def collect_results(batch_ids, out_path="results.jsonl"):
total_in = total_out = failures = 0
with open(out_path, "w", encoding="utf-8") as out:
for path, bid in batch_ids.items():
links = client.batches.results.retrieve(bid)
if not links.get("result_available"):
print(f"{bid}: results not ready, poll again later")
continue
resp = http.get(links["output_file_url"], timeout=300)
resp.raise_for_status()
for line in resp.text.splitlines():
rec = json.loads(line)
if rec.get("error"):
failures += 1
continue
usage = rec["response"]["usage"]
total_in += usage["prompt_tokens"]
total_out += usage["completion_tokens"]
content = rec["response"]["choices"][0]["message"]["content"]
try:
parsed = json.loads(content)
except json.JSONDecodeError:
failures += 1
continue
if parsed.get("category") not in
