Класифікуйте тисячі текстів за допомогою AsyncOpenAI у Python

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

31 жовтня 2026 року Андрей Карпатий (співзасновник OpenAI, колишній директор зі штучного інтелекту Tesla) описав HackerTube: він створив його приблизно за дві години та використав Sonnet, щоб звузити коментарі Hacker News до приблизно 40K, які рекомендують відео на 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 (індекси, які так і не створили дійсної мітки, порожній список, коли все пройшло) та підсумки токенів. Ідентифікатор моделі — це повний deepseek-ai/DeepSeek-V4-Flash-0731; беріть ідентифікатори для інших моделей з документації 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 від AI API.

Другий шар — ваш, і він охоплює те, чого 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. Не впевнені, як зростає такий рахунок? За що вам виставляє рахунок AI API дає картину.

Перед великим запуском#

Спочатку запустіть 20 текстів і подивіться на failed та підсумки токенів. Якщо 20 текстів використовують близько 6 000 вхідних токенів, то 5 000 використають близько 1 500 000, і ви можете помножити перед тим, як зобов’язатися.

Якщо це не працює#

• KeyError: 'CHEAPTOKENS_API_KEY': змінна не встановлена в цьому терміналі. Експортуйте її та запустіть Python знову.
• 401 або 404: неправильний ключ, базова URL-адреса або ідентифікатор моделі. Помилка API 401 або 404 розглядає кожну.
• json.JSONDecodeError на багатьох елементах: у відповіді немає {...}, часто тому, що max_tokens=100 обірвав її. Виведіть r.choices[0].finish_reason; length означає, що ліміт досягнуто, тож підвищте max_tokens.
• Багато 429 і повільний запуск: знизьте concurrency.

Загальні відомості про модель і кінцеву точку містяться в документації CheapTokens і на сторінці цін.