【LLM×非同期処理】深夜のAPIレートリミット（429 Too Many Requests）をCelery×Redisのトークンバケットで美しく制御する

はじめに

社内データを一括でベクトル化（Embedding）したり、数万件の商品データにLLMで自動タグ付けするような「大量バッチ処理」を実装したとき、高確率で我々を絶望させるのが 429 Too Many Requests (Rate Limit Exceeded) です。

OpenAI等のAPIには、1分あたりのリクエスト数（RPM）だけでなく、**1分あたりの消費トークン数（TPM）**という厳しい制限があります。
単純なループ処理や、安易な asyncio.gather で並列実行すると、一瞬でこの制限に引っかかり、タスクが途中で壊滅します。

本記事では、Pythonの定番非同期タスクキュー Celery と Redis を組み合わせ、APIの枠（RPM/TPM）を限界まで使い切りつつ、絶対に429エラーを出さない「トークンバケット型」の流量制御（Rate Limiting）を実装する方法を解説します。

1. なぜ「単なる `time.sleep()`」や「指数リトライ」ではダメなのか？

time.sleepの限界: LLMのレスポンス時間は毎回変わるため、一律で3秒待つような実装にすると、遅すぎるか、結局上限に引っかかるかの二択になります。
指数バックオフ（リトライ）の罠: エラーを検知してから引く（待つ）戦略ですが、大量のタスクが同時にリトライに回ると、今度は「リトライの波」が重なってさらに大きな429のスパイクを叩き出す原因になります（群集事故）。

理想の解決策:
リクエストを投げる「前」に、Redisで一元管理された「トークン（許可証）」をタスクが取得し、許可が出た分だけAPIを叩くトークンバケットアルゴリズムをタスクキューに組み込みます。

2. システム構成

フロー図 (Mermaid)

3. 【実装】CeleryとRedisによるカスタムRate Limiter

今回は、Redisの機能を使って「1分間に最大10,000トークン、最大50リクエスト」という枠を、複数のWorker間で共有・制御するコードを構築します。

依存関係

poetry add celery redis openai tiktoken

実装コード (`tasks.py`)

import os
import time
from celery import Celery
from redis import Redis
from openai import OpenAI
import tiktoken

# RedisとCeleryの初期化
REDIS_URL = os.getenv("REDIS_URL", "redis://localhost:6379/0")
app = Celery("llm_tasks", broker=REDIS_URL, backend=REDIS_URL)
redis_client = Redis.from_url(REDIS_URL)

openai_client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))
encoder = tiktoken.encoding_for_model("gpt-4o-mini")

# APIの制限定義 (例: 本番環境のプランに合わせて調整)
MAX_RPM = 50
MAX_TPM = 10000

def check_and_consume_limits(estimated_tokens: int) -> bool:
    """
    Redisのカウンターを使って、現在のRPM/TPMの枠に収まるかチェックする。
    収まる場合は枠を消費してTrueを返す。
    """
    current_minute = int(time.time() // 60)
    rpm_key = f"llm_limit:rpm:{current_minute}"
    tpm_key = f"llm_limit:tpm:{current_minute}"

    # パイプラインでアトミックに処理
    pipe = redis_client.pipeline()
    pipe.get(rpm_key)
    pipe.get(tpm_key)
    current_rpm, current_tpm = pipe.execute()

    current_rpm = int(current_rpm) if current_rpm else 0
    current_tpm = int(current_tpm) if current_tpm else 0

    # 制限プレチェック
    if current_rpm + 1 > MAX_RPM or current_tpm + estimated_tokens > MAX_TPM:
        return False # 枠がいっぱい

    # 枠を消費
    pipe = redis_client.pipeline()
    pipe.incr(rpm_key)
    pipe.expire(rpm_key, 120) # 2分後に自動消去
    pipe.incrby(tpm_key, estimated_tokens)
    pipe.expire(tpm_key, 120)
    pipe.execute()

    return True

@app.task(bind=True, max_retries=None)
def process_llm_heavy_task(self, text_content: str):
    """
    大量に並列実行される重いLLMタスク
    """
    # 1. 投げる前に、プロンプトのトークン数を概算（少し多めに見積もるのがコツ）
    prompt_tokens = len(encoder.encode(text_content))
    estimated_total_tokens = prompt_tokens + 500 # 出力分のバッファ

    # 2. 枠があるか確認
    if not check_and_consume_limits(estimated_total_tokens):
        # 枠がない場合は、10秒後にこのタスクをリトライ（キューに優しく戻す）
        # これによりWorkerのスレッドをロック（sleep）せずに解放できる
        print(f"【流量制御】制限に達したため、タスクを退避します（想定トークン: {estimated_total_tokens}）")
        raise self.retry(countdown=10)

    # 3. 枠が確保できた安全な状態でAPIを叩く
    try:
        response = openai_client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": text_content}]
        )
        return response.choices[0].message.content
    except Exception as e:
        # 万が一、他の要因でエラーが出た場合のセーフティネット
        print(f"APIエラー発生: {e}")
        raise self.retry(countdown=30)

4. この構成が優れている理由（実務上のメリット）

① Workerをノンブロッキングに保てる

time.sleep() でWorkerを停止させると、その間Workerのプロセスが丸々1つ無駄になります。この実装では、枠がないタスクは速やかにキューの最後尾（指定秒後）にリトライとして回り、Workerは「今すぐ処理できる別のタスク」を瞬時に引き受けられるため、インフラのCPU効率が最大化します。

② 分散システム（マルチサーバー）で完全同期する

Redisを中央カウンターにしているため、Workerが1台だろうが、オートスケーリングで10台に増えようが、システム全体での「合計TPM/RPM」を確実にコントロールできます。

まとめ

LLMのバッチ処理を安定して回すためには、APIの「外側」にあるインフラ層での流量制御が不可欠です。
429エラーを「受けてから慌てる」のではなく、Redisで「投げる前に止める」設計に切り替えるだけで、深夜のバッチ処理が驚くほど静かに、かつ最速で終わるようになります。

大量データ処理に悩むバックエンドエンジニアの参考になれば幸いです！

「このインフラ構成、知りたかった！」という方は、LGTM と ストック をお願いします！
Celery以外のキュー（SQSやCloud Tasksなど）での実装例に興味がある方は、コメント欄までどうぞ。