Batch API: Massive Asynchrone Verarbeitung

Aktualisiert am 3. November 2026 basierend auf der Batch API-Richtlinie und der Struktur der Responses API-Antworten: Das Auslesen der Ergebnisse geht nicht mehr davon aus, dass das erste Element von output die Nachricht ist (ein Reasoning-Element kann davor stehen), und der Fehlercode-Schlüssel, ohne Anführungszeichen geschrieben, führte zu einem NameError. Quelle: Batch API.

Latenz gegen Kosten abwägen

Die Batch API basiert auf einem einfachen Marktprinzip: Sie akzeptieren, dass Ihre Anfragen asynchron innerhalb eines Zeitfensters von bis zu 24 Stunden verarbeitet werden, und im Gegenzug zahlen Sie 50 % weniger. Alles, was nicht dazu dient, einem Menschen zu dienen, der auf sein Ergebnis wartet, fällt in diese Kategorie. Im großen Maßstab ist dies keine Mikro-Optimierung; sie halbiert Ihre Budgetlinie.

Die Gewohnheit, die Sie in einem Team aufbauen sollten, besteht darin, für jede neue Verarbeitungsaufgabe zu prüfen, ob jemand auf das Ergebnis wartet. Die Analyse eines Bestands an Verträgen, die Generierung von Produktdatenblättern für einen Katalog, das Durchführen eines Benchmarks auf tausenden Testfällen, die Klassifizierung eines sechsmonatigen Backlogs an Tickets oder das Extrahieren von Feldern aus einer Reihe von Rechnungen: In all diesen Fällen landet das Ergebnis in einem Dashboard oder einer Datenbank, nicht in einer konversationellen Schnittstelle. Die Verzögerung ist unsichtbar, aber die Einsparungen nicht.

Batch erstellen und starten

Das Eingabeformat ist eine JSONL-Datei: Eine Zeile pro Anfrage, wobei jede Zeile ein vollständiges, unabhängiges Objekt ist. Das Feld custom_id ermöglicht es Ihnen, jede Antwort mit ihrer ursprünglichen Anfrage zu verknüpfen, da nichts die Reihenfolge der Output-Datei garantiert. Wählen Sie eine ID, die stabil und aussagekräftig ist – beispielsweise eine Dokument-ID aus Ihrer Datenbank statt einer Iterationsnummer –, sonst wird die Abgleichphase zum Albtraum.

import json

requests = [
    {
        "custom_id": f"doc-{i}",
        "method": "POST",
        "url": "/v1/responses",
        "body": {
            "model": "gpt-5.6-luna",
            "input": f"Zusammenfassen Sie dieses Dokument in 3 Schlüsselpunkten: {doc}",
            "max_output_tokens": 500,
        },
    }
    for i, doc in enumerate(documents_list)
]

# JSONL-Datei schreiben
with open("batch_input.jsonl", "w") as f:
    for request in requests:
        f.write(json.dumps(request) + "\n")

Die Datei wird anschließend hochgeladen und beim Erstellen des Batches referenziert. Metadaten haben keinen technischen Zweck, werden Ihnen aber den Tag retten, wenn drei Batches parallel laufen und Sie herausfinden müssen, welchem Prompt-Version sie entspricht.

import openai

client = openai.OpenAI()

# Datei hochladen
file_obj = client.files.create(
    file=open("batch_input.jsonl", "rb"),
    purpose="batch",
)

# Batch erstellen
batch = client.batches.create(
    input_file_id=file_obj.id,
    endpoint="/v1/responses",
    completion_window="24h",
    metadata={"project": "contract-analysis", "version": "2.1"},
)

print(f"Batch erstellt: {batch.id}")
print(f"Status: {batch.status}")

Das Monitoring umfasst das Abfragen des Batches, bis es den Zustand „in Bearbeitung” verlässt, und das Herunterladen der Output-Datei, die ebenfalls im JSONL-Format vorliegt. Der folgende Code unterscheidet drei finale Ergebnisse – failed, expired, cancelled – und wirft für jedes eine Ausnahme aus: Ein abgelaufenes Batch ist eines, bei dem das 24-Stunden-Fenster verstrichen ist, bevor alles verarbeitet wurde, und es wäre die schlechteste Option, dies als stumme teilweise Erfolgsbehandlung zu betrachten.

import time

