LabHub

ブログ

LLMルーティング・カスケード戦略：マルチモデルオーケストレーションによるコスト最適化

2026-03-08 · 37 分で読めます · #llm#routing#cascade#model-orchestration#cost-optimization#multi-model#2026-03

한국어English日本語

LLM Routing

はじめに

LLMベースのサービスを運用していると、すべてのクエリを最高性能のモデル（GPT-4o、Claude Opusなど）へ送るのが最も安全な選択に見える。しかし現実は違う。プロダクショントラフィックの60~80%は「今日の天気を教えて」「このテキストを要約して」といった単純なクエリで構成されており、こうしたリクエストに最上位モデルを使うのはコストの無駄だ。GPT-4oのトークン単価はGPT-4o-mini比で約30倍以上、Claude OpusはHaiku比で約60倍高い。

この問題を解決する中核戦略がLLMルーティング（Routing）とカスケード（Cascade）だ。クエリの複雑度、ドメイン、要求品質をリアルタイムに分析して最適なモデルへ振り分けるか、低コストのモデルから順に試して品質基準を満たせば即座に返す方式である。RouteLLMのベンチマークによると、学習済みルーターを使えばGPT-4比で85%のコスト削減を達成しながら、95%の応答品質を維持できる。

この記事ではLLMルーティングの中核概念と主要なアプローチ（RouteLLM、FrugalGPT、Semantic Router、Martian、Not Diamond）を比較し、プロダクションで即座に適用できるマルチモデルオーケストレーションアーキテクチャをコードとともに構築する。また、運用中に発生する失敗パターンと復旧戦略、コスト最適化チェックリストまで網羅的に扱う。

LLMルーティングとは何か

ルーティングの定義と必要性

LLMルーティングは、ユーザークエリを分析して最も適したモデルへリクエストを渡す決定レイヤーである。ネットワークルーターがパケットを最適経路へ転送するように、LLMルーターは各クエリをコスト・品質・レイテンシのバランスが最も良いモデルへ振り分ける。

ルーティングが必要な根本的な理由は、モデル間のコスト性能格差が大きいからだ。単純な分類タスクにGPT-4oを使えば精度はわずかに上がるが、コストは数十倍に増える。逆に、複雑な推論が必要なタスクに小型モデルを使うと品質低下によって再試行が必要になり、かえって総コストが増えることがある。

ルーティング vs カスケード vs アンサンブル

三つのアプローチはそれぞれ異なる戦略を取る。

ルーティング（Routing）：クエリを分析したうえで単一のモデルを選び、一度だけ呼び出す。レイテンシのオーバーヘッドが低く、実装も比較的シンプルだ。ルーターの判断が外れると品質低下がすぐに現れるという欠点がある。

カスケード（Cascade）：低コストのモデルから順に呼び出し、応答品質を評価して基準に届かなければ上位モデルへエスカレーションする。品質保証には有利だが、平均レイテンシが増えることがある。

アンサンブル（Ensemble）：複数のモデルを同時に呼び出して応答を統合する。品質は最も高いが、コストとレイテンシがどちらも増えるため、医療や法律など高い信頼性が求められるドメインでのみ使うのが現実的だ。

プロダクションでは、ルーティングとカスケードを混合適用するのが最も効果的だ。ルーターが1次分類を行い、ルーターの信頼度が低い場合にのみカスケードへフォールバックする構成である。

主要なルーティングアプローチの比較

アプローチ別の詳細比較テーブル

アプローチルーティング方式コスト削減率品質維持率レイテンシオーバーヘッド実装難易度適したシナリオ
RouteLLM学習済み分類器(MF/BERT/SW)~85% (MT Bench)~95% GPT-4低い (5~15ms)中Strong/Weak 2モデルルーティング
FrugalGPTカスケード + 品質判定器~50~75%~90~95%高い (逐次呼び出し)高多段モデルパイプライン
Semantic Router埋め込み類似度ベース~40~60%~90%非常に低い (2~5ms)低ドメイン別ルーティング、Tool選択
Martianメタモデルの挙動予測~30~60%~95%低い低 (SaaS)エンタープライズのマルチモデル
Not Diamondメタモデル + 200+ モデル~30~50%~95%+低い低 (SaaS)最適モデルの自動選択
xRouter強化学習ベース~60~80%~93~96%低い高コスト制約下の最適化
ルールベースキーワード/正規表現~30~50%可変ほぼなし低MVP、初期導入

RouteLLM: 学習ベースのルーター

RouteLLMはLMSYSが開発したオープンソースのルーティングフレームワークだ。Chatbot Arenaの選好データを活用して、「このクエリに対して強いモデル（GPT-4o）が弱いモデル（GPT-4o-mini）より良い応答をする確率」を予測する分類器を学習する。

