OpenAI Batch API w Pythonie: 50% taniej za klasyfikację, embeddingi i ewaluacje (2026)

Aktualizacja: 8 lipca 2026

OpenAI Batch API to asynchroniczny endpoint, który wykonuje żądania do modeli GPT w ciągu do 24 godzin i pobiera 50% mniej niż standardowe wywołania synchroniczne. To czyni go domyślnym wyborem do klasyfikacji, ewaluacji, generowania embeddingów i innych zadań, w których nie potrzebujesz odpowiedzi w czasie rzeczywistym. W tym poradniku pokażę, jak wysłać pierwszy batch w Pythonie, jak radzić sobie z częściowymi błędami, jak łączyć batch z function callingiem oraz jak porównuje się z Message Batches API Anthropic. Wszystkie przykłady oparłem na SDK openai==1.60+ i anthropic==0.42+ (stan na lipiec 2026).

• Batch API OpenAI daje 50% zniżki na inputy i outputy dla wszystkich modeli GPT-4o, GPT-4.1, GPT-5 i o3-mini z SLA do 24 godzin.
• Format wejściowy to JSONL, gdzie każda linia to jedno kompletne żądanie w strukturze zgodnej z /v1/chat/completions, /v1/embeddings lub /v1/responses.
• Batch obsługuje function calling, structured output i wszystkie parametry samplingu. Nie musisz upraszczać zapytań, żeby skorzystać ze zniżki.
• Limity: max 50 000 requestów lub 200 MB na batch, oddzielna kolejka tokenów niż limit synchroniczny (24h queued tokens).
• Anthropic Message Batches API oferuje identyczną zniżkę 50% i podobny model 24h. Różnice sprowadzają się do formatu wyjścia i integracji z prompt cachingiem.
• Częściowe błędy w batchu są normą, więc musisz parsować plik output oraz plik error linia po linii i mieć strategię retry na poziomie pojedynczych requestów.

Czym jest OpenAI Batch API i kiedy go używać

Batch API to asynchroniczna alternatywa dla zwykłego endpointu /v1/chat/completions. Zamiast wysyłać jedno żądanie i czekać na odpowiedź w ciągu sekund, przesyłasz plik JSONL z tysiącami żądań, dostajesz identyfikator batchu, a po zakończeniu przetwarzania (deklarowane SLA: 24h; w praktyce w połowie 2026 mediana to około 40 minut) pobierasz plik wyjściowy z odpowiedziami. W zamian OpenAI liczy o 50% mniej za tokeny wejściowe i wyjściowe. Dokładnie tyle samo, ile daje prompt caching w Claude API, ale bez ograniczenia dotyczącego 5-minutowego TTL cache'a.

Szczerze? W mojej praktyce Batch API najlepiej sprawdza się w scenariuszach, w których liczy się przepustowość, nie latencja:

• Ewaluacje offline, czyli puszczanie tysięcy przykładów przez model LLM-as-a-judge nocą, żeby rano mieć nowy raport.
• Klasyfikacja i tagowanie, czyli przetwarzanie kolejek dokumentów, ticketów, feedbacku od użytkowników.
• Generowanie embeddingów, czyli indeksowanie dużych korpusów do wektorowej bazy danych (Qdrant, pgvector, Pinecone) przed uruchomieniem pipeline'u RAG.
• Wzbogacanie danych, czyli dodawanie streszczeń, tłumaczeń, metadanych do rekordów w bazie.
• Backfill, czyli jednorazowe przeliczenie historycznych danych po zmianie promptu lub modelu.

Jeśli budujesz interaktywnego czatbota lub agenta reagującego w czasie rzeczywistym, Batch API nie jest dla Ciebie. Ale jeśli Twoja aplikacja generuje kolejki „gdzieś w tle" (a większość aplikacji AI produkcyjnych taką kolejkę ma), 50% zniżki potrafi zmienić ekonomię całego projektu.

Pierwszy batch w Pythonie krok po kroku

Zacznijmy od najprostszego przypadku: klasyfikacji 100 opinii klientów jako pozytywnych, negatywnych lub neutralnych. Cały workflow ma cztery kroki: (1) przygotuj plik JSONL, (2) uploaduj go, (3) utwórz batch, (4) pobierz wyniki.

Krok 1: przygotowanie pliku JSONL

