Batch Extraction API — Classification, JSON & Summarization at Scale

Batch extraction is not one prompt applied to a CSV. It is a pipeline with input normalization, bounded concurrency, schema validation, retries, review queues, and a deliberate escalation path. The model call is only one stage. If a row cannot be parsed, exceeds a token budget, or returns a low-confidence answer, the pipeline needs a deterministic place to send it.

Use LumeAPI for real-time classification, JSON extraction, and summarization workers that need immediate responses at catalog rates. Configure an OpenAI-compatible client with base_url set to https://api.lumeapi.site/v1 and a Bearer API key from the LumeAPI Console. Do not append /chat/completions to base_url: compatible SDKs add that endpoint path themselves.

Start with a cheap, fast model on routine rows, then spend more only where the data proves it is necessary. A practical first pass is gpt-5.4-mini or gemini-3-flash. Rows that fail JSON parsing, schema validation, business-rule checks, or confidence thresholds can be retried with gpt-5.6-terra or claude-sonnet-4-6. This makes premium-model usage an exception budget instead of the default.

For workloads that can wait up to 24 hours, evaluate a provider's official Batch API separately. That is an asynchronous job model. LumeAPI is better suited to real-time worker pipelines, where your application owns queueing, retry timing, observability, and row-level escalation.

Implementation setting path: use client base_url = "https://api.lumeapi.site/v1". The SDK constructs the Chat Completions endpoint path.

A small pilot exposes schema ambiguity, noisy source fields, long-tail documents, and rate-limit behavior before they become expensive.

• 1. Select 1,000 representative rows, including short inputs, long inputs, malformed records, multiple languages, and examples that previously required manual review.
• 2. Define one output schema per job. For extraction, name every field, specify allowed values, and decide whether missing data should be null, an empty array, or a rejected row.
• 3. Keep source text and extraction instructions separate. Put the task contract in the system message and place each row's untrusted content in a clearly delimited user message.
• 4. Run the first pass with gpt-5.4-mini or gemini-3-flash. Record model id, prompt version, input tokens, output tokens, latency, HTTP status, raw response, parsed result, and validation result for every row.
• 5. Where the selected model supports it, set response_format.type to json_object. JSON mode improves transport consistency but does not replace application-side schema validation.
• 6. Validate every response after parsing. Reject unexpected keys, wrong scalar types, invalid enum values, impossible dates, and outputs that violate record-level business rules.
• 7. Retry transient failures such as HTTP 429 and temporary 5xx responses with exponential backoff and jitter. Do not retry invalid input or schema failures unchanged.
• 8. Send rows that still fail after bounded retries to an escalation queue. Re-run those rows with gpt-5.6-terra or claude-sonnet-4-6 using the same schema contract.
• 9. Sample accepted rows for human review, especially around classes that drive downstream decisions. Measure precision and recall by class, not only overall parse success.
• 10. Set production thresholds from pilot results: maximum prompt size, concurrency limit, retry count, escalation rate, acceptable validation-failure rate, and review sample size.

Verified response-format setting path: response_format.type = "json_object" where supported by the selected model.

import json
import os
import random
import time
from typing import Any

from openai import OpenAI

client = OpenAI(
    api_key=os.environ["LUMEAPI_API_KEY"],
    base_url="https://api.lumeapi.site/v1",
)

PRIMARY_MODEL = "gpt-5.4-mini"
ESCALATION_MODEL = "gpt-5.6-terra"
MAX_ATTEMPTS = 4

SYSTEM_PROMPT = """Extract structured data from the supplied record.
Return one JSON object and no markdown.
Required keys:
- category: one of billing, support, sales, other
- summary: concise factual summary
- entities: array of strings
- needs_review: boolean
Use null when a scalar value cannot be determined from the record.
Do not infer facts that are not present in the record."""


def validate_result(value: Any) -> dict:
    if not isinstance(value, dict):
        raise ValueError("result must be a JSON object")

    allowed = {"category", "summary", "entities", "needs_review"}
    if set(value) != allowed:
        raise ValueError(f"unexpected schema keys: {set(value)}")
    if value["category"] not in {"billing", "support", "sales", "other"}:
        raise ValueError("invalid category")
    if not isinstance(value["summary"], str):
        raise ValueError("summary must be a string")
    if not isinstance(value["entities"], list) or not all(isinstance(x, str) for x in value["entities"]):
        raise ValueError("entities must be an array of strings")
    if not isinstance(value["needs_review"], bool):
        raise ValueError("needs_review must be a boolean")
    return value


def call_model(record_text: str, model: str) -> dict:
    response = client.chat.completions.create(
        model=model,
        temperature=0,
        response_format={"type": "json_object"},
        messages=[
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": f"<record>\n{record_text}\n</record>"},
        ],
    )
    content = response.choices[0].message.content
    return validate_result(json.loads(content))


