multi-model-routing

マルチモデルルーティング

googleadsagent.ai™ による Agent Skills™ の一部です

説明

マルチモデルルーティングとは、タスクの特性、コスト制約、レイテンシ要件、可用性に基づいて、エージェントのタスクを最適なモデルプロバイダーへインテリジェントに振り分けることです。単一のモデルプロバイダーに依存する本番環境の AI システムは、脆弱でコストがかさみます。マルチモデルルーティングは、Claude、GPT、Gemini、オープンソースモデルの強みを活用した、堅牢でコスト効率の高いエージェントアーキテクチャを実現します。各タスクに最適なモデルを自動的に選択し、プロバイダーが利用できない場合は適切にフェイルオーバーします。

このスキルでは、googleadsagent.ai™ の Buddy™ エージェントを支えるマルチモデルルーティングアーキテクチャについて説明します。このアーキテクチャは、タスクの分類に基づいて Claude（プライマリ — 最も優れた推論能力）、GPT-4o（セカンダリ — 優れた関数呼び出し機能）、Gemini（ターシャリ — 大きなコンテキスト、低コスト）を使い分けます。ルーティングレイヤーにより、すべてのタスクに Claude を使う場合と比較してコストを 45% 削減しながら、同等の品質スコアを維持できました。これは、書式設定、要約、データ抽出といった多くのサブタスクでは、より安価なモデルでも同等の結果が得られるためです。

ルーティングの判断では、モデルの強み（コード推論、長いコンテキスト、構造化出力、創作）、トークンあたりのコスト（モデルの階層によって 100 倍の差があります）、レイテンシ目標（リアルタイムとバッチ）、可用性（レート制限、停止、性能低下）の 4 つの要素を考慮します。サーキットブレーカーのパターンにより、一時的なプロバイダーの問題がユーザー向けの障害に波及するのを防ぎます。

使用する場面

• 品質を犠牲にせず、月々の AI コストを削減したい場合
• 単一プロバイダーの停止やレート制限に対する耐障害性が必要な場合
• サブタスクごとに必要なモデルの要件が根本的に異なる場合
• 同じシステム内に、レイテンシに敏感なタスクとレイテンシに寛容なタスクが共存する場合
• 新しいモデルを全面的に採用することなく評価したい場合
• 単一の AI ベンダーにロックインされないことがコンプライアンス上必要な場合

仕組み

graph TD
    A[Incoming Task] --> B[Task Classifier]
    B --> C{Task Type}
    C -->|Code Reasoning| D[Claude Sonnet/Opus]
    C -->|Structured Extraction| E[GPT-4o / Claude Haiku]
    C -->|Long Context| F[Gemini Pro]
    C -->|Simple Format| G[Claude Haiku / GPT-4o-mini]
    
    D --> H{Provider Available?}
    E --> H
    F --> H
    G --> H
    
    H -->|Yes| I[Execute]
    H -->|No| J[Fallback Chain]
    J --> K[Next Provider]
    K --> H
    
    I --> L[Result + Metrics]
    L --> M[Router Learning]
    M --> B

ルーティングパイプラインは、受信した各タスクを種類（コード推論、構造化抽出、長いコンテキストの処理、単純な書式設定）ごとに分類し、最適なモデルプロバイダーに割り当てます。振り分けの前に、サーキットブレーカーがプロバイダーの可用性を確認します。プロバイダーで最近障害が発生している場合、タスクは直ちにフォールバックチェーンへルーティングされます。実行後は、結果の品質とパフォーマンスのメトリクスがルーターにフィードバックされ、時間の経過とともにモデルとタスクの対応付けが最適化されます。

実装

モデルプロバイダーレジストリ：

interface ModelProvider {
  id: string;
  name: string;
  models: ModelConfig[];
  circuitBreaker: CircuitBreakerState;
}

interface ModelConfig {
  id: string;
  strengths: string[];
  costPer1kInput: number;
  costPer1kOutput: number;
  maxContextTokens: number;
  avgLatencyMs: number;
}

