結論:AIエージェントのモデルルーティングは「タスク複雑度を先に分類し、必要最小限のモデルへ振る」設計で、品質を落とさずにAPIコストを大幅に圧縮できる。
- 要点1:タスクを「単純・中程度・複雑」の3段階に分類し、Haiku/Sonnet/Opusへ動的に振り分けるだけで、全件Opus処理と比べてコストを60〜80%削減できる(試算例・実際の削減率はワークロードによる)
- 要点2:orchestrator-workerパターンでHaikuを分類器、上位モデルを実行器にすると、分類コスト自体がほぼゼロになる
- 要点3:フォールバック・リトライ設計を最初から組み込むことで、特定プロバイダー障害時も自動切り替えが可能になる
対象読者:AIエージェントを本番運用しているエンジニア・アーキテクト・PM。複数モデルを使い分けてコスト効率を高めたい方。
今日やること:自社エージェントのタスクログを確認し、全タスクの何%が「単純タスク」に分類できるか計測する。
「AIエージェントのAPIコストが想定より3倍かかっている」
複数社のAIエージェント導入を支援する中で、この相談は本当によく聞きます。原因を確認すると、ほぼ例外なく「全タスクを高性能モデルに投げている」というパターンです。
実際のワークフローを分解すると、ユーザー入力の分類・意図解釈・簡単な要約といった処理は、高性能モデルでなくても十分こなせます。にもかかわらず、コード生成や複雑な推論と同じモデルに通してしまっているのです。
この記事では、タスク複雑度に基づいてモデルを動的に切り替える「モデルルーティング」設計を、コピーして使えるコードと実際の料金数値をもとに解説します。
モデルルーティングとは何か
モデルルーティングとは、受け取ったタスクの特性に応じて、処理を最適なLLMモデルへ動的に振り分ける設計パターンです。単一エージェント内で複数のモデルを使い分けることが前提になります。
たとえば、次のようなエージェントを考えてみましょう。
- ユーザーの質問を受け取り、意図を分類する
- 必要なら外部APIを呼び出してデータを取得する
- 取得したデータをもとに回答を生成する
このフローにおいて、「意図の分類」と「高品質な回答生成」を同じモデルで処理する必要はありません。分類は軽量モデル(Haiku相当)で十分で、最終回答生成にのみ高性能モデルを使えばよいのです。
ルーティングの3軸
モデルを選択する際は、次の3軸で評価します。
- タスク複雑度:単純な分類・要約か、複雑な推論・コード生成か
- コスト要件:1コール当たり許容できるAPIコストはいくらか
- レイテンシ要件:リアルタイム応答が必要か、バッチ処理でよいか
主要モデルのコスト・性能トレードオフ表(2026年6月時点)
まず料金の実数を整理します。すべてAnthropicおよびOpenAI、Google公式ドキュメントから取得した2026年6月時点の値です。
| モデル | 入力($/百万トークン) | 出力($/百万トークン) | 円換算目安(入力) | 主な用途 |
|---|---|---|---|---|
| Claude Haiku 4.5 | $1.00 | $5.00 | 約145円/百万トークン | 分類・短文要約・意図解釈 |
| Claude Sonnet 4.6 | $3.00 | $15.00 | 約435円/百万トークン | 汎用・コード生成・中程度の推論 |
| Claude Opus 4.8 | $5.00 | $25.00 | 約725円/百万トークン | 複雑な推論・長文生成・エージェント判断 |
| GPT-4.1-mini | $0.40 | $1.60 | 約58円/百万トークン | 軽量タスク・大量バッチ処理 |
| GPT-4.1 | $2.00 | $8.00 | 約290円/百万トークン | 汎用・コード生成 |
| Gemini 2.5 Flash | $0.30 | $2.50 | 約43円/百万トークン | 高速・低コスト処理 |
| Gemini 3.5 Flash | $1.50 | $9.00 | 約218円/百万トークン | バランス型・マルチモーダル |
出典:Anthropic公式Pricing(参照日: 2026-06-12)、OpenAI API Pricing(参照日: 2026-06-12)、Google Gemini API Pricing(参照日: 2026-06-12)
円換算は$1=145円として計算しています。為替レートによって変動します。
コスト試算例
以下は試算例です。実際のコストはプロンプト設計・出力長・タスク分布によって大きく異なります。参考値としてご利用ください。
事例区分: 想定シナリオ
以下は複数の導入支援経験をもとに構成した典型的な試算シナリオです。
月間10万リクエスト・平均入力1,000トークン・平均出力500トークンのエージェントを想定します。
- 全件Opus 4.8処理:入力$50 + 出力$125 = 月$175(約25,375円)
- ルーティング後(Haiku 60% / Sonnet 30% / Opus 10%):入力$22.2 + 出力$57.5 = 月$79.7(約11,557円)
- 削減率:約54%(試算例)
タスクの60%がHaikuで処理できるかどうかは、実際のワークロードを計測しなければわかりません。まず自社のログを分析するところから始めてください。
タスク複雑度の分類チェックシート
ルーティングを機能させるには、タスク複雑度を先に定義する必要があります。以下の判断基準を参考にしてください。
Level 1(軽量モデル推奨)
- 入力テキストの分類・ラベリング
- 短文の要約(500字以内)
- 定型フォーマットへの変換
- 固定パターンの意図解釈(FAQルーティング等)
- 感情分析・トーン判定
Level 2(中性能モデル推奨)
- コード生成(〜100行程度の単一関数)
- 長文の要約・構造化(1,000〜3,000字)
- RAG検索結果の統合と回答生成
- メール・レポートの下書き作成
- 複数ステップの手順説明
Level 3(高性能モデル推奨)
- 複雑なアーキテクチャ設計・技術判断
- 大規模コードのリファクタリング・デバッグ
- 複数の制約条件を伴う最適化問題
- 長文ドキュメント全体の分析・考察
- 法的・財務的判断を含む複雑な推論
このチェックシートをそのままプロンプトに組み込んで、分類器モデルに判断させることもできます。
orchestrator-worker型ルーティングの実装コード
ここからは実際に動かせるコードを紹介します。
基本構成:Haikuで分類→必要時のみ上位モデルへ
以下のコードは、タスクをまずHaiku 4.5で分類し、複雑度に応じてモデルを選択します。
import anthropic
import json
client = anthropic.Anthropic()
# 動作環境: Python 3.11+, anthropic>=0.34.0
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
ROUTING_PROMPT = """
あなたはタスク複雑度を分類するルーターです。
以下のタスクを3段階で分類し、JSONで返してください。
タスク: {task}
返すべきJSON:
{{
"level": 1 | 2 | 3,
"reason": "理由を1文で",
"estimated_output_tokens": 推定出力トークン数(整数)
}}
分類基準:
- level 1: 分類・短文要約・定型変換(軽量処理)
- level 2: コード生成・長文要約・RAG回答(中程度の処理)
- level 3: 複雑な推論・大規模コード・複数制約最適化(高度な処理)
"""
def classify_task(task: str) -> dict:
"""タスク複雑度をHaiku 4.5で分類する"""
resp = client.messages.create(
model="claude-haiku-4-5",
max_tokens=256,
messages=[{"role": "user", "content": ROUTING_PROMPT.format(task=task)}]
)
return json.loads(resp.content[0].text)
def route_task(task: str, system_prompt: str = "") -> str:
"""タスク複雑度に応じてモデルをルーティングする"""
# Step 1: 軽量モデルで分類(コストは入力1,000トークンとして約$0.001)
classification = classify_task(task)
level = classification["level"]
# Step 2: 複雑度に応じてモデルを選択
model_map = {
1: "claude-haiku-4-5", # $1/$5 per MTok
2: "claude-sonnet-4-6", # $3/$15 per MTok
3: "claude-opus-4-8", # $5/$25 per MTok
}
selected_model = model_map[level]
print(f"[ROUTER] level={level}, model={selected_model}, reason={classification['reason']}")
# Step 3: 選択されたモデルでタスクを実行
messages = [{"role": "user", "content": task}]
resp = client.messages.create(
model=selected_model,
max_tokens=2048,
system=system_prompt,
messages=messages
)
return resp.content[0].text
# 使用例
if __name__ == "__main__":
tasks = [
"このメールの感情を分類してください: 「先日の件ですが、少し不満があります」",
"PythonでBinary Searchを実装してください",
"分散型マイクロサービスアーキテクチャで、SAGA パターンとTCC パターンの適用場面を分析し、システム境界と整合性の観点で設計方針を提案してください"
]
for task in tasks:
result = route_task(task)
print(f"結果: {result[:100]}...n")
ポイント
- 分類自体もAPIコールのため、分類コストが発生します。入力1,000トークン程度のHaiku呼び出しは約$0.001(約0.14円)なので、大量リクエストでも分類コストは無視できる水準です
- 分類精度を上げたい場合は、few-shot例を追加してください
- レイテンシが重要な場合は、分類と実行を並列化するのではなく、事前に分類済みタスクをキューに溜める設計も有効です
LiteLLMを使ったルーティング(プロバイダー横断版)
LiteLLMを使うと、Claude/GPT/Geminiを同一インターフェースで扱えます。
from litellm import completion
import json
# 動作環境: Python 3.11+, litellm>=1.40.0
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
# コスト基準でモデルを定義(2026年6月時点の公式料金)
COST_TIERS = {
"ultra_low": "gemini/gemini-2.5-flash", # 入力$0.30/MTok(最安)
"low": "claude-haiku-4-5", # 入力$1.00/MTok
"mid": "gpt-4.1-mini", # 入力$0.40/MTok(OpenAI軽量)
"standard": "claude-sonnet-4-6", # 入力$3.00/MTok
"premium": "claude-opus-4-8", # 入力$5.00/MTok
}
def smart_route(task: str, max_cost_per_call: float = 0.01) -> str:
"""
コスト上限に基づいてモデルを自動選択する
max_cost_per_call: ドル単位(デフォルト$0.01)
"""
# まず軽量モデルで試行
for tier_name, model in COST_TIERS.items():
try:
resp = completion(
model=model,
messages=[{"role": "user", "content": task}],
max_tokens=1024,
)
# コスト確認(litellm.completion_costで取得可能)
actual_cost = resp._hidden_params.get("response_cost", 0)
print(f"[ROUTE] model={model}, cost=${actual_cost:.4f}")
if actual_cost <= max_cost_per_call:
return resp.choices[0].message.content
except Exception as e:
print(f"[WARN] {model} failed: {e}, trying next tier")
continue
# フォールバック: premium tier
resp = completion(
model=COST_TIERS["premium"],
messages=[{"role": "user", "content": task}],
max_tokens=2048,
)
return resp.choices[0].message.content
OpenRouterを使ったルーティング(管理画面付き)
OpenRouterはAPIキー1本で100以上のモデルにアクセスでき、ダッシュボードでコストを可視化できます。
import openai
import os
# 動作環境: Python 3.11+, openai>=1.30.0
# OpenRouter APIキーが必要(https://openrouter.ai/)
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
client = openai.OpenAI(
base_url="https://openrouter.ai/api/v1",
api_key=os.environ["OPENROUTER_API_KEY"],
)
def route_via_openrouter(task: str, complexity_level: int) -> str:
"""OpenRouterでモデルルーティングする"""
# OpenRouterのモデル名マッピング
model_map = {
1: "anthropic/claude-haiku-4-5",
2: "anthropic/claude-sonnet-4-6",
3: "anthropic/claude-opus-4-8",
}
resp = client.chat.completions.create(
model=model_map.get(complexity_level, "anthropic/claude-sonnet-4-6"),
messages=[{"role": "user", "content": task}],
extra_headers={
"X-Title": "MyAgent", # OpenRouterダッシュボードに表示される名前
}
)
return resp.choices[0].message.content
フォールバック・リトライ設計
本番エージェントでは、特定プロバイダーの障害・レート制限を想定したフォールバック設計が不可欠です。
フォールバック優先順位の設計
import anthropic
import time
from typing import Optional
client = anthropic.Anthropic()
# 動作環境: Python 3.11+, anthropic>=0.34.0
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
FALLBACK_CHAINS = {
# プライマリ: Haiku → フォールバック: Sonnet
"low": ["claude-haiku-4-5", "claude-sonnet-4-6"],
# プライマリ: Sonnet → フォールバック: Opus
"mid": ["claude-sonnet-4-6", "claude-opus-4-8"],
# プライマリ: Opus → フォールバック: Sonnet(コスト優先の緊急時)
"high": ["claude-opus-4-8", "claude-sonnet-4-6"],
}
def call_with_fallback(
task: str,
tier: str = "mid",
max_retries: int = 2,
retry_delay: float = 1.0
) -> Optional[str]:
"""フォールバック付きモデル呼び出し"""
fallback_chain = FALLBACK_CHAINS.get(tier, FALLBACK_CHAINS["mid"])
for attempt, model in enumerate(fallback_chain):
for retry in range(max_retries):
try:
resp = client.messages.create(
model=model,
max_tokens=2048,
messages=[{"role": "user", "content": task}]
)
# 成功ログ
print(f"[OK] model={model}, attempt={attempt+1}, retry={retry}")
return resp.content[0].text
except anthropic.RateLimitError:
print(f"[WARN] RateLimit: {model}, waiting {retry_delay}s")
time.sleep(retry_delay * (retry + 1)) # 指数バックオフ
except anthropic.APIError as e:
print(f"[ERROR] {model}: {e}")
break # このモデルはスキップ、次のフォールバックへ
# 全フォールバックが失敗
print("[CRITICAL] All fallback models failed")
return None
# 使用例
result = call_with_fallback(
task="PythonでQuickSortを実装してください",
tier="mid",
max_retries=2
)
if result:
print(result)
else:
print("処理失敗: 手動対応が必要です")
ポイント
max_retriesを設定することで、無限ループを防ぎます- 指数バックオフ(
retry_delay * (retry + 1))でレート制限エラーを吸収できます - フォールバック発生をログに記録し、後から「どのモデルが何回障害したか」を分析できる状態にしてください
マルチエージェントでのルーティング設計
orchestrator-workerパターンでは、orchestratorとworkerで別々のモデルを割り当てるのが基本です。
詳細な設計パターンはマルチエージェント設計パターン完全ガイドを参照してください。
import anthropic
client = anthropic.Anthropic()
# 動作環境: Python 3.11+, anthropic>=0.34.0
# 注意: 本番環境で使用する前に、必ずテスト環境で動作確認してください。
# Orchestrator: タスク分解にSonnetを使用(精度重視)
ORCHESTRATOR_MODEL = "claude-sonnet-4-6"
# Worker: サブタスク実行はタスク複雑度でルーティング
WORKER_MODELS = {
"simple": "claude-haiku-4-5",
"standard": "claude-sonnet-4-6",
"complex": "claude-opus-4-8",
}
def orchestrate(main_task: str) -> str:
"""Orchestratorがタスクを分解し、Workerへ振り分ける"""
# Step 1: OrchestratorがサブタスクリストをJSONで生成
orchestrator_prompt = f"""
以下のメインタスクを実行可能なサブタスクに分解してください。
各サブタスクにcomplexity(simple/standard/complex)を付けてください。
メインタスク: {main_task}
返すべきJSON:
[
{{"task": "サブタスク内容", "complexity": "simple|standard|complex"}},
...
]
"""
import json
orchestrator_resp = client.messages.create(
model=ORCHESTRATOR_MODEL,
max_tokens=1024,
messages=[{"role": "user", "content": orchestrator_prompt}]
)
subtasks = json.loads(orchestrator_resp.content[0].text)
# Step 2: 各サブタスクを複雑度に応じたWorkerモデルで実行
results = []
for subtask in subtasks:
worker_model = WORKER_MODELS.get(subtask["complexity"], WORKER_MODELS["standard"])
print(f"[WORKER] complexity={subtask['complexity']}, model={worker_model}")
worker_resp = client.messages.create(
model=worker_model,
max_tokens=2048,
messages=[{"role": "user", "content": subtask["task"]}]
)
results.append(worker_resp.content[0].text)
# Step 3: OrchestratorがWorker結果を統合
synthesis_prompt = f"以下のサブタスク結果を統合して最終回答を作成してください:nn" + "nn".join(results)
final_resp = client.messages.create(
model=ORCHESTRATOR_MODEL,
max_tokens=2048,
messages=[{"role": "user", "content": synthesis_prompt}]
)
return final_resp.content[0].text
【要注意】よくある失敗パターンと回避策
失敗1:分類精度が低くLevel 3に寄り過ぎる
❌ 「ほぼ全タスクがlevel 3と分類され、コスト削減ゼロ」
✅ 分類プロンプトにfew-shot例(level 1の典型例・level 3の典型例)を追加し、精度を上げる。分類ログを一週間分析してから本番適用する。
失敗2:フォールバック未設定でエージェントが止まる
❌ プライマリモデルが429エラーで全体停止
✅ 最低でも1段階のフォールバック先を設定する。max_retries=2と指数バックオフを必ず組み込む。
失敗3:ルーティングログを残さない
❌ どのタスクがどのモデルで処理されたか不明で最適化できない
✅ model・complexity_level・actual_cost・latency_msを構造化ログに記録する。1週間後に分析してルーティング基準を調整する。
失敗4:コスト削減だけを見て品質劣化を見落とす
❌ Haikuへの振り分けを増やしすぎてユーザー満足度が下がる
✅ 品質評価(ユーザーフィードバック・LLM-as-a-judge)をコスト指標と並べてモニタリングする。削減率だけを追わない。
コスト削減のさらなる手法:セマンティックキャッシュとの組み合わせ
モデルルーティングとセマンティックキャッシュを組み合わせると、類似クエリへのAPI呼び出し自体を削減できます。ルーティング→キャッシュヒット確認→キャッシュミス時のみAPI呼び出しという順序が効果的です。
またAIエージェントのコスト最適化7原則では、プロンプトキャッシュ・バッチ処理・トークン削減など、ルーティング以外の手法も詳しく解説しています。
まとめ:今日から始める3つのアクション
- 今日やること:自社エージェントの直近1週間のタスクログを確認し、タスク種別を手動で分類してみる。Level 1相当がどの程度の比率を占めるかを計測する。
- 今週中:分類器プロトタイプをローカルで実装し、サンプル50件程度で分類精度を確認する。精度80%以上を確認できたら本番適用を検討する。
- 今月中:ルーティングログを実装し、モデル別コスト・品質スコアを並べて週次でモニタリングする体制を整える。
あわせて読みたい
- LiteLLM完全ガイド|100モデル統合・LLMルーティング・コスト最適化 — ルーティング基盤の実装詳細
- マルチエージェント設計パターン完全ガイド — orchestrator-workerの全体設計
- OpenRouter完全ガイド — マルチプロバイダー管理のベストプラクティス
- AIエージェントのコスト最適化7原則 — ルーティング以外の削減手法
参考・出典
- Anthropic Claude API Pricing — Anthropic公式(参照日: 2026-06-12)
- OpenAI API Pricing — OpenAI公式(参照日: 2026-06-12)
- Gemini API Pricing — Google AI公式(参照日: 2026-06-12)
この記事を読んで導入イメージが固まってきた方へ
UravationではAIエージェント導入の研修・コンサルを行っています。
著者: 佐藤傑(さとう・すぐる)
株式会社Uravation代表取締役。X(@SuguruKun_ai)フォロワー10万人以上。著書『AIエージェント仕事術』。
ご質問・ご相談は お問い合わせフォーム からお気軽にどうぞ。
