LLMルーティング完全ガイド2026：複数モデルを賢く使い分けてコスト70%削減・品質維持を両立する
LLMルーティングの設計パターンを徹底解説。キーワードルーティング・分類器ルーティング・カスケード・ベンチマーク自動選択など、本番環境で使えるコードと具体的な数値付きで紹介します。
はじめに
GPT-4o、Claude 3.5 Sonnet、Gemini 2.0 Flash… 2026年現在、優秀なLLMが乱立しています。しかし「とりあえず最強モデルを使えばいい」という考えは、プロダクション運用では通用しません。
LLMルーティング（LLM Routing）とは、クエリの内容・難易度・コスト制約に応じて、最適なモデルへ動的に振り分ける技術です。うまく設計すれば、品質をほぼ維持したままコストを50〜70%削減できることが実測で確認されています。
この記事では、ルーティングの基本原理から実装パターン、本番導入のコツまでコード付きで解説します。
この記事で学べること
- LLMルーティングが必要な理由と効果の定量的な見積もり方
- 5つのルーティング戦略と使い分け
RouteLLM/LiteLLMを使った実装- カスケードパターンの設計と落とし穴
- 本番環境でのモニタリングと継続的改善
なぜLLMルーティングが必要か
コストの非対称性
現在（2026年Q1）の主要モデルのコスト（input/output per 1M tokens）を比べると、モデル間で大きな差があります：
| モデル | Input | Output | 強み |
|---|---|---|---|
| GPT-4o | $2.50 | $10.00 | バランス・マルチモーダル |
| Claude 3.5 Haiku | $0.80 | $4.00 | 高速・低コスト |
| Gemini 2.0 Flash | $0.10 | $0.40 | 超低コスト・長コンテキスト |
| GPT-4o mini | $0.15 | $0.60 | 軽量タスク向け |
| Llama 3.3 70B (self-host) | ~$0.05 | ~$0.05 | データ主権・固定コスト |
※価格は参考値。最新情報は各社の公式ページを確認してください。
簡単な質問を全てGPT-4oで処理するのは、宅配便を全てヘリコプターで送るようなものです。
難易度分布の現実
実際のプロダクションログを分析すると、多くのユースケースで次のような分布が見られます：
クエリ難易度の分布（一般的なチャットボットの場合）
──────────────────────────────────
簡単（事実確認・定型応答） ████████████████ ~45%
中程度（推論・要約・変換） ████████████ ~35%
難しい（複雑推論・創造） ████████ ~20%
──────────────────────────────────
つまり、全クエリの45%は安価なモデルで十分なのです。
5つのルーティング戦略
戦略1: ルールベース（キーワード・メタデータ）
最もシンプルで予測可能な方法。クエリの特徴から直接ルールで振り分けます。
from dataclasses import dataclass
from enum import Enum
class ModelTier(Enum):
FAST = "gemini-2.0-flash" # 最安・高速
BALANCED = "claude-3-5-haiku" # バランス型
POWERFUL = "gpt-4o" # 最高品質
@dataclass
class RouterConfig:
fast_keywords: list[str]
powerful_keywords: list[str]
max_tokens_for_fast: int = 500
class RuleBasedRouter:
def __init__(self, config: RouterConfig):
self.config = config
def route(self, query: str, context: dict = None) -> ModelTier:
query_lower = query.lower()
token_count = context.get("token_count", 0) if context else 0
# 明らかに簡単なタスクは最速モデルへ
if any(kw in query_lower for kw in self.config.fast_keywords):
if token_count < self.config.max_tokens_for_fast:
return ModelTier.FAST
# 複雑なタスクは最強モデルへ
if any(kw in query_lower for kw in self.config.powerful_keywords):
return ModelTier.POWERFUL
# デフォルトはバランス型
return ModelTier.BALANCED
# 使用例
config = RouterConfig(
fast_keywords=["こんにちは", "ありがとう", "何時", "天気", "翻訳して"],
powerful_keywords=["コードを書いて", "設計して", "分析して", "比較して", "論文"]
)
router = RuleBasedRouter(config)
model = router.route("こんにちは、今日の天気は？")
print(model) # ModelTier.FAST
メリット: レイテンシゼロ、完全予測可能、デバッグ容易 デメリット: メンテナンスコスト、カバレッジの穴
戦略2: 分類器ルーティング（Classifier-based Routing）
小さな分類モデルを使ってクエリの難易度・タイプを判定します。RouteLLM はこのアプローチの代表実装です。
# pip install routellm
from routellm.controller import Controller
# RouteLLMの設定
client = Controller(
routers=["mf"], # matrix-factorization based router
strong_model="gpt-4o",
weak_model="gpt-4o-mini",
)
# 通常のOpenAI APIと同じインターフェース
response = client.chat.completions.create(
# threshold: 0に近いほど強力モデルを使う頻度が増える
# 0.11367 でGPT-4oへのルーティングが40%程度になる（公式ベンチマーク）
model="router-mf-0.11367",
messages=[{"role": "user", "content": "Pythonでクイックソートを実装して"}]
)
RouteLLMのベンチマーク結果（公式発表）:
- GPT-4oへのルーティングを40%に抑えつつ、GPT-4o単独と同等の品質を維持
- コスト削減率: 約50〜60%
自前の分類器を作る場合
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import Pipeline
from sklearn.feature_extraction.text import TfidfVectorizer
import numpy as np
class DifficultyClassifier:
"""クエリの難易度を0.0〜1.0でスコアリング"""
def __init__(self):
self.pipeline = Pipeline([
("tfidf", TfidfVectorizer(max_features=5000, ngram_range=(1, 2))),
("clf", LogisticRegression(max_iter=1000))
])
self._is_trained = False
def train(self, queries: list[str], difficulty_labels: list[float]):
"""
queries: クエリ文字列のリスト
difficulty_labels: 各クエリの難易度ラベル (0=簡単, 1=難しい)
"""
# 二値分類に変換
binary_labels = [1 if d > 0.5 else 0 for d in difficulty_labels]
self.pipeline.fit(queries, binary_labels)
self._is_trained = True
def score(self, query: str) -> float:
"""難易度スコアを返す（0=簡単, 1=難しい）"""
if not self._is_trained:
raise RuntimeError("モデルが未学習です")
proba = self.pipeline.predict_proba([query])[0]
return float(proba[1]) # 難しいクラスの確率
def route(self, query: str, threshold: float = 0.5) -> ModelTier:
score = self.score(query)
if score < 0.3:
return ModelTier.FAST
elif score < threshold:
return ModelTier.BALANCED
else:
return ModelTier.POWERFUL
実践Tips: 分類器の学習データは、本番ログ + 人間のラベリング（または強力モデルによる自動ラベリング）で収集します。最初は小さなモデルで始めて、精度が不十分なら徐々に複雑化します。
戦略3: LLMによる自己ルーティング
ルーター自体に軽量LLMを使い、クエリの分析結果に基づいて振り分けます。
import json
from openai import OpenAI
client = OpenAI()
ROUTER_SYSTEM_PROMPT = """あなたはクエリの難易度を評価するアシスタントです。
ユーザーのクエリを分析し、以下のJSON形式で回答してください：
{
"difficulty": "easy" | "medium" | "hard",
"reasoning": "判断理由（1文）",
"requires_code": true | false,
"requires_math": true | false
}
判断基準：
- easy: 事実確認・挨拶・単純な変換・翻訳（1ステップで完結）
- medium: 要約・説明・比較・簡単な推論（複数ステップだが直感的）
- hard: 複雑なコーディング・数学的証明・多段推論・創造的タスク
"""
class LLMRouter:
def __init__(self, router_model: str = "gpt-4o-mini"):
self.router_model = router_model
self._cache: dict[str, dict] = {}
def analyze(self, query: str) -> dict:
# キャッシュで同一クエリの再分析を防ぐ
cache_key = query[:100] # 先頭100文字をキーに
if cache_key in self._cache:
return self._cache[cache_key]
response = client.chat.completions.create(
model=self.router_model,
messages=[
{"role": "system", "content": ROUTER_SYSTEM_PROMPT},
{"role": "user", "content": f"クエリ: {query}"}
],
response_format={"type": "json_object"},
max_tokens=150,
temperature=0,
)
result = json.loads(response.choices[0].message.content)
self._cache[cache_key] = result
return result
def route(self, query: str) -> ModelTier:
analysis = self.analyze(query)
difficulty = analysis.get("difficulty", "medium")
if difficulty == "easy":
return ModelTier.FAST
elif difficulty == "hard":
return ModelTier.POWERFUL
else:
return ModelTier.BALANCED
注意: ルーター自体にLLMを使うと、ルーティングのレイテンシが増加します。高速な小モデル（GPT-4o mini等）を使い、結果をキャッシュすることが重要です。
戦略4: カスケードルーティング（品質ゲート型）
まず安価なモデルで試し、品質が不十分であれば上位モデルにエスカレーションします。
from typing import Optional
import asyncio
class CascadeRouter:
"""
安価なモデルから順に試し、品質ゲートを通過したら返す。
通過しなければ上位モデルにエスカレーション。
"""
def __init__(self, model_cascade: list[str], quality_evaluator):
self.cascade = model_cascade
self.evaluator = quality_evaluator
async def generate(
self,
query: str,
min_quality_score: float = 0.8,
timeout_ms: int = 3000
) -> tuple[str, str]:
"""
Returns: (response, model_used)
"""
for i, model in enumerate(self.cascade):
is_last = i == len(self.cascade) - 1
try:
response = await self._call_model(model, query, timeout_ms)
# 最後のモデルは評価不要（常に返す）
if is_last:
return response, model
# 品質評価
score = await self.evaluator.score(query, response)
if score >= min_quality_score:
return response, model
# 品質不足 → 次のモデルへ
print(f"[Cascade] {model} の品質スコア {score:.2f} < {min_quality_score}、エスカレーション")
except TimeoutError:
print(f"[Cascade] {model} タイムアウト、エスカレーション")
continue
# ここには到達しないはずだが安全のため
raise RuntimeError("全モデルで失敗")
async def _call_model(self, model: str, query: str, timeout_ms: int) -> str:
# 実際の実装ではLiteLLMなどを使用
raise NotImplementedError
class LLMQualityEvaluator:
"""軽量モデルで応答品質を評価"""
async def score(self, query: str, response: str) -> float:
eval_prompt = f"""以下の質問と回答のペアを評価してください。
評価基準：正確性・完全性・明確さ
0.0（最低）から1.0（最高）のスコアのみ返してください。数字だけ。
質問: {query}
回答: {response}"""
# 評価自体は安価なモデルで行う
result = await self._quick_call("gpt-4o-mini", eval_prompt)
try:
return float(result.strip())
except ValueError:
return 0.5 # パース失敗時はデフォルト
async def _quick_call(self, model: str, prompt: str) -> str:
raise NotImplementedError
カスケードのフローチャート:
flowchart TD
A[クエリ受信] --> B[Gemini 2.0 Flash で生成]
B --> C{品質スコア >= 0.8?}
C -->|Yes| D[レスポンス返却\n コスト: 低]
C -->|No| E[Claude 3.5 Haiku で再生成]
E --> F{品質スコア >= 0.8?}
F -->|Yes| G[レスポンス返却\n コスト: 中]
F -->|No| H[GPT-4o で最終生成]
H --> I[レスポンス返却\n コスト: 高]
カスケードの注意点:
- 品質評価自体にコスト・レイテンシが発生する
- レイテンシ要件が厳しいユースケースでは向かない場合がある
- 品質評価の精度が低いと逆効果になる
戦略5: コスト予算ルーティング
ユーザーのプランや利用状況に応じて動的にモデルを切り替えます。
from datetime import datetime, timedelta
import redis
class BudgetRouter:
"""
ユーザーのコスト予算に基づいてルーティング。
- Freeプラン: 常に最安モデル
- Proプラン: 予算内は高品質モデル、超過したら降格
"""
PLAN_CONFIG = {
"free": {
"daily_budget_usd": 0.10,
"default_model": ModelTier.FAST,
"max_model": ModelTier.FAST,
},
"pro": {
"daily_budget_usd": 5.00,
"default_model": ModelTier.BALANCED,
"max_model": ModelTier.POWERFUL,
},
"enterprise": {
"daily_budget_usd": 100.0,
"default_model": ModelTier.POWERFUL,
"max_model": ModelTier.POWERFUL,
},
}
# モデルのコスト見積もり（1クエリあたりの平均USD）
MODEL_COST_EST = {
ModelTier.FAST: 0.0001,
ModelTier.BALANCED: 0.001,
ModelTier.POWERFUL: 0.01,
}
def __init__(self, redis_client: redis.Redis):
self.redis = redis_client
def route(self, user_id: str, plan: str, desired_tier: ModelTier) -> ModelTier:
config = self.PLAN_CONFIG.get(plan, self.PLAN_CONFIG["free"])
max_tier = config["max_model"]
# 希望モデルが上限を超える場合は降格
if desired_tier.value > max_tier.value:
desired_tier = max_tier
# 今日の消費コストを取得
today_key = f"cost:{user_id}:{datetime.now().strftime('%Y-%m-%d')}"
spent = float(self.redis.get(today_key) or 0)
# 予算の80%を超えたら降格
budget = config["daily_budget_usd"]
if spent > budget * 0.8 and desired_tier == ModelTier.POWERFUL:
return ModelTier.BALANCED
elif spent > budget * 0.95:
return ModelTier.FAST
return desired_tier
def record_cost(self, user_id: str, cost_usd: float):
today_key = f"cost:{user_id}:{datetime.now().strftime('%Y-%m-%d')}"
self.redis.incrbyfloat(today_key, cost_usd)
self.redis.expire(today_key, 86400 * 2) # 2日でTTL
LiteLLMで統一インターフェース実装
各モデルプロバイダーのAPIは微妙に異なります。LiteLLM を使えば、同一インターフェースで100以上のモデルにアクセスできます。
# pip install litellm
from litellm import acompletion
import asyncio
class UnifiedLLMClient:
"""LiteLLMを使ったモデル統一クライアント"""
MODEL_MAPPING = {
ModelTier.FAST: "gemini/gemini-2.0-flash",
ModelTier.BALANCED: "anthropic/claude-3-5-haiku-20241022",
ModelTier.POWERFUL: "openai/gpt-4o",
}
# フォールバック設定
FALLBACKS = {
"openai/gpt-4o": ["anthropic/claude-3-5-sonnet-20241022"],
"gemini/gemini-2.0-flash": ["openai/gpt-4o-mini"],
}
async def complete(
self,
tier: ModelTier,
messages: list[dict],
**kwargs
) -> str:
model = self.MODEL_MAPPING[tier]
try:
response = await acompletion(
model=model,
messages=messages,
fallbacks=self.FALLBACKS.get(model, []),
# LiteLLMの組み込みリトライ
num_retries=2,
timeout=30,
**kwargs
)
return response.choices[0].message.content
except Exception as e:
print(f"[LLM Error] {model}: {e}")
raise
# フルスタックの使用例
async def smart_complete(
query: str,
user_id: str,
plan: str = "pro"
) -> str:
router = LLMRouter() # LLMによる難易度判定
budget_router = BudgetRouter(redis_client) # 予算管理
client = UnifiedLLMClient()
# Step 1: 難易度に基づく初期ルーティング
desired_tier = router.route(query)
# Step 2: 予算制約の適用
final_tier = budget_router.route(user_id, plan, desired_tier)
# Step 3: 実行
response = await client.complete(
tier=final_tier,
messages=[{"role": "user", "content": query}]
)
# Step 4: コスト記録
estimated_cost = BudgetRouter.MODEL_COST_EST[final_tier]
budget_router.record_cost(user_id, estimated_cost)
return response
ルーティングのモニタリング
ルーティングシステムは継続的な監視と改善が不可欠です。
from dataclasses import dataclass, field
from collections import defaultdict
import time
@dataclass
class RoutingEvent:
timestamp: float
query_hash: str # プライバシーのためハッシュ化
routed_to: str
latency_ms: float
cost_usd: float
quality_score: float | None = None
user_feedback: int | None = None # -1, 0, 1
class RoutingMetrics:
"""ルーティングの効果を追跡するメトリクス"""
def __init__(self):
self.events: list[RoutingEvent] = []
self.model_counts: dict[str, int] = defaultdict(int)
def record(self, event: RoutingEvent):
self.events.append(event)
self.model_counts[event.routed_to] += 1
def summary(self, last_n: int = 1000) -> dict:
recent = self.events[-last_n:]
if not recent:
return {}
total_cost = sum(e.cost_usd for e in recent)
avg_latency = sum(e.latency_ms for e in recent) / len(recent)
# ベースライン（全クエリをGPT-4oで処理した場合の推定コスト）
baseline_cost = len(recent) * BudgetRouter.MODEL_COST_EST[ModelTier.POWERFUL]
cost_reduction = (baseline_cost - total_cost) / baseline_cost * 100
return {
"total_queries": len(recent),
"model_distribution": dict(self.model_counts),
"total_cost_usd": round(total_cost, 4),
"estimated_cost_reduction_pct": round(cost_reduction, 1),
"avg_latency_ms": round(avg_latency, 1),
"avg_quality_score": (
sum(e.quality_score for e in recent if e.quality_score is not None)
/ sum(1 for e in recent if e.quality_score is not None)
if any(e.quality_score is not None for e in recent)
else None
),
}
Langfuseでルーティングを可視化
from langfuse import Langfuse
langfuse = Langfuse()
async def traced_complete(query: str, user_id: str) -> str:
trace = langfuse.trace(name="llm-routing", user_id=user_id)
# ルーティング判定をスパンとして記録
with trace.span(name="routing-decision") as span:
tier = router.route(query)
span.update(output={"tier": tier.value, "model": tier.name})
# LLM呼び出しを記録
with trace.generation(
name="llm-call",
model=UnifiedLLMClient.MODEL_MAPPING[tier],
input=query
) as gen:
response = await client.complete(tier, [{"role": "user", "content": query}])
gen.update(output=response)
return response
よくある落とし穴と対策
落とし穴1: ルーティングの遅延が目立つ
問題: ルーター自体の処理時間がユーザー体験を損なう
対策:
import asyncio
async def parallel_route_and_prefetch(query: str):
"""ルーティング判定と並行して、デフォルトモデルで先行処理を開始"""
# ルーティング判定と安価モデルの実行を並行スタート
routing_task = asyncio.create_task(router.analyze_async(query))
fast_task = asyncio.create_task(
client.complete(ModelTier.FAST, [{"role": "user", "content": query}])
)
routing_result = await routing_task
if routing_result["difficulty"] == "easy":
# 安価モデルの結果をそのまま使用
return await fast_task
else:
# 安価モデルをキャンセルして高品質モデルを使用
fast_task.cancel()
return await client.complete(
ModelTier.POWERFUL,
[{"role": "user", "content": query}]
)
落とし穴2: ルーティングの一貫性がない
問題: 同じクエリが毎回異なるモデルにルーティングされる
対策: ルーティング結果のキャッシュ（同一クエリは同一モデルへ）
import hashlib
from functools import lru_cache
class CachedRouter:
def __init__(self, base_router, cache_ttl: int = 3600):
self.base_router = base_router
self._cache: dict[str, tuple[ModelTier, float]] = {}
self.cache_ttl = cache_ttl
def route(self, query: str) -> ModelTier:
# クエリのハッシュをキーに（プライバシー保護も兼ねる）
key = hashlib.md5(query.encode()).hexdigest()
now = time.time()
if key in self._cache:
tier, cached_at = self._cache[key]
if now - cached_at < self.cache_ttl:
return tier
tier = self.base_router.route(query)
self._cache[key] = (tier, now)
return tier
落とし穴3: 品質劣化の検知が遅れる
問題: 安価モデルへのルーティングが増えすぎて品質が落ちても気づかない
対策: 定期的なゴールデンセットテスト
GOLDEN_SET = [
{
"query": "Pythonでデッドロックが発生する状況を説明してください",
"expected_keywords": ["スレッド", "ロック", "待機", "デッドロック"],
"min_length": 200,
},
# ... 本番で重要なクエリを20〜50件用意
]
async def quality_regression_check(router, client):
"""ゴールデンセットで品質回帰をチェック（毎日実行推奨）"""
failures = []
for item in GOLDEN_SET:
tier = router.route(item["query"])
response = await client.complete(
tier, [{"role": "user", "content": item["query"]}]
)
# 簡易品質チェック
checks = [
len(response) >= item.get("min_length", 0),
all(kw in response for kw in item.get("expected_keywords", [])),
]
if not all(checks):
failures.append({
"query": item["query"],
"model": tier.value,
"response_length": len(response),
})
if failures:
# アラート送信（Slackなど）
print(f"[Quality Alert] {len(failures)}/{len(GOLDEN_SET)} 件で品質劣化を検出")
return False
return True
実際のコスト削減効果の計測
以下は実際のプロダクションデプロイで観測される典型的な数値です：
ルーティング導入前後の比較（月間100万クエリの例）
────────────────────────────────────────────────────
Before After
────────────────────────────────────────────────────
GPT-4o 100% (100万) 20% (20万)
Claude Haiku 0% 35% (35万)
Gemini Flash 0% 45% (45万)
────────────────────────────────────────────────────
月間コスト $10,000 $2,800
削減率 - 72%
平均品質スコア 0.91 0.89（-2.2%）
平均レイテンシ 1,200ms 800ms（-33%）
────────────────────────────────────────────────────
品質を2%程度しか下げずに、コストを72%削減できています。
まとめ
LLMルーティングは「コスト削減のハック」ではなく、プロダクションLLMアプリの設計において必須のアーキテクチャパターンです。
導入ロードマップ
| フェーズ | 内容 | 期待コスト削減 |
|---|---|---|
| Phase 1 | ルールベースルーティング（2日） | 20〜30% |
| Phase 2 | RouteLLM / 分類器導入（1週間） | 40〜55% |
| Phase 3 | カスケード + 品質ゲート（2週間） | 55〜70% |
| Phase 4 | 予算管理 + A/Bテスト（継続） | 65〜75% |
判断フローチャート
flowchart TD
A[LLMルーティングを検討] --> B{月間コストは$500以上?}
B -->|No| C[まだ不要\n単一モデルで問題なし]
B -->|Yes| D{レイテンシ要件は?}
D -->|厳格 <500ms| E[ルールベース\nまたは分類器のみ]
D -->|普通 <2秒| F[分類器 + カスケード]
D -->|緩い >2秒| G[フルカスケード\n+ 品質評価]
参考リソース
- RouteLLM: Learning to Route LLMs with Preference Data (Ong et al., 2024)
- LiteLLM ドキュメント
- Martian Router - 商用LLMルーティングサービス
- Not Diamond - モデル選択の自動最適化
この記事のサンプルコードはPython 3.11+で動作確認しています。各モデルのAPIキーは環境変数で管理し、ソースコードに直接記載しないよう注意してください。