Python で AsyncOpenAI を使って数千のテキストを分類する

数千のテキストを分類するには、AsyncOpenAI を使い、各呼び出しの周りに asyncio.Semaphore を置いて同時に N リクエストだけが実行中になるようにし、1行の JSON ラベルを要求し、失敗した項目だけを再実行します。以下のスクリプトはまさにそれを https://api.cheaptokens.dev/v1 に対して行い、最後に usage フィールドから合計トークン数を出力します。

2026年10月31日、Andrej Karpathy（OpenAI 共同創業者、元 Tesla AI ディレクター）は HackerTube について説明しました。彼は約2時間でそれを構築し、Sonnet を使って Hacker News のコメントを、YouTube 動画を推薦する約40K件に絞り込みました。彼の投稿にはプロンプト、バッチサイズ、並行数は記載されていないため、以下はこの種の作業の標準パターンであり、彼のレシピではありません。

スクリプト#

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（有効なラベルを生成できなかったインデックス。すべて成功した場合は空リスト）、そしてトークン合計です。モデル ID は完全な deepseek-ai/DeepSeek-V4-Flash-0731 です。他のモデルの ID は CheapTokens のドキュメント から取得してください。短い名前はプロバイダー間で異なるためです。

パイプラインを順番に並べて、頭の中で全体像を確認しましょう：

なぜセマフォを使い、一度に5,000リクエストを送らないのか#

asyncio.gather は与えられたものを何でも喜んで開始します。制限なしで数千のコルーチンを渡すと、それらはすべて一度に開始され、リクエストのバーストは 429 エラーの簡単な原因になります。セマフォは改札口です：concurrency=5 は、任意の時点で最大5つの呼び出しが async with sem 内にあり、残りは順番を待つことを意味します。Python は依然としてすべてのタスクを作成しますが、API と通信するのは5つだけです。

concurrency はいくつにすべきでしょうか？上記の5のように低く始め、エラーが出ない間に上げていきます。これはジョブごとに調整する設定であり、コピーする定数ではありません。

リトライ：2つの層、2つの役割#

最初の層は SDK に組み込まれています。openai-python README（2026年11月1日確認）によると、接続エラー、408、409、429、5xx レスポンスはデフォルトで2回、短い指数バックオフでリトライされます。スクリプトの max_retries=2 はそれを可視化するだけです。client.with_options(max_retries=5) で増やすか、0 で無効にできます。デフォルトのタイムアウトは10分で、タイムアウトは APITimeoutError を発生させるため、スクリプトでは短いラベルのために timeout=60 を設定しています。詳細は OpenAI Python SDK のタイムアウトとリトライ に、その背後にあるバックオフの考え方は AI API からの 429 Too Many Requests にあります。

2番目の層はあなたのもので、SDK が見えないものをカバーします：呼び出しは成功したが、モデルが散文で答え JSON が全くない（I can't classify this）、または complaint というラベルを発明した場合です。{...} の周りのテキストは問題ありません。コードが答えから JSON を切り出すからです。その後、json.loads または LABELS チェックが例外を発生させ、gather(..., return_exceptions=True) がバッチを殺す代わりにエラーを値として保持し、項目は次のラウンドに進みます。rounds 回の試行後、failed に残っているものは手動で検査してください。5,000件中1件の悪い答えは、ジョブ全体ではなく、1回の余分なリクエストのコストで済みます。

スクリプトはプロンプトで JSON を要求し、コードで検証します。response_format でフォーマットをさらに強制したい場合は、LLM API から有効な JSON を取得する を参照してください。

ジョブのコスト#

トークンだけがメーターなので、スクリプトは usage を合計し、コストが続きます。一般的なケースを考えます：短いテキスト、短いラベル。10,000件のテキストがそれぞれ約300入力トークン（テキストと指示）で、JSON ラベルに約20出力トークンと仮定します：

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 フィールドは1リクエストあたりの値で、runs は送信するリクエスト数です。

ここでは入力側が支配的です：テキストが入り、1単語が出てきます。だからこそ、入力が安く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、モデル ID が間違っています。API 401 または 404 エラー がそれぞれを説明します。
• 多くの項目で json.JSONDecodeError：答えに {...} がありません。多くの場合 max_tokens=100 がそれを切り捨てたためです。r.choices[0].finish_reason を出力してください。length は制限に達したことを意味するので、max_tokens を上げてください。
• 多くの 429 と遅い実行：concurrency を下げてください。

一般的なモデルとエンドポイントの詳細は、CheapTokens のドキュメント と 料金ページ にあります。