エージェントルーティングの実装方法｜Gemini 3.7 FlashとClaude Opus 5を使い分ける設計

エージェントルーティングとは

エージェントルーティングとは、入力されたタスクや現在の処理状態を判定し、適切なAIモデル、ツール、サブエージェントへ処理を振り分ける仕組みです。単純なモデル切り替えだけでなく、計画、実行、検証、再試行まで含めたワークフロー制御を意味します。

例えば、複数ファイルにまたがる設計、曖昧な要件の整理、長期的な依存関係の判断はClaude Opus 5へ送ります。一方、計画に沿ったコード生成、定型データ抽出、分類、並列処理などはGemini 3.7 Flashへ送ります。

エージェントルーティングの主要パターン比較

最初から複雑なLLMルーターを構築する必要はありません。初期構成では、Planner-Executor＋ルールベース＋失敗時のエスカレーションが最も実装しやすく、改善効果も測定しやすい組み合わせです。

Gemini 3.7 Flash＋Claude Opus 5の推奨アーキテクチャ

2026年8月時点の公式モデルIDは、Gemini APIがgemini-3.7-flash、Claude APIがclaude-opus-5です。モデルIDは将来変更される可能性があるため、ソースコードへ直接埋め込まず、環境変数や設定ファイルで管理します。

1. PLAN：Claude Opus 5が要件を分解し、実行手順と完了条件を作成
2. EXECUTE：Gemini 3.7 Flashが各ステップを高速に実行
3. VALIDATE：JSON Schema、テスト、ルールで結果を機械検証
4. ESCALATE：検証不合格または高難度タスクだけClaude Opus 5へ昇格
5. REVIEW：Claude Opus 5が成果物全体を最終確認

ポイントは、Flashの自己申告による「自信があります」という文章だけで品質を判断しないことです。必須項目、JSON Schema、単体テスト、禁止条件、引用の有無など、アプリケーション側で確認できる客観的な判定基準を用意します。

Pythonによるルールベースのエージェントルーティング実装

最初に導入したいのが、処理フェーズと難易度スコアを使うシンプルなルーターです。APIキーはソースコードへ書かず、ANTHROPIC_API_KEYとGEMINI_API_KEYを環境変数へ設定してください。

import os
from enum import Enum
from anthropic import Anthropic
from google import genai

OPUS_MODEL = os.getenv("OPUS_MODEL", "claude-opus-5")
FLASH_MODEL = os.getenv("FLASH_MODEL", "gemini-3.7-flash")

anthropic_client = Anthropic(
    api_key=os.environ["ANTHROPIC_API_KEY"]
)
gemini_client = genai.Client(
    api_key=os.environ["GEMINI_API_KEY"]
)

class Phase(str, Enum):
    PLAN = "plan"
    EXECUTE = "execute"
    REVIEW = "review"

def route_model(
    phase: Phase,
    complexity: float = 0.5,
    previous_failed: bool = False
) -> str:
    if phase in {Phase.PLAN, Phase.REVIEW}:
        return OPUS_MODEL

    if previous_failed or complexity >= 0.75:
        return OPUS_MODEL

    return FLASH_MODEL

def call_opus(prompt: str) -> str:
    response = anthropic_client.messages.create(
        model=OPUS_MODEL,
        max_tokens=8192,
        messages=[
            {"role": "user", "content": prompt}
        ]
    )

    return "".join(
        block.text
        for block in response.content
        if block.type == "text"
    )

def call_flash(prompt: str) -> str:
    response = gemini_client.models.generate_content(
        model=FLASH_MODEL,
        contents=prompt
    )
    return response.text

def run_task(
    prompt: str,
    phase: Phase,
    complexity: float = 0.5,
    previous_failed: bool = False
) -> dict:
    model = route_model(
        phase=phase,
        complexity=complexity,
        previous_failed=previous_failed
    )

    if model == OPUS_MODEL:
        output = call_opus(prompt)
    else:
        output = call_flash(prompt)

    return {
        "model": model,
        "phase": phase.value,
        "output": output
    }

result = run_task(
    prompt="商品データを指定されたJSON形式へ変換してください",
    phase=Phase.EXECUTE,
    complexity=0.3
)

print(result["model"])
print(result["output"])

この構成なら、ルーティング理由をログに残しやすく、意図しない高コストモデルの連続呼び出しも防げます。閾値の0.75は固定の正解ではないため、実際の成功率、所要時間、トークン使用量から調整してください。

構造化出力を使ったLLMルーターの実装

キーワードだけでは分類できないタスクが増えたら、Gemini 3.7 Flashをルーターとして利用します。ルーティング結果は自由文ではなく、PydanticとJSON Schemaを使って構造化します。

from typing import Literal
from pydantic import BaseModel, Field

