Classifying 50,000 Rows Without Melting the Budget

The naive version of this job is df["label"] = df["text"].apply(classify), which takes fourteen hours, fails at row 34,000, loses everything and costs more than it needed to. Four changes fix all of that, and they go in a specific order.

The order of operations that saves the money

1. Filter. Drop rows the job does not need at all — empty text, rows already labelled, rows outside the date range. Free.
2. Deduplicate. Classify unique texts, not rows. On real data this is routinely the biggest single saving on the page, and it costs one groupby.
3. Triage with something cheaper. If a regex, a keyword list or a small local classifier can decide half the rows with high precision, those rows never reach a model. See the cheap filter pattern.
4. Batch. Several items per request amortises the system prompt across all of them.
5. Run concurrently. Only now, and only up to the rate limit.
6. Checkpoint every batch. So that steps 1 to 5 never have to happen twice.

Reversing steps 2 and 5 is the common mistake. Making a job that sends duplicate work go faster is optimising the wrong thing — you finish sooner and pay exactly as much.

Deduplicate first

# prepare.py
import hashlib
import re

import pandas as pd

WHITESPACE = re.compile(r"\s+")


def normalise(text: str) -> str:
    return WHITESPACE.sub(" ", str(text)).strip().lower()


def prepare(df: pd.DataFrame, column: str) -> tuple[pd.DataFrame, pd.DataFrame]:
    """Return (df with a key column, one row per distinct text)."""
    df = df[df[column].notna() & (df[column].astype(str).str.strip() != "")].copy()
    df["_norm"] = df[column].map(normalise)
    df["_key"] = df["_norm"].map(
        lambda t: hashlib.sha1(t.encode("utf-8")).hexdigest()[:16]
    )
    unique = (
        df.drop_duplicates("_key")[["_key", column]]
        .rename(columns={column: "text"})
        .reset_index(drop=True)
    )
    print(f"{len(df):,} rows -> {len(unique):,} distinct texts"
          f" ({100 * (1 - len(unique) / len(df)):.1f}% saved)")
    return df, unique

Normalising before hashing is what makes this work on real data: “Order not received”, “order not received” and “Order not received ” are three rows and one classification. Keep the original text in the unique frame for sending — you normalise to match, not to send a mangled prompt.

Print the saving. It is the number that justifies the whole step, and on support tickets, product reviews and log messages it is frequently 30 per cent or more. If it is near zero, you have learnt something useful for one groupby.

Estimating the bill before you spend it

Do this on a 200-row sample and extrapolate. It takes a minute and it is the difference between a surprise and a decision.

# estimate.py
PRICE_IN_PER_M = 0.15    # <- from your provider's pricing page, per 1M input tokens
PRICE_OUT_PER_M = 0.60   # <- per 1M output tokens


