Classifique milhares de textos com AsyncOpenAI em Python

Para classificar milhares de textos, use AsyncOpenAI, coloque um asyncio.Semaphore em torno de cada chamada para que apenas N requisições estejam em andamento, peça um rótulo JSON de uma linha e execute novamente apenas os itens que falharam. O script abaixo faz exatamente isso contra https://api.cheaptokens.dev/v1 e imprime o total de tokens do campo usage no final.

Em 31 de outubro de 2026, Andrej Karpathy (cofundador da OpenAI, ex-diretor de IA da Tesla) descreveu o HackerTube: ele o construiu em cerca de duas horas e usou o Sonnet para reduzir os comentários do Hacker News a cerca de 40 mil que recomendam um vídeo do YouTube. O post dele não fornece o prompt, o tamanho do lote ou a concorrência, então o que segue é o padrão comum para esse tipo de tarefa, não a receita dele.

O script#

Digamos que você tenha 5.000 tickets de suporte e queira marcar cada um como bug, question ou praise. Aqui está o código completo. Instale o SDK com pip install openai (Python 3.10+; python -c "import openai; print(openai.__version__)" mostra o que você tem) e coloque sua chave em uma variável de ambiente primeiro.

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

Você recebe de volta labels (índice do texto → rótulo), failed (índices que nunca produziram um rótulo válido, uma lista vazia quando tudo passou) e os totais de tokens. O id do modelo é o deepseek-ai/DeepSeek-V4-Flash-0731 completo; pegue ids de outros modelos na documentação do CheapTokens, porque os nomes curtos diferem entre provedores.

Coloque o pipeline em ordem para verificar se você entendeu a estrutura:

Por que um semáforo e não 5.000 requisições de uma vez#

asyncio.gather inicia alegremente tudo o que você entregar. Passe milhares de corrotinas sem limite e todas começam de uma vez, uma rajada de requisições que é um caminho rápido para erros 429. O semáforo é uma catraca: concurrency=5 significa que no máximo cinco chamadas estão dentro de async with sem a qualquer momento, e o resto espera a sua vez. O Python ainda cria todas as tarefas, mas apenas cinco falam com a API.

Qual número concurrency deve ser? Comece baixo, como o 5 acima, e aumente enquanto os erros ficarem longe. Trate-o como uma configuração que você ajusta por tarefa, não uma constante para copiar.

Retentativas: duas camadas, duas funções#

A primeira camada é embutida no SDK. De acordo com o README do openai-python (verificado em 1 de novembro de 2026), erros de conexão, 408, 409, 429 e respostas 5xx são repetidos duas vezes por padrão com um curto backoff exponencial. max_retries=2 no script apenas torna isso visível; defina um valor maior com client.with_options(max_retries=5) ou 0 para desativar. O timeout padrão é de 10 minutos e um timeout gera APITimeoutError, por isso o script define timeout=60 para rótulos curtos. Os detalhes estão em Timeout e retentativas do OpenAI Python SDK, e a ideia de backoff por trás disso em 429 Too Many Requests de uma API de IA.

A segunda camada é sua, e cobre o que o SDK não vê: a chamada foi bem-sucedida, mas o modelo respondeu com texto corrido e nenhum JSON (I can't classify this) ou inventou um rótulo chamado complaint. Texto em volta de um {...} não importa, porque o código extrai o JSON da resposta. Então json.loads ou a verificação LABELS gera erro, gather(..., return_exceptions=True) mantém o erro como valor em vez de matar o lote, e o item vai para a próxima rodada. Após rounds tentativas, o que ainda estiver em failed é seu para inspecionar manualmente. Uma resposta ruim entre 5.000 custa uma requisição extra, não a tarefa inteira.

O script pede JSON no prompt e o valida no código. Se você quiser forçar mais o formato com response_format, veja Obtenha JSON válido de uma API de LLM.

Quanto custa a tarefa#

Tokens são o único medidor, então o script soma usage e o custo vem em seguida. Considere o caso comum: textos curtos, rótulos curtos. Suponha 10.000 textos com cerca de 300 tokens de entrada cada (o texto mais a instrução) e cerca de 20 tokens de saída para o rótulo JSON:

10.000 × 300 = 3.000.000 tokens de entrada; 10.000 × 20 = 200.000 tokens de saída.

No DeepSeek-V4-Flash essa tarefa fica em $0.065, no GLM-5.3-Flash em $0.068, no MiniMax-M2.7 em $0.18. A tabela mostra a mesma tarefa para cada modelo, ao lado do OpenRouter; com entrada e saída contadas juntas, o total sai mais barato que no OpenRouter.

Mova os controles da calculadora para os seus próprios números. Os campos in e out são para uma requisição, e runs é quantas requisições você envia.

O lado da entrada domina aqui: o texto entra, uma palavra sai. É por isso que um modelo com entrada barata e janela de contexto de 380K, como o DeepSeek-V4-Flash com um preço a qualquer hora, serve para filtragem em massa; para rótulos curtos, GLM-5.3-Flash e MiniMax-M2.7 são alternativas para o mesmo script, você só muda MODEL. Não tem certeza de como uma conta dessas cresce? O que uma API de IA cobra de você tem o panorama.

Antes da execução grande#

Execute 20 textos primeiro e observe failed e os totais de tokens. Se 20 textos usam cerca de 6.000 tokens de entrada, 5.000 usarão cerca de 1.500.000, e você pode multiplicar antes de se comprometer.

Se não funcionar#

• KeyError: 'CHEAPTOKENS_API_KEY': a variável não está definida neste terminal. Exporte-a e inicie o Python novamente.
• 401 ou 404: chave, URL base ou id de modelo errado. Erro 401 ou 404 da API explica cada um.
• json.JSONDecodeError em muitos itens: a resposta não tem {...}, muitas vezes porque max_tokens=100 a cortou. Imprima r.choices[0].finish_reason; length significa que o limite foi atingido, então aumente max_tokens.
• Muitos 429 e uma execução lenta: diminua concurrency.

Os detalhes gerais do modelo e do endpoint estão na documentação do CheapTokens e na página de preços.