AI

LLM APIコスト削減：キャッシュとレート制限の実装ガイド

LLM APIの料金が月末に想定の2〜3倍になっていた、という経験をお持ちの方は少なくないはずです。特に「ちょっと試してみただけ」の開発フェーズが長引くほど、コストは静かに積み上がっていきます。この記事では、LLM APIコストを抑えるために実践的に機能するキャッシュ戦略とレート制限の実装方法を、コードを交えながら解説します。

LLM APIのコスト構造：何が料金を決めるのか

多くのLLM APIは入力トークン数 × 単価 ＋ 出力トークン数 × 単価という課金モデルを採用しています。モデルのグレードによってトークン単価が異なり、高性能なモデルほど単価が上がります。

コスト増の主な要因は次の3つです。

  • 同一または類似プロンプトの重複リクエスト：ユーザーが同じ質問を何度もする、バッチ処理でほぼ同じ内容を繰り返し送るケース
  • 冗長なシステムプロンプト：毎リクエストに長い指示文を含める設計
  • 不必要な高性能モデルの使用：単純なタスクにも最上位モデルを使い続ける

まずこの3点を計測・整理するだけで、改善余地がどこにあるかが見えてきます。

プロンプトキャッシングの仕組みと効果測定

プロンプトキャッシングとは、同一の入力（またはプレフィックス）に対するAPIレスポンスをキャッシュしておき、次回は実際のAPI呼び出しを省略する手法です。

Claude APIでは、Anthropicが提供するキャッシュコントロール機能（cache_control）を使うことで、繰り返し送信するシステムプロンプトやコンテキストのトークン処理コストを大幅に削減できるとされています。APIレイヤーでのキャッシュと、アプリケーションレイヤーでのキャッシュを組み合わせるのが実践的なアプローチです。

import anthropic

client = anthropic.Anthropic()

# システムプロンプトをキャッシュ対象として明示する
response = client.messages.create(
    model="claude-opus-4-5",
    max_tokens=1024,
    system=[
        {
            "type": "text",
            "text": "あなたはECサイトのカスタマーサポート担当です。以下の商品カタログと返品ポリシーに基づいて回答してください。\n\n[長大な商品カタログや規定文書がここに入る]",
            "cache_control": {"type": "ephemeral"}  # このブロックをキャッシュ
        }
    ],
    messages=[{"role": "user", "content": "注文をキャンセルしたいのですが"}]
)

# キャッシュの効き具合を確認する
usage = response.usage
print(f"入力トークン: {usage.input_tokens}")
print(f"キャッシュ作成トークン: {usage.cache_creation_input_tokens}")
print(f"キャッシュ読み込みトークン: {usage.cache_read_input_tokens}")

cache_read_input_tokensが増えているほど、キャッシュが有効に機能しています。この値を定期的にログに記録しておくと、効果測定が容易になります。

キャッシュ戦略の選択：どのデータを、どのレイヤーで保持するか

キャッシュには「APIレイヤー」と「アプリケーションレイヤー」の2段構えが基本です。

APIレイヤーのキャッシュ（プレフィックスキャッシュ）

前述のcache_controlがこれに当たります。システムプロンプトや固定のコンテキスト（マニュアル文書、コードベースのスナップショットなど）が長い場合に特に効果的です。

アプリケーションレイヤーのキャッシュ（セマンティックキャッシュ）

完全一致するプロンプトだけでなく、意味的に近い質問に対して同じ回答を返す仕組みです。RedisやFAISSなどを使って、入力テキストの埋め込みベクトルを保存・検索します。

import hashlib
import json
import redis

r = redis.Redis(host="localhost", port=6379, decode_responses=True)

def get_cache_key(prompt: str, model: str) -> str:
    """プロンプトとモデル名からキャッシュキーを生成"""
    content = json.dumps({"prompt": prompt, "model": model}, ensure_ascii=False)
    return f"llm_cache:{hashlib.sha256(content.encode()).hexdigest()}"

def cached_llm_call(prompt: str, model: str = "claude-opus-4-5", ttl: int = 3600):
    key = get_cache_key(prompt, model)
    cached = r.get(key)
    if cached:
        return json.loads(cached)  # キャッシュヒット

    # キャッシュミス → 実際のAPI呼び出し
    response = call_llm_api(prompt, model)
    r.setex(key, ttl, json.dumps(response))
    return response

TTL（有効期限）の設定は用途によって変わります。ニュース要約なら短く（数分〜1時間）、商品説明のような変化の少ない内容なら長く（1日〜1週間）設定するのが妥当です。

レート制限の実装パターン：バックオフ戦略とキューイング