Każda linia pliku wejściowego to jeden request. Wymagane pola: custom_id (Twój identyfikator do dopasowania odpowiedzi), method (zawsze POST), url (endpoint) i body (dokładnie taki sam JSON, jaki wysłałbyś do sync API).

import json
from pathlib import Path

opinions = [
    {"id": "op_001", "text": "Świetny produkt, kupiłem trzy sztuki dla rodziny."},
    {"id": "op_002", "text": "Dostawa spóźniona o tydzień, obsługa nie odpowiada."},
    # ... 98 kolejnych
]

def build_request(op):
    return {
        "custom_id": op["id"],
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-4o-mini",
            "messages": [
                {"role": "system", "content": "Zwróć wyłącznie jedno słowo: pozytywna, negatywna lub neutralna."},
                {"role": "user", "content": op["text"]},
            ],
            "max_tokens": 5,
            "temperature": 0,
        },
    }

Path("requests.jsonl").write_text(
    "\n".join(json.dumps(build_request(op), ensure_ascii=False) for op in opinions),
    encoding="utf-8",
)

Krok 2: upload i utworzenie batchu

from openai import OpenAI

client = OpenAI()

batch_file = client.files.create(
    file=open("requests.jsonl", "rb"),
    purpose="batch",
)

batch = client.batches.create(
    input_file_id=batch_file.id,
    endpoint="/v1/chat/completions",
    completion_window="24h",
    metadata={"job": "opinion_classification", "run": "2026-07-08"},
)

print(f"Batch ID: {batch.id}, status: {batch.status}")

Krok 3: polling statusu i pobranie wyników

Batch ma następujące stany: validating, in_progress, finalizing, completed, failed, expired, cancelled. W praktyce robię polling co 60 sekund, ale w środowisku produkcyjnym warto ustawić webhook w Assistants Dashboard albo trigger w kolejce zadań.

import time

while True:
    batch = client.batches.retrieve(batch.id)
    print(f"{batch.status}: {batch.request_counts.completed}/{batch.request_counts.total}")
    if batch.status in {"completed", "failed", "expired", "cancelled"}:
        break
    time.sleep(60)

if batch.status == "completed":
    output = client.files.content(batch.output_file_id).text
    results = [json.loads(line) for line in output.splitlines()]
    for r in results:
        label = r["response"]["body"]["choices"][0]["message"]["content"].strip()
        print(r["custom_id"], "->", label)

To wszystko. Bez żadnych dodatkowych bibliotek. Zwróć uwagę, że custom_id pozwala mi zmapować odpowiedź do rekordu, bo kolejność w pliku wyjściowym nie jest gwarantowana. Dlatego zawsze używaj custom_id.

Batch API z function callingiem i structured output

To jest część, w której Batch API bywa niedocenione. Wielu deweloperów zakłada, że jeśli chcesz zniżki, musisz zrezygnować z zaawansowanych funkcji. Nieprawda. Dokumentacja OpenAI Batch API jasno stwierdza, że wszystkie parametry endpointów synchronicznych są obsługiwane, w tym tools, tool_choice, response_format i parallel_tool_calls.

W schema-first workflow, który polecam, robię tak: definiuję schemat Pydanticiem, generuję z niego JSON Schema i puszczam batch z response_format. Poniżej ekstrakcja strukturalna z 10 000 opinii, gdzie chcę wyciągnąć sentyment, temat i intencję zakupową:

from pydantic import BaseModel, Field
from typing import Literal
import json

class OpinionAnalysis(BaseModel):
    sentiment: Literal["pozytywny", "negatywny", "neutralny"]
    topic: Literal["cena", "jakość", "dostawa", "obsługa", "inne"]
    purchase_intent: bool = Field(description="Czy autor prawdopodobnie kupi ponownie")
    key_phrase: str = Field(max_length=80)

schema = OpinionAnalysis.model_json_schema()

def build_structured_request(op):
    return {
        "custom_id": op["id"],
        "method": "POST",
        "url": "/v1/chat/completions",
        "body": {
            "model": "gpt-4o-2024-08-06",
            "messages": [
                {"role": "system", "content": "Przeanalizuj opinię klienta zgodnie ze schematem."},
                {"role": "user", "content": op["text"]},
            ],
            "response_format": {
                "type": "json_schema",
                "json_schema": {
                    "name": "OpinionAnalysis",
                    "schema": schema,
                    "strict": True,
                },
            },
        },
    }

