Gids: batchverwerking

Batchtaken (een productcatalogus vertalen, supporttickets classificeren, velden uit duizenden documenten halen) sturen dezelfde soort request vele malen. Het werk zelf is eenvoudig; de lastige onderdelen zijn doorvoer, fouten en kosten. Deze gids behandelt alle drie met het officiële pakket openai en base URL https://api14.com/api/v1.

Er is geen apart batch-endpoint: de Batch API van OpenAI is niet beschikbaar op api14 (zie Compatibiliteit met de OpenAI-SDK). Je voert batches uit als gewone requests met gecontroleerde gelijktijdigheid.

Ken je rate limit

Limieten worden geteld in requests per minuut (RPM) per API-key, begrensd door je abonnement:

Een key met een eigen RPM-limiet krijgt de laagste van de twee. Controleer die van jou met GET /key (rate_limit.requests is null wanneer de key de limiet van het abonnement gebruikt), zie Saldo en key-info. Bij overschrijding komt 429 rate_limited terug met Retry-After: 60.

RPM begrenst hoeveel requests per minuut starten, niet hoeveel er tegelijk lopen. Als een gemiddeld request 4 seconden duurt en je limiet 100 RPM is, bereiken ongeveer 7 gelijktijdige requests de limiet al. Pace requests op starttijd en houd de gelijktijdigheid gematigd.

Een gepaceerde, gelijktijdige runner (Python)

import asyncio, json, os, time
from openai import AsyncOpenAI, APIStatusError

client = AsyncOpenAI(base_url="https://api14.com/api/v1", api_key=os.environ["API14_API_KEY"], max_retries=4)

RPM = 100           # je abonnements- of keylimiet
CONCURRENCY = 10
FATAL = {"daily_limit_reached", "insufficient_balance", "invalid_api_key", "model_not_allowed"}

class Pacer:
    """Spaces request starts evenly so we never exceed RPM."""
    def __init__(self, rpm):
        self.interval, self.next, self.lock = 60 / rpm, 0.0, asyncio.Lock()
    async def wait(self):
        async with self.lock:
            now = time.monotonic()
            self.next = max(self.next, now)
            delay, self.next = self.next - now, self.next + self.interval
        await asyncio.sleep(delay)

SYSTEM = "Translate the user's text to German. Keep product names and units unchanged. Reply with the translation only."

async def translate(item, pacer, sem):
    async with sem:
        await pacer.wait()
        r = await client.chat.completions.create(
            model="openai/gpt-5-mini",
            messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": item["text"]}],
            max_tokens=500,
        )
        return {"id": item["id"], "text": r.choices[0].message.content, "cost": r.usage.cost}

async def main(items, out_path="out.jsonl"):
    done = set()
    if os.path.exists(out_path):
        done = {json.loads(line)["id"] for line in open(out_path)}
    todo = [i for i in items if i["id"] not in done]

    pacer, sem = Pacer(RPM), asyncio.Semaphore(CONCURRENCY)
    with open(out_path, "a") as out:
        for task in asyncio.as_completed([translate(i, pacer, sem) for i in todo]):
            try:
                out.write(json.dumps(await task) + "\n")
            except APIStatusError as e:
                if e.code in FATAL:
                    raise SystemExit(f"Stopping batch: {e.code}")
                print("failed:", e.status_code, e.code)  # gelogd; draai opnieuw om het nog eens te proberen

items = [json.loads(line) for line in open("items.jsonl")]  # [{"id": "...", "text": "..."}]
asyncio.run(main(items))

Wat dit doet:

• Pacing: de Pacer start maximaal RPM requests per minuut; de semafoor begrenst hoeveel er tegelijk lopen.
• Nieuwe pogingen: de SDK probeert 429 en 5xx opnieuw met exponentiële backoff en respecteert Retry-After (max_retries=4). 429's van de upstream-provider worden op dezelfde manier opnieuw geprobeerd.
• Fatale codes stoppen de taak: daily_limit_reached en insufficient_balance verdwijnen niet door opnieuw te proberen, dus de taak sluit af in plaats van pogingen te verspillen.
• Checkpointing: resultaten worden als JSONL aan een bestand toegevoegd zodra ze klaar zijn. Bij opnieuw draaien slaat het script id's over die al klaar zijn, zodat een crash of gestopte taak niets verliest.