def extract_row(row_id: str, record_text: str) -> dict:
    last_error = None

    for attempt in range(MAX_ATTEMPTS):
        try:
            result = call_model(record_text, PRIMARY_MODEL)
            return {"row_id": row_id, "status": "accepted", "model": PRIMARY_MODEL, "result": result}
        except Exception as error:
            last_error = error
            # Restrict this retry loop to transient transport/rate-limit errors in production.
            # Log HTTP status and provider error details before deciding whether to retry.
            delay = min(30, (2 ** attempt) + random.uniform(0, 1))
            time.sleep(delay)

    try:
        result = call_model(record_text, ESCALATION_MODEL)
        return {"row_id": row_id, "status": "escalated", "model": ESCALATION_MODEL, "result": result}
    except Exception as escalation_error:
        return {
            "row_id": row_id,
            "status": "failed",
            "primary_error": str(last_error),
            "escalation_error": str(escalation_error),
            "model": ESCALATION_MODEL,
        }

# Do not append /chat/completions to the base_url above; the SDK adds the endpoint path.
# Export LUMEAPI_API_KEY from the Bearer API key created in the LumeAPI Console.

For production, classify errors by HTTP status: retry 429 and temporary 5xx responses; route parse and validation failures to escalation or review instead of repeating the identical request.

Use a primary model for the common path and reserve higher-cost models for rows that demonstrate a need for them.

Catalog pricing updated 2026-07-22. Prices shown are LumeAPI catalog input / output prices and should be paired with measured token usage from your own pilot.

The largest avoidable cost in extraction pipelines is unnecessary input text. Strip navigation, boilerplate, duplicate email chains, repeated headers, and fields unrelated to the requested schema before sending a row. Preserve source identifiers so results can be traced back to the original record without carrying irrelevant context into every prompt.

Set hard limits for both input and output. An extraction response should be small because its schema is small. Ask for concise summaries, bounded arrays, normalized labels, and null for unavailable values. If the job requires evidence, request short source spans rather than a restatement of the entire record.

Track the escalation rate as a first-class cost and quality metric. A 2% escalation rate may be an efficient design. A 35% escalation rate usually means the first-pass prompt, schema, source normalization, or routing rules need work. Do not hide this by silently sending every difficult row to a premium model.

Pricing alone does not determine total cost. Repeated retries, oversized prompts, invalid JSON, and unbounded summaries can erase the benefit of a cheaper model. The useful unit is cost per accepted validated record: total model spend divided by rows that passed parsing, schema validation, and downstream rules.

Keep model ids exact in requests, including gpt-5.4-mini, gemini-3-flash, gpt-5.6-terra, and claude-sonnet-4-6.

A worker pool should recover from bursts without duplicating work or turning a temporary 429 into an outage.

• 1. Put source records into a durable queue with a stable row id and an idempotency key derived from the source version and extraction-schema version.
• 2. Start with a small fixed worker pool, then raise concurrency only after observing latency, 429 frequency, queue age, and downstream database capacity.
• 3. On HTTP 429, reduce pressure with exponential backoff and randomized jitter. Requeue the row rather than holding a worker forever on a long sleep.
• 4. Cap retry attempts and record every attempt. A row that repeatedly receives the same invalid result needs a different model, a different prompt, or human review—not infinite retries.
• 5. Persist raw model output before parsing when your data-handling policy permits it. This is essential for diagnosing schema drift, prompt regressions, and unexpected refusal patterns.
• 6. Parse JSON, validate the schema, and run domain checks before marking a row complete. A syntactically valid JSON object is not proof that the extracted data is usable.
• 7. Route schema failures and low-confidence cases to an escalation queue using gpt-5.6-terra or claude-sonnet-4-6. Preserve the original source, prompt version, and primary failure reason.
• 8. Send unresolved rows to a human review queue with the source record, proposed output, validation errors, and model history attached.
• 9. Publish dashboards for accepted rows, failed rows, retry count, escalation rate, p50 and p95 latency, tokens per accepted row, and cost per accepted row.
• 10. Version prompts and schemas together. When either changes, run a held-out regression set before mixing new outputs with records produced by the prior version.

Use LumeAPI for real-time worker processing. For jobs explicitly designed to complete asynchronously within 24 hours, compare the relevant provider's official Batch API operating model separately.

Where supported, response_format.type set to json_object is the right default for machine-consumed extraction. It reduces the chance that a model wraps the payload in prose or markdown. It does not guarantee that all required keys exist, that values match your taxonomy, or that the model has correctly interpreted the source.

Your application remains responsible for a parser, a strict schema validator, and business rules. For example, a ticket can be valid JSON while still containing a category outside your allowed routing list, a date before the source document date, or entities that do not appear in the source. These are product-level failures, not transport-level failures.

Be explicit about uncertainty. Require null for unsupported scalar claims, empty arrays when no entities are found, and needs_review for records with conflicting or insufficient evidence. This is more useful than forcing the model to manufacture complete-looking objects for every row.

Finally, preserve an audit trail. Store the source version, extraction schema version, prompt version, model id, timestamp, validation outcome, and escalation history. That record lets teams explain a downstream classification, compare model routes fairly, and reprocess only the rows affected when requirements change.

LumeAPI is independent and is not OpenAI, Anthropic, or Google. Use LumeAPI catalog model ids exactly as listed in your request.