const PROVIDERS: ModelProvider[] = [
  {
    id: "anthropic",
    name: "Anthropic",
    models: [
      { id: "claude-opus", strengths: ["reasoning", "code", "analysis"], costPer1kInput: 0.015, costPer1kOutput: 0.075, maxContextTokens: 200000, avgLatencyMs: 3000 },
      { id: "claude-sonnet", strengths: ["reasoning", "code", "balanced"], costPer1kInput: 0.003, costPer1kOutput: 0.015, maxContextTokens: 200000, avgLatencyMs: 1500 },
      { id: "claude-haiku", strengths: ["speed", "extraction", "formatting"], costPer1kInput: 0.00025, costPer1kOutput: 0.00125, maxContextTokens: 200000, avgLatencyMs: 500 },
    ],
    circuitBreaker: { failures: 0, lastFailure: 0, state: "closed" },
  },
  {
    id: "openai",
    name: "OpenAI",
    models: [
      { id: "gpt-4o", strengths: ["function-calling", "structured-output", "general"], costPer1kInput: 0.005, costPer1kOutput: 0.015, maxContextTokens: 128000, avgLatencyMs: 1200 },
      { id: "gpt-4o-mini", strengths: ["speed", "extraction", "formatting"], costPer1kInput: 0.00015, costPer1kOutput: 0.0006, maxContextTokens: 128000, avgLatencyMs: 400 },
    ],
    circuitBreaker: { failures: 0, lastFailure: 0, state: "closed" },
  },
  {
    id: "google",
    name: "Google",
    models: [
      { id: "gemini-pro", strengths: ["long-context", "multimodal", "general"], costPer1kInput: 0.00125, costPer1kOutput: 0.005, maxContextTokens: 1000000, avgLatencyMs: 2000 },
    ],
    circuitBreaker: { failures: 0, lastFailure: 0, state: "closed" },
  },
];

インテリジェントルーター：

class MultiModelRouter:
    TASK_MODEL_MAP = {
        "code_reasoning": ["claude-sonnet", "claude-opus", "gpt-4o"],
        "structured_extraction": ["claude-haiku", "gpt-4o-mini", "gpt-4o"],
        "long_context": ["gemini-pro", "claude-sonnet", "gpt-4o"],
        "simple_formatting": ["claude-haiku", "gpt-4o-mini", "gemini-pro"],
        "creative_writing": ["claude-opus", "claude-sonnet", "gpt-4o"],
        "analysis": ["claude-sonnet", "gpt-4o", "gemini-pro"],
    }

    def __init__(self, providers: dict, circuit_breakers: dict):
        self.providers = providers
        self.breakers = circuit_breakers

    def route(self, task_type: str, constraints: dict = None) -> str:
        candidates = self.TASK_MODEL_MAP.get(task_type, ["claude-sonnet"])
        constraints = constraints or {}

        for model_id in candidates:
            provider = self.get_provider(model_id)
            if self.breakers[provider].is_open():
                continue
            if constraints.get("max_cost") and self.estimate_cost(model_id, constraints) > constraints["max_cost"]:
                continue
            if constraints.get("max_latency_ms") and self.avg_latency(model_id) > constraints["max_latency_ms"]:
                continue
            return model_id

        return candidates[0]  # Last resort: try primary anyway

    def estimate_cost(self, model_id: str, constraints: dict) -> float:
        config = self.get_model_config(model_id)
        est_input = constraints.get("estimated_input_tokens", 1000)
        est_output = constraints.get("estimated_output_tokens", 500)
        return (est_input / 1000 * config["costPer1kInput"] +
                est_output / 1000 * config["costPer1kOutput"])

サーキットブレーカー：

class CircuitBreaker:
    def __init__(self, failure_threshold=3, recovery_timeout=60):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.failures = 0
        self.last_failure = 0
        self.state = "closed"

    def record_success(self):
        self.failures = 0
        self.state = "closed"

    def record_failure(self):
        self.failures += 1
        self.last_failure = time.time()
        if self.failures >= self.failure_threshold:
            self.state = "open"

    def is_open(self) -> bool:
        if self.state == "open":
            if time.time() - self.last_failure > self.recovery_timeout:
                self.state = "half-open"
                return False
            return True
        return False