def wait_for_batch(batch_id: str, interval: int = 60) -> list:
    """Wartet auf das Ende eines Batches und gibt die Ergebnisse zurück."""
    while True:
        batch = client.batches.retrieve(batch_id)
        print(
            f"Status: {batch.status} | "
            f"Abgeschlossen: {batch.request_counts.completed}/"
            f"{batch.request_counts.total} | "
            f"Gefehlt: {batch.request_counts.failed}"
        )

        if batch.status == "completed":
            break
        elif batch.status in ("failed", "expired", "cancelled"):
            raise RuntimeError(f"Batch fehlgeschlagen: {batch.status}")

        time.sleep(interval)

    # Ergebnisse herunterladen
    content = client.files.content(batch.output_file_id)
    results = []
    for line in content.text.strip().split("\n"):
        results.append(json.loads(line))

    return results


results = wait_for_batch(batch.id)

# Ergebnisse verarbeiten
for result in results:
    custom_id = result["custom_id"]
    response = result["response"]
    if response["status_code"] == 200:
        # output kann andere Elemente vor der Nachricht enthalten
        # (z. B. Reasoning): Suchen Sie nach dem Nachrichtentext
        text = next(
            part["text"]
            for item in response["body"]["output"] if item["type"] == "message"
            for part in item["content"] if part["type"] == "output_text"
        )
        print(f"{custom_id}: {text[:100]}...")
    else:
        print(f"{custom_id}: ERROR {response['status_code']}")

Mit Teilfehlern planen

Bei einem großen Batch werden einige Anfragen fehlschlagen: ein abgeschnittenes Dokument, ein Kontext-Overflow oder eine vorübergehende Serverfehler. Ein Batch ist daher nie einfach nur „erfolgreich” oder „gefehlt” – es ist teilweise erfolgreich, und Sie müssen wissen, wie Sie den fehlenden Teil isolieren. Das zu adoptierende Muster besteht darin, Erfolge und Fehler sofort zu trennen und eine JSONL-Datei neu zu schreiben, die bereit für einen Neustart ist, anstatt drei Tage später durch Logs zu wühlen, um das Fehlende zu rekonstruieren.

def handle_batch_errors(results: list[dict]) -> tuple[list, list]:
    """Trennt Erfolge von Fehlern zur Neuverarbeitung."""
    successes = []
    failures = []

    for r in results:
        if r["response"]["status_code"] == 200:
            successes.append(r)
        else:
            failures.append(r)

    if failures:
        print(f"Warnung: {len(failures)} fehlgeschlagene Anfragen von {len(results)}")
        # Schreiben einer JSONL-Datei mit Fehlern für den Wiederlauf
        with open("batch_retry.jsonl", "w") as f:
            for failure in failures:
                original_request = {
                    "custom_id": failure["custom_id"],
                    "method": "POST",
                    "url": "/v1/responses",
                    "body": failure["request"]["body"],
                }
                f.write(json.dumps(original_request) + "\n")

    return successes, failures

Ihre Einsparungen potenzieren

Bei einer Verarbeitungslauf von zehntausend Dokumenten mit GPT-5.6 Luna stellt die synchrone API die Basiskosten dar, die Batch API die Hälfte davon, und Prompt-Caching fügt sich hinzu, da auch zwischengespeicherte Tokens von ihrer eigenen Reduktion profitieren. Die beiden Mechanismen konkurrieren nicht: Wenn Ihre zehntausend Anfragen denselben langen System-Prompt teilen, strukturieren Sie ihn als stabiles Präfix, und Sie potenzieren beide Effekte.

Schließlich stellt sich die Frage der Aufteilung. Ein Batch sollte nicht unendlich sein, und es ist angenehmer, mehrere handhabbare Batches zu überwachen als einen einzelnen Monolithen, dessen Fehler einen ganzen Tag kostet. Die folgende Funktion schneidet eine Liste von Anfragen in Batches begrenzter Größe auf.

def create_optimized_batches(
    requests: list[dict],
    max_size: int = 50_000,
) -> list[list[dict]]:
    """Teilt in Batches optimaler Größe auf."""
    batches = []
    current_batch = []

    for request in requests:
        current_batch.append(request)
        if len(current_batch) >= max_size:
            batches.append(current_batch)
            current_batch = []

    if current_batch:
        batches.append(current_batch)

    return batches

Kernaussagen

• Die Batch API bietet eine 50 %ige Reduzierung für nicht-Echtzeit-Verarbeitung
• Bereiten Sie Ihre Anfragen in einer JSONL-Datei vor mit einer eindeutigen custom_id pro Anfrage
• Planen Sie einen Wiederlauf-Mechanismus für fehlgeschlagene Anfragen
• Kombinieren Sie dies mit Prompt-Caching, um die Einsparungen zu maximieren