Classer des milliers de textes avec AsyncOpenAI en Python

Pour classer des milliers de textes, utilisez AsyncOpenAI, placez un asyncio.Semaphore autour de chaque appel pour que seules N requêtes soient en vol, demandez une étiquette JSON sur une ligne et relancez uniquement les éléments qui ont échoué. Le script ci-dessous fait exactement cela contre https://api.cheaptokens.dev/v1 et affiche le total de tokens à partir du champ usage à la fin.

Le 31 octobre 2026, Andrej Karpathy (cofondateur d'OpenAI, ancien directeur IA de Tesla) a décrit HackerTube : il l'a construit en environ deux heures et a utilisé Sonnet pour réduire les commentaires de Hacker News à environ 40K qui recommandent une vidéo YouTube. Son post ne donne ni le prompt, ni la taille de lot, ni la concurrence, donc ce qui suit est le schéma standard pour ce type de tâche, pas sa recette.

Le script#

Supposez que vous avez 5 000 tickets de support et que vous voulez étiqueter chacun comme bug, question ou praise. Voici l'ensemble. Installez le SDK avec pip install openai (Python 3.10+ ; python -c "import openai; print(openai.__version__)" montre ce que vous avez) et placez d'abord votre clé dans une variable d'environnement.

import asyncio
import json
import os

from openai import AsyncOpenAI

client = AsyncOpenAI(
    base_url='https://api.cheaptokens.dev/v1',
    api_key=os.environ['CHEAPTOKENS_API_KEY'],
    max_retries=2,
    timeout=60,
)
MODEL = 'deepseek-ai/DeepSeek-V4-Flash-0731'
LABELS = {'bug', 'question', 'praise'}
SYSTEM = (
    'Classify the text as bug, question or praise. '
    'Reply with JSON only: {"label": "<bug|question|praise>"}'
)


async def classify(text, sem, usage):
    async with sem:
        r = await client.chat.completions.create(
            model=MODEL,
            temperature=0,
            max_tokens=100,
            messages=[
                {'role': 'system', 'content': SYSTEM},
                {'role': 'user', 'content': text},
            ],
        )
    usage['in'] += r.usage.prompt_tokens
    usage['out'] += r.usage.completion_tokens
    raw = r.choices[0].message.content
    label = json.loads(raw[raw.find('{'):raw.rfind('}') + 1])['label']
    if label not in LABELS:
        raise ValueError(f'unexpected label: {label!r}')
    return label


async def run(texts, concurrency=5, rounds=3):
    sem = asyncio.Semaphore(concurrency)
    usage = {'in': 0, 'out': 0}
    labels, pending = {}, list(range(len(texts)))
    for _ in range(rounds):
        if not pending:
            break
        results = await asyncio.gather(
            *(classify(texts[i], sem, usage) for i in pending),
            return_exceptions=True,
        )
        failed = []
        for i, res in zip(pending, results):
            if isinstance(res, Exception):
                failed.append(i)
            else:
                labels[i] = res
        pending = failed
    return labels, pending, usage


texts = [
    'The export button crashes the app on Android 15.',
    'Does the Pro plan include API access?',
    'Love the new dark theme, thank you!',
]
labels, failed, usage = asyncio.run(run(texts))
print(labels)
print('failed:', failed)
print('tokens in:', usage['in'], 'out:', usage['out'])

Vous récupérez labels (index du texte → étiquette), failed (les index qui n'ont jamais produit d'étiquette valide, une liste vide quand tout est passé) et les totaux de tokens. L'id du modèle est le deepseek-ai/DeepSeek-V4-Flash-0731 complet ; prenez les ids des autres modèles dans la documentation CheapTokens, car les noms courts diffèrent selon les fournisseurs.

Remettez le pipeline dans l'ordre pour vérifier que vous avez bien la structure en tête :

Pourquoi un sémaphore et pas 5 000 requêtes d'un coup#

asyncio.gather démarre volontiers tout ce que vous lui donnez. Donnez-lui des milliers de coroutines sans limite et elles démarrent toutes en même temps, une rafale de requêtes qui est un moyen rapide d'obtenir des erreurs 429. Le sémaphore est un tourniquet : concurrency=5 signifie qu'au plus cinq appels sont dans async with sem à un instant donné, et les autres attendent leur tour. Python crée quand même toutes les tâches, mais seules cinq parlent à l'API.

Quelle valeur donner à concurrency ? Commencez bas, comme le 5 ci-dessus, et augmentez-la tant que les erreurs restent absentes. Traitez-la comme un réglage que vous ajustez par tâche, pas comme une constante à copier.

Les nouvelles tentatives : deux couches, deux rôles#

La première couche est intégrée au SDK. D'après le README openai-python (vérifié le 1er novembre 2026), les erreurs de connexion, les réponses 408, 409, 429 et 5xx sont retentées deux fois par défaut avec un court backoff exponentiel. max_retries=2 dans le script rend simplement cela visible ; augmentez-la avec client.with_options(max_retries=5) ou mettez-la à 0 pour la désactiver. Le timeout par défaut est de 10 minutes et un timeout lève APITimeoutError, c'est pourquoi le script définit timeout=60 pour des étiquettes courtes. Les détails sont dans Timeout et nouvelles tentatives du SDK Python OpenAI, et l'idée de backoff derrière dans 429 Too Many Requests d'une API IA.

La seconde couche est la vôtre, et elle couvre ce que le SDK ne peut pas voir : l'appel a réussi, mais le modèle a répondu en prose sans aucun JSON (I can't classify this) ou a inventé une étiquette appelée complaint. Le texte autour d'un {...} n'a pas d'importance, car le code extrait le JSON de la réponse. Ensuite json.loads ou la vérification LABELS lève une erreur, gather(..., return_exceptions=True) conserve l'erreur comme valeur au lieu de tuer le lot, et l'élément passe au tour suivant. Après rounds tentatives, tout ce qui reste dans failed est à inspecter à la main. Une mauvaise réponse sur 5 000 vous coûte une requête supplémentaire, pas toute la tâche.