Parsowanie wyników przez Pydantic daje mi walidację za darmo. Jeśli model zwróci coś, co nie pasuje do schematu (co przy strict: true praktycznie się nie zdarza), dostanę wyjątek i mogę wrzucić rekord do retry queue:

parsed = []
errors = []
for line in output.splitlines():
    r = json.loads(line)
    try:
        raw = r["response"]["body"]["choices"][0]["message"]["content"]
        parsed.append((r["custom_id"], OpinionAnalysis.model_validate_json(raw)))
    except Exception as e:
        errors.append((r["custom_id"], str(e)))

Więcej o tym wzorcu piszę w poradniku o structured output z Pydantic i Instructorem. Tam znajdziesz też porównanie strict mode z podejściem re-ask, które ma sens, gdy pracujesz z modelami bez natywnego structured output.

Jak obsługiwać błędy i częściowe niepowodzenia

Jeden z najczęstszych błędów, które widzę u zespołów wdrażających Batch API po raz pierwszy: zakładają, że jeśli batch.status == "completed", to wszystkie requesty się powiodły. Nieprawda. Batch może być completed z 5% requestów, które zwróciły błąd (rate limit, content policy, timeout na pojedynczym prompcie). Te błędy trafiają do osobnego pliku dostępnego pod batch.error_file_id.

Na tym się kiedyś sparzyłem, jeszcze w projekcie z zeszłego roku. Puściliśmy 40 000 klasyfikacji, statust był completed, wynik trafił do bazy, a 2 tygodnie później okazało się, że ~800 rekordów po prostu nie miało etykiety. Od tamtego czasu zawsze porównuję listę custom_id na wejściu i wyjściu:

def collect_results(batch):
    successes, failures = {}, {}

    if batch.output_file_id:
        output = client.files.content(batch.output_file_id).text
        for line in output.splitlines():
            r = json.loads(line)
            if r.get("error"):
                failures[r["custom_id"]] = r["error"]
            else:
                successes[r["custom_id"]] = r["response"]["body"]

    if batch.error_file_id:
        errors = client.files.content(batch.error_file_id).text
        for line in errors.splitlines():
            r = json.loads(line)
            failures[r["custom_id"]] = r.get("error", {"message": "unknown"})

    return successes, failures

Moja standardowa strategia retry: (1) zbierz wszystkie custom_id z failures, (2) sprawdź kod błędu, (3) requesty z rate_limit_exceeded puść w kolejnym batchu, (4) requesty z invalid_request_error lub content_policy_violation zaloguj i przekaż do przeglądu ręcznego. Nie próbuj retry na content policy. Jest deterministyczny i będzie failował w kółko.

Batch API vs Anthropic Message Batches, porównanie

W 2026 obie firmy oferują funkcjonalnie równoważne API batchowe. Bez flame-warowania, oto porównanie na podstawie tego, co realnie robię w projektach:

Cecha	OpenAI Batch API	Anthropic Message Batches API
Zniżka na tokeny	50% input + 50% output	50% input + 50% output
SLA	do 24h (mediana ~40 min)	do 24h (mediana ~1h)
Format wejścia	JSONL uploadowany do Files API	Lista requestów bezpośrednio w wywołaniu API
Max requestów na batch	50 000 lub 200 MB	100 000 lub 256 MB
Function calling / tools	Pełne wsparcie	Pełne wsparcie
Structured output	response_format json_schema, strict mode	Tool use z JSON Schema
Prompt caching	Nie stackuje się z batchem	Stackuje się, dodatkowe 90% zniżki na cached tokens
Retencja plików	Output file dostępny 29 dni	Wyniki dostępne 29 dni
Endpointy	chat/completions, embeddings, responses	messages (bez natywnych embeddingów)

Krótka wersja. Jeśli robisz masowe podsumowania długich dokumentów z powtarzającymi się instrukcjami systemowymi, Anthropic wygrywa dzięki stackowaniu batch + prompt caching. Realny koszt spada nawet do 5% ceny bazowej. Jeśli generujesz embeddingi lub potrzebujesz jednego API do wszystkiego, OpenAI wygrywa. Do agnostycznej warstwy używam LiteLLM w Pythonie z fallbackami i trackingiem kosztów, choć tryb batch w LiteLLM w połowie 2026 jest jeszcze niepełny.

