Classify thousands of texts with AsyncOpenAI in Python

To classify thousands of texts, use AsyncOpenAI, put an asyncio.Semaphore around every call so only N requests are in flight, ask for a one-line JSON label, and re-run only the items that failed. The script below does exactly that against https://api.cheaptokens.dev/v1 and prints the total tokens from the usage field at the end.

On 31 October 2026 Andrej Karpathy (OpenAI co-founder, former Tesla AI director) described HackerTube: he built it in about two hours and used Sonnet to narrow Hacker News comments down to about 40K that recommend a YouTube video. His post doesn't give the prompt, batch size or concurrency, so what follows is the standard pattern for this kind of job, not his recipe.

The script#

Say you have 5,000 support tickets and want each one tagged bug, question or praise. Here is the whole thing. Install the SDK with pip install openai (Python 3.10+; python -c "import openai; print(openai.__version__)" shows what you have) and put your key in an environment variable first.

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

You get back labels (text index → label), failed (indexes that never produced a valid label, an empty list when everything went through) and the token totals. The model id is the full deepseek-ai/DeepSeek-V4-Flash-0731; take ids for other models from the CheapTokens docs, because the short names differ between providers.

Put the pipeline in order to check you've got the shape in your head:

Why a semaphore and not 5,000 requests at once#

asyncio.gather happily starts everything you give it. Hand it thousands of coroutines without a limit and they all start at once, a burst of requests that is a quick way to 429 errors. The semaphore is a turnstile: concurrency=5 means at most five calls are inside async with sem at any moment, and the rest wait their turn. Python still creates all the tasks, but only five talk to the API.

What number should concurrency be? Start low, like the 5 above, and raise it while the errors stay away. Treat it as a setting you tune per job, not a constant to copy.

Retries: two layers, two jobs#

The first layer is built into the SDK. According to the openai-python README (checked 1 November 2026), connection errors, 408, 409, 429 and 5xx responses are retried twice by default with a short exponential backoff. max_retries=2 in the script just makes that visible; set it higher with client.with_options(max_retries=5) or to 0 to turn it off. The default timeout is 10 minutes and a timeout raises APITimeoutError, which is why the script sets timeout=60 for short labels. The details are in OpenAI Python SDK timeout and retries, and the backoff idea behind it in 429 Too Many Requests from an AI API.

The second layer is yours, and it covers what the SDK can't see: the call succeeded, but the model answered with prose and no JSON at all (I can't classify this) or invented a label called complaint. Text around a {...} doesn't matter, because the code cuts the JSON out of the answer. Then json.loads or the LABELS check raises, gather(..., return_exceptions=True) keeps the error as a value instead of killing the batch, and the item goes to the next round. After rounds attempts, whatever is still in failed is yours to inspect by hand. One bad answer out of 5,000 costs you one extra request, not the whole job.

The script asks for JSON in the prompt and validates it in code. If you want to push the format harder with response_format, see Get valid JSON from an LLM API.

What the job costs#

Tokens are the only meter, so the script sums usage and the cost follows. Take the common case: short texts, short labels. Assume 10,000 texts at about 300 input tokens each (the text plus the instruction) and about 20 output tokens for the JSON label:

10,000 × 300 = 3,000,000 input tokens; 10,000 × 20 = 200,000 output tokens.

On DeepSeek-V4-Flash that job comes to $0.065, on GLM-5.3-Flash to $0.068, on MiniMax-M2.7 to $0.18. The table shows the same job for every model, next to OpenRouter; with input and output counted together it comes out cheaper than OpenRouter in total.

Move the sliders in the calculator to your own numbers. The in and out fields are for one request, and runs is how many requests you send.

The input side dominates here: the text goes in, one word comes out. That's why a model with cheap input and a 380K context window, like DeepSeek-V4-Flash with one price at any hour, suits bulk filtering; for short labels, GLM-5.3-Flash and MiniMax-M2.7 are alternatives for the same script, you only change MODEL. Not sure how a bill like this grows? What an AI API bills you for has the picture.

Before the big run#

Run 20 texts first and look at failed and the token totals. If 20 texts use about 6,000 input tokens, 5,000 will use about 1,500,000, and you can multiply before you commit.

If it doesn't work#

• KeyError: 'CHEAPTOKENS_API_KEY': the variable isn't set in this terminal. Export it and start Python again.
• 401 or 404: wrong key, base URL or model id. API 401 or 404 error walks through each.
• json.JSONDecodeError on many items: the answer has no {...}, often because max_tokens=100 cut it off. Print r.choices[0].finish_reason; length means the limit hit, so raise max_tokens.
• Many 429 and a slow run: lower concurrency.

The general model and endpoint details live in the CheapTokens docs and on the pricing page.