In Node werkt dezelfde structuur met Promise.all over een vast aantal workers die uit een gedeelde wachtrij putten, waarbij elk 60_000 / RPM ms tussen starts wacht.

Meerdere items in één request

Bij korte items verspilt één request per item RPM en herhaalt het de instructies elke keer. Classificeer 20 tot 50 items per request met een gestructureerde uitvoer, en je deelt het aantal requests door dezelfde factor:

schema = {
    "type": "object",
    "properties": {"labels": {"type": "array", "items": {
        "type": "object",
        "properties": {"id": {"type": "string"}, "label": {"type": "string", "enum": ["billing", "bug", "feature", "other"]}},
        "required": ["id", "label"], "additionalProperties": False,
    }}},
    "required": ["labels"], "additionalProperties": False,
}

tickets = "\n".join(f"{t['id']}: {t['text']}" for t in batch_of_30)
r = await client.chat.completions.create(
    model="openai/gpt-5-mini",
    messages=[{"role": "system", "content": "Label each ticket."}, {"role": "user", "content": tickets}],
    response_format={"type": "json_schema", "json_schema": {"name": "labels", "strict": True, "schema": schema}},
)
labels = json.loads(r.choices[0].message.content)["labels"]

Controleer of elk invoer-id is teruggekomen; stuur ontbrekende opnieuw. Als finish_reason "length" is, was de batch te groot voor max_tokens: splits hem. Zie Structured outputs. Geef voor embeddings tot 100 strings mee in één input-array, zie Embeddings.

Kosten verlagen met prompt caching

In een batch zijn de instructies, voorbeelden en het schema in elk request identiek. Zet ze vooraan en het wisselende item achteraan, zodat providers de gecachete prefix kunnen hergebruiken:

• OpenAI, DeepSeek, Gemini 2.5 en andere cachen automatisch zodra de prefix lang genoeg is (1.024 tokens bij OpenAI). Cachelezingen kosten een fractie van normale invoer.
• Anthropic-modellen hebben een expliciet cache_control-breakpoint nodig op het statische deel van de prompt.

Een lange, gedetailleerde systeemprompt met enkele uitgewerkte voorbeelden kan dus nauwelijks meer kosten dan een korte, en verbetert meestal de kwaliteit. Bevestig treffers in usage.prompt_tokens_details.cached_tokens. Zie Prompt caching.

Kies een goedkoop model en controleer het dan

Voor vertaling, classificatie en extractie zijn kleine modellen vaak goed genoeg voor een fractie van de prijs. Toon tekstmodellen op invoerprijs:

curl -s "https://api14.com/api/v1/models?category=text" \
  | jq -r '.data | sort_by(.pricing.prompt | tonumber) | .[:15][] | "\(.id)  \(.pricing.prompt)  \(.pricing.completion)"'

Meet daarna voordat je je vastlegt:

1. Neem 50 tot 100 representatieve items en label de verwachte uitvoer met de hand.
2. Laat ze door twee of drie kandidaatmodellen lopen en vergelijk de nauwkeurigheid en de opgetelde usage.cost.
3. Vermenigvuldig de kosten per item met je totale aantal items om de hele taak te schatten, en controleer met GET /credits of je saldo dat dekt.

:free-varianten gebruiken het gratis endpoint van een model waar dat bestaat, maar hebben strengere upstream rate limits en zijn daarom geschikter voor tests dan voor grote batches. Zie Model-id's en varianten.

Voor een grote run

• Maak een aparte key met een dagelijkse USD-limiet, zodat een bug je saldo niet kan leegtrekken. De taak stopt dan met 429 daily_limit_reached. Zie Rate limits.
• Stel max_tokens in op elk request.
• Draai eerst de eerste 1% van de items, bekijk de uitvoer en kosten, en start daarna de rest.
• Mislukte requests (elke 4xx of 5xx) worden niet in rekening gebracht, dus mislukkingen opnieuw draaien is veilig. Zie Fouten.