四種類のルーターを提供する。MF（Matrix Factorization）ルーターはクエリ埋め込みとモデル特性を行列分解して勝率を予測する。SW（Similarity-Weighted）ルーターは類似クエリの過去の勝率を加重平均する。BERTルーターはBERT分類器で直接ルーティングを学習する。Causal LLMルーターはLLM自体をファインチューニングしてルーティング判断を行う。

# RouteLLMサーバーの実行とクライアントからの利用例
# pip install routellm

# 1. サーバーを実行 (OpenAI API 互換)
# python -m routellm.openai_server \
#   --routers mf \
#   --strong-model gpt-4o \
#   --weak-model gpt-4o-mini

# 2. クライアントから利用
import openai

client = openai.OpenAI(
    base_url="http://localhost:6060/v1",
    api_key="not-needed",  # RouteLLM サーバー側で管理
)

# ルーターが strong/weak モデルを自動で選択
# threshold 値でルーティングの感度を調整 (0.0~1.0)
response = client.chat.completions.create(
    model="router-mf-0.11593",  # router-{ルーター名}-{threshold}
    messages=[
        {"role": "user", "content": "量子コンピューティングにおける量子ビットの誤り訂正手法を説明して"}
    ],
)

# threshold が低いほど strong モデルの使用比率が上がる
# 0.5 → ~50% strong モデルを使用
# 0.1 → ~90% strong モデルを使用 (品質優先)
# 0.9 → ~10% strong モデルを使用 (コスト優先)
print(response.choices[0].message.content)
print(f"使用されたモデル: {response.model}")

FrugalGPT: カスケードによるコスト最適化

FrugalGPTはスタンフォード大学が提案したアプローチで、モデルカスケードと品質判定器（Quality Estimator）を組み合わせる。中核となる考え方は、最も安いモデルから試し、応答品質が十分なら即座に返すという点だ。

動作の流れは次の通りである。クエリが入ってくると、まず最も低コストのモデル（例: GPT-4o-mini）へ渡す。応答を品質判定器が評価し、スコアが閾値以上なら返す。届かなければ次の上位モデル（例: Claude Sonnet）へ同じクエリを渡し、再び品質判定を行う。最上位モデル（例: GPT-4o）まで到達したら無条件で返す。

# FrugalGPTスタイルのカスケード実装
from openai import OpenAI
from anthropic import Anthropic
import time
from dataclasses import dataclass
from typing import Optional

@dataclass
class ModelTier:
    name: str
    provider: str
    cost_per_1k_tokens: float
    quality_threshold: float  # このスコア以上なら応答を返す

# コストの昇順でモデルティアを定義
MODEL_CASCADE = [
    ModelTier("gpt-4o-mini", "openai", 0.00015, 0.7),
    ModelTier("claude-3-5-haiku-20241022", "anthropic", 0.001, 0.8),
    ModelTier("claude-sonnet-4-20250514", "anthropic", 0.003, 0.85),
    ModelTier("gpt-4o", "openai", 0.005, 0.0),  # 最終段: 常に返す
]

openai_client = OpenAI()
anthropic_client = Anthropic()


def call_model(model: ModelTier, query: str) -> str:
    """モデルプロバイダーに応じて適切なAPIを呼び出す"""
    if model.provider == "openai":
        resp = openai_client.chat.completions.create(
            model=model.name,
            messages=[{"role": "user", "content": query}],
            temperature=0.3,
        )
        return resp.choices[0].message.content
    elif model.provider == "anthropic":
        resp = anthropic_client.messages.create(
            model=model.name,
            max_tokens=2048,
            messages=[{"role": "user", "content": query}],
        )
        return resp.content[0].text


def estimate_quality(query: str, response: str) -> float:
    """応答品質の判定器 - 軽量モデルで品質スコアを返す"""
    judge_prompt = f"""次の質問と回答の品質を0.0~1.0のスコアで評価してください。
評価基準: 正確性、完全性、関連性、明確性
質問: {query}
回答: {response}
数値のみを返してください (例: 0.85)"""

    resp = openai_client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": judge_prompt}],
        temperature=0.0,
        max_tokens=10,
    )
    try:
        return float(resp.choices[0].message.content.strip())
    except ValueError:
        return 0.0  # パース失敗時は低スコア → 上位モデルへエスカレーション


def frugal_cascade(query: str) -> dict:
    """FrugalGPTカスケードを実行する"""
    results = []

    for tier in MODEL_CASCADE:
        start = time.time()
        response = call_model(tier, query)
        latency = time.time() - start

        # 最終ティアは品質判定を省略
        if tier.quality_threshold == 0.0:
            return {
                "response": response,
                "model": tier.name,
                "latency": latency,
                "cascade_depth": len(results) + 1,
                "total_cost_ratio": sum(r["cost"] for r in results) + tier.cost_per_1k_tokens,
            }

        quality = estimate_quality(query, response)

        results.append({
            "model": tier.name,
            "quality": quality,
            "cost": tier.cost_per_1k_tokens,
            "latency": latency,
        })

        if quality >= tier.quality_threshold:
            return {
                "response": response,
                "model": tier.name,
                "quality_score": quality,
                "latency": latency,
                "cascade_depth": len(results),
            }

    # 理論上は到達しない (最後のティアで無条件に返すため)
    return {"response": response, "model": MODEL_CASCADE[-1].name}

Semantic Router: 埋め込みベースの超高速ルーティング

Semantic RouterはAurelio Labsが開発したライブラリで、クエリの意味的な類似度をもとにルーティングを決定する。LLMを呼び出さず、埋め込みベクトル間のコサイン類似度だけで判断するため、レイテンシのオーバーヘッドは2~5ms程度と極めて低い。

# Semantic Routerを活用したドメイン別モデルルーティング
# pip install semantic-router

from semantic_router import Route, RouteLayer
from semantic_router.encoders import OpenAIEncoder

# ルート定義: 各ルートに代表発話(utterance)を指定
simple_route = Route(
    name="simple",
    utterances=[
        "今日の天気はどう?",
        "ソウルの人口は?",
        "こんにちは",
        "この単語の意味は?",
        "1+1は?",
        "現在時刻を教えて",
    ],
)

coding_route = Route(
    name="coding",
    utterances=[
        "Pythonでクイックソートを実装して",
        "Reactコンポーネントの useEffect のメモリリークを解決",
        "Kubernetes Pod の OOMKilled をデバッグ",
        "SQLクエリの最適化方法",
        "gRPCとREST APIの性能比較コード",
    ],
)

reasoning_route = Route(
    name="reasoning",
    utterances=[
        "この論文の方法論を批判的に分析して",
        "GDP成長率と失業率の相関関係を経済学的に説明して",
        "量子もつれと量子テレポーテーションの違いを数学的に導出して",
        "RLHFとDPOの理論的限界を比較分析して",
    ],
)

# エンコーダーとルートレイヤーの初期化
encoder = OpenAIEncoder(name="text-embedding-3-small")
route_layer = RouteLayer(
    encoder=encoder,
    routes=[simple_route, coding_route, reasoning_route],
)

# ルートに応じたモデルマッピング
MODEL_MAP = {
    "simple": "gpt-4o-mini",        # 低コストモデル
    "coding": "claude-sonnet-4-20250514",  # コーディング特化
    "reasoning": "gpt-4o",          # 高性能な推論
    None: "claude-sonnet-4-20250514",      # 既定のフォールバック
}


def route_query(query: str) -> dict:
    """クエリを分析して最適なモデルを決定する"""
    route_result = route_layer(query)
    selected_model = MODEL_MAP.get(route_result.name, MODEL_MAP[None])

    return {
        "query": query,
        "route": route_result.name,
        "confidence": route_result.similarity_score,
        "model": selected_model,
    }


# 使用例
queries = [
    "Pythonでリストをソートする方法",
    "トランスフォーマーアーキテクチャのアテンション機構を数学的に説明して",
    "明日の釜山の天気",
]

for q in queries:
    result = route_query(q)
    print(f"クエリ: {q}")
    print(f"  ルート: {result['route']} → モデル: {result['model']}")
    print(f"  信頼度: {result['confidence']:.3f}")

プロダクションのマルチモデルオーケストレーションアーキテクチャ

システム全体の構造

プロダクション環境でマルチモデルオーケストレーションを構築するときは、単純なルーティングを超えて可観測性（Observability）、フォールバック（Fallback）、キャッシング、レートリミットなどを統合する必要がある。以下はプロダクション水準のオーケストレーション層の設計だ。

ユーザーリクエスト → API Gateway → セマンティックキャッシュ検索
                              ↓ (キャッシュミス)
                         クエリ分類器 (複雑度/ドメイン分析)
                              ↓
                    ┌─────────┼─────────┐
                    ↓         ↓         ↓
                小型モデル  中型モデル  大型モデル
              (GPT-4o-mini) (Sonnet)  (GPT-4o/Opus)
                    ↓         ↓         ↓
                    └─────────┼─────────┘
                              ↓
                         品質ゲート (カスケード判定)
                              ↓
                    ┌── 品質を満たす → 応答を返す + キャッシュ保存
                    └── 品質が不足   → 上位モデルへエスカレーション
                              ↓
                         メトリクス収集 (コスト/レイテンシ/品質)

TypeScriptベースのオーケストレーションエンジン

// multi-model-orchestrator.ts
import OpenAI from 'openai'
import Anthropic from '@anthropic-ai/sdk'

interface ModelConfig {
  id: string
  provider: 'openai' | 'anthropic'
  costPer1kInput: number
  costPer1kOutput: number
  maxTokens: number
  tier: 'small' | 'medium' | 'large'
}

interface RoutingDecision {
  model: ModelConfig
  reason: string
  confidence: number
}

interface OrchestratorResult {
  response: string
  model: string
  tier: string
  latencyMs: number
  estimatedCost: number
  cascadeDepth: number
  cacheHit: boolean
}

// モデルカタログの定義
const MODEL_CATALOG: ModelConfig[] = [
  {
    id: 'gpt-4o-mini',
    provider: 'openai',
    costPer1kInput: 0.00015,
    costPer1kOutput: 0.0006,
    maxTokens: 16384,
    tier: 'small',
  },
  {
    id: 'claude-3-5-haiku-20241022',
    provider: 'anthropic',
    costPer1kInput: 0.001,
    costPer1kOutput: 0.005,
    maxTokens: 8192,
    tier: 'small',
  },
  {
    id: 'claude-sonnet-4-20250514',
    provider: 'anthropic',
    costPer1kInput: 0.003,
    costPer1kOutput: 0.015,
    maxTokens: 8192,
    tier: 'medium',
  },
  {
    id: 'gpt-4o',
    provider: 'openai',
    costPer1kInput: 0.005,
    costPer1kOutput: 0.015,
    maxTokens: 16384,
    tier: 'large',
  },
]

class QueryClassifier {
  private openai: OpenAI

  constructor(openai: OpenAI) {
    this.openai = openai
  }

  async classify(query: string): Promise<RoutingDecision> {
    // ルールベースの1次分類 (LLM 呼び出しなしで即座に判断)
    const ruleResult = this.ruleBasedClassify(query)
    if (ruleResult) return ruleResult

    // 軽量モデルで2次分類
    const resp = await this.openai.chat.completions.create({
      model: 'gpt-4o-mini',
      messages: [
        {
          role: 'system',
          content: `クエリの複雑度を分類してください。
JSONで応答: {"tier": "small|medium|large", "reason": "...", "confidence": 0.0~1.0}
- small: 単純な質問、挨拶、翻訳、要約
- medium: コーディング、分析、比較、構造化された説明
- large: 複合推論、数学的証明、マルチステップ分析、創造的な長文`,
        },
        { role: 'user', content: query },
      ],
      response_format: { type: 'json_object' },
      temperature: 0.0,
      max_tokens: 100,
    })

    const parsed = JSON.parse(resp.choices[0].message.content || '{}')
    const tier = parsed.tier || 'medium'
    const model = MODEL_CATALOG.find((m) => m.tier === tier) || MODEL_CATALOG[2]

    return {
      model,
      reason: parsed.reason || 'LLM分類器の判断',
      confidence: parsed.confidence || 0.5,
    }
  }

  private ruleBasedClassify(query: string): RoutingDecision | null {
    const len = query.length

    // 非常に短いクエリ → small
    if (len < 30) {
      return {
        model: MODEL_CATALOG[0],
        reason: '短いクエリ (ルールベース)',
        confidence: 0.9,
      }
    }

    // コード関連のキーワード → medium
    const codeKeywords = /\b(コード|実装|関数|クラス|デバッグ|エラー|API|SQL|React|Python)\b/i
    if (codeKeywords.test(query)) {
      return {
        model: MODEL_CATALOG[2], // claude-sonnet
        reason: 'コーディング関連のクエリ (ルールベース)',
        confidence: 0.8,
      }
    }

    // 複雑な推論のキーワード → large
    const reasoningKeywords = /\b(証明|分析|比較.*違い|数学的|論理的|戦略.*策定|アーキテクチャ.*設計)\b/
    if (reasoningKeywords.test(query)) {
      return {
        model: MODEL_CATALOG[3], // gpt-4o
        reason: '複合推論のクエリ (ルールベース)',
        confidence: 0.75,
      }
    }

    return null // ルールでは判断不可 → LLM分類器を使用
  }
}

セマンティックキャッシュの統合

同一または類似のクエリに対して以前の応答を再利用すれば、コストを劇的に減らせる。厳密な文字列一致ではなく埋め込みベースの類似度検索を使えば、「Pythonのリストをソートする方法」と「パイソンでリストはどうやってソートしますか」を同一のクエリとして認識できる。

# セマンティックキャッシュの実装 (Redis + ベクトル類似度)
import hashlib
import json
import time
import numpy as np
from openai import OpenAI
from redis import Redis

client = OpenAI()
redis_client = Redis(host="localhost", port=6379, db=0)

CACHE_TTL = 3600  # 1時間
SIMILARITY_THRESHOLD = 0.92  # 類似度の閾値


def get_embedding(text: str) -> list[float]:
    """テキストの埋め込みベクトルを生成する"""
    resp = client.embeddings.create(
        model="text-embedding-3-small",
        input=text,
    )
    return resp.data[0].embedding


def cosine_similarity(a: list[float], b: list[float]) -> float:
    """コサイン類似度を計算する"""
    a_np, b_np = np.array(a), np.array(b)
    return float(np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np)))


class SemanticCache:
    def __init__(self, namespace: str = "llm_cache"):
        self.namespace = namespace

    def _cache_key(self, idx: int) -> str:
        return f"{self.namespace}:entry:{idx}"

    def _counter_key(self) -> str:
        return f"{self.namespace}:counter"

    def get(self, query: str) -> dict | None:
        """類似クエリのキャッシュ済み応答を検索する"""
        query_embedding = get_embedding(query)
        counter = int(redis_client.get(self._counter_key()) or 0)

        best_match = None
        best_similarity = 0.0

        for i in range(counter):
            entry_raw = redis_client.get(self._cache_key(i))
            if not entry_raw:
                continue

            entry = json.loads(entry_raw)
            similarity = cosine_similarity(query_embedding, entry["embedding"])

            if similarity > best_similarity and similarity >= SIMILARITY_THRESHOLD:
                best_similarity = similarity
                best_match = entry

        if best_match:
            return {
                "response": best_match["response"],
                "model": best_match["model"],
                "similarity": best_similarity,
                "cached_at": best_match["timestamp"],
            }

        return None

    def put(self, query: str, response: str, model: str):
        """応答をキャッシュに保存する"""
        embedding = get_embedding(query)
        counter = int(redis_client.get(self._counter_key()) or 0)

        entry = {
            "query": query,
            "response": response,
            "model": model,
            "embedding": embedding,
            "timestamp": time.time(),
        }

        redis_client.setex(
            self._cache_key(counter),
            CACHE_TTL,
            json.dumps(entry),
        )
        redis_client.incr(self._counter_key())

コスト最適化の深掘り戦略

トークンコスト分析フレームワーク

コスト最適化の第一歩は、現在のコスト構造を正確に把握することだ。モデル別、機能別、時間帯別のコストを追跡し、最適化の余地が大きい領域を特定する必要がある。

# コスト追跡・分析システム
from dataclasses import dataclass, field
from collections import defaultdict
from datetime import datetime, timedelta
import json


@dataclass
class TokenUsage:
    model: str
    input_tokens: int
    output_tokens: int
    timestamp: datetime
    route: str  # どのルートに分類されたか
    cascade_depth: int = 1
    quality_score: float = 0.0


class CostAnalyzer:
    # 2026年3月時点の主要モデル価格 (USD per 1K tokens)
    PRICING = {
        "gpt-4o": {"input": 0.0025, "output": 0.01},
        "gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
        "claude-opus-4-20250514": {"input": 0.015, "output": 0.075},
        "claude-sonnet-4-20250514": {"input": 0.003, "output": 0.015},
        "claude-3-5-haiku-20241022": {"input": 0.001, "output": 0.005},
    }

    def __init__(self):
        self.usage_log: list[TokenUsage] = []

    def log(self, usage: TokenUsage):
        self.usage_log.append(usage)

    def calculate_cost(self, usage: TokenUsage) -> float:
        """単件のコスト計算"""
        pricing = self.PRICING.get(usage.model)
        if not pricing:
            return 0.0

        input_cost = (usage.input_tokens / 1000) * pricing["input"]
        output_cost = (usage.output_tokens / 1000) * pricing["output"]
        return input_cost + output_cost

    def daily_report(self, date: datetime = None) -> dict:
        """日次コストレポートを生成する"""
        date = date or datetime.now()
        day_start = date.replace(hour=0, minute=0, second=0)
        day_end = day_start + timedelta(days=1)

        day_logs = [
            u for u in self.usage_log
            if day_start <= u.timestamp < day_end
        ]

        model_costs = defaultdict(float)
        route_costs = defaultdict(float)
        total_cost = 0.0
        total_requests = len(day_logs)

        for usage in day_logs:
            cost = self.calculate_cost(usage)
            model_costs[usage.model] += cost
            route_costs[usage.route] += cost
            total_cost += cost

        # ルーティングなしで全リクエストをGPT-4oへ送った場合のコスト推定
        baseline_cost = sum(
            (u.input_tokens / 1000) * self.PRICING["gpt-4o"]["input"]
            + (u.output_tokens / 1000) * self.PRICING["gpt-4o"]["output"]
            for u in day_logs
        )

        return {
            "date": date.strftime("%Y-%m-%d"),
            "total_requests": total_requests,
            "total_cost_usd": round(total_cost, 4),
            "baseline_cost_usd": round(baseline_cost, 4),
            "savings_pct": round((1 - total_cost / baseline_cost) * 100, 1) if baseline_cost > 0 else 0,
            "cost_by_model": dict(model_costs),
            "cost_by_route": dict(route_costs),
            "avg_cost_per_request": round(total_cost / total_requests, 6) if total_requests > 0 else 0,
        }

プロンプト圧縮とトークン節約

ルーティング以外にも、プロンプト自体を最適化してトークン使用量を減らせる。システムプロンプトの圧縮、不要なコンテキストの削除、出力トークンの制限などを適用すれば、同じモデルでも20~40%のコストを削減できる。

主な手法は次の通りである。システムプロンプトのキャッシング：AnthropicのPrompt Caching機能を活用すれば、繰り返されるシステムプロンプトのコストを90%削減できる。LLMLinguaによるプロンプト圧縮：元のプロンプトから重要度の低いトークンを取り除き、2~5倍に圧縮する。出力長の制御：max_tokensをタスクに合わせて制限し、簡潔な応答を促すプロンプトを書く。

失敗パターンと復旧戦略

主な失敗シナリオ

1. ルーターの誤分類 (Misrouting)

ルーターが複雑なクエリを単純なクエリと誤って分類し、小型モデルへ送ってしまうと品質が大きく低下する。逆に、単純なクエリを複雑だと判断すれば不要なコストが発生する。

復旧戦略: ユーザーフィードバックのループを構築して誤分類の事例を集め、ルーターを定期的に再学習する。信頼度が低い分類結果（confidence < 0.6）については、自動的に中間ティアのモデルを選ぶ安全装置を置く。

2. カスケードのレイテンシ爆発

カスケード方式では、低コストのモデルが連続して品質基準を満たせないとすべてのティアを巡回することになり、レイテンシが急増する。4段のカスケードで各段に1~2秒かかると、最悪の場合8秒以上の応答時間になる。

復旧戦略: カスケードの最大深度を制限する（通常は2~3段）。全体のタイムアウトを設定し、タイムアウト発生時はそれまでで最善の応答を返す。カスケード深度の分布を監視し、平均深度が1.5を超えたらルーターを再較正する。

3. プロバイダー障害 (Provider Outage)

特定のモデルプロバイダー（OpenAI、Anthropic、Googleなど）に障害が発生すると、そのプロバイダーを使うすべてのルーティング経路が停止する。

復旧戦略: プロバイダー別のヘルスチェックを実装し、障害を検知したら同一ティアの代替モデルへ自動でフォールバックする。サーキットブレーカーパターンを適用し、連続失敗時はそのプロバイダーを一時的に遮断する。

4. 品質判定器のドリフト

FrugalGPTの品質判定器が時間とともに不正確になっていく現象である。モデルの更新、データ分布の変化、判定器自体のバイアスなどが原因だ。

復旧戦略: 定期的に人間の評価者のラベルと判定器のスコアを比較してドリフトを検知する。判定器モデルも定期的に再学習するか、A/Bテストで判定器のバージョンを比較する。

サーキットブレーカーパターンの実装

# プロバイダー別サーキットブレーカーの実装
import time
from enum import Enum
from threading import Lock


class CircuitState(Enum):
    CLOSED = "closed"      # 正常状態
    OPEN = "open"          # 遮断状態
    HALF_OPEN = "half_open"  # 試験状態


class CircuitBreaker:
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 60,
        half_open_max_calls: int = 3,
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.half_open_max_calls = half_open_max_calls

        self.state = CircuitState.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = 0.0
        self.half_open_calls = 0
        self._lock = Lock()

    def can_execute(self) -> bool:
        """現在の状態でリクエストを実行できるか"""
        with self._lock:
            if self.state == CircuitState.CLOSED:
                return True

            if self.state == CircuitState.OPEN:
                # 復旧タイムアウト経過時に HALF_OPEN へ遷移
                if time.time() - self.last_failure_time >= self.recovery_timeout:
                    self.state = CircuitState.HALF_OPEN
                    self.half_open_calls = 0
                    return True
                return False

            if self.state == CircuitState.HALF_OPEN:
                return self.half_open_calls < self.half_open_max_calls

            return False

    def record_success(self):
        """成功を記録する"""
        with self._lock:
            if self.state == CircuitState.HALF_OPEN:
                self.success_count += 1
                if self.success_count >= self.half_open_max_calls:
                    self.state = CircuitState.CLOSED
                    self.failure_count = 0
                    self.success_count = 0
            elif self.state == CircuitState.CLOSED:
                self.failure_count = 0

    def record_failure(self):
        """失敗を記録する"""
        with self._lock:
            self.failure_count += 1
            self.last_failure_time = time.time()

            if self.state == CircuitState.HALF_OPEN:
                self.state = CircuitState.OPEN  # 再び遮断
            elif self.failure_count >= self.failure_threshold:
                self.state = CircuitState.OPEN


# プロバイダー別サーキットブレーカーの管理
provider_circuits: dict[str, CircuitBreaker] = {
    "openai": CircuitBreaker(failure_threshold=5, recovery_timeout=60),
    "anthropic": CircuitBreaker(failure_threshold=5, recovery_timeout=60),
    "google": CircuitBreaker(failure_threshold=3, recovery_timeout=120),
}

# 同一ティアのフォールバックマッピング
FALLBACK_MAP = {
    "gpt-4o-mini": ["claude-3-5-haiku-20241022"],
    "claude-3-5-haiku-20241022": ["gpt-4o-mini"],
    "claude-sonnet-4-20250514": ["gpt-4o"],
    "gpt-4o": ["claude-sonnet-4-20250514"],
}


def get_available_model(
    primary_model: str,
    primary_provider: str,
) -> tuple[str, str]:
    """利用可能なモデルを返す (サーキットブレーカーの確認を含む)"""
    # 1次: 基本モデルのプロバイダーを確認
    if provider_circuits[primary_provider].can_execute():
        return primary_model, primary_provider

    # 2次: フォールバックモデルを探索
    fallbacks = FALLBACK_MAP.get(primary_model, [])
    for fb_model in fallbacks:
        fb_provider = "anthropic" if "claude" in fb_model else "openai"
        if provider_circuits[fb_provider].can_execute():
            return fb_model, fb_provider

    # 3次: すべてのプロバイダーが障害の場合はエラー
    raise RuntimeError(
        f"すべてのプロバイダーのサーキットが OPEN 状態。 "
        f"primary={primary_model}, fallbacks={fallbacks}"
    )

トラブルシューティングガイド

ルーティング品質低下の診断

症状：ユーザー不満の増加、再試行率の上昇、特定のクエリ種別での品質急落

診断手順は次の通りである。

  1. ルーティング分布の確認：各モデルティアのトラフィック比率が想定範囲にあるかを確認する。小型モデルの比率が80%を超えるなら、ルーターが過度にコスト最適化されている可能性が高い。

  2. 誤分類率の測定：ランダムサンプリングで100~500件のクエリを人間の評価者が再分類し、ルーターの分類結果と比較する。誤分類率が15%を超えるならルーターの再学習が必要だ。

  3. ティア別の品質スコア比較：各ティアで処理されたクエリの平均品質スコアを比較する。小型モデルティアの平均スコアが0.7未満なら閾値の調整が必要だ。

  4. カスケード深度の分析：平均カスケード深度が1.5を超えるなら、1次ルーティングの精度が低いという意味だ。

コスト急増の原因分析

症状：日次コストが突然2倍以上に増加

確認事項は次の通りである。

  1. トラフィック急増：全体のリクエスト量が増えていないか確認する。
  2. ルーティング分布の移動：大型モデルの比率が急増していないか確認する。ルーターの更新やクエリ分布の変化が原因のことがある。
  3. キャッシュヒット率の低下：キャッシュTTLの失効、キャッシュサーバーの障害、新しい種別のクエリの流入などを確認する。
  4. カスケードループ：品質判定器の誤作動ですべてのクエリが最上位モデルまでエスカレーションされていないか確認する。
  5. プロンプトの肥大化：システムプロンプトやコンテキストが異常に大きくなっていないか確認する。

レイテンシの最適化

ルーティングレイヤー自体のレイテンシを最小化することが重要だ。ルーターに50ms以上かかると、ユーザーの体感性能に影響する。

最適化の方法は次の通りである。ルールベースの分類を1次で行い、70%のクエリを即座にルーティングする（レイテンシ < 1ms）。残りの30%についてのみ埋め込みベースまたはLLMベースの分類を行う。分類器モデルは可能ならローカルにデプロイしてネットワークレイテンシを取り除く。Semantic Routerの埋め込み演算はバッチ処理して処理量を高める。

運用上の注意点

コストと品質のトレードオフ管理

ルーティング戦略を導入すると、コストと品質のあいだに継続的な緊張が生じる。コストを積極的に削れば品質が下がり、品質を上げればルーティングによるコスト削減効果が減る。

運用原則は次の通りである。

モデル更新への対応

LLMプロバイダーがモデルを更新（GPT-4o → GPT-4o-2024-11-20 など）すると、ルーターの学習データと現在のモデル挙動が食い違うことがある。モデル更新時は次の手順に従う。

  1. ステージング環境で新しいモデルバージョンのベンチマークを実施する。
  2. ルーターのモデルプロファイル（コスト、性能特性）を更新する。
  3. A/Bテストで既存のルーティングと新モデルベースのルーティングを比較する。
  4. 有意な差があればルーターを再学習する。

マルチプロバイダーのAPIキー管理

複数のプロバイダーを使うとAPIキー管理の複雑度が増す。環境変数、シークレットマネージャー、キーのローテーションなどを体系的に管理する必要がある。

// 安全なマルチプロバイダーAPIキー管理
// provider-config.ts
import { SecretManagerServiceClient } from '@google-cloud/secret-manager'

interface ProviderCredentials {
  apiKey: string
  orgId?: string
  rateLimit: number // RPM
  lastRotated: Date
}

class ProviderKeyManager {
  private secretClient: SecretManagerServiceClient
  private cache: Map<string, ProviderCredentials> = new Map()
  private cacheTTL = 300_000 // 5分
  private lastFetch: Map<string, number> = new Map()

  constructor() {
    this.secretClient = new SecretManagerServiceClient()
  }

  async getCredentials(provider: string): Promise<ProviderCredentials> {
    const now = Date.now()
    const lastFetched = this.lastFetch.get(provider) || 0

    // キャッシュが有効ならキャッシュから返す
    if (now - lastFetched < this.cacheTTL && this.cache.has(provider)) {
      return this.cache.get(provider)!
    }

    // Secret Manager からキーを取得
    const secretName = `projects/my-project/secrets/llm-${provider}-key/versions/latest`
    const [version] = await this.secretClient.accessSecretVersion({
      name: secretName,
    })

    const apiKey = version.payload?.data?.toString() || ''
    const credentials: ProviderCredentials = {
      apiKey,
      rateLimit: this.getDefaultRateLimit(provider),
      lastRotated: new Date(),
    }

    this.cache.set(provider, credentials)
    this.lastFetch.set(provider, now)

    return credentials
  }

  private getDefaultRateLimit(provider: string): number {
    const limits: Record<string, number> = {
      openai: 500,
      anthropic: 300,
      google: 200,
    }
    return limits[provider] || 100
  }
}

プロダクションチェックリスト

導入前チェックリスト

実装チェックリスト

運用チェックリスト

性能目標のガイドライン

指標目標値危険閾値
コスト削減率 (baseline比)40~70%< 20%
品質維持率 (baseline比)> 93%< 88%
ルーターレイテンシ< 15ms> 50ms
キャッシュヒット率25~40%< 10%
平均カスケード深度< 1.3> 1.8
プロバイダー可用性> 99.5%< 98%
誤分類率< 10%> 20%

xRouterと強化学習ベースのルーティングの未来

最近の研究で注目すべきアプローチがxRouterだ。xRouterはルーティングを逐次的な意思決定問題として定式化し、強化学習（RL）でルーターを学習する。既存の方式がクエリとモデルの静的なマッピングを学習するのに対し、xRouterはコスト予算の制約下でセッション全体の累積報酬を最大化する方策を学習する。

このアプローチの要点は、コストを明示的な制約としてモデリングするという点にある。「総コスト $X 以下で品質を最大化せよ」という目標を直接最適化するため、予算の範囲内で自動的にコストと品質のトレードオフを調整する。コスト予算に余裕があるときは大型モデルを積極的に活用し、予算が尽きかけているときは小型モデルをより多く使うという適応的な振る舞いを見せる。

Pick and Spinフレームワークも注目に値する。KubernetesベースのセルフホストLLM環境で、適応的な scale-to-zero 自動化とハイブリッドルーティングモジュールを組み合わせ、静的デプロイ比で21.6%高い成功率、30%低いレイテンシ、33%低いGPUコストを達成した。

エージェンティックAIワークフローでは、ルーティングの重要性がさらに高まる。エージェントが複数のステップを順に実行する構造では各ステップごとに誤りが積み重なるため、ステップ別に最適なモデルを選ぶルーティングが全体の成功率に決定的な影響を与える。MartianのExpert Orchestration AI Architectureは、「判事(judge)」モデルが「専門家(expert)」モデルの能力を評価し、ルーターが最も信頼できる専門家へクエリを割り当てる構造を提案する。

実戦導入シナリオ別ガイド

シナリオ1: SaaSチャットボットサービス (月間コスト $5,000 → $1,500 が目標)

最も一般的な導入シナリオだ。カスタマーサポートのチャットボットでは、全クエリの70%がFAQ水準の単純な質問であることが多い。

推奨戦略は次の通りである。Semantic RouterでFAQ/一般/専門の3つのルートを定義する。FAQはキャッシュ + GPT-4o-mini、一般はClaude Haiku、専門はClaude Sonnetへルーティングする。セマンティックキャッシュを適用し、FAQの繰り返しクエリのキャッシュヒット率50%以上を目標にする。この構成で一般的に60~70%のコスト削減を達成できる。

シナリオ2: コードレビューツール (品質 > コスト)

コード分析は精度が要になるため、積極的なコスト削減よりも品質保証に焦点を合わせる。

推奨戦略は次の通りである。ルールベースの分類で、変更行数とファイル種別に応じてルーティングする。10行未満の単純な変更はGPT-4o-miniへ、50行以上の複合的な変更やセキュリティ関連ファイルはGPT-4oまたはClaude Opusへルーティングする。カスケードは2段に制限し、最初のモデルの応答に「確信なし」「追加レビューが必要」といった不確実性のマーカーがあれば上位モデルへエスカレーションする。

シナリオ3: RAGパイプライン (大量文書処理)

RAGシステムでは、検索された文書chunkの数とクエリの複雑度によって必要なモデル性能が変わる。

推奨戦略は次の通りである。検索結果が1~2個のchunkで直接回答できる場合はGPT-4o-miniで処理する。5個以上のchunkを統合する必要がある場合や、比較・分析が必要な場合は中型モデルを使う。複数文書間の衝突解決や推論が必要な場合は大型モデルを使う。文書要約と埋め込み生成は常に小型モデルで処理する。

参考資料

原文を読む: https://www.youngju.dev/blog/llm/2026-03-08-llm-routing-cascade-multi-model-orchestration.ja

コメント

まだコメントはありません。

ログインするとコメントできます