ガイド: バッチ処理

バッチジョブ(商品カタログの翻訳、サポートチケットの分類、数千のドキュメントからのフィールド抽出)は、同じ種類のリクエストを何度も送信します。作業自体は単純ですが、難しいのはスループット、失敗、コストです。このガイドでは、公式の openai パッケージをベース URL https://api14.com/api/v1 で使い、この 3 つすべてを扱います。

専用のバッチエンドポイントはありません。OpenAI の Batch API は api14 では利用できません(OpenAI SDK 互換性を参照)。バッチは、同時実行数を制御した通常のリクエストとして実行します。

レート制限を把握する

上限は API キーごとの毎分リクエスト数(RPM)で数えられ、プランによって制限されます。

独自の RPM 上限を持つキーには、2 つのうち低いほうが適用されます。自分の上限は GET /key で確認できます(キーがプランの上限を使う場合、rate_limit.requests は null です)。残高とキー情報を参照してください。上限を超えると、Retry-After: 60 付きの 429 rate_limited が返ります。

RPM が制限するのは、1 分間に開始されるリクエスト数であり、同時に実行される数ではありません。平均的なリクエストに 4 秒かかり、上限が 100 RPM の場合、約 7 件のリクエストが同時に進行しているだけで上限に達します。開始時刻でリクエストのペースを配分し、同時実行数は控えめにしてください。

ペース配分付きの並行ランナー(Python)

import asyncio, json, os, time
from openai import AsyncOpenAI, APIStatusError

client = AsyncOpenAI(base_url="https://api14.com/api/v1", api_key=os.environ["API14_API_KEY"], max_retries=4)

RPM = 100           # プランまたはキーの上限
CONCURRENCY = 10
FATAL = {"daily_limit_reached", "insufficient_balance", "invalid_api_key", "model_not_allowed"}

class Pacer:
    """リクエストの開始を均等に間隔を空けて、RPM を超えないようにします。"""
    def __init__(self, rpm):
        self.interval, self.next, self.lock = 60 / rpm, 0.0, asyncio.Lock()
    async def wait(self):
        async with self.lock:
            now = time.monotonic()
            self.next = max(self.next, now)
            delay, self.next = self.next - now, self.next + self.interval
        await asyncio.sleep(delay)

SYSTEM = "Translate the user's text to German. Keep product names and units unchanged. Reply with the translation only."

async def translate(item, pacer, sem):
    async with sem:
        await pacer.wait()
        r = await client.chat.completions.create(
            model="openai/gpt-5-mini",
            messages=[{"role": "system", "content": SYSTEM}, {"role": "user", "content": item["text"]}],
            max_tokens=500,
        )
        return {"id": item["id"], "text": r.choices[0].message.content, "cost": r.usage.cost}

async def main(items, out_path="out.jsonl"):
    done = set()
    if os.path.exists(out_path):
        done = {json.loads(line)["id"] for line in open(out_path)}
    todo = [i for i in items if i["id"] not in done]

    pacer, sem = Pacer(RPM), asyncio.Semaphore(CONCURRENCY)
    with open(out_path, "a") as out:
        for task in asyncio.as_completed([translate(i, pacer, sem) for i in todo]):
            try:
                out.write(json.dumps(await task) + "\n")
            except APIStatusError as e:
                if e.code in FATAL:
                    raise SystemExit(f"Stopping batch: {e.code}")
                print("failed:", e.status_code, e.code)  # ログに記録。再実行で再試行されます

items = [json.loads(line) for line in open("items.jsonl")]  # [{"id": "...", "text": "..."}]
asyncio.run(main(items))

このコードの動作は次のとおりです。

• ペース配分: Pacer は、1 分間に最大 RPM 件のリクエストを開始します。セマフォは、同時に進行する数を制限します。
• 再試行: SDK は、Retry-After に従いながら、429 と 5xx を指数バックオフで再試行します(max_retries=4)。上流プロバイダーの 429 も同様に再試行されます。
• 致命的なコードではジョブを停止: daily_limit_reached と insufficient_balance は再試行しても解消されないため、ジョブは試行を浪費せずに終了します。
• チェックポイント: 結果は完了するたびに JSONL ファイルへ追記されます。スクリプトを再実行すると、完了済みの ID はスキップされるため、クラッシュやジョブの停止があっても何も失われません。

