Классификация тысяч текстов через AsyncOpenAI на Python

Чтобы разметить тысячи текстов, возьмите AsyncOpenAI, оберните каждый вызов в asyncio.Semaphore, чтобы в полёте было не больше N запросов, попросите ответ в виде короткого JSON-ярлыка и повторяйте только те тексты, что не получились. Скрипт ниже делает именно это через https://api.cheaptokens.dev/v1 и в конце печатает сумму токенов из поля usage.

31 октября 2026 года Андрей Карпати (сооснователь OpenAI, бывший директор по ИИ в Tesla) рассказал про HackerTube: он собрал проект примерно за два часа и с помощью Sonnet сузил комментарии Hacker News до около 40 тысяч тех, где советуют видео на YouTube. Промпт, размер пачек и параллельность в посте не приводятся, поэтому дальше — обычный приём для такой работы, а не его рецепт.

Скрипт#

Допустим, у вас 5 000 обращений в поддержку и каждое нужно пометить как bug, question или praise. Вот всё целиком. Поставьте SDK командой pip install openai (нужен Python 3.10+; версию покажет python -c "import openai; print(openai.__version__)") и заранее положите ключ в переменную окружения.

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

На выходе: labels (номер текста → ярлык), failed (номера, по которым так и не пришёл корректный ярлык; если всё прошло, список пустой) и суммы токенов. Id модели — полный, deepseek-ai/DeepSeek-V4-Flash-0731; id других моделей берите из документации CheapTokens: у разных провайдеров короткие имена отличаются.

Расставьте шаги по порядку — так проверите, что схема у вас в голове:

Зачем семафор, а не 5 000 запросов сразу#

asyncio.gather запустит всё, что ему дали. Отдайте ему тысячи корутин без ограничения — они стартуют разом, и такой всплеск запросов быстро приводит к ошибкам 429. Семафор — это турникет: concurrency=5 значит, что внутри async with sem одновременно не больше пяти вызовов, остальные ждут очереди. Задачи Python создаёт все, но с API разговаривают только пять.

Какое число ставить в concurrency? Начните с малого, как 5 выше, и повышайте, пока ошибок нет. Это настройка, которую подбирают под задачу, а не константа для копирования.

Повторы: два слоя и две задачи#

Первый слой встроен в SDK. По README openai-python (сверено 1 ноября 2026 года), ошибки соединения, ответы 408, 409, 429 и 5xx повторяются по умолчанию дважды с короткой экспоненциальной паузой. max_retries=2 в скрипте просто делает это видимым; больше задаётся через client.with_options(max_retries=5), а 0 отключает повторы. Таймаут по умолчанию — 10 минут, при его превышении возникает APITimeoutError; для коротких ярлыков в скрипте стоит timeout=60. Подробности — в статье таймаут и повторы в OpenAI Python SDK, а идея паузы между повторами — в разборе 429 Too Many Requests.

Второй слой ваш, и он ловит то, чего SDK не видит: запрос прошёл, но модель ответила текстом без JSON (I can't classify this) или придумала ярлык complaint. Текст вокруг {...} не мешает: код вырезает JSON из ответа. Тогда json.loads или проверка по LABELS выбрасывает ошибку, gather(..., return_exceptions=True) возвращает её как значение, а не роняет всю пачку, и текст уходит в следующий круг. После rounds попыток всё, что осталось в failed, можно разобрать вручную. Один плохой ответ из 5 000 стоит одного лишнего запроса, а не всей работы.

Скрипт просит JSON в промпте и проверяет его в коде. Если хотите жёстче задать формат через response_format, смотрите валидный JSON от LLM API.

Сколько стоит работа#

Счётчик один — токены: скрипт суммирует usage, а цена следует из них. Возьмём типичный случай — короткие тексты и короткие ярлыки. Допустим, 10 000 текстов по примерно 300 входных токенов (текст плюс инструкция) и около 20 выходных на JSON-ярлык:

10 000 × 300 = 3 000 000 входных токенов; 10 000 × 20 = 200 000 выходных.

На DeepSeek-V4-Flash такая работа обойдётся в $0.065, на GLM-5.3-Flash — в $0.068, на MiniMax-M2.7 — в $0.18. Таблица показывает ту же работу по всем моделям рядом с OpenRouter; со входом и выходом вместе выходит дешевле, чем у OpenRouter, в сумме.

Подвигайте значения в калькуляторе под свои числа. Поля in и out — токены одного запроса, runs — сколько запросов вы отправите.

Здесь перевешивает вход: текст заходит, наружу выходит одно слово. Поэтому для массовой фильтрации подходит модель с дешёвым входом и контекстом 380K, как DeepSeek-V4-Flash с единой ценой в любое время суток; для коротких ярлыков тот же скрипт работает с GLM-5.3-Flash и MiniMax-M2.7, меняется только MODEL. Как растёт такой счёт, показано в статье за что API берёт деньги: вход и выход.

Перед большим прогоном#

Сначала запустите 20 текстов и посмотрите на failed и суммы токенов. Если 20 текстов дали около 6 000 входных токенов, 5 000 текстов дадут около 1 500 000 — умножайте до того, как решились.

Если не работает#

• KeyError: 'CHEAPTOKENS_API_KEY': переменная не задана в этом терминале. Экспортируйте её и запустите Python заново.
• 401 или 404: неверный ключ, base URL или id модели. Каждый случай разобран в статье ошибка API 401 или 404.
• json.JSONDecodeError на многих текстах: в ответе нет {...}, часто потому, что max_tokens=100 обрезал ответ. Выведите r.choices[0].finish_reason: length значит, что лимит сработал, — поднимите max_tokens.
• Много 429 и медленный прогон: уменьшите concurrency.

Общие сведения об адресе API и моделях — в документации CheapTokens и на странице цен.