統合実行インターフェース：

class UnifiedModelClient:
    def __init__(self, router: MultiModelRouter, clients: dict):
        self.router = router
        self.clients = clients

    async def generate(self, task_type: str, messages: list, **kwargs) -> dict:
        model_id = self.router.route(task_type, kwargs.get("constraints"))
        client = self.clients[self.router.get_provider(model_id)]

        try:
            result = await client.generate(model=model_id, messages=messages, **kwargs)
            self.router.breakers[self.router.get_provider(model_id)].record_success()
            return {"result": result, "model": model_id, "provider": self.router.get_provider(model_id)}
        except Exception as e:
            self.router.breakers[self.router.get_provider(model_id)].record_failure()
            fallback_model = self.router.route(task_type, {**kwargs.get("constraints", {}), "exclude": [model_id]})
            fallback_client = self.clients[self.router.get_provider(fallback_model)]
            result = await fallback_client.generate(model=fallback_model, messages=messages, **kwargs)
            return {"result": result, "model": fallback_model, "fallback": True}

ベストプラクティス

1. モデルの強みを実証的に把握する — 特定のタスクタイプごとに各モデルをベンチマークします。公開されているベンチマークは、特定のドメインでの性能を反映していないことがほとんどです。
2. プロバイダーごとにサーキットブレーカーを実装する — あるプロバイダーの停止が再試行の嵐に波及しないようにします。迅速に失敗させ、代替手段へ切り替えます。
3. タスクの成功あたりのコストを追跡する — トークン単価だけでは誤解を招きます。再試行、フォールバック、失敗した試行を含む総コストを測定します。
4. 品質のしきい値を満たす最も安価なモデルを使う — 多くのタスクでは、Haiku クラスのモデルは Opus クラスのモデルと同じ結果を 1/60 のコストで生成します。
5. プロバイダー間でメッセージ形式を統一する — 内部では統一されたメッセージ形式を維持し、アダプターレイヤーでプロバイダー固有の形式に変換します。
6. 分析のためにルーティングの判断を記録する — ルーターを継続的に最適化できるよう、タスクタイプ、選択されたモデル、制約、結果とともに、すべてのルーティング判断を記録します。
7. フォールバックチェーンを四半期ごとに見直す — モデルの能力、価格、可用性は頻繁に変わります。それに応じて、プロバイダーレジストリとルーティングマップを更新します。

プラットフォームの互換性

機能	Claude Code	Cursor	Codex	Gemini CLI
複数モデルの選択	✅ --model flag	✅ モデルピッカー	⚠️ OpenAI のみ	⚠️ Google のみ
プログラムによるルーティング	✅ MCP/API 経由	✅ 拡張機能経由	✅ API 経由	✅ API 経由
サーキットブレーカー	✅ カスタム	✅ カスタム	✅ カスタム	✅ カスタム
コスト追跡	✅ カスタム	✅ カスタム	✅ カスタム	✅ カスタム
フォールバックチェーン	✅ カスタム	✅ カスタム	✅ カスタム	✅ カスタム

関連スキル

• トークンの最適化 - モデルルーティングは、トークンコストを最適化するうえで最も効果の高い手法です
• 並列エージェントのオーケストレーション - 並列サブエージェントでは、品質を安定させるためにモデル選択を統一することが効果的です
• MCP サーバーの作成 - モデルルーティングのロジックは、MCP サーバーのツール実行パイプラインに統合できます
• 予算の最適化 - ポートフォリオ最適化の原則は、モデルのコストと品質のトレードオフに応用できます

キーワード

multi-model-routing, model-selection, cost-optimization, circuit-breaker, fallback-chains, provider-resilience, task-classification, latency-routing, load-balancing, agent-skills

---

© 2026 googleadsagent.ai™ | Agent Skills™ | MIT License