Classificare migliaia di testi con AsyncOpenAI in Python

Per classificare migliaia di testi, usa AsyncOpenAI, metti un asyncio.Semaphore attorno a ogni chiamata così solo N richieste sono in volo, chiedi un'etichetta JSON su una riga e riesegui solo gli elementi falliti. Lo script qui sotto fa esattamente questo contro https://api.cheaptokens.dev/v1 e stampa il totale dei token dal campo usage alla fine.

Il 31 ottobre 2026 Andrej Karpathy (co-fondatore di OpenAI, ex direttore AI di Tesla) ha descritto HackerTube: l'ha costruito in circa due ore e ha usato Sonnet per ridurre i commenti di Hacker News a circa 40K che consigliano un video di YouTube. Il suo post non riporta il prompt, la dimensione del batch o la concorrenza, quindi quello che segue è il pattern standard per questo tipo di lavoro, non la sua ricetta.

Lo script#

Supponi di avere 5.000 ticket di supporto e di voler etichettare ciascuno come bug, question o praise. Ecco tutto. Installa l'SDK con pip install openai (Python 3.10+; python -c "import openai; print(openai.__version__)" mostra cosa hai) e metti prima la tua chiave in una variabile d'ambiente.

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'])

Ottieni labels (indice del testo → etichetta), failed (indici che non hanno mai prodotto un'etichetta valida, una lista vuota quando è andato tutto a buon fine) e i totali dei token. L'id del modello è il deepseek-ai/DeepSeek-V4-Flash-0731 completo; prendi gli id per altri modelli dalla documentazione di CheapTokens, perché i nomi brevi differiscono tra i provider.

Metti in ordine la pipeline per verificare di aver chiaro lo schema:

Perché un semaforo e non 5.000 richieste in una volta#

asyncio.gather avvia volentieri tutto ciò che gli dai. Passagli migliaia di coroutine senza un limite e partono tutte insieme, un'ondata di richieste che è un modo rapido per ottenere errori 429. Il semaforo è un tornello: concurrency=5 significa che al massimo cinque chiamate sono dentro async with sem in qualsiasi momento, e le altre aspettano il loro turno. Python crea comunque tutti i task, ma solo cinque parlano con l'API.

Che numero dovrebbe essere concurrency? Inizia basso, come il 5 qui sopra, e alzalo finché gli errori restano lontani. Trattalo come un'impostazione da tarare per ogni lavoro, non una costante da copiare.

Retry: due livelli, due compiti#

Il primo livello è integrato nell'SDK. Secondo il README di openai-python (verificato il 1 novembre 2026), gli errori di connessione, le risposte 408, 409, 429 e 5xx vengono ritentati due volte per impostazione predefinita con un breve backoff esponenziale. max_retries=2 nello script serve solo a renderlo visibile; impostalo più alto con client.with_options(max_retries=5) o a 0 per disattivarlo. Il timeout predefinito è di 10 minuti e un timeout genera APITimeoutError, ed è per questo che lo script imposta timeout=60 per etichette brevi. I dettagli sono in Timeout e retry dell'SDK Python di OpenAI, e l'idea di backoff alla base in 429 Too Many Requests da un'API AI.

Il secondo livello è tuo, e copre ciò che l'SDK non può vedere: la chiamata è riuscita, ma il modello ha risposto con prosa e nessun JSON (I can't classify this) o ha inventato un'etichetta chiamata complaint. Il testo attorno a un {...} non conta, perché il codice ritaglia il JSON dalla risposta. Poi json.loads o il controllo LABELS genera un'eccezione, gather(..., return_exceptions=True) conserva l'errore come valore invece di uccidere il batch, e l'elemento passa al round successivo. Dopo rounds tentativi, ciò che resta in failed è da ispezionare a mano. Una risposta sbagliata su 5.000 ti costa una richiesta in più, non l'intero lavoro.

Lo script chiede JSON nel prompt e lo valida nel codice. Se vuoi spingere di più sul formato con response_format, vedi Ottenere JSON valido da un'API LLM.

Quanto costa il lavoro#

I token sono l'unico contatore, quindi lo script somma usage e il costo ne consegue. Prendi il caso comune: testi brevi, etichette brevi. Supponi 10.000 testi a circa 300 token di input ciascuno (il testo più l'istruzione) e circa 20 token di output per l'etichetta JSON:

10.000 × 300 = 3.000.000 token di input; 10.000 × 20 = 200.000 token di output.

Su DeepSeek-V4-Flash questo lavoro arriva a $0.065, su GLM-5.3-Flash a $0.068, su MiniMax-M2.7 a $0.18. La tabella mostra lo stesso lavoro per ogni modello, accanto a OpenRouter; contando input e output insieme, il totale risulta più economico di OpenRouter.

Sposta i cursori nel calcolatore sui tuoi numeri. I campi in e out sono per una singola richiesta, e runs è quante richieste invii.

Il lato input domina qui: entra il testo, esce una parola. Ecco perché un modello con input economico e una finestra di contesto da 380K, come DeepSeek-V4-Flash con un unico prezzo a qualsiasi ora, si adatta al filtraggio in blocco; per etichette brevi, GLM-5.3-Flash e MiniMax-M2.7 sono alternative per lo stesso script, cambi solo MODEL. Non sei sicuro di come cresce una bolletta così? Per cosa ti fattura un'API AI ha il quadro.

Prima della grande esecuzione#

Esegui prima 20 testi e guarda failed e i totali dei token. Se 20 testi usano circa 6.000 token di input, 5.000 ne useranno circa 1.500.000, e puoi moltiplicare prima di impegnarti.

Se non funziona#

• KeyError: 'CHEAPTOKENS_API_KEY': la variabile non è impostata in questo terminale. Esportala e riavvia Python.
• 401 o 404: chiave, base URL o id del modello sbagliati. Errore API 401 o 404 li esamina uno per uno.
• json.JSONDecodeError su molti elementi: la risposta non ha {...}, spesso perché max_tokens=100 l'ha troncata. Stampa r.choices[0].finish_reason; length significa che il limite è stato raggiunto, quindi alza max_tokens.
• Molti 429 e un'esecuzione lenta: abbassa concurrency.

I dettagli generali su modello ed endpoint sono nella documentazione di CheapTokens e nella pagina dei prezzi.