Guide : traitement par lots

Les traitements par lots (traduire un catalogue de produits, classer des tickets de support, extraire des champs de milliers de documents) envoient de nombreuses fois le même type de requête. Le travail est simple ; les difficultés sont le débit, les échecs et le coût. Ce guide couvre les trois avec le paquet officiel openai et l'URL de base https://api14.com/api/v1.

Il n'y a pas d'endpoint de lot séparé : l'API Batch d'OpenAI n'est pas disponible sur api14 (voir Compatibilité avec le SDK OpenAI). Vous exécutez les lots comme des requêtes normales avec une concurrence maîtrisée.

Connaître sa limite de débit

Les limites sont comptées en requêtes par minute (RPM) par clé d'API, plafonnées par votre forfait :

Une clé avec sa propre limite de RPM reçoit la plus basse des deux. Vérifiez la vôtre avec GET /key (rate_limit.requests vaut null quand la clé utilise la limite du forfait), voir Solde et infos de clé. Dépasser la limite renvoie 429 rate_limited avec Retry-After: 60.

Le RPM plafonne le nombre de requêtes qui démarrent par minute, pas le nombre de requêtes simultanées. Si une requête moyenne dure 4 secondes et que votre limite est de 100 RPM, environ 7 requêtes en cours atteignent déjà la limite. Cadencez les requêtes selon leur heure de démarrage et gardez une concurrence modérée.

Un exécuteur cadencé et concurrent (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           # la limite de votre forfait ou de votre clé
CONCURRENCY = 10
FATAL = {"daily_limit_reached", "insufficient_balance", "invalid_api_key", "model_not_allowed"}

class Pacer:
    """Espace régulièrement les démarrages de requêtes pour ne jamais dépasser le 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)  # journalisé ; relancez pour réessayer

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

Ce que fait ce code :

• Cadencement : le Pacer démarre au plus RPM requêtes par minute ; le sémaphore borne le nombre de requêtes en cours.
• Nouvelles tentatives : le SDK réessaie les 429 et 5xx avec un backoff exponentiel, en respectant Retry-After (max_retries=4). Les 429 du fournisseur en amont sont réessayés de la même façon.
• Les codes fatals arrêtent le traitement : daily_limit_reached et insufficient_balance ne disparaîtront pas en réessayant : le traitement s'arrête donc au lieu de gaspiller des tentatives.
• Points de reprise : les résultats sont ajoutés à un fichier JSONL au fur et à mesure. Relancer le script ignore les identifiants déjà traités ; un plantage ou un arrêt ne fait donc rien perdre.

En Node, la même structure fonctionne avec Promise.all sur un nombre fixe de workers qui puisent dans une file partagée, chacun attendant 60_000 / RPM ms entre deux démarrages.

Mettre plusieurs éléments dans une requête

Pour des éléments courts, une requête par élément gaspille du RPM et répète les instructions à chaque fois. Classez 20 à 50 éléments par requête avec une sortie structurée, et vous divisez le nombre de requêtes par le même facteur :

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

Vérifiez que chaque identifiant d'entrée est revenu ; renvoyez ceux qui manquent. Si finish_reason vaut "length", le lot était trop grand pour max_tokens : découpez-le. Voir Sorties structurées. Pour les embeddings, passez jusqu'à 100 chaînes dans un seul tableau input, voir Embeddings.

Réduire le coût avec le cache de prompt

Dans un lot, les instructions, les exemples et le schéma sont identiques à chaque requête. Placez-les en premier et l'élément variable en dernier, pour que les fournisseurs puissent réutiliser le préfixe en cache :

• OpenAI, DeepSeek, Gemini 2.5 et d'autres mettent en cache automatiquement dès que le préfixe est assez long (1 024 tokens pour OpenAI). Les lectures du cache coûtent une fraction de l'entrée normale.
• Les modèles Anthropic nécessitent un point d'arrêt cache_control explicite sur la partie statique du prompt.

Un long prompt système détaillé avec quelques exemples travaillés peut donc coûter à peine plus qu'un prompt court, et améliore généralement la qualité. Confirmez les succès dans usage.prompt_tokens_details.cached_tokens. Voir Cache de prompt.

Choisir un modèle économique, puis vérifier

Pour la traduction, la classification et l'extraction, les petits modèles sont souvent suffisants pour une fraction du prix. Listez les modèles de texte par prix d'entrée :

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

Puis mesurez avant de vous engager :

1. Prenez 50 à 100 éléments représentatifs et étiquetez à la main le résultat attendu.
2. Passez-les dans deux ou trois modèles candidats et comparez la précision et la somme des usage.cost.
3. Multipliez le coût par élément par votre nombre total d'éléments pour estimer le traitement complet, et vérifiez que votre solde le couvre avec GET /credits.

Les variantes :free utilisent l'endpoint gratuit d'un modèle quand il existe, mais elles ont des limites de débit plus strictes en amont : elles conviennent donc mieux aux tests qu'aux gros lots. Voir Identifiants et variantes de modèles.

Avant un gros traitement

• Créez une clé dédiée avec une limite quotidienne en USD pour qu'un bogue ne puisse pas vider votre solde. Le traitement s'arrête alors avec 429 daily_limit_reached. Voir Limites de débit.
• Définissez max_tokens sur chaque requête.
• Lancez le premier 1 % des éléments, examinez le résultat et le coût, puis démarrez le reste.
• Les requêtes en échec (tout 4xx ou 5xx) ne sont pas facturées : relancer les échecs est donc sans risque. Voir Erreurs.