LLMシステムの費用最適化:コストが実際にどこに発生しているか

本当に重要な箇所にトークンを集中させましょう。

目次

LLM(大規模言語モデル)のコストは使用量に比例して線形に増加します。1日10,000リクエストを処理し、1リクエストあたりのコストが$0.01の場合、日額コストは$100、年間では$365になります。エンタープライズ規模では、これは1万ドルを超えます。

コスト最適化とは、切り捨てを行うことではありません。重要箇所においてトークンを適切に配分することです。

無駄遣いされた1トークンは、より良い回答を得るために使えたはずの1トークンです。

LLM cost optimization strategies

トークンの予算管理

コストを制御する最も簡単な方法は、制限を設定することです。セッションごと、タスクごと、または日次で設定します。

戦略1: セッションごとの予算

セッションごとの予算は直截的です:

class SessionBudget:
    def __init__(self, budget_tokens: int = 10000):
        self.budget = budget_tokens
        self.used = 0

    def allocate(self, tokens: int) -> bool:
        if self.used + tokens <= self.budget:
            self.used += tokens
            return True
        return False

    def remaining(self) -> int:
        return self.budget - self.used

戦略2: タスクごとの予算

タスクごとの予算はより実用的です。異なるタスクには異なる量のコンテキストが必要です:

task_budgets:
  classify:
    max_tokens: 100
    model: qwen2.5-1.5b
  summarize:
    max_tokens: 500
    model: qwen2.5-7b
  code_review:
    max_tokens: 2000
    model: qwen2.5-coder-7b
  reason:
    max_tokens: 4000
    model: qwen2.5-32b

戦略3: 適応型予算

適応型予算は、実際に発生した使用量に基づいて調整されます。分類タスクが常に80トークンを使用している場合、100トークンを割り当てるのを止めます:

class AdaptiveBudget:
    def __init__(self):
        self.task_history = {}

    def allocate(self, task_type: str) -> int:
        if task_type in self.task_history:
            return int(self.task_history[task_type] * 1.5)
        return 1000

    def record(self, task_type: str, tokens_used: int):
        if task_type not in self.task_history:
            self.task_history[task_type] = tokens_used
        else:
            self.task_history[task_type] = (
                0.9 * self.task_history[task_type] + 0.1 * tokens_used
            )

指数移動平均(重み0.9)を用いることで、過去の履歴よりも最近の使用量が重視されます。ワークロードの変動性に応じて重みを調整してください。

APIとローカル推論

大規模な利用では、ローカル推論の方がコストが安くなります。損益分岐点は、ハードウェアとAPIのレートに依存します。

モデル API ($/Mトークン) ローカルコスト/時間 損益分岐点
GPT-4o $2.50 / $10.00 N/A
Claude Sonnet 4 $3.00 / $15.00 N/A
Qwen2.5-72B $0.50 / $2.00 ~$0.50 ~1日4時間
Qwen2.5-32B $0.30 / $1.20 ~$0.20 ~1日2時間
Qwen2.5-7B $0.10 / $0.40 ~$0.05 ~1日1時間

ハードウェアに関する計算:

ハードウェア 初期投資 月額電気代 APIとの損益分岐点
RTX 3090 (中古) $600 $15 ~4ヶ月
RTX 4090 $1,500 $20 ~6ヶ月
RTX 5080 $1,000 $18 ~5ヶ月
DGX Spark $2,000 $30 ~8ヶ月

中程度の利用(1日1時間以上)であれば、ローカル推論で初期投資を回収できます。高頻度利用の場合は、節約額は劇的になります。ただし、課題は初期の資本支出です。RTX 5080は$1,000します。APIの請求は一時停止できますが、ハードウェアはそうはいきません。

フォールバック戦略

希望するモデルが高すぎるか、低速すぎる場合、より安価なモデルにフォールバックします。重要なのは、品質が「十分」と判断できるタイミングを知ることです。

戦略1: 品質ベースのフォールバック

品質ベースのフォールバックは、出力が閾値を満たすまでモデルを試行します:

class QualityFallback:
    def __init__(self, quality_threshold: float = 0.8):
        self.threshold = quality_threshold
        self.models = [
            {"model": "claude-sonnet-4", "cost": 0.015},
            {"model": "qwen2.5-72b", "cost": 0.002},
            {"model": "qwen2.5-32b", "cost": 0.001},
            {"model": "qwen2.5-7b", "cost": 0.0004},
        ]

    def route(self, prompt: str) -> str:
        for model_config in self.models:
            result = self.call_model(model_config["model"], prompt)
            if self.evaluate_quality(result) >= self.threshold:
                return result
        return self.call_model(self.models[0]["model"], prompt)

