Anleitung: Batch-Verarbeitung

Batch-Jobs (einen Produktkatalog übersetzen, Support-Tickets klassifizieren, Felder aus Tausenden von Dokumenten extrahieren) senden dieselbe Art von Anfrage viele Male. Die Arbeit ist einfach; schwierig sind Durchsatz, Fehler und Kosten. Diese Anleitung behandelt alle drei mit dem offiziellen Paket openai und der Base URL https://api14.com/api/v1.

Es gibt keinen separaten Batch-Endpunkt: Die Batch API von OpenAI ist bei api14 nicht verfügbar (siehe OpenAI-SDK-Kompatibilität). Sie führen Batches als normale Anfragen mit kontrollierter Parallelität aus.

Ihr Rate Limit kennen

Limits werden in Anfragen pro Minute (RPM) pro API-Schlüssel gezählt und durch Ihren Plan begrenzt:

Ein Schlüssel mit eigenem RPM-Limit erhält den niedrigeren der beiden Werte. Prüfen Sie Ihr Limit mit GET /key (rate_limit.requests ist null, wenn der Schlüssel das Limit des Plans verwendet), siehe Guthaben und Schlüsselinfo. Bei Überschreitung des Limits kommt 429 rate_limited mit Retry-After: 60.

RPM begrenzt, wie viele Anfragen pro Minute starten, nicht wie viele gleichzeitig laufen. Wenn eine durchschnittliche Anfrage 4 Sekunden dauert und Ihr Limit 100 RPM beträgt, erreichen schon etwa 7 gleichzeitige Anfragen das Limit. Takten Sie die Anfragen nach Startzeit und halten Sie die Parallelität moderat.

Ein getakteter, paralleler 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           # Limit Ihres Plans oder Schlüssels
CONCURRENCY = 10
FATAL = {"daily_limit_reached", "insufficient_balance", "invalid_api_key", "model_not_allowed"}

class Pacer:
    """Verteilt die Anfragestarts gleichmäßig, damit RPM nie überschritten wird."""
    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)  # protokolliert; erneut ausführen, um zu wiederholen

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

Was das Skript tut:

• Taktung: Der Pacer startet höchstens RPM Anfragen pro Minute; das Semaphor begrenzt, wie viele gleichzeitig laufen.
• Wiederholungen: Das SDK wiederholt 429 und 5xx mit exponentiellem Backoff und beachtet Retry-After (max_retries=4). 429-Antworten des Upstream-Providers werden genauso wiederholt.
• Fatale Codes stoppen den Job: daily_limit_reached und insufficient_balance verschwinden durch Wiederholen nicht, daher beendet sich der Job, statt Versuche zu verbrennen.
• Checkpointing: Ergebnisse werden beim Fertigwerden an eine JSONL-Datei angehängt. Ein erneuter Lauf überspringt bereits erledigte IDs, sodass bei einem Absturz oder gestoppten Job nichts verloren geht.

In Node funktioniert dieselbe Struktur mit Promise.all über eine feste Anzahl von Workern, die aus einer gemeinsamen Warteschlange ziehen und zwischen den Starts jeweils 60_000 / RPM ms warten.

Mehrere Elemente in eine Anfrage packen

Bei kurzen Elementen verschwendet eine Anfrage pro Element RPM und wiederholt jedes Mal die Anweisungen. Klassifizieren Sie 20 bis 50 Elemente pro Anfrage mit einer strukturierten Ausgabe, und die Anzahl der Anfragen sinkt um denselben Faktor:

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"]

Prüfen Sie, dass jede Eingabe-ID zurückgekommen ist, und senden Sie fehlende erneut. Ist finish_reason gleich "length", war der Batch für max_tokens zu groß; teilen Sie ihn auf. Siehe Strukturierte Ausgaben. Für Embeddings übergeben Sie bis zu 100 Strings in einem input-Array, siehe Embeddings.

Kosten mit Prompt Caching senken

In einem Batch sind Anweisungen, Beispiele und Schema in jeder Anfrage identisch. Stellen Sie sie an den Anfang und das wechselnde Element ans Ende, damit Anbieter das zwischengespeicherte Präfix wiederverwenden können:

• OpenAI, DeepSeek, Gemini 2.5 und andere cachen automatisch, sobald das Präfix lang genug ist (1.024 Tokens bei OpenAI). Cache-Lesezugriffe kosten einen Bruchteil der normalen Eingabe.
• Anthropic-Modelle benötigen einen expliziten cache_control-Breakpoint am statischen Teil des Prompts.

Ein langer, detaillierter System-Prompt mit einigen ausgearbeiteten Beispielen kann daher kaum mehr kosten als ein kurzer und verbessert meist die Qualität. Treffer bestätigen Sie in usage.prompt_tokens_details.cached_tokens. Siehe Prompt Caching.

Ein günstiges Modell wählen, dann verifizieren

Für Übersetzung, Klassifizierung und Extraktion reichen kleine Modelle oft zu einem Bruchteil des Preises. Listen Sie Textmodelle nach Eingabepreis auf:

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

Messen Sie dann, bevor Sie sich festlegen:

1. Nehmen Sie 50 bis 100 repräsentative Elemente und legen Sie die erwartete Ausgabe von Hand fest.
2. Lassen Sie sie durch zwei oder drei Kandidatenmodelle laufen und vergleichen Sie Genauigkeit und die summierten usage.cost.
3. Multiplizieren Sie die Kosten pro Element mit Ihrer Gesamtzahl an Elementen, um den ganzen Job abzuschätzen, und prüfen Sie mit GET /credits, dass Ihr Guthaben ausreicht.

:free-Varianten nutzen den kostenlosen Endpunkt eines Modells, sofern vorhanden, haben aber strengere Upstream-Rate-Limits und eignen sich daher eher für Tests als für große Batches. Siehe Modell-IDs & Varianten.

Vor einem großen Lauf

• Erstellen Sie einen eigenen Schlüssel mit einem täglichen USD-Limit, damit ein Fehler nicht Ihr Guthaben leert. Der Job stoppt dann mit 429 daily_limit_reached. Siehe Rate Limits.
• Setzen Sie max_tokens bei jeder Anfrage.
• Führen Sie die ersten 1 % der Elemente aus, prüfen Sie Ausgabe und Kosten und starten Sie dann den Rest.
• Fehlgeschlagene Anfragen (jedes 4xx oder 5xx) werden nicht berechnet, daher ist das erneute Ausführen von Fehlern sicher. Siehe Fehler.