How to Run Bulk Record Classification With GPT-6 Luna
Labeling 100,000 messy records by hand takes a team weeks. With GPT-6 Luna classification run through the Batch API, the same job can cost less than a coffee. By the end you'll have a pipeline that tags each record with a category, a PII flag and a dedupe key in strict JSON, a way to measure accuracy on a labeled sample, and a rule for sending only the doubtful rows to GPT-6 Sol.
GPT-6 Luna classification costs at Batch rates
Luna's model page lists $0.10 per 1M input tokens and $0.50 per 1M output tokens, with Batch priced at 50% of Standard: $0.05 and $0.25. Sol lists at $2 and $10, so a Sol token costs 20 times a Luna token.
Take 100,000 records at roughly 300 input and 60 output tokens each (your numbers will differ). That is 30M input tokens for $1.50 and 6M output tokens for $1.50, about $3 in total. Batch jobs finish within 24 hours, so this suits backfills and nightly jobs, not live requests.
Define a strict JSON schema for the labels
Structured Outputs with strict: true makes the model follow your schema, so you can drop the retry-on-bad-JSON code. Strict mode needs every field in required and additionalProperties set to false. The example labels support tickets; swap in your own enums.
RECORD_LABEL_SCHEMA = {
"type": "object",
"properties": {
"category": {
"type": "string",
"enum": ["billing", "shipping", "product_defect", "account_access", "other"],
"description": "Best single category. Use other only if none fit.",
},
"contains_pii": {
"type": "boolean",
"description": "True if the text has names, emails, phones, addresses, ID or card numbers.",
},
"pii_types": {
"type": "array",
"items": {
"type": "string",
"enum": ["name", "email", "phone", "address", "government_id", "payment_card"],
},
},
"dedupe_key": {
"type": "string",
"description": "Main entity, lowercase, no punctuation or legal suffixes.",
},
"confidence": {"type": "string", "enum": ["low", "medium", "high"]},
},
"required": ["category", "contains_pii", "pii_types", "dedupe_key", "confidence"],
"additionalProperties": False,
}
The other bucket and the confidence field are your escalation signals later. The model can't compare 100,000 rows in one call, so it emits a normalized dedupe_key instead. Group on that key in your warehouse, then confirm real duplicates with deterministic matching.
Build the Batch API input file
Each line of the input file is one request with a custom_id, method, url and body. Use your record ID as the custom_id. Setting reasoning effort to none means you pay for the JSON, not for thinking tokens. A batch file is capped at 50,000 requests and 200 MB, so the script writes chunks of 10,000.
import json
SYSTEM = (
"You label customer records. Classify the record, flag PII, "
"and give a normalized dedupe key. Return only the requested JSON."
)
def request_line(record_id, text, model="gpt-6-luna", effort="none"):
return json.dumps({
"custom_id": record_id,
"method": "POST",
"url": "/v1/responses",
"body": {
"model": model,
"reasoning": {"effort": effort},
"instructions": SYSTEM,
"input": text,
"max_output_tokens": 300,
"text": {"format": {
"type": "json_schema",
"name": "record_label",
"strict": True,
"schema": RECORD_LABEL_SCHEMA,
}},
},
})
def write_chunks(records, chunk_size=10_000):
# records: list of {"id": str, "text": str}
for i in range(0, len(records), chunk_size):
path = f"batch_{i // chunk_size:03d}.jsonl"
with open(path, "w", encoding="utf-8") as f:
for rec in records[i:i + chunk_size]:
print(request_line(rec["id"], rec["text"]), file=f)
Submit the batch and collect results
Upload the file with purpose batch, create the batch against /v1/responses with a 24h completion window, and poll until it ends. The Batch guide warns that output order may not match input order, so key everything on custom_id. Failures land in a separate error file.
import json
import time
from openai import OpenAI
client = OpenAI()
def run_batch(path):
uploaded = client.files.create(file=open(path, "rb"), purpose="batch")
batch = client.batches.create(
input_file_id=uploaded.id,
endpoint="/v1/responses",
completion_window="24h",
metadata={"description": f"luna labeling {path}"},
)
while batch.status not in {"completed", "failed", "expired", "cancelled"}:
time.sleep(60)
batch = client.batches.retrieve(batch.id)
return batch
def parse_results(batch):
labels, failed = {}, []
if batch.output_file_id:
for line in client.files.content(batch.output_file_id).text.splitlines():
row = json.loads(line)
resp = row.get("response") or {}
body = resp.get("body") or {}
text = None
for item in body.get("output", []):
if item.get("type") == "message":
for part in item["content"]:
if part.get("type") == "output_text":
text = part["text"]
if (row.get("error") or resp.get("status_code") != 200
or body.get("status") != "completed" or text is None):
failed.append(row["custom_id"])
else:
labels[row["custom_id"]] = json.loads(text)
if batch.error_file_id:
for line in client.files.content(batch.error_file_id).text.splitlines():
failed.append(json.loads(line)["custom_id"])
return labels, failed
Measure accuracy on a labeled sample first
Hand-label about 300 records, making sure rare categories appear, and run them through the same pipeline before the full job. At that size the margin of error is a few points, so read the result as a gauge, not a guarantee.
Check PII recall first, because a missed email costs more than a false flag. The groupby shows whether low confidence really predicts errors; self-reported confidence is a hint, not a probability. If accuracy falls short, tighten the label descriptions in the schema before reaching for a bigger model.
import pandas as pd
from sklearn.metrics import classification_report
gold = pd.read_csv("gold_sample.csv", dtype={"id": str}, index_col="id") # category, contains_pii
pred = pd.DataFrame.from_dict(labels, orient="index")
df = gold.join(pred, lsuffix="_gold", rsuffix="_pred", how="inner")
print(classification_report(df["category_gold"], df["category_pred"]))
print(classification_report(df["contains_pii_gold"], df["contains_pii_pred"]))
df["correct"] = df["category_gold"] == df["category_pred"]
print(df.groupby("confidence")["correct"].agg(["mean", "count"]))
Escalate low-confidence rows to GPT-6 Sol
Collect the rows that came back low, landed in other, or failed. Resubmit them with the same schema to gpt-6-sol at medium reasoning effort, then overwrite the Luna labels.
by_id = {rec["id"]: rec for rec in records}
retry_ids = set(failed) | {
rid for rid, label in labels.items()
if label["confidence"] == "low" or label["category"] == "other"
}
with open("escalate.jsonl", "w", encoding="utf-8") as f:
for rid in retry_ids:
print(request_line(rid, by_id[rid]["text"], model="gpt-6-sol", effort="medium"), file=f)
sol_labels, sol_failed = parse_results(run_batch("escalate.jsonl"))
labels.update(sol_labels)
At Batch rates Sol costs $1 and $5 per 1M tokens. If 10% of rows escalate, that adds about twice the Luna pass before reasoning tokens, so the total lands near 3x the Luna-only cost instead of 20x for all-Sol. Sol's reasoning tokens bill as output, so measure on the escalated rows before you commit.
Gotchas
- Queue limits. Luna's enqueued-token cap is 5M at Tier 1 and 20M at Tier 2. The example job is about 30M prompt tokens, so on lower tiers submit chunks one after another.
- Refusals and incomplete responses don't follow your schema. The parser sends them to the
failedlist. - Expired batches cancel unfinished requests, and the error file marks them
batch_expired. Resubmit those IDs. - Higher reasoning effort means removing
temperature,top_pandtop_logprobsfrom the request. - PII leaves your network. Check your residency needs; the model page lists EU data residency for Batch.
Keep the labeled sample as a regression test whenever you change the prompt, schema or model. For a production data pipeline, that gives you a first pass at a twentieth of the Sol price and a measured safety net for the hard rows.