Tausende von Texten mit AsyncOpenAI in Python klassifizieren

Um Tausende von Texten zu klassifizieren, verwende AsyncOpenAI, lege ein asyncio.Semaphore um jeden Aufruf, sodass nur N Anfragen gleichzeitig unterwegs sind, frage nach einem einzeiligen JSON-Label und führe nur die fehlgeschlagenen Elemente erneut aus. Das folgende Skript macht genau das gegen https://api.cheaptokens.dev/v1 und gibt am Ende die Gesamtzahl der Tokens aus dem Feld usage aus.

Am 31. Oktober 2026 beschrieb Andrej Karpathy (OpenAI-Mitbegründer, ehemaliger Tesla-KI-Direktor) HackerTube: Er baute es in etwa zwei Stunden und nutzte Sonnet, um Hacker-News-Kommentare auf etwa 40K einzugrenzen, die ein YouTube-Video empfehlen. Sein Beitrag nennt weder den Prompt, die Batch-Größe noch die Parallelität, daher ist das Folgende das Standardmuster für diese Art von Aufgabe, nicht sein Rezept.

Das Skript#

Angenommen, du hast 5.000 Support-Tickets und möchtest jedes mit bug, question oder praise taggen. Hier ist das Ganze. Installiere das SDK mit pip install openai (Python 3.10+; python -c "import openai; print(openai.__version__)" zeigt, was du hast) und lege deinen Schlüssel zuerst in einer Umgebungsvariable ab.

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

Du erhältst labels (Textindex → Label), failed (Indizes, die nie ein gültiges Label erzeugt haben, eine leere Liste, wenn alles durchlief) und die Token-Summen. Die Modell-ID ist die vollständige deepseek-ai/DeepSeek-V4-Flash-0731; IDs für andere Modelle findest du in der CheapTokens-Dokumentation, da sich die Kurznamen zwischen Anbietern unterscheiden.

Bringe die Pipeline in die richtige Reihenfolge, um zu prüfen, ob du die Struktur im Kopf hast:

Warum ein Semaphor und nicht 5.000 Anfragen auf einmal#

asyncio.gather startet bereitwillig alles, was du ihm gibst. Übergib ihm Tausende von Coroutinen ohne Limit, und sie starten alle gleichzeitig – ein Ansturm von Anfragen, der ein schneller Weg zu 429-Fehlern ist. Das Semaphor ist ein Drehkreuz: concurrency=5 bedeutet, dass sich höchstens fünf Aufrufe gleichzeitig in async with sem befinden, und der Rest wartet. Python erstellt zwar alle Tasks, aber nur fünf sprechen mit der API.

Welche Zahl sollte concurrency sein? Beginne niedrig, wie die 5 oben, und erhöhe sie, solange die Fehler ausbleiben. Behandle sie als eine Einstellung, die du pro Aufgabe abstimmst, nicht als eine zu kopierende Konstante.

Wiederholungen: zwei Ebenen, zwei Aufgaben#

Die erste Ebene ist im SDK integriert. Laut der openai-python README (geprüft am 1. November 2026) werden Verbindungsfehler, 408, 409, 429 und 5xx-Antworten standardmäßig zweimal mit kurzem exponentiellem Backoff wiederholt. max_retries=2 im Skript macht das nur sichtbar; setze es mit client.with_options(max_retries=5) höher oder mit 0 auf aus. Das Standard-Timeout beträgt 10 Minuten, und ein Timeout löst APITimeoutError aus, weshalb das Skript timeout=60 für kurze Labels setzt. Die Details stehen in OpenAI Python SDK Timeout und Wiederholungen, und die Idee hinter dem Backoff in 429 Too Many Requests von einer KI-API.

Die zweite Ebene ist deine, und sie deckt ab, was das SDK nicht sehen kann: Der Aufruf war erfolgreich, aber das Modell antwortete mit Prosa und gar keinem JSON (I can't classify this) oder erfand ein Label namens complaint. Text um ein {...} herum spielt keine Rolle, da der Code das JSON aus der Antwort herausschneidet. Dann löst json.loads oder die LABELS-Prüfung einen Fehler aus, gather(..., return_exceptions=True) behält den Fehler als Wert, anstatt den Batch abzubrechen, und das Element geht in die nächste Runde. Nach rounds Versuchen liegt alles, was noch in failed steht, bei dir zur manuellen Prüfung. Eine schlechte Antwort von 5.000 kostet dich eine zusätzliche Anfrage, nicht den ganzen Job.

Das Skript fordert JSON im Prompt an und validiert es im Code. Wenn du das Format mit response_format stärker erzwingen willst, siehe Gültiges JSON von einer LLM-API erhalten.

Was der Job kostet#

Tokens sind der einzige Maßstab, daher summiert das Skript usage und die Kosten folgen. Nehmen wir den häufigen Fall: kurze Texte, kurze Labels. Angenommen 10.000 Texte mit etwa 300 Eingabe-Tokens jeweils (der Text plus die Anweisung) und etwa 20 Ausgabe-Tokens für das JSON-Label:

10.000 × 300 = 3.000.000 Eingabe-Tokens; 10.000 × 20 = 200.000 Ausgabe-Tokens.

Auf DeepSeek-V4-Flash kostet dieser Job $0.065, auf GLM-5.3-Flash $0.068, auf MiniMax-M2.7 $0.18. Die Tabelle zeigt denselben Job für jedes Modell, neben OpenRouter; mit Eingabe und Ausgabe zusammen gerechnet ist es insgesamt günstiger als OpenRouter.

Verschiebe die Schieberegler im Rechner auf deine eigenen Zahlen. Die Felder in und out gelten für eine Anfrage, und runs ist die Anzahl der Anfragen, die du sendest.

Die Eingabeseite dominiert hier: Der Text geht rein, ein Wort kommt raus. Deshalb eignet sich ein Modell mit günstiger Eingabe und einem 380K-Kontextfenster wie DeepSeek-V4-Flash mit einem Preis zu jeder Stunde für die Massenfilterung; für kurze Labels sind GLM-5.3-Flash und MiniMax-M2.7 Alternativen für dasselbe Skript, du änderst nur MODEL. Unsicher, wie eine solche Rechnung wächst? Wofür eine KI-API dir berechnet hat das Bild.

Vor dem großen Lauf#

Führe zuerst 20 Texte aus und schaue dir failed und die Token-Summen an. Wenn 20 Texte etwa 6.000 Eingabe-Tokens verbrauchen, werden 5.000 etwa 1.500.000 verbrauchen, und du kannst vor dem Commit multiplizieren.

Wenn es nicht funktioniert#

• KeyError: 'CHEAPTOKENS_API_KEY': Die Variable ist in diesem Terminal nicht gesetzt. Exportiere sie und starte Python erneut.
• 401 oder 404: Falscher Schlüssel, falsche Basis-URL oder falsche Modell-ID. API 401 oder 404 Fehler führt durch jeden Fall.
• json.JSONDecodeError bei vielen Elementen: Die Antwort enthält kein {...}, oft weil max_tokens=100 sie abgeschnitten hat. Gib r.choices[0].finish_reason aus; length bedeutet, das Limit wurde erreicht, also erhöhe max_tokens.
• Viele 429 und ein langsamer Lauf: Senke concurrency.

Die allgemeinen Modell- und Endpunktdetails findest du in der CheapTokens-Dokumentation und auf der Preisseite.