複数AIモデル使用時のAPIコスト最適化方法：開発者向けプレイブック

単一のAIモデルを使用するだけでも十分に高額です。5つ使用すれば、CFOから質問が飛んでくるでしょう。

しかし、直感に反する真実があります：モデルを追加することで総コストを削減できるのです——インテリジェントにルーティングすれば。Notionのチームは、単純なクエリを安価なモデルに、複雑なクエリをフラッグシップモデルにルーティングすることで、AIコストを45%削減しました。Cursorは安価から高価なモデルへのカスケードを使用し、安価なモデルが失敗した場合にのみエスカレーションします。これは単なるコスト削減ではありません。コストエンジニアリングです。

このプレイブックでは、複数のAIモデルにわたるコストを最適化するために必要なパターン、コード、監視について、出力品質を低下させることなく説明します。

---

マルチモデルコスト最適化が異なる理由

単一モデルの最適化は簡単です：プロバイダーを切り替え、プロンプトを圧縮し、レスポンスをキャッシュします。マルチモデル最適化には2つの次元が追加されます：

1. リクエスト時のモデル選択：この特定のプロンプトをどのモデルが処理するか？
2. クロスモデルコスト帰属：Kimi K2.5のコード生成はDeepSeek V4 Flashの12倍のコストに見合う価値があるか？

目標はコスト認識ルーターを構築することです——各リクエストを処理できる最も安価なモデルを選択し、モデル別・タスク種別ごとのコストを追跡し、情報に基づいたトレードオフ判断に必要なデータを可視化するミドルウェアです。

---

パターン1：階層型モデルルーティング（80/20の勝利）

すべての受信プロンプトを3つの階層に分類します。それに応じてルーティングします。

| 階層 | 複雑度 | モデル | コスト/100万トークン | 例 | |------|-----------|--------|---------------|---------| | Hot | 簡単なチャット、FAQ、要約、分類 | GA-Economy, Qwen3-8B | $0.01-$0.125 | 「このメールを要約して」 | | Warm | コードレビュー、分析、コンテンツ生成 | DeepSeek V4 Flash, Qwen3-235B | $0.25-$1.82 | 「このプルリクエストをレビューして」 | | Cold | 複雑な推論、マルチステップエージェント、ビジョン | Kimi K2.5, GLM-5, MiniMax M2.5 | $1.15-$3.00 | 「マイクロサービスアーキテクチャを設計して...」 |

分類器の構築

プロンプトを分類するために別のLLM呼び出しは必要ありません。シンプルなヒューリスティックが90%のケースで機能します：

# router.py — コスト認識型の階層モデルルーティング
from dataclasses import dataclass
from typing import Optional
import re
from openai import OpenAI

@dataclass
class ModelTier:
    name: str
    model_id: str
    cost_per_million: float
    max_tokens: int = 4096

# モデル階層 — すべての価格は global-apis.com から
HOT_MODELS = [
    ModelTier("GA-Economy", "ga-economy", 0.125),
    ModelTier("Qwen3-8B", "qwen3-8b", 0.01),
]

WARM_MODELS = [
    ModelTier("DeepSeek V4 Flash", "deepseek-v4-flash", 0.25),
    ModelTier("Qwen3-235B", "qwen3-235b-a22b", 1.82),
]

COLD_MODELS = [
    ModelTier("GLM-5", "glm-5", 1.92),
    ModelTier("MiniMax M2.5", "minimax-m2.5", 1.15),
    ModelTier("Kimi K2.5", "kimi-k2.5", 3.00),
]

def classify_prompt(prompt: str, context_length: int) -> str:
    """追加のLLM呼び出しなしでプロンプトの複雑度を分類します。"""
    word_count = len(prompt.split())

    # コード生成またはデバッグ
    code_keywords = [
        "write a function", "debug this", "refactor", "implement",
        "fix the bug", "generate code", "write a script"
    ]
    if any(kw in prompt.lower() for kw in code_keywords):
        return "warm" if word_count < 200 else "cold"

    # アーキテクチャ、設計、複雑な推論
    complex_keywords = [
        "design a system", "architecture", "explain how",
        "compare", "trade-offs", "strategy", "roadmap"
    ]
    if any(kw in prompt.lower() for kw in complex_keywords):
        return "cold" if word_count > 100 else "warm"

    # 長いコンテキスト — 通常はより大きなモデルが必要
    if context_length > 8000:
        return "cold"

    # マルチステップまたはエージェント指示
    if re.search(r"step \d|first.*then.*finally", prompt.lower()):
        return "cold" if word_count > 150 else "warm"

    # デフォルト：簡単なチャット、FAQ、分類
    return "hot"

def estimate_cost(model_id: str, prompt_tokens: int, completion_tokens: int) -> float:
    """モデル価格に基づいてリクエストのコストを見積もります。"""
    pricing = {
        "ga-economy": 0.125,
        "qwen3-8b": 0.01,
        "deepseek-v4-flash": 0.25,
        "qwen3-235b-a22b": 1.82,
        "glm-5": 1.92,
        "minimax-m2.5": 1.15,
        "kimi-k2.5": 3.00,
    }
    rate = pricing.get(model_id, 0.50)
    total_tokens = prompt_tokens + completion_tokens
    return (total_tokens / 1_000_000) * rate

ルーターの動作

class CostAwareRouter:
    """最も安価で対応可能なモデルにプロンプトをルーティングします。"""

    def __init__(self, api_key: str, base_url: str = "https://global-apis.com/v1"):
        self.client = OpenAI(api_key=api_key, base_url=base_url)
        self.spend_log: list[dict] = []

    def route(self, prompt: str, context_length: int = 0) -> dict:
        tier = classify_prompt(prompt, context_length)

        if tier == "hot":
            model = HOT_MODELS[0]  # 最も安価なものを優先
        elif tier == "warm":
            model = WARM_MODELS[0]  # 最高のバリューを優先
        else:
            model = COLD_MODELS[0]

        response = self.client.chat.completions.create(
            model=model.model_id,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=model.max_tokens,
        )

        usage = response.usage
        cost = estimate_cost(
            model.model_id,
            usage.prompt_tokens,
            usage.completion_tokens,
        )

        self.spend_log.append({
            "model": model.model_id,
            "tier": tier,
            "prompt_tokens": usage.prompt_tokens,
            "completion_tokens": usage.completion_tokens,
            "cost_usd": cost,
        })

        return {
            "content": response.choices[0].message.content,
            "model": model.model_id,
            "tier": tier,
            "cost": cost,
        }

    def total_spend(self) -> dict:
        """モデル別・階層別にコストを集計します。"""
        by_model = {}
        by_tier = {}
        total = 0.0
        for entry in self.spend_log:
            model = entry["model"]
            tier = entry["tier"]
            cost = entry["cost_usd"]
            by_model[model] = by_model.get(model, 0) + cost
            by_tier[tier] = by_tier.get(tier, 0) + cost
            total += cost
        return {"total": total, "by_model": by_model, "by_tier": by_tier}


# 使用例
router = CostAwareRouter(api_key="a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")

result = router.route("What is the capital of France?")
print(f"Model: {result['model']}, Cost: ${result['cost']:.6f}")
# Model: ga-economy, Cost: $0.000028

result = router.route("Design a distributed caching layer for a microservice system")
print(f"Model: {result['model']}, Cost: ${result['cost']:.6f}")
# Model: glm-5, Cost: $0.001536

print(router.total_spend())
# {'total': 0.001564, 'by_model': {'ga-economy': 2.8e-05, 'glm-5': 0.001536}, ...}

---

パターン2：セマンティックキャッシュ（繰り返しクエリで40〜70%のコスト削減）

キャッシングはマルチモデル構成において単一で最もROIの高い最適化です。2人のユーザーが本質的に同じ質問をした場合、LLMの呼び出しは1回だけ支払えば済みます。

仕組み

User: "What is a Python decorator?"
  → キャッシュをチェック（埋め込み類似度 >0.95）
  → キャッシュ MISS → DeepSeek V4 Flashを呼び出し → キャッシュに保存
  → コスト: $0.0001

