Przewodnik: przetwarzanie wsadowe

Zadania wsadowe (tłumaczenie katalogu produktów, klasyfikacja zgłoszeń do supportu, wyodrębnianie pól z tysięcy dokumentów) wysyłają wiele razy ten sam rodzaj żądania. Sama praca jest prosta; trudne są przepustowość, awarie i koszt. Ten przewodnik omawia wszystkie trzy, używając oficjalnego pakietu openai z bazowym adresem URL https://api14.com/api/v1.

Nie ma osobnego endpointu wsadowego: Batch API od OpenAI nie jest dostępne w api14 (zobacz zgodność z SDK OpenAI). Wsady uruchamiasz jako zwykłe żądania z kontrolowaną współbieżnością.

Poznaj swój limit zapytań

Limity są liczone w zapytaniach na minutę (RPM) dla każdego klucza API i ograniczone przez Twój plan:

Klucz z własnym limitem RPM dostaje niższy z dwóch limitów. Sprawdź swój przez GET /key (rate_limit.requests ma wartość null, gdy klucz używa limitu planu), zobacz Saldo i informacje o kluczu. Przekroczenie limitu zwraca 429 rate_limited z Retry-After: 60.

RPM ogranicza, ile żądań startuje na minutę, a nie ile działa jednocześnie. Jeśli średnie żądanie trwa 4 sekundy, a Twój limit to 100 RPM, około 7 żądań w toku już osiąga limit. Rozkładaj żądania według czasu startu i utrzymuj umiarkowaną współbieżność.

Rozłożony w czasie, współbieżny 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           # your plan or key limit
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)  # logged; rerun to retry

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

Co to robi:

• Rozkładanie w czasie: Pacer startuje co najwyżej RPM żądań na minutę; semafor ogranicza liczbę żądań w toku.
• Ponowienia: SDK ponawia 429 i 5xx z wykładniczym opóźnieniem, respektując Retry-After (max_retries=4). Odpowiedzi 429 od dostawcy są ponawiane tak samo.
• Kody krytyczne zatrzymują zadanie: daily_limit_reached i insufficient_balance nie znikną po ponowieniu, więc zadanie kończy się zamiast marnować próby.
• Punkty kontrolne: wyniki są dopisywane do pliku JSONL, gdy tylko się pojawią. Ponowne uruchomienie skryptu pomija identyfikatory, które są już gotowe, więc awaria lub zatrzymane zadanie nie traci niczego.

W Node ta sama struktura działa z Promise.all na stałej liczbie workerów pobierających z wspólnej kolejki, z których każdy czeka 60_000 / RPM ms między startami.

Umieść kilka elementów w jednym żądaniu

Dla krótkich elementów jedno żądanie na element marnuje RPM i za każdym razem powtarza instrukcje. Klasyfikuj 20–50 elementów w jednym żądaniu z ustrukturyzowanym wyjściem, a liczbę żądań podzielisz przez ten sam czynnik:

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

Sprawdź, czy wróciły wszystkie identyfikatory wejściowe; brakujące wyślij ponownie. Jeśli finish_reason ma wartość "length", wsad był za duży dla max_tokens, podziel go. Zobacz Ustrukturyzowane wyjście. Dla embeddingów przekaż do 100 ciągów w jednej tablicy input, zobacz Embeddingi.

Obniż koszt dzięki cache promptów

We wsadzie instrukcje, przykłady i schemat są identyczne w każdym żądaniu. Umieść je na początku, a zmienny element na końcu, aby dostawcy mogli użyć prefiksu z cache:

• OpenAI, DeepSeek, Gemini 2.5 i inni stosują cache automatycznie, gdy prefiks jest wystarczająco długi (1,024 tokenów w OpenAI). Odczyty z cache kosztują ułamek zwykłych tokenów wejściowych.
• Modele Anthropic wymagają jawnego punktu podziału cache_control na statycznej części promptu.

Długi, szczegółowy prompt systemowy z kilkoma opracowanymi przykładami może więc kosztować niewiele więcej niż krótki, a zwykle poprawia jakość. Trafienia potwierdzisz w usage.prompt_tokens_details.cached_tokens. Zobacz Cache promptów.

Wybierz tani model, a potem zweryfikuj

W tłumaczeniu, klasyfikacji i ekstrakcji małe modele są często wystarczająco dobre przy ułamku ceny. Wyświetl modele tekstowe według ceny wejścia:

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

Potem zmierz, zanim się zdecydujesz:

1. Weź 50–100 reprezentatywnych elementów i ręcznie oznacz oczekiwane wyjście.
2. Uruchom je na dwóch lub trzech modelach kandydatach i porównaj dokładność oraz zsumowany usage.cost.
3. Pomnóż koszt na element przez łączną liczbę elementów, aby oszacować całe zadanie, i sprawdź przez GET /credits, czy Twoje saldo je pokryje.

Warianty :free używają bezpłatnego endpointu modelu, jeśli istnieje, ale mają surowsze limity zapytań po stronie dostawcy, więc nadają się bardziej do testów niż dużych wsadów. Zobacz Identyfikatory modeli i warianty.

Przed dużym uruchomieniem

• Utwórz dedykowany klucz z dziennym limitem w USD, aby błąd nie opróżnił Twojego salda. Zadanie zatrzyma się wtedy z 429 daily_limit_reached. Zobacz Limity zapytań.
• Ustaw max_tokens w każdym żądaniu.
• Uruchom pierwszy 1% elementów, sprawdź wynik i koszt, a dopiero potem start reszty.
• Nieudane żądania (dowolne 4xx lub 5xx) nie są rozliczane, więc ponowne uruchomienie nieudanych jest bezpieczne. Zobacz Błędy.