class RouteDecision(BaseModel):
    destination: Literal["flash", "opus"]
    complexity: float = Field(ge=0.0, le=1.0)
    task_type: Literal[
        "planning",
        "execution",
        "review",
        "data_processing",
        "tool_use"
    ]
    reason: str



def llm_router(task: str) -> RouteDecision:
    prompt = f"""
あなたはAIエージェントのルーターです。
次の基準で宛先を決定してください。

opus:
- 複雑な設計や計画
- 曖昧な要件整理
- 複数工程にまたがる深い推論
- 最終レビュー
- 失敗時の原因分析

flash:
- 計画済み手順の実行
- データ抽出、分類、変換
- 定型コード生成
- 大量または並列処理
- 明確なツール呼び出し

タスク:
{task}
"""

    response = gemini_client.interactions.create(
        model=FLASH_MODEL,
        input=prompt,
        response_format={
            "type": "text",
            "mime_type": "application/json",
            "schema": RouteDecision.model_json_schema()
        }
    )

    return RouteDecision.model_validate_json(
        response.output_text
    )

構造化出力は形式を安定させますが、意味的な正しさを保証するものではありません。complexityが範囲内か、許可された宛先か、セキュリティ上の制約に違反していないかを、必ずアプリケーション側でも検証します。

LangGraphによる動的ルーティング実装

状態を持つエージェントや、実行とレビューを繰り返すワークフローにはLangGraphが適しています。LangGraphでは、ノードが処理を行い、エッジが次の処理先を決定します。

ルーティングだけを行う場合はadd_conditional_edgesを使い、状態更新と遷移を同時に行う場合はCommandを使うのが基本です。両者を中途半端に混在させないようにします。

from typing import TypedDict, Literal
from langgraph.graph import StateGraph, START, END

class AgentState(TypedDict):
    task: str
    complexity: float
    failed: bool
    output: str

def select_agent(
    state: AgentState
) -> Literal["opus_agent", "flash_agent"]:
    if state["failed"] or state["complexity"] >= 0.75:
        return "opus_agent"
    return "flash_agent"

def opus_agent(state: AgentState):
    output = call_opus(state["task"])
    return {"output": output, "failed": False}

def flash_agent(state: AgentState):
    output = call_flash(state["task"])
    return {"output": output}

builder = StateGraph(AgentState)

builder.add_node("opus_agent", opus_agent)
builder.add_node("flash_agent", flash_agent)

builder.add_conditional_edges(
    START,
    select_agent,
    {
        "opus_agent": "opus_agent",
        "flash_agent": "flash_agent"
    }
)

builder.add_edge("opus_agent", END)
builder.add_edge("flash_agent", END)

graph = builder.compile()

result = graph.invoke({
    "task": "既存仕様から実装手順を作成してください",
    "complexity": 0.8,
    "failed": False,
    "output": ""
})

実際の本番環境では、チェックポイント、最大ステップ数、タイムアウト、冪等性キーも追加します。LangGraphではノードが再実行される可能性があるため、データベース登録や外部API更新は、同じ処理が複数回走っても破損しない設計が必要です。

フォールバックと品質判定の実装方法

Cascade方式では、まずGemini 3.7 Flashを実行し、検証に失敗した場合のみClaude Opus 5へエスカレーションします。ただし、モデル自身が出力する確信度だけを使うと、誤った回答を高い確信度で返す可能性があります。

def validate_result(result: str) -> tuple[bool, list[str]]:
    errors = []

    if not result or len(result.strip()) < 50:
        errors.append("出力が短すぎます")

    if "TODO" in result:
        errors.append("未完了項目が残っています")

    return len(errors) == 0, errors

def execute_with_escalation(prompt: str) -> dict:
    flash_output = call_flash(prompt)
    passed, errors = validate_result(flash_output)

    if passed:
        return {
            "model": FLASH_MODEL,
            "escalated": False,
            "output": flash_output
        }

    review_prompt = f"""
次のタスクを再実行してください。
高速モデルの出力には検証エラーがあります。

元のタスク:
{prompt}

高速モデルの出力:
{flash_output}

検証エラー:
{errors}
"""

    opus_output = call_opus(review_prompt)

    return {
        "model": OPUS_MODEL,
        "escalated": True,
        "errors": errors,
        "output": opus_output
    }

エスカレーション条件の具体例

• JSON SchemaまたはPydanticの検証に失敗した
• 単体テスト、静的解析、型チェックに失敗した
• 必須項目や引用元が不足している
• ツール呼び出しがタイムアウトした
• 同じ処理が規定回数以上失敗した
• 法務、金融、個人情報など高リスク領域に該当した
• 予算、最大ステップ数、コンテキスト上限を超えた

LiteLLM・OpenRouter・Google Cloudを使ったGateway構成