Node では、共有キューから取り出す一定数のワーカーに対する Promise.all で同じ構造が使えます。各ワーカーは、開始の間に 60_000 / RPM ミリ秒待ちます。

1 つのリクエストに複数の項目を入れる

短い項目では、項目ごとに 1 リクエストにすると RPM を無駄にし、そのたびに指示を繰り返すことになります。構造化出力で 1 リクエストあたり 20〜50 件を分類すれば、リクエスト数を同じ倍率で割れます。

schema = {
    "type": "object",
    "properties": {"labels": {"type": "array", "items": {
        "type": "object",
        "properties": {"id": {"type": "string"}, "label": {"type": "string", "enum": ["billing", "bug", "feature", "other"]}},
        "required": ["id", "label"], "additionalProperties": False,
    }}},
    "required": ["labels"], "additionalProperties": False,
}

tickets = "\n".join(f"{t['id']}: {t['text']}" for t in batch_of_30)
r = await client.chat.completions.create(
    model="openai/gpt-5-mini",
    messages=[{"role": "system", "content": "Label each ticket."}, {"role": "user", "content": tickets}],
    response_format={"type": "json_schema", "json_schema": {"name": "labels", "strict": True, "schema": schema}},
)
labels = json.loads(r.choices[0].message.content)["labels"]

すべての入力 ID が返ってきたかを確認し、欠けているものは再送してください。finish_reason が "length" の場合、バッチは max_tokens に対して大きすぎたため、分割してください。構造化出力を参照してください。埋め込みでは、1 つの input 配列に最大 100 個の文字列を渡せます。埋め込みを参照してください。

プロンプトキャッシュでコストを削減する

バッチでは、指示、例、スキーマがすべてのリクエストで同一です。これらを先頭に、変化する項目を末尾に置くと、プロバイダーはキャッシュされたプレフィックスを再利用できます。

• OpenAI、DeepSeek、Gemini 2.5 などは、プレフィックスが十分な長さになると自動でキャッシュします(OpenAI では 1,024 トークン)。キャッシュの読み取りは、通常の入力の何分の一かの料金です。
• Anthropic のモデルでは、プロンプトの静的な部分に明示的な cache_control のブレークポイントが必要です。

そのため、いくつかの実例を含む長く詳細なシステムプロンプトでも、短いものとほとんど変わらないコストで済み、品質も通常は向上します。ヒットは usage.prompt_tokens_details.cached_tokens で確認してください。プロンプトキャッシュを参照してください。

低価格のモデルを選び、検証する

翻訳、分類、抽出では、小さなモデルでも十分な品質を、わずかな価格で得られることがよくあります。テキストモデルを入力価格順に一覧表示します。

curl -s "https://api14.com/api/v1/models?category=text" \
  | jq -r '.data | sort_by(.pricing.prompt | tonumber) | .[:15][] | "\(.id)  \(.pricing.prompt)  \(.pricing.completion)"'

決定する前に、次の手順で測定します。

1. 代表的な 50〜100 件の項目を選び、期待される出力を手作業でラベル付けします。
2. 2〜3 個の候補モデルで実行し、精度と usage.cost の合計を比較します。
3. 項目あたりのコストに全項目数を掛けてジョブ全体を見積もり、GET /credits で残高がまかなえることを確認します。

:free バリアントは、無料のエンドポイントがあるモデルでそれを使いますが、上流のレート制限がより厳しいため、大規模なバッチよりテストに向いています。モデル ID とバリアントを参照してください。

大規模な実行の前に

• バグで残高を使い果たさないように、1 日の USD 上限を設定した専用のキーを作成します。上限に達すると、ジョブは 429 daily_limit_reached で停止します。レート制限を参照してください。
• すべてのリクエストに max_tokens を設定します。
• 最初の 1% の項目を実行し、出力とコストを確認してから、残りを開始します。
• 失敗したリクエスト(4xx または 5xx)には課金されないため、失敗したものを再実行しても安全です。エラーを参照してください。