Project walkthrough · 12 min read

Batch AI Jobs in Python: Queue, Retry, Resume, and Cost

A batch job is not a for loop around an API call. This walkthrough builds the four parts that keep a long run honest: stable job keys, a concurrency ceiling, a resume file, and a price you can defend before you launch.

Start from job records, not from a loop

A batch is a list of inputs with stable identities. Hash the input text to get a key, so the same ticket always maps to the same key; that is what makes a rerun safe and a duplicate visible. Count rows and distinct keys before you spend a token.

import hashlib

def job_key(text):
    return hashlib.sha256(text.encode("utf-8")).hexdigest()[:12]

texts = ["first ticket", "second ticket", "first ticket"]
jobs = [{"key": job_key(t), "text": t} for t in texts]
print(len(jobs), len({job["key"] for job in jobs}))

Keys make duplicates visible

Cap concurrency at the provider's limit

Your CPU is not the bottleneck; the quota is. Wrap every call in a semaphore so at most N requests are in flight, then gather the results. N belongs in a config file, because the number that works at 9 a.m. rarely works at peak.

import asyncio

async def call(text, gate):
    async with gate:
        await asyncio.sleep(0.01)   # stands in for the API call
        return text.upper()

async def run(texts, limit=4):
    gate = asyncio.Semaphore(limit)
    return await asyncio.gather(*(call(t, gate) for t in texts))

print(asyncio.run(run(["a", "b", "c"])))

One gate for the whole batch

Treat 429 as slow down, not fail

A rate limit is a request to wait. Double the delay after each attempt, add jitter so workers do not retry in lockstep, and give up after three or four tries. Business errors such as 400 or 401 must not be retried at all: they will fail identically every time.

import asyncio, random

async def with_backoff(call, attempts=4):
    delay = 0.5
    for attempt in range(attempts):
        try:
            return await call()
        except RuntimeError:            # stand-in for HTTP 429
            if attempt == attempts - 1:
                raise
            await asyncio.sleep(delay + random.uniform(0, delay / 2))
            delay *= 2

Capped backoff with jitter

Resume from an append-only file

Write one JSON line per finished job and flush immediately. On restart, read the file into a set of done keys and skip them. Appending is the whole trick: a crash costs the current job, never the previous five hundred.

import json, pathlib

def load_done(path):
    p = pathlib.Path(path)
    if not p.exists():
        return set()
    rows = [json.loads(line) for line in p.read_text(encoding="utf-8").splitlines() if line.strip()]
    return {row["key"] for row in rows if row.get("ok")}

def save(path, key, text):
    with open(path, "a", encoding="utf-8") as fh:
        fh.write(json.dumps({"key": key, "ok": True, "text": text}, ensure_ascii=False) + "\n")

JSONL survives a kill

Validate before you persist

Even with a JSON response format, output can arrive wrapped in code fences or cut off mid-object. Strip the fence, parse, and fall back to the raw prefix instead of raising inside the loop. One bad row should not end an 800-row batch.

import json

def parse(text, fallback_chars=200):
    raw = text.strip()
    if raw.startswith("```"):
        raw = raw.split("\n", 1)[-1].rsplit("```", 1)[0]
    try:
        return json.loads(raw)
    except json.JSONDecodeError:
        return {"title": raw[:fallback_chars], "fallback": True}

print(parse('```json\n{"title": "ok"}\n```'))
print(parse("cut off mid")["fallback"])

Parse with a fallback

Look at one request before you manage a thousand

Everything above assumes you know what a single call looks like. It is one POST with two headers and a JSON body: the model name, the message list, and the output contract. response_format only promises parseable JSON, so the schema check from the previous section still has to run.

import os, requests

resp = requests.post(
    "https://api.example.com/v1/chat/completions",
    headers={
        "Authorization": f"Bearer {os.environ['OPENAI_API_KEY']}",
        "Content-Type": "application/json",
    },
    json={
        "model": "gpt-4o-mini",
        "messages": [{"role": "user", "content": "Return the ticket title as JSON."}],
        "response_format": {"type": "json_object"},
        "max_tokens": 120,
    },
    timeout=(3.05, 60),
)
resp.raise_for_status()
data = resp.json()
print(data["choices"][0]["message"]["content"], data["usage"]["total_tokens"])

One request, four decisions

Price the batch before you launch it

Estimate prompt tokens from characters, multiply by the unit price, and print the estimate. Then log actual usage and compare; the gap is usually the prompt template you forgot to count, and it grows with every retry.

PRICE_IN = 0.15 / 1_000_000
PRICE_OUT = 0.60 / 1_000_000

def batch_cost(prompt_tokens, completion_tokens):
    return round(prompt_tokens * PRICE_IN + completion_tokens * PRICE_OUT, 4)

print(batch_cost(1_200_000, 300_000))

A number you can defend

Your practice checklist

Every job has a content-derived key, so a rerun never pays twice.
Concurrency is a config value, set below the provider quota.
429 retries use doubling delay plus jitter; 4xx business errors do not retry.
Results are appended as JSONL and flushed per row.
Structured output has an explicit parse-failure branch.
A single request is written down: auth header, model, output contract, timeout.
The run prints job count, failure count, and actual spend at the end.

Common questions

Should I use ThreadPoolExecutor instead?

If your SDK is synchronous and the calls are pure network waits, threads are simpler and fine. asyncio wins when you need hundreds of concurrent waits in one process. Pick one and do not nest them.

How large does the results file get?

Roughly 200 bytes per row, so a million jobs is about 200 MB. Rotate by day or by source, and keep a separate key index if you only need the set of completed work.

How many retries is enough?

Three or four attempts with a total wait under your per-job timeout. If a job still fails, write it to failed.jsonl with the error and move on.

Why hash the input instead of using a row number?

Row numbers shift as soon as the input file changes, and they cannot detect the same ticket arriving twice. A content hash does both jobs.

Practice retries and structured output

Questions on retry budgets, timeouts, streaming chunks, and JSON validation.

Open practice

Keep reading: