モデル間の負荷分散
安価なプロンプトを小規模モデルへ、難しいプロンプトを大規模モデルへルーティングします。
「モデル間の負荷分散」はCoddyKit上の無料AI Prompt Engineeringレッスンです。 これはレッスン3/4です。 下記で完全なレッスンを無料で読むことができます。その後、ブラウザ内の組み込みコードエディタと24時間対応のAIチューターでハンズオン演習できます。 これはAI Prompt Engineering学習パスの一部であり、ウェブとCoddyKitアプリ全体で進捗が同期されます。 AI Prompt Engineeringコースには全4レッスンが含まれています。
モデル間でルーティングする理由
すべてのタスクに最も高性能（かつ高価）なモデルが必要なわけではありません。単純なあいさつへの応答にGPT-4oは必要ありません。モデルルーティングでは、各リクエストを適切に処理できる最も安価なモデルへ振り分けます。これにより、重要な場面で品質を維持しながらコストを50～90%削減できます。
複雑度に基づくルーティング
APIを呼び出す前に、タスクの複雑度を分類します。単純なタスクは安価なモデルに、複雑なタスクは高性能なモデルに振り分けます。軽量な分類器やヒューリスティックを使えば、この判断を素早く行えます。
COMPLEXITY_CLASSIFIER_PROMPT = '''Classify the complexity of this user request.
Return ONLY one word: SIMPLE, MODERATE, or COMPLEX.
SIMPLE: greeting, factual lookup, single-step question, direct answer needed
MODERATE: multi-step explanation, comparison, short analysis, code snippet
COMPLEX: deep analysis, long code generation, reasoning chain, specialized domain
Request: {request}'''
import openai
client_mini = openai.OpenAI(api_key='YOUR_API_KEY')
def classify_complexity(request):
response = client_mini.chat.completions.create(
model='gpt-4o-mini', # always use cheap model for classifier
messages=[{'role': 'user', 'content':
COMPLEXITY_CLASSIFIER_PROMPT.format(request=request)}],
max_tokens=5,
temperature=0
)
label = response.choices[0].message.content.strip().upper()
if label not in ('SIMPLE', 'MODERATE', 'COMPLEX'):
label = 'MODERATE' # safe default
return label
for req in ['Hi', 'Explain quicksort', 'Design a distributed systems architecture']:
print(f'{req[:40]}: {classify_complexity(req)}')モデルルータークラス
モデルルーターは複雑度のラベルをモデルに対応付け、それに応じて各リクエストを振り分けます。ルーティングの判断は決定論的で、設定可能なしきい値に基づきます。
MODEL_ROUTES = {
'SIMPLE': {
'model': 'gpt-4o-mini',
'max_tokens': 300,
'cost_per_1k_input': 0.00015,
'use_case': 'greetings, FAQ, simple factual questions'
},
'MODERATE': {
'model': 'gpt-4o',
'max_tokens': 1500,
'cost_per_1k_input': 0.0025,
'use_case': 'explanations, analysis, code snippets'
},
'COMPLEX': {
'model': 'claude-opus-4-5',
'max_tokens': 4096,
'cost_per_1k_input': 0.015,
'use_case': 'deep reasoning, long code, specialized domains'
}
}
class ModelRouter:
def __init__(self):
self.routes = MODEL_ROUTES
self.call_counts = {k: 0 for k in MODEL_ROUTES}
def route(self, request, messages):
complexity = classify_complexity(request)
route = self.routes[complexity]
self.call_counts[complexity] += 1
print(f'Routing "{request[:40]}" -> {route["model"]} ({complexity})')
return route['model'], route['max_tokens']
def cost_report(self):
total = sum(self.call_counts.values())
for complexity, count in self.call_counts.items():
pct = count / total * 100 if total else 0
print(f'{complexity}: {count} calls ({pct:.0f}%)')コストを考慮したルーティング
複雑度だけでなく、コストを考慮したルーティングでは、トークン予算、ユーザー層（無料と有料）、1日の支出上限も考慮し、システム全体でコストを予測可能にします。
class CostAwareRouter(ModelRouter):
def __init__(self, daily_budget_usd=100.0):
super().__init__()
self.daily_budget = daily_budget_usd
self.daily_spent = 0.0
def estimate_cost(self, model, input_tokens, max_output_tokens):
route = next((r for r in self.routes.values() if r['model'] == model), None)
if not route:
return 0.0
return (
(input_tokens / 1000) * route['cost_per_1k_input'] +
(max_output_tokens / 1000) * route['cost_per_1k_input'] * 3
)
def route_with_budget(self, request, messages, user_tier='free'):
complexity = classify_complexity(request)
# Downgrade if budget is exhausted or user is on free tier
budget_remaining = self.daily_budget - self.daily_spent
if budget_remaining < 0.01 or user_tier == 'free':
complexity = 'SIMPLE' # downgrade to cheapest model
print('Budget constraint: routing to SIMPLE model')
route = self.routes[complexity]
input_tokens = sum(len(m['content'].split()) for m in messages) * 1.3
cost = self.estimate_cost(route['model'], input_tokens, route['max_tokens'])
self.daily_spent += cost
return route['model'], route['max_tokens']レイテンシを考慮したルーティング
モデルごとにレイテンシの特性は異なります。時間に制約がある場合（たとえば、3秒のSLAが設定されたチャットボットなど）は、性能が低くても高速なモデルに振り分けます。
import time
# Model latency profiles (approximate P95 values)
MODEL_LATENCY_P95 = {
'gpt-4o-mini': 1.5, # seconds
'gpt-4o': 4.0,
'claude-haiku-4-5': 1.2,
'claude-sonnet-4-5': 3.0,
'claude-opus-4-5': 6.0
}
SLA_LATENCY_BUDGET = 3.0 # seconds
def route_with_latency_constraint(complexity, sla_seconds=SLA_LATENCY_BUDGET):
route = MODEL_ROUTES[complexity]
p95_latency = MODEL_LATENCY_P95.get(route['model'], 5.0)
if p95_latency > sla_seconds:
# Find fastest model under SLA
affordable_models = [
(lat, m) for m, lat in MODEL_LATENCY_P95.items()
if lat <= sla_seconds
]
if affordable_models:
fastest = min(affordable_models)[1]
print(f'Latency constraint: downgrading from {route["model"]} to {fastest}')
return fastest
return route['model']
print('COMPLEX request under 3s SLA:', route_with_latency_constraint('COMPLEX'))機能を考慮したルーティング
タスクによっては、ビジョン、function calling、長いコンテキスト、コードインタープリターなど、特定のモデル機能が必要です。機能を考慮したルーティングにより、選択したモデルが実際にタスクを処理できることを保証します。
MODEL_CAPABILITIES = {
'gpt-4o-mini': {
'vision': True,
'function_calling': True,
'context_window': 128000,
'code_interpreter': False
},
'gpt-4o': {
'vision': True,
'function_calling': True,
'context_window': 128000,
'code_interpreter': True
},
'claude-opus-4-5': {
'vision': True,
'function_calling': True,
'context_window': 200000,
'code_interpreter': False
}
}
def capability_aware_route(required_capabilities, context_length=0):
candidates = []
for model, caps in MODEL_CAPABILITIES.items():
if context_length > caps['context_window']:
continue
if all(caps.get(cap, False) for cap in required_capabilities):
candidates.append(model)
if not candidates:
raise ValueError(f'No model supports: {required_capabilities}')
# Among capable models, pick cheapest
cost_rank = ['gpt-4o-mini', 'claude-opus-4-5', 'gpt-4o']
for model in cost_rank:
if model in candidates:
return model
return candidates[0]
print(capability_aware_route(['vision', 'function_calling'], context_length=5000))フォールバックチェーン
フォールバックチェーンでは、主となるモデルが失敗した場合に試すモデルの順序を定義します。これにより、個々のプロバイダーで障害やレート制限の問題が発生しても、高い可用性を維持できます。
FALLBACK_CHAINS = {
'primary': 'claude-opus-4-5',
'fallback': 'gpt-4o',
'emergency': 'gpt-4o-mini'
}
def call_with_fallback(messages, chain=FALLBACK_CHAINS):
providers = [
('anthropic', chain['primary']),
('openai', chain['fallback']),
('openai', chain['emergency'])
]
for provider, model in providers:
try:
print(f'Trying {model}...')
if provider == 'anthropic':
import anthropic
ac = anthropic.Anthropic(api_key='YOUR_KEY')
resp = ac.messages.create(
model=model, max_tokens=500, messages=messages
)
return resp.content[0].text
else:
import openai
oc = openai.OpenAI(api_key='YOUR_KEY')
resp = oc.chat.completions.create(
model=model, messages=messages, max_tokens=500
)
return resp.choices[0].message.content
except Exception as e:
print(f'{model} failed: {e}. Trying next...')
raise RuntimeError('All models in fallback chain failed')ルーティング判断のログ記録
監査やしきい値の調整、コスト分布の把握に十分なコンテキストを含め、すべてのルーティング判断をログに記録します。このデータは、時間の経過に伴ってルーティングロジックを最適化するために不可欠です。
import json
from datetime import datetime
ROUTING_LOG_FILE = 'routing_decisions.jsonl'
def log_routing_decision(request_id, request_text, complexity,
model_selected, cost_estimate, latency_ms,
user_tier='free'):
entry = {
'timestamp': datetime.utcnow().isoformat(),
'request_id': request_id,
'request_preview': request_text[:50],
'complexity': complexity,
'model': model_selected,
'cost_estimate_usd': round(cost_estimate, 6),
'latency_ms': round(latency_ms),
'user_tier': user_tier
}
with open(ROUTING_LOG_FILE, 'a') as f:
f.write(json.dumps(entry) + '\n')
# Analyze routing log to tune thresholds
def analyze_routing_log():
from collections import Counter
model_counts = Counter()
total_cost = 0.0
with open(ROUTING_LOG_FILE) as f:
for line in f:
e = json.loads(line)
model_counts[e['model']] += 1
total_cost += e['cost_estimate_usd']
print('Model distribution:', dict(model_counts))
print(f'Total estimated cost: ${total_cost:.4f}')本番環境でのモデルのA/Bテスト
モデルルーティングでは、A/Bテストも実施できます。トラフィックの一部を新しいモデルに振り分け、全面展開の前に品質を比較します。モニタリングと組み合わせることで、データに基づいてモデルを選択できます。
import random
class ABModelRouter:
def __init__(self, control_model, treatment_model, treatment_pct=10):
self.control = control_model
self.treatment = treatment_model
self.treatment_pct = treatment_pct
self.assignment_log = {} # request_id: 'control' | 'treatment'
def route(self, request_id):
if request_id in self.assignment_log:
# Sticky assignment: same user always gets same model
return self.assignment_log[request_id]
if random.random() * 100 < self.treatment_pct:
assignment = 'treatment'
model = self.treatment
else:
assignment = 'control'
model = self.control
self.assignment_log[request_id] = assignment
return model, assignment
# Usage
ab_router = ABModelRouter(
control_model='gpt-4o',
treatment_model='claude-opus-4-5',
treatment_pct=10 # 10% get new model
)
for user_id in range(5):
result = ab_router.route(f'user_{user_id}')
print(f'user_{user_id}: {result}')モデルのヘルスチェック
モデルにトラフィックをルーティングする前に、正しく応答していることを確認します。ヘルスチェックでは、既知のプロンプトをモデルに送信して応答を検証し、プロバイダーが利用可能であることを確認します。
import time
def health_check(model, provider='openai', timeout=5):
'''
Returns True if model is healthy, False if timed out or errored.
'''
test_prompt = 'Reply with exactly: OK'
try:
start = time.time()
if provider == 'openai':
import openai
client = openai.OpenAI(api_key='YOUR_API_KEY')
resp = client.chat.completions.create(
model=model,
messages=[{'role': 'user', 'content': test_prompt}],
max_tokens=5,
timeout=timeout
)
text = resp.choices[0].message.content.strip()
elif provider == 'anthropic':
import anthropic
client = anthropic.Anthropic(api_key='YOUR_API_KEY')
resp = client.messages.create(
model=model, max_tokens=5,
messages=[{'role': 'user', 'content': test_prompt}],
)
text = resp.content[0].text.strip()
latency = (time.time() - start) * 1000
healthy = 'ok' in text.lower()
print(f'{model}: {"HEALTHY" if healthy else "DEGRADED"} ({latency:.0f}ms)')
return healthy
except Exception as e:
print(f'{model}: UNHEALTHY ({e})')
return False
# Run health checks before routing critical traffic
# health_check('gpt-4o-mini', provider='openai')
# health_check('claude-haiku-4-5', provider='anthropic')コスト影響分析
モデルルーティングによるコスト削減効果を定量化します。トラフィックがSIMPLE 60%、MODERATE 30%、COMPLEX 10%の場合、インテリジェントなルーティングによって、すべてに最良のモデルを使用する場合と比べてコストを70～80%削減できます。
def cost_impact_analysis(daily_requests=10000):
# Traffic distribution
traffic = {'SIMPLE': 0.60, 'MODERATE': 0.30, 'COMPLEX': 0.10}
# Avg tokens per request (input + output)
avg_tokens = {'SIMPLE': 500, 'MODERATE': 2000, 'COMPLEX': 5000}
# Pricing per 1K tokens (blended input+output)
pricing = {'SIMPLE': 0.00030, 'MODERATE': 0.01000, 'COMPLEX': 0.04500}
premium_price = 0.04500 # if we used COMPLEX model for everything
routed_cost = 0.0
premium_cost = 0.0
for complexity, pct in traffic.items():
requests = daily_requests * pct
tokens = avg_tokens[complexity]
routed_cost += requests * (tokens / 1000) * pricing[complexity]
premium_cost += requests * (tokens / 1000) * premium_price
savings_pct = (1 - routed_cost / premium_cost) * 100
print(f'Daily requests: {daily_requests:,}')
print(f'With routing: ${routed_cost:,.2f}/day')
print(f'Without routing: ${premium_cost:,.2f}/day')
print(f'Savings: {savings_pct:.0f}% (${premium_cost - routed_cost:,.2f}/day)')
cost_impact_analysis()理解度チェック
ユーザーが「Hi, how are you?」と尋ねました。モデルルーターはこれをSIMPLEに分類します。これをgpt-4oではなくgpt-4o-miniにルーティングするのが適切なのはなぜですか。
モデルルーティングのまとめ
モデル間で負荷分散を行うと、コストを削減しながらタスクに適した機能を確保できます：
- 複雑度ルーティング：タスクをSIMPLE/MODERATE/COMPLEXに分類し、対応するモデル層に振り分けます
- コストを考慮したルーティング：予算を使い切った場合や無料ユーザー層の場合は、より低コストのモデルに切り替えます
- レイテンシを考慮したルーティング：SLAが厳しい場合は、より高速なモデルを使用します
- 機能ルーティング：選択したモデルが必要な機能（ビジョン、function calling）をサポートしていることを確認します
- フォールバックチェーン：高い可用性を実現するため、主モデル → フォールバック → 緊急用の順に試します
- A/Bテスト：全面展開の前に、トラフィックの一部で新しいモデルをテストします
- コストへの影響：常に最良のモデルを使用する場合と比べ、ルーティングによってコストを70～80%削減できます
よくある質問
「モデル間の負荷分散」レッスンは無料ですか？
はい。「モデル間の負荷分散」の完全なテキストはこのウェブで無料で読めます。インタラクティブに演習し（組み込みコードエディタと24時間対応のAIチューター）、AI Prompt Engineeringコースの残りをアンロックするには、CoddyKit PROにアップグレードしてください。 AI Prompt Engineeringコースには全4レッスンが含まれています。
「モデル間の負荷分散」で何を学びますか？
安価なプロンプトを小規模モデルへ、難しいプロンプトを大規模モデルへルーティングします。 ブラウザで直接実行するハンズオンコードでAI Prompt Engineeringを演習し、24時間対応のAIチューターがレッスンを進める中での質問に答えます。
AI Prompt Engineeringを始めるのに経験は必要ですか？
事前経験は必要ありません。CoddyKitのAI Prompt Engineeringは初級者から上級者向けに構成されているため、ここから始めるか最初から始めて、自分のペースで進むことができます。 これはレッスン3/4です。
「モデル間の負荷分散」レッスンにはどのくらい時間がかかりますか？
ほとんどのCoddyKitレッスンは約5～10分かかります。各レッスンはコンパクトでインタラクティブなので、着実に進歩し、ウェブとアプリ全体で正確に前回の場所から再開できます。
このAI Prompt Engineeringレッスンでコードを書いて実行できますか？
はい。すべてのAI Prompt Engineeringレッスンに組み込みコードエディタが含まれているため、ブラウザでリアルコードを書いて実行し、即座のAIフィードバックを取得できます。ローカル設定は不要です。