User: "Explain Python decorators to me"
  → キャッシュをチェック → 埋め込み類似度 0.97 ✓
  → キャッシュ HIT → キャッシュされたレスポンスを返す
  → コスト: $0（節約）

リファレンス実装

# semantic_cache.py — 埋め込みベースのレスポンスキャッシュ
import hashlib
import json
from typing import Optional
import numpy as np
from openai import OpenAI

class SemanticCache:
    """埋め込み類似度によってLLMレスポンスをキャッシュします。"""

    def __init__(
        self,
        api_key: str,
        base_url: str = "https://global-apis.com/v1",
        similarity_threshold: float = 0.92,
        max_entries: int = 10_000,
    ):
        self.client = OpenAI(api_key=api_key, base_url=base_url)
        self.threshold = similarity_threshold
        self.max_entries = max_entries
        self.cache: dict[str, dict] = {}  # hash → {embedding, response, tokens}

    def _hash(self, text: str) -> str:
        return hashlib.sha256(text.encode()).hexdigest()[:16]

    def _cosine_similarity(self, a: list[float], b: list[float]) -> float:
        a_arr = np.array(a)
        b_arr = np.array(b)
        return float(np.dot(a_arr, b_arr) / (np.linalg.norm(a_arr) * np.linalg.norm(b_arr)))

    def _get_embedding(self, text: str) -> list[float]:
        response = self.client.embeddings.create(
            model="text-embedding-3-small",
            input=text,
        )
        return response.data[0].embedding

    def lookup(self, prompt: str) -> Optional[str]:
        """意味的に類似したクエリがキャッシュされているか確認します。"""
        try:
            query_embedding = self._get_embedding(prompt)
        except Exception:
            return None  # 埋め込み失敗 — キャッシュをスキップ

        for entry in self.cache.values():
            sim = self._cosine_similarity(query_embedding, entry["embedding"])
            if sim >= self.threshold:
                return entry["response"]

        return None

    def store(self, prompt: str, response: str, tokens_used: int):
        """レスポンスをキャッシュします。"""
        try:
            embedding = self._get_embedding(prompt)
        except Exception:
            return

        key = self._hash(prompt)
        self.cache[key] = {
            "embedding": embedding,
            "response": response,
            "tokens": tokens_used,
            "stored_at": __import__("time").time(),
        }

        # LRUエビクション
        if len(self.cache) > self.max_entries:
            oldest = min(self.cache.items(), key=lambda x: x[1]["stored_at"])
            del self.cache[oldest[0]]

    def stats(self) -> dict:
        total_tokens_saved = sum(e["tokens"] for e in self.cache.values())
        total_entries = len(self.cache)
        return {
            "cached_entries": total_entries,
            "tokens_saved": total_tokens_saved,
            "estimated_cost_saved": (total_tokens_saved / 1_000_000) * 0.25,
        }

JavaScript版:

// semantic-cache.js — Node.js版
import OpenAI from "openai";

class SemanticCache {
  constructor(apiKey, baseURL = "https://global-apis.com/v1", threshold = 0.92) {
    this.client = new OpenAI({ apiKey, baseURL });
    this.threshold = threshold;
    this.cache = new Map();
  }

  _cosineSimilarity(a, b) {
    let dot = 0, normA = 0, normB = 0;
    for (let i = 0; i < a.length; i++) {
      dot += a[i] * b[i];
      normA += a[i] * a[i];
      normB += b[i] * b[i];
    }
    return dot / (Math.sqrt(normA) * Math.sqrt(normB));
  }

  async _getEmbedding(text) {
    const res = await this.client.embeddings.create({
      model: "text-embedding-3-small",
      input: text,
    });
    return res.data[0].embedding;
  }

  async lookup(prompt) {
    const queryEmb = await this._getEmbedding(prompt);
    for (const [, entry] of this.cache) {
      if (this._cosineSimilarity(queryEmb, entry.embedding) >= this.threshold) {
        return entry.response;
      }
    }
    return null;
  }

