Classify a million rows with an LLM: cost, file splitting, retries

Labeling a million rows with an LLM is mostly a plumbing problem. The prompt takes an afternoon. The rest is keeping the output parseable, splitting the input into files the API will accept, and making sure the 0.3% of lines that fail get rerun without redoing everything else.

This is the setup I'd use, end to end, with the OpenAI Python SDK pointed at Batchrate.

What it costs first

Classification is close to the ideal batch job: lots of input, a very short answer, and nobody waiting on any single row. With a JSON schema, the output is just a label or two, so output tokens stay small.

Example numbers: 1,000,000 rows, about 300 input tokens per row (instructions plus the text), about 20 output tokens per row, on qwen3.6-35b-a3b at $0.05 per 1M input tokens and $0.40 per 1M output tokens.

Example numbers	Tokens	Price / 1M	Cost
Input (1M × 300)	300,000,000	$0.05	$15.00
Output (1M × 20)	20,000,000	$0.40	$8.00
Total			$23.00

If your rows are longer, input dominates: at 600 input tokens per row the input line doubles to $30.00 and the total is $38.00. Keep the instructions short, because they're repeated on every row.

New accounts get 10,000,000 free tokens after email verification. That covers roughly 30,000 rows at the sizes above, which is plenty to check accuracy on a sample before running the whole thing.

Step 1: a schema for the label

Free-text answers are where large classification jobs go wrong: "Billing.", "billing issue", "Category: billing". Use response_format with a JSON schema and an enum, and the model can only answer with one of your labels.

SCHEMA = {
    "type": "json_schema",
    "json_schema": {
        "name": "label",
        "strict": True,
        "schema": {
            "type": "object",
            "properties": {
                "label": {"type": "string",
                          "enum": ["billing", "bug", "feature_request", "account", "other"]},
                "confidence": {"type": "string", "enum": ["low", "medium", "high"]},
            },
            "required": ["label", "confidence"],
            "additionalProperties": False,
        },
    },
}

This schema requires confidence, so every row comes back as low, medium, or high. Low-confidence rows are the ones to spot-check by hand or rerun with a second prompt.

Step 2: build the JSONL

One request per row. The custom_id is how you join results back to your data, so use your row's primary key. A custom_id can be up to 128 characters.

import csv, json

INSTRUCTIONS = ("Classify the customer message into one category. "
                "billing: charges, refunds, invoices. bug: something broken. "
                "feature_request: asks for something new. account: login, access, settings. "
                "other: anything else.")

def build_jsonl(csv_path, out_path, model="qwen3.6-35b-a3b"):
    with open(csv_path, newline="") as fin, open(out_path, "w") as fout:
        for row in csv.DictReader(fin):
            line = {
                "custom_id": f"row-{row['id']}",
                "method": "POST",
                "url": "/v1/chat/completions",
                "body": {
                    "model": model,
                    "max_tokens": 50,
                    "messages": [
                        {"role": "system", "content": INSTRUCTIONS},
                        {"role": "user", "content": row["text"][:4000]},
                    ],
                    "response_format": SCHEMA,
                },
            }
            fout.write(json.dumps(line) + "\n")

build_jsonl("messages.csv", "all.jsonl")

A single line looks like this:

{"custom_id":"row-104233","method":"POST","url":"/v1/chat/completions","body":{"model":"qwen3.6-35b-a3b","max_tokens":50,"messages":[{"role":"system","content":"Classify the customer message into one category. ..."},{"role":"user","content":"I was charged twice for October."}],"response_format":{"type":"json_schema","json_schema":{"name":"label","strict":true,"schema":{"type":"object","properties":{"label":{"type":"string","enum":["billing","bug","feature_request","account","other"]},"confidence":{"type":"string","enum":["low","medium","high"]}},"required":["label","confidence"],"additionalProperties":false}}}}}

Set max_tokens close to what you need. The cost estimate uses it for output, and a schema answer only cut off by max_tokens can come back as incomplete JSON. 50 is comfortable for the schema above. Thinking is off by default on JSON-schema lines, which is what you want here.