問題は評価そのものです。別のモデルを呼び出さずに、どのように品質を測定するのでしょうか?一部のシステムでは小さな分類器を使用します。他のシステムでは、長さ、構造、キーワードの存在などのヘuristikなチェックを用います。これらは完璧ではありません。

戦略2: 遅延ベースのフォールバック

遅延ベースのフォールバックはシンプルです。時間予算を満たす最も高速なモデルへルーティングします:

class LatencyFallback:
    def __init__(self, max_latency: float = 5.0):
        self.max_latency = max_latency
        self.models = [
            {"model": "qwen2.5-1.5b", "latency": 0.5},
            {"model": "qwen2.5-7b", "latency": 2.0},
            {"model": "qwen2.5-32b", "latency": 10.0},
            {"model": "claude-sonnet-4", "latency": 5.0},
        ]

    def route(self, prompt: str) -> str:
        for model_config in sorted(self.models, key=lambda x: x["latency"]):
            if model_config["latency"] <= self.max_latency:
                return self.call_model(model_config["model"], prompt)
        return self.call_model(self.models[0]["model"], prompt)

キャッシング

キャッシングは、最も評価されていないコスト最適化手法です。同一のプロンプトは想像以上に頻繁に発生します。分類リクエスト、FAQスタイルのクエリ、繰り返されるツール呼び出しなどです。

戦略1: プロンプトキャッシング

完全一致のプロンプトキャッシングはシンプルです:

import hashlib

class PromptCache:
    def __init__(self, max_size: int = 1000):
        self.cache = {}
        self.max_size = max_size

    def get(self, prompt: str) -> str | None:
        key = hashlib.sha256(prompt.encode()).hexdigest()
        return self.cache.get(key)

    def set(self, prompt: str, response: str):
        key = hashlib.sha256(prompt.encode()).hexdigest()
        if len(self.cache) >= self.max_size:
            self.cache.pop(next(iter(self.cache)))
        self.cache[key] = response

戦略2: 意味的キャッシング

意味的キャッシングはより実用的です。意味は同じだが、テキストが異なるプロンプトをキャッチします:

from sentence_transformers import SentenceTransformer

class SemanticCache:
    def __init__(self, similarity_threshold: float = 0.95):
        self.model = SentenceTransformer('all-MiniLM-L6-v2')
        self.cache = {}
        self.threshold = similarity_threshold

    def get(self, prompt: str) -> str | None:
        prompt_embedding = self.model.encode([prompt])[0]
        for cached_prompt, cached_response in self.cache.items():
            cached_embedding = self.model.encode([cached_prompt])[0]
            similarity = self.cosine_similarity(
                prompt_embedding, cached_embedding
            )
            if similarity >= self.threshold:
                return cached_response
        return None

    def set(self, prompt: str, response: str):
        self.cache[prompt] = response

閾値は重要です。0.95は厳格で、非常に類似したプロンプトのみが一致します。0.85は寛容ですが、誤った回答を返すリスクがあります。ミスレートを測定し、調整してください。

一般的なクエリに対するレスポンスキャッシングも価値があります。ユーザーが「天気は何ですか」や「何時ですか」を繰り返し質問する場合、正確なプロンプトだけでなく、パターンをキャッシュします:

class ResponseCache:
    def __init__(self):
        self.common_queries = {
            "what is the weather": "Check weather API",
            "what is the time": "Check system time",
            "who is the president": "Check current president",
        }

    def get(self, query: str) -> str | None:
        query_lower = query.lower()
        for common_query, response in self.common_queries.items():
            if common_query in query_lower:
                return response
        return None

これは高度ではありませんが、有効です。一般的なクエリが一般的であるには理由があります。

最適化が効果的な場面

最適化は、高容量の処理、混合ワークロードの実行、または累積するAPIコストを支払っている場合に重要です。

プロトタイピング中、単一のモデルを使用している、または低容量の処理を行っている場合は、重要ではありません。1日100リクエストしか行わないシステムにとって、予算管理、フォールバック、キャッシングの複雑さはそれに見合う価値がありません。

まず基本的なフローを動作させましょう。請求書が届いた時点で最適化を追加します。

トレードオフ

戦略 コスト 品質 複雑さ
最適化なし 最高 一定 最低
トークン予算管理 中程度 変動 中程度
フォールバックモデル 低〜中 変動 中程度
キャッシング 最低 高(キャッシュヒット時) 中程度
ハイブリッド 最適化済み 最適化済み 最高

本番環境のシステムは通常、ハイブリッド方式で動作します。セッションごとに予算を設定し、品質または遅延に基づいてフォールバックし、可能な限りキャッシングを行います。複雑さは確かに存在しますが、節約効果もまた現実的なものです。

関連項目

購読する

システム、インフラ、AIエンジニアリングの新記事をお届けします。