Python'da AsyncOpenAI ile binlerce metni sınıflandırın

Binlerce metni sınıflandırmak için AsyncOpenAI kullanın, her çağrının etrafına bir asyncio.Semaphore koyun ki aynı anda yalnızca N istek uçuşta olsun, tek satırlık bir JSON etiketi isteyin ve yalnızca başarısız olan öğeleri yeniden çalıştırın. Aşağıdaki betik tam olarak bunu https://api.cheaptokens.dev/v1 üzerinde yapar ve sonunda usage alanından toplam token sayısını yazdırır.

31 Ekim 2026'da Andrej Karpathy (OpenAI kurucu ortağı, eski Tesla AI direktörü) HackerTube'u tanımladı: yaklaşık iki saatte inşa etti ve Hacker News yorumlarını bir YouTube videosu öneren yaklaşık 40K'ya indirmek için Sonnet kullandı. Gönderisinde istem, toplu iş boyutu veya eşzamanlılık yok, bu yüzden aşağıdakiler onun tarifi değil, bu tür bir iş için standart kalıptır.

Betik#

Diyelim ki 5.000 destek talebiniz var ve her birini bug, question veya praise olarak etiketlemek istiyorsunuz. İşte tamamı. SDK'yı pip install openai ile kurun (Python 3.10+; python -c "import openai; print(openai.__version__)" neye sahip olduğunuzu gösterir) ve anahtarınızı önce bir ortam değişkenine koyun.

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

Geri aldığınız şeyler: labels (metin indeksi → etiket), failed (geçerli bir etiket üretmeyen indeksler, her şey yolunda gittiğinde boş liste) ve token toplamları. Model kimliği tam deepseek-ai/DeepSeek-V4-Flash-0731'dir; diğer modeller için kimlikleri CheapTokens belgelerinden alın, çünkü kısa adlar sağlayıcılar arasında farklılık gösterir.

Şekli kafanızda canlandırdığınızdan emin olmak için ardışık düzeni sıraya koyun:

Neden 5.000 istek birden değil de semafor#

asyncio.gather verdiğiniz her şeyi memnuniyetle başlatır. Binlerce eşyordamı sınırsız verirseniz hepsi birden başlar, bu da 429 hatalarına hızlı bir yol olan bir istek patlamasıdır. Semafor bir turnikedir: concurrency=5, herhangi bir anda en fazla beş çağrının async with sem içinde olduğu ve geri kalanın sırasını beklediği anlamına gelir. Python yine de tüm görevleri oluşturur, ancak yalnızca beşi API ile konuşur.

concurrency kaç olmalı? Yukarıdaki 5 gibi düşük başlayın ve hatalar uzak durdukça yükseltin. Bunu kopyalanacak bir sabit değil, işe göre ayarladığınız bir ayar olarak görün.

Yeniden denemeler: iki katman, iki iş#

İlk katman SDK'ya yerleşiktir. openai-python README dosyasına göre (1 Kasım 2026'da kontrol edildi), bağlantı hataları, 408, 409, 429 ve 5xx yanıtları varsayılan olarak kısa bir üstel geri çekilme ile iki kez yeniden denenir. Betikteki max_retries=2 bunu yalnızca görünür kılar; client.with_options(max_retries=5) ile daha yükseğe ayarlayın veya kapatmak için 0 yapın. Varsayılan zaman aşımı 10 dakikadır ve zaman aşımı APITimeoutError hatası verir, bu yüzden betik kısa etiketler için timeout=60 ayarlar. Ayrıntılar OpenAI Python SDK zaman aşımı ve yeniden denemeler sayfasında ve arkasındaki geri çekilme fikri Bir AI API'sinden 429 Too Many Requests sayfasında.

İkinci katman sizindir ve SDK'nın göremediğini kapsar: çağrı başarılı oldu ama model düzyazıyla yanıt verdi ve hiç JSON yok (I can't classify this) veya complaint adında bir etiket uydurdu. Bir {...} etrafındaki metin önemli değildir, çünkü kod JSON'u yanıttan kesip çıkarır. Sonra json.loads veya LABELS kontrolü hata verir, gather(..., return_exceptions=True) hatayı toplu işi öldürmek yerine bir değer olarak tutar ve öğe bir sonraki tura gider. rounds denemeden sonra failed içinde kalan her şey elle incelemeniz için sizindir. 5.000'de bir kötü yanıt size tüm işi değil, yalnızca bir ekstra isteğe mal olur.

Betik istemde JSON ister ve kodu içinde doğrular. Formatı response_format ile daha da zorlamak isterseniz, Bir LLM API'sinden geçerli JSON alın sayfasına bakın.

İşin maliyeti#

Tokenlar tek ölçüdür, bu yüzden betik usage toplar ve maliyet bunu takip eder. Yaygın durumu ele alalım: kısa metinler, kısa etiketler. Her biri yaklaşık 300 giriş tokenı (metin artı talimat) ve JSON etiketi için yaklaşık 20 çıkış tokenı olan 10.000 metin varsayalım:

10.000 × 300 = 3.000.000 giriş tokenı; 10.000 × 20 = 200.000 çıkış tokenı.

DeepSeek-V4-Flash'te bu iş $0.065, GLM-5.3-Flash'te $0.068, MiniMax-M2.7'de $0.18 tutar. Tablo aynı işi her model için OpenRouter'ın yanında gösterir; giriş ve çıkış birlikte sayıldığında toplamda OpenRouter'dan daha ucuz çıkar.

Hesaplayıcıdaki kaydırıcıları kendi sayılarınıza taşıyın. in ve out alanları tek bir istek içindir ve runs kaç istek gönderdiğinizdir.

Burada giriş tarafı baskındır: metin girer, tek kelime çıkar. Bu yüzden ucuz girişi ve 380K bağlam penceresi olan bir model, örneğin her saat aynı fiyatlı DeepSeek-V4-Flash, toplu filtrelemeye uygundur; kısa etiketler için GLM-5.3-Flash ve MiniMax-M2.7 aynı betiğin alternatifleridir, yalnızca MODEL değiştirirsiniz. Böyle bir faturanın nasıl büyüdüğünden emin değil misiniz? Bir AI API'sinin sizden ne ücret aldığı resmi gösterir.

Büyük çalıştırmadan önce#

Önce 20 metin çalıştırın ve failed ile token toplamlarına bakın. 20 metin yaklaşık 6.000 giriş tokenı kullanıyorsa, 5.000 metin yaklaşık 1.500.000 kullanır ve taahhütte bulunmadan önce çarpabilirsiniz.

Çalışmazsa#

• KeyError: 'CHEAPTOKENS_API_KEY': değişken bu terminalde ayarlanmamış. Dışa aktarın ve Python'u yeniden başlatın.
• 401 veya 404: yanlış anahtar, temel URL veya model kimliği. API 401 veya 404 hatası her birini adım adım açıklar.
• Birçok öğede json.JSONDecodeError: yanıtta {...} yok, genellikle max_tokens=100 onu kestiği için. r.choices[0].finish_reason yazdırın; length sınıra ulaşıldığı anlamına gelir, bu yüzden max_tokens yükseltin.
• Birçok 429 ve yavaş bir çalıştırma: concurrency düşürün.

Genel model ve uç nokta ayrıntıları CheapTokens belgelerinde ve fiyatlandırma sayfasında bulunur.