  async store(prompt, response, tokensUsed) {
    const embedding = await this._getEmbedding(prompt);
    this.cache.set(`${Date.now()}-${Math.random()}`, {
      embedding, response, tokens: tokensUsed,
    });
  }
}

---

パターン3：プロンプト圧縮（20〜40%のトークン削減）

短いプロンプト = 少ないトークン = 低コスト。しかし、省略されたプロンプトは品質の低い結果を生みます。解決策：送信前に長いコンテキストを圧縮します。

チェーン要約

長い会話の場合、モデルに渡す前に履歴を要約します：

def compress_conversation(messages: list[dict], router: CostAwareRouter) -> str:
    """会話履歴を要約してトークン使用量を削減します。"""
    if len(messages) < 6:
        # 十分短い — そのまま送信
        return "\n".join(f"{m['role']}: {m['content']}" for m in messages)

    # 古いメッセージを最も安価なモデルで要約
    history_text = "\n".join(
        f"{m['role']}: {m['content']}" for m in messages[:-2]
    )

    result = router.route(
        f"Summarize this conversation in 2-3 sentences, "
        f"preserving key facts and decisions:\n\n{history_text}"
    )
    # 要約を安く保つためhot階層に強制
    summary = result["content"]

    # 要約 + 最近のメッセージを結合
    recent = "\n".join(
        f"{m['role']}: {m['content']}" for m in messages[-2:]
    )
    return f"[Previous conversation summary]: {summary}\n\n{recent}"

---

パターン4：モデル別コスト追跡（計測なくして最適化なし）

計測できないものは最適化できません。軽量なコストトラッカーを構築しましょう：

# cost_tracker.py
import json
import time
from collections import defaultdict

class CostTracker:
    def __init__(self, log_file: str = "llm_costs.jsonl"):
        self.log_file = log_file
        self.session: list[dict] = []

    def record(self, model: str, task_type: str, prompt_tokens: int,
               completion_tokens: int, cost_usd: float, latency_ms: float):
        entry = {
            "timestamp": time.time(),
            "model": model,
            "task": task_type,
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "cost_usd": cost_usd,
            "latency_ms": latency_ms,
        }
        self.session.append(entry)
        with open(self.log_file, "a") as f:
            f.write(json.dumps(entry) + "\n")

    def report(self) -> dict:
        by_model = defaultdict(lambda: {"calls": 0, "tokens": 0, "cost": 0.0})
        by_task = defaultdict(lambda: {"calls": 0, "cost": 0.0})

        for entry in self.session:
            m = entry["model"]
            t = entry["task"]
            by_model[m]["calls"] += 1
            by_model[m]["tokens"] += entry["prompt_tokens"] + entry["completion_tokens"]
            by_model[m]["cost"] += entry["cost_usd"]
            by_task[t]["calls"] += 1
            by_task[t]["cost"] += entry["cost_usd"]

        total = sum(e["cost_usd"] for e in self.session)
        return {
            "total_cost": round(total, 4),
            "total_calls": len(self.session),
            "by_model": dict(by_model),
            "by_task": dict(by_task),
            "avg_cost_per_call": round(total / max(len(self.session), 1), 6),
        }

tracker.report()を毎週実行してください。どのタスクが高価なモデルに値し、どのタスクがそうでないかがすぐに明確になります。

---

パターン5：フォールバックカスケード

最も安価なモデルが失敗することがあります——ハルシネーション、拒否、またはタイムアウト。リクエストを失敗させるのではなく、上位にカスケードします：

def cascade_call(prompt: str, router: CostAwareRouter) -> dict:
    """安価なモデルを最初に試し、失敗時にエスカレーションします。"""
    # コスト順
    model_sequence = [
        ("ga-economy", 0.125),
        ("deepseek-v4-flash", 0.25),
        ("glm-5", 1.92),
    ]

    for model_id, rate in model_sequence:
        try:
            response = router.client.chat.completions.create(
                model=model_id,
                messages=[{"role": "user", "content": prompt}],
                max_tokens=2048,
                timeout=15,
            )
            content = response.choices[0].message.content

            # 拒否または空のレスポンスをチェック
            if not content or len(content.strip()) < 5:
                continue

            return {
                "content": content,
                "model": model_id,
                "cost": (response.usage.total_tokens / 1_000_000) * rate,
                "escalated": model_id != "ga-economy",
            }
        except Exception:
            continue

    raise RuntimeError("All models failed for prompt")

