Klasifikasikan ribuan teks dengan AsyncOpenAI di Python

Untuk mengklasifikasikan ribuan teks, gunakan AsyncOpenAI, letakkan asyncio.Semaphore di sekitar setiap panggilan agar hanya N permintaan yang berjalan bersamaan, minta label JSON satu baris, dan jalankan ulang hanya item yang gagal. Skrip di bawah melakukan persis itu terhadap https://api.cheaptokens.dev/v1 dan mencetak total token dari field usage di akhir.

Pada 31 Oktober 2026 Andrej Karpathy (co-founder OpenAI, mantan direktur AI Tesla) mendeskripsikan HackerTube: ia membangunnya dalam sekitar dua jam dan menggunakan Sonnet untuk menyaring komentar Hacker News menjadi sekitar 40K yang merekomendasikan video YouTube. Postingannya tidak memberikan prompt, ukuran batch, atau konkurensi, jadi yang berikut adalah pola standar untuk pekerjaan semacam ini, bukan resepnya.

Skrip#

Katakanlah Anda memiliki 5.000 tiket dukungan dan ingin masing-masing diberi tag bug, question, atau praise. Berikut keseluruhannya. Instal SDK dengan pip install openai (Python 3.10+; python -c "import openai; print(openai.__version__)" menunjukkan versi yang Anda miliki) dan letakkan kunci Anda di variabel lingkungan terlebih dahulu.

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

Anda mendapatkan kembali labels (indeks teks → label), failed (indeks yang tidak pernah menghasilkan label valid, daftar kosong saat semuanya berhasil) dan total token. Id model adalah deepseek-ai/DeepSeek-V4-Flash-0731 lengkap; ambil id untuk model lain dari dokumentasi CheapTokens, karena nama pendeknya berbeda antar penyedia.

Susun pipeline dalam urutan untuk memeriksa apakah Anda sudah memahami bentuknya:

Mengapa semaphore dan bukan 5.000 permintaan sekaligus#

asyncio.gather dengan senang hati memulai semua yang Anda berikan. Berikan ribuan coroutine tanpa batas dan semuanya mulai sekaligus, ledakan permintaan yang merupakan cara cepat untuk mendapatkan error 429. Semaphore adalah pintu putar: concurrency=5 berarti paling banyak lima panggilan berada di dalam async with sem pada saat bersamaan, dan sisanya menunggu giliran. Python tetap membuat semua task, tetapi hanya lima yang berbicara dengan API.

Berapa angka untuk concurrency? Mulai dari yang rendah, seperti 5 di atas, dan naikkan selama error tidak muncul. Perlakukan sebagai pengaturan yang Anda sesuaikan per pekerjaan, bukan konstanta untuk disalin.

Percobaan ulang: dua lapisan, dua tugas#

Lapisan pertama tertanam di SDK. Menurut README openai-python (diperiksa 1 November 2026), error koneksi, 408, 409, 429, dan respons 5xx dicoba ulang dua kali secara default dengan backoff eksponensial singkat. max_retries=2 dalam skrip hanya membuatnya terlihat; atur lebih tinggi dengan client.with_options(max_retries=5) atau ke 0 untuk mematikannya. Timeout default adalah 10 menit dan timeout memunculkan APITimeoutError, itulah sebabnya skrip menetapkan timeout=60 untuk label pendek. Detailnya ada di Timeout dan percobaan ulang OpenAI Python SDK, dan ide backoff di baliknya di 429 Too Many Requests dari API AI.

Lapisan kedua adalah milik Anda, dan mencakup apa yang tidak bisa dilihat SDK: panggilan berhasil, tetapi model menjawab dengan prosa dan sama sekali tanpa JSON (I can't classify this) atau mengarang label bernama complaint. Teks di sekitar {...} tidak masalah, karena kode memotong JSON dari jawaban. Kemudian json.loads atau pemeriksaan LABELS memunculkan error, gather(..., return_exceptions=True) menyimpan error sebagai nilai alih-alih mematikan batch, dan item tersebut masuk ke putaran berikutnya. Setelah rounds percobaan, apa pun yang masih ada di failed menjadi milik Anda untuk diperiksa secara manual. Satu jawaban buruk dari 5.000 hanya menghabiskan satu permintaan tambahan, bukan seluruh pekerjaan.

Skrip meminta JSON dalam prompt dan memvalidasinya dalam kode. Jika Anda ingin memaksakan format lebih keras dengan response_format, lihat Dapatkan JSON valid dari API LLM.

Berapa biaya pekerjaan ini#

Token adalah satu-satunya meter, jadi skrip menjumlahkan usage dan biaya mengikuti. Ambil kasus umum: teks pendek, label pendek. Asumsikan 10.000 teks dengan sekitar 300 token input masing-masing (teks plus instruksi) dan sekitar 20 token output untuk label JSON:

10.000 × 300 = 3.000.000 token input; 10.000 × 20 = 200.000 token output.

Pada DeepSeek-V4-Flash pekerjaan itu menjadi $0.065, pada GLM-5.3-Flash menjadi $0.068, pada MiniMax-M2.7 menjadi $0.18. Tabel menunjukkan pekerjaan yang sama untuk setiap model, di samping OpenRouter; dengan input dan output dihitung bersama, totalnya lebih murah daripada OpenRouter.

Geser slider di kalkulator ke angka Anda sendiri. Field in dan out adalah untuk satu permintaan, dan runs adalah berapa banyak permintaan yang Anda kirim.

Sisi input mendominasi di sini: teks masuk, satu kata keluar. Itulah sebabnya model dengan input murah dan jendela konteks 380K, seperti DeepSeek-V4-Flash dengan satu harga di jam berapa pun, cocok untuk penyaringan massal; untuk label pendek, GLM-5.3-Flash dan MiniMax-M2.7 adalah alternatif untuk skrip yang sama, Anda hanya mengubah MODEL. Tidak yakin bagaimana tagihan seperti ini tumbuh? Apa yang ditagih API AI kepada Anda memiliki gambaran lengkapnya.

Sebelum menjalankan besar#

Jalankan 20 teks terlebih dahulu dan lihat failed dan total token. Jika 20 teks menggunakan sekitar 6.000 token input, 5.000 akan menggunakan sekitar 1.500.000, dan Anda dapat mengalikannya sebelum berkomitmen.

Jika tidak berhasil#

• KeyError: 'CHEAPTOKENS_API_KEY': variabel tidak diatur di terminal ini. Ekspor dan mulai Python lagi.
• 401 atau 404: kunci, base URL, atau id model salah. Error API 401 atau 404 membahas masing-masing.
• json.JSONDecodeError pada banyak item: jawaban tidak memiliki {...}, sering karena max_tokens=100 memotongnya. Cetak r.choices[0].finish_reason; length berarti batas tercapai, jadi naikkan max_tokens.
• Banyak 429 dan jalannya lambat: turunkan concurrency.

Detail model dan endpoint umum ada di dokumentasi CheapTokens dan di halaman harga.