Step 3: split into files of 5,000 lines or 10 MB

A batch input file can hold up to 5,000 lines and 10 MB (10,000,000 bytes). A million rows becomes about 200 files.

def split_jsonl(path, prefix, max_lines=5000, max_bytes=10_000_000):
    names, buf, size = [], [], 0
    def flush():
        name = f"{prefix}-{len(names):04d}.jsonl"
        open(name, "wb").writelines(buf)
        names.append(name)
    for line in open(path, "rb"):
        if buf and (len(buf) >= max_lines or size + len(line) > max_bytes):
            flush(); buf, size = [], 0
        buf.append(line); size += len(line)
    if buf: flush()
    return names

parts = split_jsonl("all.jsonl", "part")
print(len(parts), "files")

Smaller files also make failures cheaper to deal with. If one file has a bad line, upload rejects that file with invalid_file and names the line, and the other files are unaffected.

Step 4: submit, keeping under the running-batch limit

import os, time
from openai import OpenAI

client = OpenAI(api_key=os.environ["BATCHRATE_API_KEY"],
                base_url="https://api.batchrate.ai/v1")
DONE = ("completed", "failed", "expired", "cancelled")

def run_all(paths, max_running=90):
    pending, running, finished = list(paths), {}, {}
    while pending or running:
        while pending and len(running) < max_running:
            p = pending.pop(0)
            f = client.files.create(file=open(p, "rb"), purpose="batch")
            b = client.batches.create(input_file_id=f.id, endpoint="/v1/chat/completions",
                                      completion_window="24h", metadata={"part": p})
            running[b.id] = p
        time.sleep(120)
        for bid in list(running):
            b = client.batches.retrieve(bid)
            if b.status in DONE:
                finished[running.pop(bid)] = b
    return finished

batches = run_all(parts)

Account limits matter at this size. Before you add credit, an account can run 3 batches at a time and submit about 10M tokens per 24 hours, so use the free tokens on a sample. After the first top-up it's 100 running batches and 1B tokens per 24 hours, which fits a 320M-token job. If you go over, POST /v1/batches returns active_batch_limit or daily_token_limit, so in a real script, catch that, wait, and try again.

Each batch has a 24h completion window. Results typically come back within hours. metadata values have to be strings, which a file name is, up to 512 characters.

Step 5: collect results

import json

results = {}
for part, b in batches.items():
    if not b.output_file_id:
        continue
    for line in client.files.content(b.output_file_id).text.splitlines():
        row = json.loads(line)
        content = row["response"]["body"]["choices"][0]["message"]["content"]
        results[row["custom_id"]] = json.loads(content)
print(len(results), "labeled")

Step 6: retry failed lines by custom_id

You're only charged for requests that complete. Failed lines land in each batch's error file with response: null and an error object with a code and message. If a batch reaches the end of its window, the unfinished lines are listed there with code batch_expired.

def collect_failures(batches):
    failed = {}
    for b in batches.values():
        if b.error_file_id:
            for line in client.files.content(b.error_file_id).text.splitlines():
                row = json.loads(line)
                failed[row["custom_id"]] = row["error"]["code"]
    return failed

failed = collect_failures(batches)
retry_ids = {cid for cid, code in failed.items()}
with open("all.jsonl") as fin, open("retry.jsonl", "w") as fout:
    for line in fin:
        if json.loads(line)["custom_id"] in retry_ids:
            fout.write(line)

retry_batches = run_all(split_jsonl("retry.jsonl", "retry"))

Look at the codes before rerunning. batch_expired lines were not charged and can go back as-is. If many lines fail with the same code, fix the cause first. Also make sure every custom_id in your input is present in either results or failures; that's a quick check that nothing went missing.

A few things I'd do differently the second time

Run 2,000 rows first and compare against a hand-labeled sample. Fixing the category definitions is far cheaper at 2,000 rows than at a million.
Keep the instructions in the system message, short. At 300 tokens a row, every 50 tokens of instruction is about $2.50 across a million rows.
Store the raw output files. Output and error files are deleted 30 days after they're created, so copy them somewhere you control.