複数プロバイダーの認証、モデル名、タイムアウト、再試行、監視をアプリケーションから分離したい場合は、AI Gatewayまたはプロキシを導入します。

LiteLLM

LiteLLM Proxyは、複数モデルに対して統一的なエンドポイントを提供し、ロードバランシング、リトライ、クールダウン、フォールバックを設定できます。プロバイダー障害やレート制限への対策をGateway側へ集約したい場合に向いています。

OpenRouter

OpenRouterは複数プロバイダーのモデルを共通APIから利用できるため、モデル切り替えの実装量を減らせます。一方、データ保持条件、利用可能リージョン、プロバイダー選択、障害時の挙動は導入前に確認が必要です。

Google Cloud

Google Cloudでは、Geminiに加えてパートナーモデルとしてClaudeを利用できる構成があります。認証、IAM、監査ログ、請求先をGoogle Cloudへ寄せたい企業システムと相性があります。

ただし、一般的なAPI Gatewayを置くだけでタスク難易度に応じたルーティングが自動化されるわけではありません。Planner-Executorや品質判定のロジックは、バックエンド、ワークフロー基盤、または専用ルーターとして実装する必要があります。

本番運用で失敗しないためのポイント

1．状態をモデル非依存の形式で保存する

会話履歴をそのまま引き継ぐだけでなく、目的、計画、完了済みステップ、ツール結果、エラー、成果物URLを共通Stateとして保存します。モデルを切り替えても必要な情報を再現できることが重要です。

2．ツール定義を共通JSON Schemaへ寄せる

ClaudeとGeminiではAPIの細部が異なります。アプリケーション内部では共通のツール定義を持ち、各プロバイダー用のアダプターで変換すると保守しやすくなります。

3．コスト・遅延・品質を同時に記録する

モデル名、ルーティング理由、入力・出力トークン、キャッシュ利用、レイテンシ、再試行回数、検証結果、ユーザー評価を記録します。単純なAPI単価ではなく、成功タスク1件あたりの総コストで比較してください。

4．プロンプトキャッシュを活用する

長いシステム指示、共通仕様、ツール定義を繰り返し送信する場合は、ClaudeのPrompt CachingとGeminiのContext Cachingを検討します。キャッシュ条件や料金はモデルごとに異なるため、公式ドキュメントを確認します。

5．無限ループを防止する

最大ステップ数、最大再試行回数、最大予算、処理時間、同一ツールの連続実行回数を設定します。「失敗したらOpusへ送る」だけではなく、Opusでも失敗した場合の人間への引き継ぎ経路を用意します。

おすすめの導入手順

1. Claude Opus 5をPLAN・REVIEW、Gemini 3.7 FlashをEXECUTEへ固定する
2. JSON Schema、単体テスト、必須項目チェックを追加する
3. Flashの失敗時だけOpusへ昇格する
4. ルーティング結果と成功率をログへ保存する
5. 誤振り分けが増えた段階でLLMルーターを追加する
6. 処理量が増えたらLangGraphやLiteLLMで基盤化する

結論として、Gemini 3.7 FlashとClaude Opus 5を組み合わせる場合は、Claude Opus 5に考えさせ、Gemini 3.7 Flashに大量実行させ、機械検証に失敗した処理だけをClaudeへ戻す構成が実用的です。

エージェントルーティングに関するよくある質問

Q1．最初からLLMルーターを導入すべきですか？

いいえ。初期段階ではPLAN、EXECUTE、REVIEWのフェーズで固定するルールベースが適しています。ログが蓄積し、誤振り分けのパターンを把握してからLLMルーターを追加する方が安全です。

Q2．Gemini 3.7 Flashの出力は必ずClaude Opus 5でレビューすべきですか？

必ずしも必要ではありません。JSON Schemaやテストで自動検証できるタスクは、合格時にそのまま採用できます。高リスク領域、複雑な成果物、最終納品物だけをClaude Opus 5へ送るとコストを抑えられます。

Q3．LangGraphとLiteLLMはどちらを選べばよいですか？

LangGraphは状態とワークフローの制御、LiteLLMは複数モデルへの接続、ロードバランシング、リトライ、フォールバックに向いています。役割が異なるため、LangGraphの接続先としてLiteLLMを使う構成も可能です。

Q4．ルーティング精度を改善するには何を記録すべきですか？

タスク種別、選択モデル、選択理由、難易度、成功・失敗、検証エラー、処理時間、トークン数、再試行回数、ユーザー評価を記録します。成功タスク1件あたりのコストを指標にすると改善効果を判断しやすくなります。

参考資料

• 
  Google AI for Developers：Gemini 3.7 Flash
  
• 
  Google AI for Developers：Structured outputs
  
• 
  Anthropic：Claude models overview
  
• 
  Anthropic：Prompt caching
  
• 
  LangChain：LangGraph Graph API