Klasyfikuj tysiące tekstów za pomocą AsyncOpenAI w Pythonie

Aby sklasyfikować tysiące tekstów, użyj AsyncOpenAI, otocz każde wywołanie asyncio.Semaphore, aby tylko N żądań było w locie, poproś o etykietę JSON w jednej linii i ponownie uruchom tylko te elementy, które zawiodły. Poniższy skrypt robi dokładnie to względem https://api.cheaptokens.dev/v1 i na końcu wypisuje sumę tokenów z pola usage.

31 października 2026 Andrej Karpathy (współzałożyciel OpenAI, były dyrektor ds. AI w Tesli) opisał HackerTube: zbudował go w około dwie godziny i użył Sonnet, aby zawęzić komentarze z Hacker News do około 40 tys. tych, które polecają film z YouTube. Jego wpis nie podaje promptu, rozmiaru partii ani współbieżności, więc to, co następuje, to standardowy wzorzec dla tego typu zadania, a nie jego przepis.

Skrypt#

Załóżmy, że masz 5000 zgłoszeń do wsparcia i chcesz każde oznaczyć jako bug, question lub praise. Oto całość. Zainstaluj SDK za pomocą pip install openai (Python 3.10+; python -c "import openai; print(openai.__version__)" pokazuje, co masz) i najpierw umieść klucz w zmiennej środowiskowej.

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

Otrzymujesz z powrotem labels (indeks tekstu → etykieta), failed (indeksy, które nigdy nie wygenerowały poprawnej etykiety, pusta lista, gdy wszystko przeszło) oraz sumy tokenów. Identyfikator modelu to pełny deepseek-ai/DeepSeek-V4-Flash-0731; identyfikatory dla innych modeli weź z dokumentacji CheapTokens, ponieważ krótkie nazwy różnią się między dostawcami.

Ułóż potok w kolejności, aby sprawdzić, czy masz w głowie właściwy kształt:

Dlaczego semafor, a nie 5000 żądań naraz#

asyncio.gather chętnie uruchamia wszystko, co mu dasz. Przekaż mu tysiące korutyn bez limitu, a wszystkie zaczną się naraz — wybuch żądań, który jest szybkim sposobem na błędy 429. Semafor to bramka: concurrency=5 oznacza, że w dowolnym momencie co najwyżej pięć wywołań jest w async with sem, a reszta czeka na swoją kolej. Python nadal tworzy wszystkie zadania, ale tylko pięć rozmawia z API.

Jaka liczba powinna być w concurrency? Zacznij nisko, jak w powyższej piątce, i zwiększaj, dopóki błędy nie znikną. Traktuj to jako ustawienie dostrajane dla każdego zadania, a nie stałą do skopiowania.

Ponowienia: dwie warstwy, dwa zadania#

Pierwsza warstwa jest wbudowana w SDK. Według README openai-python (sprawdzone 1 listopada 2026) błędy połączenia, 408, 409, 429 i odpowiedzi 5xx są domyślnie ponawiane dwukrotnie z krótkim wykładniczym backoffem. max_retries=2 w skrypcie tylko to uwidacznia; ustaw wyżej przez client.with_options(max_retries=5) lub na 0, aby wyłączyć. Domyślny limit czasu to 10 minut, a przekroczenie limitu zgłasza APITimeoutError, dlatego skrypt ustawia timeout=60 dla krótkich etykiet. Szczegóły są w OpenAI Python SDK timeout and retries, a idea backoffu w 429 Too Many Requests from an AI API.

Druga warstwa należy do Ciebie i obejmuje to, czego SDK nie widzi: wywołanie się powiodło, ale model odpowiedział prozą bez żadnego JSON-a (I can't classify this) lub wymyślił etykietę o nazwie complaint. Tekst wokół {...} nie ma znaczenia, ponieważ kod wycina JSON z odpowiedzi. Wtedy json.loads lub sprawdzenie LABELS zgłasza wyjątek, gather(..., return_exceptions=True) zachowuje błąd jako wartość zamiast zabijać partię, a element trafia do następnej rundy. Po rounds próbach to, co pozostaje w failed, jest do ręcznego sprawdzenia. Jedna zła odpowiedź na 5000 kosztuje jedno dodatkowe żądanie, a nie całe zadanie.

Skrypt prosi o JSON w prompcie i waliduje go w kodzie. Jeśli chcesz mocniej wymusić format za pomocą response_format, zobacz Get valid JSON from an LLM API.

Ile kosztuje zadanie#

Tokeny to jedyny licznik, więc skrypt sumuje usage i koszt wynika z tego. Weź typowy przypadek: krótkie teksty, krótkie etykiety. Załóż 10 000 tekstów po około 300 tokenów wejściowych każdy (tekst plus instrukcja) i około 20 tokenów wyjściowych na etykietę JSON:

10 000 × 300 = 3 000 000 tokenów wejściowych; 10 000 × 20 = 200 000 tokenów wyjściowych.

W DeepSeek-V4-Flash to zadanie kosztuje $0.065, w GLM-5.3-Flash $0.068, w MiniMax-M2.7 $0.18. Tabela pokazuje to samo zadanie dla każdego modelu, obok OpenRouter; przy sumowaniu wejścia i wyjścia wychodzi taniej niż OpenRouter łącznie.

Przesuń suwaki w kalkulatorze na własne liczby. Pola in i out dotyczą jednego żądania, a runs to liczba wysyłanych żądań.

Strona wejściowa dominuje: tekst wchodzi, jedno słowo wychodzi. Dlatego model z tanim wejściem i oknem kontekstu 380K, jak DeepSeek-V4-Flash z jedną ceną o każdej porze, nadaje się do masowego filtrowania; dla krótkich etykiet GLM-5.3-Flash i MiniMax-M2.7 są alternatywami dla tego samego skryptu, zmieniasz tylko MODEL. Nie wiesz, jak rośnie taki rachunek? What an AI API bills you for ma obraz.

Przed dużym uruchomieniem#

Najpierw uruchom 20 tekstów i sprawdź failed oraz sumy tokenów. Jeśli 20 tekstów zużywa około 6000 tokenów wejściowych, 5000 zużyje około 1 500 000, i możesz pomnożyć przed zobowiązaniem.

Jeśli nie działa#

• KeyError: 'CHEAPTOKENS_API_KEY': zmienna nie jest ustawiona w tym terminalu. Wyeksportuj ją i uruchom Pythona ponownie.
• 401 lub 404: zły klucz, bazowy URL lub identyfikator modelu. API 401 or 404 error omawia każdy z nich.
• json.JSONDecodeError na wielu elementach: odpowiedź nie ma {...}, często ponieważ max_tokens=100 ją uciął. Wypisz r.choices[0].finish_reason; length oznacza, że limit został osiągnięty, więc zwiększ max_tokens.
• Wiele 429 i wolne uruchomienie: zmniejsz concurrency.

Ogólne szczegóły modelu i punktu końcowego znajdują się w dokumentacji CheapTokens i na stronie cen.