カスケードにより、常にレスポンスが得られます——可能な限り安価に、必要な場合のみ高価に。

---

すべてを統合：マルチモデルオプティマイザー

# optimizer.py — すべてのパターンを組み合わせます
class MultiModelOptimizer:
    def __init__(self, api_key: str, base_url: str = "https://global-apis.com/v1"):
        self.router = CostAwareRouter(api_key, base_url)
        self.cache = SemanticCache(api_key, base_url, similarity_threshold=0.92)
        self.tracker = CostTracker()

    def call(self, prompt: str, task_type: str = "general") -> dict:
        # ステップ1：キャッシュをチェック
        cached = self.cache.lookup(prompt)
        if cached:
            self.tracker.record("cache", task_type, 0, 0, 0, 0)
            return {"content": cached, "model": "cache", "cost": 0, "cached": True}

        # ステップ2：最も安価で対応可能なモデルにルーティング
        result = self.router.route(prompt)

        # ステップ3：レスポンスをキャッシュ
        tokens = result.get("prompt_tokens", 0) + result.get("completion_tokens", 0)
        self.cache.store(prompt, result["content"], tokens)

        # ステップ4：コストを追跡
        self.tracker.record(
            result["model"], task_type,
            result.get("prompt_tokens", 0),
            result.get("completion_tokens", 0),
            result["cost"],
            result.get("latency_ms", 0),
        )

        return {**result, "cached": False}

    def dashboard(self) -> dict:
        return {
            "spend": self.router.total_spend(),
            "cache": self.cache.stats(),
            "usage": self.tracker.report(),
        }


# 使用例
optimizer = MultiModelOptimizer(api_key="a1b2c3d4e5f6a7b8c9d0e1f2a3b4c5d6")

# これらはすべてオプティマイザーを通過します
queries = [
    ("What is 2+2?", "math"),
    ("Write a Python function to parse JSON", "code"),
    ("Design a notification system for 1M users", "architecture"),
]

for query, task in queries:
    result = optimizer.call(query, task)
    print(f"[{task}] model={result['model']} cost=${result['cost']:.6f} cached={result['cached']}")

# 週次コストレポート
print(json.dumps(optimizer.dashboard(), indent=2))

---

実際のコスト Before/After

月間50,000件の会話を処理するカスタマーサポートチャットボットを運営する典型的なSaaSスタートアップの場合：

| 戦略 | 月間コスト | 節約額 | |----------|-------------|---------| | 全クエリ → GPT-4o | $3,200 | — | | 全クエリ → DeepSeek V4 Flash | $180 | GPT-4o比94%削減 | | 階層型ルーティングのみ | $95 | 単一モデル比47%削減 | | + セマンティックキャッシュ（35%ヒット率） | $62 | さらに35%削減 | | + プロンプト圧縮 | $48 | さらに23%削減 | | GPT-4o比の総削減額 | $3,152 | 98.5% |

重要な洞察：最適化は一度きりの切り替えではありません。計測、ルーティング、キャッシング、圧縮の継続的なプロセスであり、各層が前の層の上に積み重なっていきます。

---

今日から最適化を始めましょう

このプレイブックのパターンは、あらゆるOpenAI互換APIで動作します。複数のAIモデルを管理している場合、Global APIを使用すると、DeepSeek、Qwen、GLM、Kimi、MiniMaxなどに単一のAPIキーでアクセスでき、$0.01/100万トークンからの定額料金で利用できます。

• 100無料クレジットで登録 — クレジットカード不要
• 全料金プランを表示 — 全モデル、定額制、透明
• APIドキュメント — OpenAI互換、ドロップイン置き換え

1つのAPIキー、すべてのモデル、無駄ゼロ。