Guida: elaborazione batch

I job batch (tradurre un catalogo prodotti, classificare ticket di assistenza, estrarre campi da migliaia di documenti) inviano molte volte lo stesso tipo di richiesta. Il lavoro è semplice; le parti difficili sono la velocità di elaborazione, i guasti e i costi. Questa guida tratta tutti e tre usando il pacchetto ufficiale openai con base URL https://api14.com/api/v1.

Non esiste un endpoint batch separato: la Batch API di OpenAI non è disponibile su api14 (vedi Compatibilità con l'SDK OpenAI). Esegui i batch come normali richieste con concorrenza controllata.

Conosci il tuo limite di frequenza

I limiti sono conteggiati in richieste al minuto (RPM) per chiave API, con un tetto dato dal tuo piano:

Una chiave con un proprio limite RPM ottiene il più basso dei due. Controlla il tuo con GET /key (rate_limit.requests è null quando la chiave usa il limite del piano), vedi Saldo e informazioni sulla chiave. Superare il limite restituisce 429 rate_limited con Retry-After: 60.

L'RPM limita quante richieste iniziano al minuto, non quante girano contemporaneamente. Se una richiesta media dura 4 secondi e il tuo limite è 100 RPM, bastano circa 7 richieste in corso per raggiungere il limite. Distanzia le richieste in base al momento di avvio e mantieni una concorrenza moderata.

Un esecutore concorrente con ritmo controllato (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           # il limite del tuo piano o della tua chiave
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)  # registrato; riesegui per riprovare

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

Cosa fa:

• Ritmo: il Pacer avvia al massimo RPM richieste al minuto; il semaforo limita quante sono in corso.
• Retry: l'SDK ripete 429 e 5xx con backoff esponenziale, rispettando Retry-After (max_retries=4). I 429 del provider upstream vengono ripetuti allo stesso modo.
• I codici fatali fermano il job: daily_limit_reached e insufficient_balance non si risolvono riprovando, quindi il job esce invece di bruciare tentativi.
• Checkpoint: i risultati vengono aggiunti a un file JSONL man mano che finiscono. Rieseguendo lo script gli id già completati vengono saltati, quindi un crash o un job interrotto non fa perdere nulla.

In Node, la stessa struttura funziona con Promise.all su un numero fisso di worker che prelevano da una coda condivisa, ciascuno in attesa di 60_000 / RPM ms tra un avvio e l'altro.

Metti più elementi in una sola richiesta

Per gli elementi brevi, una richiesta per elemento spreca RPM e ripete ogni volta le istruzioni. Classifica 20-50 elementi per richiesta con un output strutturato e dividi il numero di richieste per lo stesso fattore:

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

Verifica che ogni id di input sia tornato; rinvia quelli mancanti. Se finish_reason è "length", il batch era troppo grande per max_tokens: dividilo. Vedi Output strutturati. Per gli embeddings, passa fino a 100 stringhe in un solo array input, vedi Embeddings.

Riduci i costi con il prompt caching

In un batch, le istruzioni, gli esempi e lo schema sono identici in ogni richiesta. Mettili all'inizio e l'elemento variabile alla fine, in modo che i provider possano riutilizzare il prefisso in cache:

• OpenAI, DeepSeek, Gemini 2.5 e altri usano la cache automaticamente quando il prefisso è abbastanza lungo (1,024 token per OpenAI). Le letture dalla cache costano una frazione del normale input.
• I modelli Anthropic richiedono un punto di interruzione cache_control esplicito sulla parte statica del prompt.

Un prompt di sistema lungo e dettagliato con alcuni esempi svolti può quindi costare poco più di uno breve, e di solito migliora la qualità. Conferma i riscontri in usage.prompt_tokens_details.cached_tokens. Vedi Prompt caching.

Scegli un modello economico, poi verifica

Per traduzione, classificazione ed estrazione, i modelli piccoli spesso bastano a una frazione del prezzo. Elenca i modelli di testo per prezzo di input:

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

Poi misura prima di impegnarti:

1. Prendi 50-100 elementi rappresentativi ed etichetta a mano l'output atteso.
2. Eseguili su due o tre modelli candidati e confronta accuratezza e usage.cost sommato.
3. Moltiplica il costo per elemento per il numero totale di elementi per stimare l'intero job, e verifica con GET /credits che il tuo saldo lo copra.

Le varianti :free usano l'endpoint gratuito di un modello dove esiste, ma hanno limiti di frequenza upstream più severi, quindi si adattano ai test più che ai grandi batch. Vedi Id dei modelli e varianti.

Prima di un'esecuzione di grandi dimensioni

• Crea una chiave dedicata con un limite giornaliero in USD, così un bug non può svuotare il tuo saldo. Il job si ferma allora con 429 daily_limit_reached. Vedi Limiti di frequenza.
• Imposta max_tokens su ogni richiesta.
• Esegui il primo 1% degli elementi, controlla output e costo, poi avvia il resto.
• Le richieste fallite (qualsiasi 4xx o 5xx) non vengono addebitate, quindi rieseguire i fallimenti è sicuro. Vedi Errori.