Clasifica miles de textos con AsyncOpenAI en Python

Para clasificar miles de textos, usa AsyncOpenAI, pon un asyncio.Semaphore alrededor de cada llamada para que solo haya N solicitudes en vuelo, pide una etiqueta JSON de una línea y vuelve a ejecutar solo los elementos que fallaron. El script de abajo hace exactamente eso contra https://api.cheaptokens.dev/v1 e imprime el total de tokens del campo usage al final.

El 31 de octubre de 2026 Andrej Karpathy (cofundador de OpenAI, exdirector de IA de Tesla) describió HackerTube: lo construyó en unas dos horas y usó Sonnet para reducir los comentarios de Hacker News a unos 40K que recomiendan un video de YouTube. Su publicación no da el prompt, el tamaño del lote ni la concurrencia, así que lo que sigue es el patrón estándar para este tipo de trabajo, no su receta.

El script#

Supón que tienes 5,000 tickets de soporte y quieres etiquetar cada uno como bug, question o praise. Aquí está todo. Instala el SDK con pip install openai (Python 3.10+; python -c "import openai; print(openai.__version__)" muestra lo que tienes) y primero pon tu clave en una variable de entorno.

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

Obtienes labels (índice de texto → etiqueta), failed (índices que nunca produjeron una etiqueta válida, una lista vacía cuando todo salió bien) y los totales de tokens. El id del modelo es el deepseek-ai/DeepSeek-V4-Flash-0731 completo; toma los ids de otros modelos de la documentación de CheapTokens, porque los nombres cortos difieren entre proveedores.

Ordena el pipeline para comprobar que tienes la forma en la cabeza:

Por qué un semáforo y no 5,000 solicitudes a la vez#

asyncio.gather inicia felizmente todo lo que le des. Pásale miles de corrutinas sin límite y todas empiezan a la vez, una ráfaga de solicitudes que es una forma rápida de obtener errores 429. El semáforo es un torniquete: concurrency=5 significa que como máximo cinco llamadas están dentro de async with sem en cualquier momento, y el resto espera su turno. Python sigue creando todas las tareas, pero solo cinco hablan con la API.

¿Qué número debería ser concurrency? Empieza bajo, como el 5 de arriba, y súbelo mientras los errores no aparezcan. Trátalo como un ajuste que calibras por trabajo, no como una constante que copiar.

Reintentos: dos capas, dos funciones#

La primera capa está integrada en el SDK. Según el README de openai-python (consultado el 1 de noviembre de 2026), los errores de conexión, las respuestas 408, 409, 429 y 5xx se reintentan dos veces por defecto con un breve backoff exponencial. max_retries=2 en el script solo hace eso visible; súbelo con client.with_options(max_retries=5) o ponlo en 0 para desactivarlo. El tiempo de espera predeterminado es de 10 minutos y un tiempo de espera agotado lanza APITimeoutError, por eso el script establece timeout=60 para etiquetas cortas. Los detalles están en Tiempo de espera y reintentos del SDK de OpenAI para Python, y la idea de backoff detrás de esto en 429 Too Many Requests de una API de IA.

La segunda capa es tuya, y cubre lo que el SDK no puede ver: la llamada tuvo éxito, pero el modelo respondió con prosa y sin JSON alguno (I can't classify this) o inventó una etiqueta llamada complaint. El texto alrededor de un {...} no importa, porque el código recorta el JSON de la respuesta. Entonces json.loads o la comprobación LABELS lanza una excepción, gather(..., return_exceptions=True) guarda el error como un valor en lugar de matar el lote, y el elemento pasa a la siguiente ronda. Después de rounds intentos, lo que quede en failed es tuyo para inspeccionarlo a mano. Una respuesta mala de 5,000 te cuesta una solicitud extra, no todo el trabajo.

El script pide JSON en el prompt y lo valida en el código. Si quieres forzar más el formato con response_format, consulta Obtener JSON válido de una API de LLM.

Cuánto cuesta el trabajo#

Los tokens son el único medidor, así que el script suma usage y el costo se deriva. Toma el caso común: textos cortos, etiquetas cortas. Supón 10,000 textos de unos 300 tokens de entrada cada uno (el texto más la instrucción) y unos 20 tokens de salida para la etiqueta JSON:

10,000 × 300 = 3,000,000 tokens de entrada; 10,000 × 20 = 200,000 tokens de salida.

En DeepSeek-V4-Flash ese trabajo sale por $0.065, en GLM-5.3-Flash por $0.068, en MiniMax-M2.7 por $0.18. La tabla muestra el mismo trabajo para cada modelo, junto a OpenRouter; contando entrada y salida juntas, sale más barato que OpenRouter en total.

Mueve los controles de la calculadora a tus propios números. Los campos in y out son para una solicitud, y runs es cuántas solicitudes envías.

El lado de la entrada domina aquí: entra el texto, sale una palabra. Por eso un modelo con entrada barata y una ventana de contexto de 380K, como DeepSeek-V4-Flash con un precio a cualquier hora, sirve para filtrado masivo; para etiquetas cortas, GLM-5.3-Flash y MiniMax-M2.7 son alternativas para el mismo script, solo cambias MODEL. ¿No estás seguro de cómo crece una factura así? Qué te cobra una API de IA tiene la imagen.

Antes de la gran ejecución#

Ejecuta primero 20 textos y mira failed y los totales de tokens. Si 20 textos usan unos 6,000 tokens de entrada, 5,000 usarán unos 1,500,000, y puedes multiplicar antes de comprometerte.

Si no funciona#

• KeyError: 'CHEAPTOKENS_API_KEY': la variable no está configurada en esta terminal. Expórtala e inicia Python de nuevo.
• 401 o 404: clave, URL base o id de modelo incorrectos. Error 401 o 404 de la API recorre cada uno.
• json.JSONDecodeError en muchos elementos: la respuesta no tiene {...}, a menudo porque max_tokens=100 la cortó. Imprime r.choices[0].finish_reason; length significa que se alcanzó el límite, así que sube max_tokens.
• Muchos 429 y una ejecución lenta: baja concurrency.

Los detalles generales del modelo y del endpoint están en la documentación de CheapTokens y en la página de precios.