Anthropic Message Batches, minimalny przykład

from anthropic import Anthropic
from anthropic.types.messages.batch_create_params import Request
from anthropic.types.message_create_params import MessageCreateParamsNonStreaming

client = Anthropic()

batch = client.messages.batches.create(
    requests=[
        Request(
            custom_id=f"op_{i:03d}",
            params=MessageCreateParamsNonStreaming(
                model="claude-sonnet-5",
                max_tokens=100,
                system=[
                    {
                        "type": "text",
                        "text": "Zwróć jedno słowo: pozytywna, negatywna lub neutralna.",
                        "cache_control": {"type": "ephemeral"},
                    }
                ],
                messages=[{"role": "user", "content": op["text"]}],
            ),
        )
        for i, op in enumerate(opinions)
    ],
)

# polling
while True:
    b = client.messages.batches.retrieve(batch.id)
    if b.processing_status == "ended":
        break
    time.sleep(60)

for result in client.messages.batches.results(batch.id):
    if result.result.type == "succeeded":
        print(result.custom_id, result.result.message.content[0].text)

Batchowe generowanie embeddingów do RAG

Osobny use case, który zasługuje na własną sekcję. Jeśli indeksujesz korpus do wektorowej bazy (powiedzmy 500 000 chunków po średnio 400 tokenów), synchroniczne wywołania /v1/embeddings zajmą kilka godzin i kosztują pełną stawkę. Batch API robi to nocą i za połowę ceny. Przy text-embedding-3-large w 2026 to różnica około 30 vs 15 USD na milion tokenów, więc dla korpusów rzędu miliardów tokenów to poważna zmiana ekonomiczna.

def embedding_request(chunk_id, text):
    return {
        "custom_id": chunk_id,
        "method": "POST",
        "url": "/v1/embeddings",
        "body": {
            "model": "text-embedding-3-large",
            "input": text,
            "dimensions": 1024,  # Matryoshka truncation dla mniejszego indeksu
        },
    }

with open("embeddings_batch.jsonl", "w", encoding="utf-8") as f:
    for chunk in chunks:
        f.write(json.dumps(embedding_request(chunk.id, chunk.text), ensure_ascii=False) + "\n")

batch_file = client.files.create(file=open("embeddings_batch.jsonl", "rb"), purpose="batch")
batch = client.batches.create(
    input_file_id=batch_file.id,
    endpoint="/v1/embeddings",
    completion_window="24h",
)

Po zakończeniu każdy wynik zawiera data[0].embedding, który wrzucasz bezpośrednio do Qdranta lub pgvector. W kombinacji z rerankerem opisanym w poradniku o hybrid search w RAG dostajesz kompletny pipeline: batch do indeksacji, sync do zapytań, hybrid search do jakości.

Monitorowanie statusu, kosztów i limitów

Trzy rzeczy, które musisz mieć w produkcji.

Monitoring statusu bez pollingu

Polling co minutę na 1000 batchy dziennie to niepotrzebne obciążenie i zaśmiecone logi. W lipcu 2026 OpenAI udostępnia webhooki batch eventów: batch.completed, batch.failed, batch.expired. Skonfiguruj webhook endpoint, weryfikuj podpis HMAC i puszczaj resztę pipeline'u dopiero po zdarzeniu.

Śledzenie kosztów

Zniżka jest naliczana automatycznie i widoczna w Usage dashboard jako osobna linia „Batch". W kodzie warto liczyć koszt lokalnie na podstawie usage.prompt_tokens i usage.completion_tokens zwracanych w każdej odpowiedzi. Prosty helper:

PRICING_2026 = {
    "gpt-4o-mini":   {"in": 0.150,  "out": 0.600},   # USD za 1M tokenów, sync
    "gpt-4o":        {"in": 2.500,  "out": 10.000},
    "gpt-5":         {"in": 5.000,  "out": 20.000},
}

def batch_cost(model, usage):
    p = PRICING_2026[model]
    return 0.5 * (usage["prompt_tokens"] * p["in"] + usage["completion_tokens"] * p["out"]) / 1_000_000

Limity i quoty