def estimate(sample_usages: list[dict], n_total: int, batch_size: int) -> None:
    """sample_usages: the usage object from each of N sample requests."""
    n_sample = len(sample_usages)
    in_per_req = sum(u["prompt_tokens"] for u in sample_usages) / n_sample
    out_per_req = sum(u["completion_tokens"] for u in sample_usages) / n_sample

    requests = -(-n_total // batch_size)          # ceiling division
    total_in = in_per_req * requests
    total_out = out_per_req * requests
    cost = (total_in / 1e6) * PRICE_IN_PER_M + (total_out / 1e6) * PRICE_OUT_PER_M

    print(f"requests      {requests:,}")
    print(f"input tokens  {total_in:,.0f}")
    print(f"output tokens {total_out:,.0f}")
    print(f"estimated     {cost:,.2f} (currency of your price constants)")

Take the prices from your provider’s own page rather than from any article, this one included — per-token prices move, and a stale constant in an estimator is worse than no estimator because it is believed. How LLM pricing works explains the units.

The estimate is a lower bound, not a forecast. It excludes retries, repairs after a validation failure, and the second pass you will probably run after seeing the first results. Multiplying by 1.3 before quoting a number to anybody else has never yet been the wrong call.

Batching, and the size to choose

Putting several items in one request amortises the system prompt. Concretely, with a 400-token system prompt and 60-token items: one item per request is 460 input tokens for 60 tokens of content, so 87 per cent of what you buy is the instruction. At ten items per request it is 1,000 tokens for 600 of content — 40 per cent overhead. At twenty, 25 per cent. The curve flattens fast, which is why the useful range is roughly five to twenty rather than as many as fit.

# batch.py
import json

SYSTEM = (
    "Classify each item into exactly one of: billing, technical, account, other.\n"
    "Reply with a JSON array of objects: {\"id\": <id>, \"label\": <label>}.\n"
    "Return one object per input item, in the same order. No other text."
)


def build_batch_prompt(items: list[tuple[str, str]]) -> list[dict]:
    """items: [(id, text)]"""
    lines = [json.dumps({"id": item_id, "text": text[:1500]},
                        ensure_ascii=False)
             for item_id, text in items]
    return [
        {"role": "system", "content": SYSTEM},
        {"role": "user", "content": "\n".join(lines)},
    ]


def parse_batch(raw: str, expected_ids: list[str]) -> dict[str, str]:
    """Return {id: label}. Missing ids are simply absent — the caller re-queues them."""
    labels: dict[str, str] = {}
    parsed = json.loads(raw)                       # see the parsing recipe
    if not isinstance(parsed, list):
        raise ValueError("expected a JSON array")
    known = set(expected_ids)
    for entry in parsed:
        item_id = str(entry.get("id", ""))
        label = entry.get("label")
        if item_id in known and isinstance(label, str):
            labels[item_id] = label
    return labels

Giving every item an explicit id, and matching on it rather than on position, is the difference between a batch job that works and one that silently shifts every label by one when the model returns nine results for ten items. Never zip a model’s output list against your input list by index.

The trade to be aware of: a batch fails as a unit. One unparseable reply costs you the whole batch rather than one row, and per-item quality can drop slightly as the batch grows because attention is spread across more items. Ten is a reasonable default, and re-queue the failures individually.

Checkpointing so a crash costs nothing

One SQLite table keyed by the text hash. It doubles as a cache: a second run over overlapping data sends only what is genuinely new.

# checkpoint.py
import sqlite3


def open_store(path: str = "labels.db") -> sqlite3.Connection:
    conn = sqlite3.connect(path, isolation_level=None)
    conn.execute("PRAGMA journal_mode=WAL")
    conn.execute(
        "CREATE TABLE IF NOT EXISTS labels ("
        " key TEXT PRIMARY KEY, label TEXT NOT NULL,"
        " model TEXT NOT NULL, created_at REAL NOT NULL)"
    )
    return conn


def already_done(conn: sqlite3.Connection) -> set[str]:
    return {row[0] for row in conn.execute("SELECT key FROM labels")}


def save(conn: sqlite3.Connection, results: dict[str, str], model: str) -> None:
    import time
    now = time.time()
    conn.executemany(
        "INSERT OR REPLACE INTO labels (key, label, model, created_at)"
        " VALUES (?, ?, ?, ?)",
        [(key, label, model, now) for key, label in results.items()],
    )

# run.py
conn = open_store()
done = already_done(conn)
todo = unique[~unique["_key"].isin(done)]
print(f"{len(unique):,} distinct, {len(done):,} already labelled,"
      f" {len(todo):,} to do")

for start in range(0, len(todo), BATCH):
    chunk = todo.iloc[start:start + BATCH]
    items = list(zip(chunk["_key"], chunk["text"]))
    raw = call_model(build_batch_prompt(items))
    labels = parse_batch(raw, [k for k, _ in items])
    save(conn, labels, MODEL)                    # commit before moving on
    missing = {k for k, _ in items} - labels.keys()
    if missing:
        print(f"batch at {start}: {len(missing)} items missing, will retry")
    print(f"{start + len(chunk):,}/{len(todo):,}", flush=True)

Two properties fall out of this that are worth naming. The script is idempotent: run it twice and the second run does almost nothing. And it is interruptible on purpose — Ctrl-C is now a safe operation, which means you can stop a job to check the early labels instead of letting fourteen hours of work commit to a prompt you have not verified.

PRAGMA journal_mode=WAL matters here: it lets you query the labels table from a second terminal while the job is still writing to it, which is how you check the early labels without stopping anything.

Joining the results back

labels = pd.read_sql("SELECT key AS _key, label FROM labels", conn)
out = df.merge(labels, on="_key", how="left")

unlabelled = out["label"].isna().sum()
print(f"{unlabelled:,} rows unlabelled ({100 * unlabelled / len(out):.2f}%)")
out.drop(columns=["_norm", "_key"]).to_parquet("classified.parquet", index=False)

Use how="left" and then count the nulls. An inner join hides exactly the rows you most need to know about — the ones no label came back for — by silently shrinking the frame, and the row count is not something anybody checks by eye.

Write Parquet rather than CSV for anything you will read again. It preserves dtypes, so a column of ids that happen to look numeric comes back as strings rather than as floats with the leading zeroes gone.

The pandas mistakes that cost the most

Four of these are specific to putting a model in the middle of a dataframe job, and each one has cost somebody a re-run.

• df.apply with a network call inside. It is strictly sequential, it cannot be interrupted safely, and it gives you no progress. The loop over batches above is longer to write and is the only version that can checkpoint. Reserve apply for the cheap local transformations.
• Reading ids as numbers. pd.read_csv infers dtypes, so an order reference like 00417 becomes the integer 417 and no longer joins against anything. Pass dtype=str for identifier columns, or dtype=str for the whole frame and convert what you actually need.
• Losing the join key to whitespace. The hash is computed from normalised text but the merge is on _key, so a key column that was written and re-read with different whitespace produces a silent zero-match join. This is why the join above counts its nulls rather than trusting the result.
• Holding the whole frame while the job runs. A 50,000-row frame with a long text column plus a copy per intermediate step is easily several gigabytes. Select the two columns you need into the unique frame — as prepare does — and let the rest stay on disk until the join.
• Writing back to the source file. Never overwrite the input. Write a new file, diff the row counts and a sample of the labels, and only then replace anything. A batch job with a wrong prompt is recoverable; a batch job that overwrote its own input is not.

One habit that removes most of these at once: keep the labelled results in their own table and join at read time rather than mutating the source frame. The classification then becomes an annotation you can recompute, compare across models, or throw away, instead of an edit you cannot undo.

Checking the labels are worth having

A finished job is not a correct job. Three checks, in this order, because each is cheaper than the next:

1. Distribution. out["label"].value_counts(normalize=True). If one class is 95 per cent, or a class you defined has zero rows, the prompt is broken rather than the data being unusual. This costs nothing and catches most failures.
2. Out-of-schema labels. Count rows whose label is not in your defined set. It should be zero; anything above that is a parsing or prompting problem you can still fix cheaply.
3. A hand-labelled sample. Take 100 rows stratified by predicted label, label them yourself without looking at the prediction, and compare. This is the only step that measures accuracy rather than plausibility, and it is the one everybody skips — a golden dataset makes it repeatable for the next model or prompt.