Le script demande du JSON dans le prompt et le valide dans le code. Si vous voulez forcer davantage le format avec response_format, voir Obtenir du JSON valide d'une API LLM.

Ce que coûte la tâche#

Les tokens sont le seul compteur, donc le script additionne usage et le coût suit. Prenons le cas courant : textes courts, étiquettes courtes. Supposez 10 000 textes à environ 300 tokens d'entrée chacun (le texte plus l'instruction) et environ 20 tokens de sortie pour l'étiquette JSON :

10 000 × 300 = 3 000 000 tokens d'entrée ; 10 000 × 20 = 200 000 tokens de sortie.

Sur DeepSeek-V4-Flash cette tâche revient à $0.065, sur GLM-5.3-Flash à $0.068, sur MiniMax-M2.7 à $0.18. Le tableau montre la même tâche pour chaque modèle, à côté d'OpenRouter ; en comptant entrée et sortie ensemble, le total est moins cher qu'OpenRouter.

Déplacez les curseurs du calculateur vers vos propres chiffres. Les champs in et out concernent une requête, et runs est le nombre de requêtes que vous envoyez.

Le côté entrée domine ici : le texte entre, un mot sort. C'est pourquoi un modèle avec une entrée bon marché et une fenêtre de contexte de 380K, comme DeepSeek-V4-Flash avec un prix unique à toute heure, convient au filtrage en masse ; pour des étiquettes courtes, GLM-5.3-Flash et MiniMax-M2.7 sont des alternatives pour le même script, vous ne changez que MODEL. Vous ne savez pas comment une facture comme celle-ci grandit ? Ce qu'une API IA vous facture donne le tableau.

Avant la grosse exécution#

Lancez d'abord 20 textes et regardez failed et les totaux de tokens. Si 20 textes utilisent environ 6 000 tokens d'entrée, 5 000 en utiliseront environ 1 500 000, et vous pouvez multiplier avant de vous engager.

Si ça ne marche pas#

• KeyError: 'CHEAPTOKENS_API_KEY' : la variable n'est pas définie dans ce terminal. Exportez-la et relancez Python.
• 401 ou 404 : mauvaise clé, URL de base ou id de modèle. Erreur API 401 ou 404 détaille chacune.
• json.JSONDecodeError sur de nombreux éléments : la réponse n'a pas de {...}, souvent parce que max_tokens=100 l'a coupée. Affichez r.choices[0].finish_reason ; length signifie que la limite a été atteinte, donc augmentez max_tokens.
• Beaucoup de 429 et une exécution lente : baissez concurrency.

Les détails généraux sur les modèles et les endpoints se trouvent dans la documentation CheapTokens et sur la page de tarifs.