APIプロバイダーはレート制限（Rate Limit）を設けており、超過するとエラーが返ります。適切な実装があれば、制限に引っかかってもサービスを止めずに済みます。

エクスポネンシャルバックオフ

import time
import random
from anthropic import RateLimitError

def call_with_retry(prompt: str, max_retries: int = 5):
    for attempt in range(max_retries):
        try:
            return call_llm_api(prompt)
        except RateLimitError:
            if attempt == max_retries - 1:
                raise
            # ジッターを加えたエクスポネンシャルバックオフ
            wait = (2 ** attempt) + random.uniform(0, 1)
            print(f"Rate limit hit. Waiting {wait:.1f}s before retry {attempt + 1}")
            time.sleep(wait)

ジッター（ランダムな待機時間の揺らぎ）を加えることで、複数のワーカーが同時にリトライしてさらに負荷をかける「thundering herd」問題を防げます。

トークンバケットによるクライアント側スロットリング

エラーが出てから対処するのではなく、あらかじめリクエスト頻度を制御する方法です。

import threading
import time

class TokenBucket:
    def __init__(self, rate: float, capacity: float):
        self.rate = rate          # 1秒あたりの補充量
        self.capacity = capacity  # バケットの最大容量
        self.tokens = capacity
        self.lock = threading.Lock()
        self.last_refill = time.monotonic()

    def acquire(self, tokens: float = 1.0) -> bool:
        with self.lock:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(self.capacity, self.tokens + elapsed * self.rate)
            self.last_refill = now

            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False

    def wait_and_acquire(self, tokens: float = 1.0):
        while not self.acquire(tokens):
            time.sleep(0.05)

# 1秒あたり最大2リクエストに制限する例
bucket = TokenBucket(rate=2.0, capacity=5.0)

def rate_limited_call(prompt: str):
    bucket.wait_and_acquire()
    return call_llm_api(prompt)

バッチ処理や非同期ワーカーを複数立てる場合は、このバケットをRedisなどの共有ストレージで管理すると、プロセス間でレートを均等に分散できます。

開発環境でのコスト検証：モック・ダミーレスポンスの使い分け

開発中にLLM APIを毎回呼び出すのは、コストの無駄であり、テストの再現性も下がります。環境変数などのフラグでモックと本番を切り替える仕組みを早めに用意しましょう。

import os

USE_MOCK_LLM = os.getenv("USE_MOCK_LLM", "false").lower() == "true"

def call_llm_api(prompt: str) -> dict:
    if USE_MOCK_LLM:
        # 固定のダミーレスポンスを返す（ユニットテスト向け）
        return {
            "content": "これはモックレスポンスです。",
            "usage": {"input_tokens": 0, "output_tokens": 0}
        }
    # 実際のAPI呼び出し
    return _real_api_call(prompt)

テスト戦略としては、次の3段階を使い分けるのが現実的です。

フェーズ方法用途
ユニットテスト固定モックロジックのテスト
統合テスト録画済みレスポンス（VCR）APIとの接続確認
本番前検証実API＋小規模サンプルコスト・品質の最終確認

pytest-recordingやvcrpyなどのライブラリを使うと、一度取得したAPIレスポンスをカセットとして保存し、以降はそれを再生する形で統合テストが回せます。

本番環境への段階的な移行と監視ポイント

コスト削減の施策は実装して終わりではなく、本番環境で実際に機能しているかを継続的に監視する必要があります。

監視すべきメトリクス

  • キャッシュヒット率：低すぎる場合、TTLの見直しやキャッシュキー設計の改善が必要
  • 月次トークン消費量の推移：急増のアラートを設定しておく
  • レート制限エラーの頻度：急増時はバックオフ設定やバケット容量を見直す
  • 平均レイテンシ：キャッシュ追加後に遅延が増えていないか確認

段階的な移行の流れ

本番移行は一気に切り替えるのではなく、フィーチャーフラグでトラフィックの一部（例：10%）だけを新しい実装に流しながら比較する方法が安全です。

import random

def should_use_cached_path(user_id: str, rollout_rate: float = 0.1) -> bool:
    """ユーザーIDに基づいて一貫したフラグを返す（同じユーザーは常に同じ経路）"""
    hash_val = int(hashlib.md5(user_id.encode()).hexdigest(), 16)
    return (hash_val % 100) < (rollout_rate * 100)

キャッシュ導入後にレスポンス品質が下がっていないかを人間またはLLM自体で評価するフローも、できれば自動化しておくと長期的な運用が楽になります。コスト削減と品質維持は両立できますが、そのためには数字で追える仕組みが不可欠です。

#llm-api #cost-optimization #caching-strategy #rate-limiting