Batch queue mierzy się w tokenach w oknie 24-godzinnym, nie w tokenach na minutę. Twój tier określa, ile tokenów możesz mieć „w powietrzu" naraz. Jeśli utworzysz batch z token_limit_exceeded, dostaniesz status failed od razu i musisz podzielić plik na mniejsze części. Praktyczna reguła: shard po 5000 requestów na batch, uruchamiaj równolegle 3–5 batchy i będziesz mieć płynny throughput.

Kiedy Batch API to zły pomysł

Sytuacje, w których Batch API pogorszy Twój produkt zamiast go poprawić:

• Interaktywne aplikacje, gdzie user czeka na odpowiedź w UI. 24h okno oznacza konieczność zbudowania kolejki „w tle" z powiadomieniem, a jeśli i tak nie budujesz tej architektury, sync jest prostszy.
• Agenci wieloetapowi. Batch nie potrafi zrobić pętli think-act. Każdy krok musi być osobnym batchem, co dodaje wiele godzin latencji na krok. Do agentów wybierz sync, zobacz porównanie CrewAI i LangGraph.
• Testowanie zmian promptu. Cykl iteracyjny „zmień prompt, sprawdź wynik" wymaga sekund, nie minut. Iteruj na sync API na małej próbce, dopiero potem puszczaj batch na pełny dataset.
• Bardzo małe wolumeny. Poniżej ~100 requestów zysk 50% jest niższy niż koszt operacyjny obsługi asynchronicznego workflow. Sync + asyncio.gather jest wystarczające.
• Ścisłe SLA. Jeśli klient płaci Ci za odpowiedź w < 1h, nie ryzykuj batchu. OpenAI gwarantuje tylko 24h.

Reguła kciuka: Batch API opłaca się, kiedy masz kolejkę zadań, którą i tak masz zamiar puścić „nocą", a wolumen przekracza kilka tysięcy requestów. Wtedy 50% zniżki to czysty zysk bez kompromisów jakościowych. A dokładnie taki argument potrzebujesz, żeby przekonać zespół finansowy do skalowania AI w firmie.

Najczęściej zadawane pytania

Ile kosztuje Batch API w OpenAI?

Dokładnie 50% ceny synchronicznej dla wszystkich modeli i wszystkich obsługiwanych endpointów, zarówno inputy, jak i outputy. Ceny są liczone automatycznie w Usage dashboard jako osobna kategoria „Batch" i widoczne w każdym wywołaniu jako usage.prompt_tokens i usage.completion_tokens w odpowiedzi.

Jak długo trwa przetwarzanie batchu w praktyce?

Deklarowane SLA to 24 godziny, ale w połowie 2026 mediana dla batchy poniżej 10 000 requestów to około 40 minut na modelach GPT-4o i 1–2 godziny na GPT-5. Duże batche (50 000 requestów) potrafią zająć 3–8 godzin. Przy zwiększonym ruchu OpenAI wydłuża kolejkę i wtedy realny czas zbliża się do 24h.

Czy Batch API obsługuje function calling i structured output?

Tak, oba w pełni. Możesz używać tools, tool_choice, parallel_tool_calls oraz response_format z JSON Schema w trybie strict. Batch traktuje pole body identycznie jak sync API. Nie ma parametrów wyłącznych dla jednego z trybów.

Czym różni się Batch API od Anthropic Message Batches?

Oba oferują 50% zniżki i SLA 24h. Anthropic pozwala stackować batch z prompt cachingiem, co przy powtarzalnych instrukcjach systemowych daje realny koszt bliski 5% ceny bazowej. OpenAI używa uploadowanego pliku JSONL, Anthropic wysyła requesty bezpośrednio w wywołaniu API. OpenAI obsługuje embeddingi w batchu, Anthropic nie.

Co się dzieje, kiedy batch wygaśnie po 24 godzinach?

Batch przechodzi w stan expired. Requesty, które zdążyły się wykonać, są dostępne w output file, reszta jest bezpowrotnie tracona. Musisz porównać listę custom_id z pliku wejściowego z listą wyników i wysłać brakujące requesty ponownie. W praktyce dzieje się to rzadko, ale kod produkcyjny musi to obsłużyć.

Czy mogę anulować batch po jego utworzeniu?

Tak, przez wywołanie client.batches.cancel(batch_id). Cancelling to stan przejściowy: OpenAI kończy requesty, które są już w trakcie, i zwraca je w output file. Nie płacisz za requesty, które nie zostały wykonane, ale płacisz za te, które już wystartowały.