0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@nogataka

トークン節約 ── 最適な LLM ルーティングについて考えてみた

0
Last updated at Posted at 2026-09-16

要点

  • 「LLMルーティング」という言葉は1つの技術を指しません。私が調べた3つの shunt という名前のOSSは、ルーティング対象がそれぞれ違いました。具体的には「アカウント」「モデル・Provider」「作業そのもの」です。
  • 「タスク難易度に応じてモデルを選ぶ」設計を作るなら、難易度とモデルを直接結びつけないことが重要です。その間に「必要な能力」という抽象レイヤーを挟むのが、良い設計だと考えています。
  • 難易度を判定するClassifierには、モデル名や価格を考えさせないことが重要です。価格の最適化は、その後段の役割にするのが良いと考えています。
  • 依存ゼロで動くPythonのサンプルを作り、実際に claude CLIをLLM Judgeとして呼び出すところまで動かしました。
  • ルーティングを一発で決め切らず、Agentの実行中にツールのエラーや進捗の停滞を見て、必要ならモデルを引き上げる仕組みも含めています。

はじめに

対象読者は、Claude CodeなどのAIコーディングエージェントを業務で使っていて、API費用やレート制限に悩んでいる方です。前提環境はPython 3.11以降とします。

私は普段、Claude Codeを中心に複数のAIコーディングツールを併用しています。使う中で「どのタスクにどのモデルを割り当てるか」を考える機会が増えました。その過程で、GitHub上に同じ「shunt」という名前を持つ複数のOSSプロジェクトがあることに気づきました。この記事では、それらを調べて分かった違いと、自分なりに「タスク難易度からモデルを選ぶ」ロジックをPythonで組んでみた過程をまとめます。

「ルーティング」と一口に言っても3種類ある

調べた3つのプロジェクトを並べると、主目的とルーティング対象がまったく違うことが分かりました。

プロジェクト 主目的 ルーティング対象 重要指標
ramc10/shunt レート制限の回避とアカウントプーリング アカウント Quota、Rate Limit、5h/7dウィンドウ、Cooldown、RPM
pleaseai/shunt LLM GatewayとModel Routing モデル・Provider Model ID、Provider、Failover、Stage
spotify/portal-ai-plugins shunt Task・Context委譲 作業そのもの 委譲した処理の入出力トークン量

ramc10/shunt: アカウントプール

ramc10/shuntは、複数のClaude Codeアカウントを1つのローカルプロキシに束ねるツールです。公式READMEでは、利用余力の大きいアカウントへ自動でルーティングすると説明されています。レート制限に達したアカウントは、静かにフェイルオーバーする仕組みです。Anthropicだけでなく、複数のProviderに対応しています。対応例はOpenAI・Gemini・Groq・Mistral・DeepSeekです。ほかにOpenRouter・Together・Fireworks・Ollamaにも対応しています。

READMEには明記されていませんが、ソースコード（src/router.rs）を確認すると分かることがあります。ルーティング戦略が4種類のenumとして実装されている点です。

戦略 動き
Carousel Round Robinで順にアカウントを回す
Cushion Quota残量が多いアカウントを優先する
Maximus 5h/7dの利用率とreset時間、burst状況を合わせてスコア化する
Reaper まもなくresetされるQuotaから優先的に使い切る

同じくソースコードには、会話単位のstickinessという仕組みがあります。これはsystem文やツール定義のハッシュでアカウントを固定する仕組みです。ほかにも5h/7dウィンドウの考慮、cooldown、RPMペーシングといった実装があります。これらはREADME本文ではなく、コードを読んで初めて分かる内容です。

ramc10/shuntの主眼は「どのモデルが最適か」ではありません。「どのアカウント・Providerに流せばレート制限を回避しながら効率よく使えるか」にあります。ルーティング層(router.rs)とProvider層(provider.rs)は分かれています。そのため、この上に「タスク難易度からモデルを選ぶ」ロジックを足すこと自体は、相性が良さそうだと感じました。

pleaseai/shunt: LLM Gateway

pleaseai/shuntは、README自身が「Claude Code LLM gateway」と定義しています。その定義の通り、モデル・ProviderのRoutingに明確に設計されているツールです。Claude Codeから来たリクエストのモデルIDを見て、別のProviderへ振り分けます。対応例はOpenAI・Codex・xAI・Gemini・Anthropicなどです。

設定の中心は [[routes]] と [models.upstream_model] です。README上では [[routes]] はレガシー扱いです。代わりに、公開モデルIDから複数の実バックエンドへ順序付きでマッピングする方式が推奨されています。この方式の設定名が [models.upstream_model] です。モデルIDの解決順序は、ARCHITECTURE.mdに説明があります。順序は「exact route → prefix route → default_provider」です。複数のupstreamを指定すると、順序付きのフェイルオーバーチェーンとしても機能します。

さらに面白いのが [models.stage_router] です。READMEでは「opt-in, content-aware tier selection」と説明されています。会話の直近のツール実行結果などから、その時点の作業段階を推定する仕組みです。推定結果に応じて、efficientなモデルとcapableなモデルのどちらを使うかをターンごとに切り替えます。これは、プロンプト本文をLLM分類器で解析するような方式とは異なります。ツール実行結果の構造化されたメタデータから判定する仕組みです。私が最終的に組みたいのは「プロンプト内容から難易度を判定するLLM Classifier」です。stage_routerとは、着眼点が違う点に注意が必要です。

spotify/portal-ai-plugins shunt: 作業委譲

もう1つのshuntは、SpotifyのClaude Codeプラグインです。こちらはモデルやアカウントのルーティングではありません。大きなファイルの読み込みや定型的なコード生成といった作業そのものを、Claude本体から軽量なワーカーへ委譲する仕組みでした。委譲によるトークン削減の考え方は、モデル選択とは別レイヤーの話で、組み合わせる価値があると考えています。この記事ではモデルルーティングに絞るため詳しくは扱いません。詳細は関連記事にまとめる予定です。

「タスク難易度→モデル選択」を考える

3つのshuntを見て、自分が欲しいものが分かりました。「アカウントのルーティング」でも「Providerのルーティング」でもありません。「このタスクにはどれくらいの能力を持つモデルが必要か」を判定してから、コストやレイテンシを見てモデルを決める仕組みです。全体の流れを図にすると次のようになります。

参考にした OSS

このロジックを組むにあたって、参考になったOSSを5つ挙げます。

リポジトリ 方式 見どころ
NVIDIA-NeMo/Switchyard LLM Judge + Stage signals 判定ロジック、閾値、Escalationの設計
maynewong/pi-model-auto Heuristic + LLM Classifier 難易度を判定するローカルHeuristicの考え方
lm-sys/RouteLLM 学習型Router strong/weakモデルの判定手法、評価方法
vllm-project/semantic-router Signal/ML/Router-R1 本格的なModel Selectionの構成
openziti/llm-gateway Rule → Embedding → LLM の3層cascade シンプルな多層判定の考え方

NVIDIA-NeMo/Switchyardの公式ドキュメントでは、シグナルによって判定を寄せる先が異なると説明されています。エラーの深刻度や手詰まり・探索のみの状態といったシグナルはcapableモデルへ判定を寄せます。直近の生産的な編集が続いている状態は、efficientモデルへ判定を寄せます。この「実行中の状態を難易度判断に使う」という考え方は、後述するLayer 6のEscalationに反映しました。

maynewong/pi-model-auto は、low/medium/high/ultraという4段階で難易度を分類する設計です。ソースコード（src/router-core.ts）を確認したところ、ローカルのHeuristicの仕組みが分かりました。コンテキスト量・直近のユーザー入力長・ツール利用密度を、それぞれ0.3・0.5・0.2の重みで合成しています。合成したスコアは、0.30/0.52/0.74という3つの境界値で4段階に振り分けていました。このうち3つの境界値は、私が組んだサンプルでもそのまま採用しています。重みは、サンプルではプロンプト長・コンテキスト量・ファイル数・ツール数という4つの特徴量を使います。そのため、0.25・0.30・0.20・0.25と独自に置き直しました。READMEには、次の一文があります。

The mode says how capable the model must be. It does not describe price.

難易度とモデル価格をClassifierの中で一緒に判断させない、という設計思想が明確に示されています。

lm-sys/RouteLLMは、Routerの共通インターフェースを持っています。そのインターフェースは calculate_strong_win_rate(prompt) という関数です。この戻り値が閾値以上ならstrongモデル、未満ならweakモデルへルーティングする構造です。Router実装には複数の方式があります。前半はMatrix FactorizationとSimilarity Weighted Rankingです。後半はBERTとCausal LLM Classifierです。これらを比較できるようになっています。「difficulty」という主観的な軸で難易度を直接学習するのではありません。「strongモデルがweakモデルに勝つ確率」という学習しやすい軸に落とし込んでいる点が参考になりました。

vllm-project/semantic-routerというプロジェクトがあります。Signal抽出からDecision、Model Selectorへという構造を持っています。複数の選択アルゴリズムを備えていると説明されています。その1つがRouter-R1です。これはLLM自身にモデル選択の思考過程を持たせる方式として紹介されています。この方式については、公式リポジトリでの直接の記述までは確認できていないため、断定は避けます。

openziti/llm-gatewayは、3層のcascadeでリクエストを分類する設計です。層はHeuristic・Embedding・LLM Classifierの順です。confidenceが高ければ早い段階でルートを決め、曖昧な場合だけ後段のLLM Classifierへエスカレーションします。具体的な閾値は設定ファイルに依存するため、この記事では数値までは踏み込みません。3層に分けてコストの低い判定から順に試す、という構造自体が参考になりました。

設計の要点3つ

これらを踏まえて、自分なりに次の3点を設計の軸にしました。

1つ目は、Difficultyとモデルを直接結びつけないことです。難易度がHIGHだからモデルXを使う、とハードコードすると、モデルが半年後に入れ替わるたびにロジックを書き直すことになります。難易度からいったん「必要な能力（Required Capability）」という抽象値に変換します。その能力を満たす候補の中から、コストとレイテンシで選ぶという2段構えにしました。

2つ目は、難易度を判定するClassifierに、モデル名や価格を考えさせないことです。分類の役割とコスト最適化の役割を1つのプロンプトに混ぜると、Classifierの出力が不安定になります。分類は分類だけに専念させ、モデル選択は別のレイヤーに任せます。

3つ目は、ルーティングを一発で決め切らないことです。最初のリクエストの時点で分かる情報は限られています。Agentが実際に動き出してから、実行中のシグナルを見ます。ツールのエラーやテストの失敗、作業の停滞といったシグナルがあれば、必要に応じてモデルを引き上げる仕組みを組み込みました。

Python サンプル

上記の考え方を、依存ゼロの1ファイルPythonスクリプトにまとめました。外部ライブラリを使わず、標準ライブラリだけで動きます。全体は6つの層に分かれています。

Layer 1: Cheap Heuristic

最初の層は、LLMを呼ばずに計算できる軽量なHeuristicです。4つの特徴量（プロンプト長・コンテキスト量・対象ファイル数・ツール利用数）を、0.25・0.30・0.20・0.25の重みで合成します。合成したスコアを、pi-model-autoと同じ0.30/0.52/0.74の境界で判定します。境界に応じて、LOW/MEDIUM/HIGH/ULTRAの4段階に振り分けます。

router.py(Layer1抜粋)
class HeuristicClassifier:
    def classify(self, ctx: RequestContext) -> DifficultyResult:
        prompt_length_score = min(len(ctx.prompt) / 4000, 1.0)
        context_score = min(max(ctx.context_tokens, ctx.estimated_input_tokens) / 100_000, 1.0)
        file_score = min(ctx.file_count / 10, 1.0)
        tool_score = min(ctx.tool_count / 8, 1.0)

        score = (
            0.25 * prompt_length_score
            + 0.30 * context_score
            + 0.20 * file_score
            + 0.25 * tool_score
        )

        if ctx.security_sensitive:
            score += 0.20
        if ctx.requires_vision:
            score += 0.10
        score = min(score, 1.0)

        difficulty = _score_to_difficulty(score)

        # 閾値から離れているほどconfidenceが高い
        distance = min(abs(score - b) for b in THRESHOLDS)
        confidence = min(0.50 + distance * 3, 0.95)

境界値からの距離をconfidenceに変換しているのがポイントです。スコアが境界のすぐ近くにある場合は、Heuristicだけでは判断が心もとないので、confidenceを下げて次の層に判断を委ねます。

Layer 2: LLM Classifier

Heuristicのconfidenceが低いとき（0.70未満）だけ、LLM Classifierを呼びます。サンプルにはキーワードマッチの疑似Classifierに加えて、claude CLIを実際のJudgeとして呼び出す実装も入れました。

router.py(Layer2抜粋)
    # 生のprompt文をそのまま渡すと「タスクとして実行しよう」としてしまうため、
    # 分類対象のデータであることが分かるようタグで包む。
    message = (
        f"<task_to_classify>\n{ctx.prompt.strip()}\n</task_to_classify>\n\n"
        "Output the JSON classification now."
    )

    last_error: Optional[Exception] = None
    for attempt in range(1, 3):
        try:
            proc = subprocess.run(
                [
                    "claude", "-p", message,
                    "--model", "haiku",
                    "--output-format", "json",
                    "--allowedTools", "",  # Workerにツールは使わせない
                    # --append-system-prompt だとデフォルトの
                    # コーディングアシスタント人格が残り分類を無視されることがあるため
                    # --system-prompt で完全に差し替える。
                    "--system-prompt", CLASSIFIER_SYSTEM_PROMPT,
                ],

分類対象のプロンプトを <task_to_classify> タグで包んでいます。これは、claude CLIがユーザーメッセージをそのまま「実行すべきタスク」として受け取ってしまうのを防ぐためです。タグで包むことで、「これは実行対象ではなく分類対象のデータである」ことを明示しています。

Layer 3: RouteLLM-style Strong Model Probability

3つ目の層では、difficultyに加えてコンテキスト量・ツール利用の有無・セキュリティ機微度を加味します。これらから、「strongモデルが必要になる確率」をシグモイド関数で算出します。RouteLLMの calculate_strong_win_rate の考え方を、シンプルな式に落とし込んだものです。

router.py(Layer3抜粋)
class StrongModelPredictor:
    def predict(self, ctx: RequestContext, difficulty: DifficultyResult) -> float:
        difficulty_value = {
            Difficulty.LOW: 0.15,
            Difficulty.MEDIUM: 0.40,
            Difficulty.HIGH: 0.70,
            Difficulty.ULTRA: 0.92,
        }[difficulty.difficulty]

        x = difficulty_value
        if ctx.context_tokens > 100_000:
            x += 0.08
        if ctx.requires_tools:
            x += 0.05
        if ctx.security_sensitive:
            x += 0.12

        probability = 1 / (1 + math.exp(-6 * (x - 0.5)))
        return min(max(probability, 0), 1)

difficultyだけでなく、コンテキストの大きさやセキュリティ機微度といった個別の要素も、確率に反映させています。これにより、同じHIGHでも状況によって必要な能力が変わる、という幅を持たせています。

Layer 4: Capability Mapper

4つ目の層で、difficultyとstrong probabilityを「必要な能力値（0.0から1.0）」に変換します。ここがDifficultyとModelを直接結びつけない、設計の要となる部分です。

router.py(Layer4抜粋)
class CapabilityMapper:
    def required_capability(self, difficulty: Difficulty, strong_probability: float) -> float:
        base = {
            Difficulty.LOW: 0.25,
            Difficulty.MEDIUM: 0.50,
            Difficulty.HIGH: 0.72,
            Difficulty.ULTRA: 0.90,
        }[difficulty]

        adjustment = (strong_probability - 0.5) * 0.20
        return min(max(base + adjustment, 0), 1)

difficultyごとのベース値に、Layer 3で求めたstrong probabilityによる微調整を加えています。モデル名の代わりに数値の能力値を扱うことで、モデルカタログが変わってもこの層のロジックは書き直さずに済みます。

Layer 5: Policy / Constraint Filter

5つ目の層で、実際のモデルカタログから候補を絞り込みます。必要な能力値・コンテキストウィンドウ・ツール対応・Vision対応を満たすモデルだけを候補にし、その中からコストとレイテンシが最小のものを選びます。

router.py(Layer5抜粋)
class ModelSelector:
    def select(self, ctx: RequestContext, required_capability: float) -> Model:
        candidates = []
        for model in MODELS:
            if model.capability < required_capability:
                continue
            if model.max_context_tokens < ctx.context_tokens:
                continue
            if ctx.requires_tools and not model.supports_tools:
                continue
            if ctx.requires_vision and not model.supports_vision:
                continue
            candidates.append(model)

        if not candidates:
            # 最強モデルへfall-open
            return max(MODELS, key=lambda m: m.capability)

        # capabilityを満たした中でcost + latencyを最小化
        return min(candidates, key=lambda m: 0.7 * m.cost + 0.3 * m.latency)

条件を満たす候補が1つもない場合は、最も能力の高いモデルへフォールバックしています。安全側に倒す設計です。

Layer 6: Switchyard-style Runtime Escalation

最後の層は、Agentが実際に動き出したあとの再評価です。ツールのエラーが続く、同じ失敗を繰り返す、テストが通らない、作業が停滞している、といったシグナルが出ることがあります。そのときは、現在のモデルより能力の高いモデルへ引き上げます。

router.py(Layer6抜粋)
class EscalationPolicy:
    def should_escalate(self, current_model: Model, signals: RuntimeSignals) -> tuple[bool, list[str]]:
        reasons = []

        if signals.tool_errors >= 2:
            reasons.append("multiple tool errors")
        if signals.repeated_failures >= 2:
            reasons.append("repeated failures")
        if signals.test_failures >= 2:
            reasons.append("repeated test failures")
        if signals.spinning_score >= 0.70:
            reasons.append("agent appears to be spinning")
        if signals.tool_calls >= 5 and signals.progress_score < 0.30:
            reasons.append("low progress despite tool usage")
        if signals.architecture_work and current_model.capability < 0.80:
            reasons.append("architecture work requires higher capability")

        return bool(reasons), reasons

NVIDIA-NeMo/Switchyardの「WRONG系シグナルはcapableモデルへ寄せる」という考え方があります。これをシンプルなルールベースの条件に置き換えた形です。

router.py 全文

router.py 全文（約490行）
router.py
"""依存ゼロ・1ファイルで動く6層LLMルーターのサンプル。

Layer 1 Heuristic -> Layer 2 LLM Classifier(曖昧時のみ) -> Layer 3 Strong-model
probability -> Layer 4 Required capability -> Layer 5 Constraint filter /
cost・latency最小化 -> Layer 6 Runtime escalation、という構成です。

    python3 router.py                  # 疑似Classifierでデモ実行
    python3 router.py --classifier claude   # claude CLI(サブスク認証)で分類
"""

from __future__ import annotations

import argparse
import json
import math
import re
import shutil
import subprocess
import sys
from dataclasses import dataclass, field
from enum import Enum
from typing import Callable, Optional

# ---- Domain model ----

class Difficulty(str, Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    ULTRA = "ultra"

@dataclass
class RequestContext:
    prompt: str

    # 事前に取得できる軽量な特徴量
    context_tokens: int = 0
    estimated_input_tokens: int = 0
    file_count: int = 0
    tool_count: int = 0

    requires_tools: bool = False
    requires_vision: bool = False
    security_sensitive: bool = False

@dataclass
class RuntimeSignals:
    """実行中エージェントの「苦戦度合い」を表すシグナル。"""

    tool_calls: int = 0
    tool_errors: int = 0
    repeated_failures: int = 0
    test_failures: int = 0
    files_read: int = 0
    files_modified: int = 0
    planning: bool = False
    architecture_work: bool = False

    # 同じような行動を繰り返している場合など
    spinning_score: float = 0.0
    # 作業が順調に進んでいるか
    progress_score: float = 1.0

@dataclass
class DifficultyResult:
    difficulty: Difficulty
    score: float
    confidence: float
    source: str

@dataclass
class Model:
    name: str
    capability: float  # 0.0 - 1.0
    cost: float  # 相対コスト
    latency: float  # 相対latency
    max_context_tokens: int
    supports_tools: bool = True
    supports_vision: bool = False

@dataclass
class RouteDecision:
    model: Model
    difficulty: Difficulty
    required_capability: float
    strong_model_probability: float
    reason: list[str] = field(default_factory=list)

# ---- Model Catalog ----

MODELS = [
    Model(name="small", capability=0.35, cost=0.10, latency=0.20, max_context_tokens=128_000),
    Model(name="medium", capability=0.60, cost=0.30, latency=0.40, max_context_tokens=200_000),
    Model(
        name="strong",
        capability=0.82,
        cost=0.70,
        latency=0.65,
        max_context_tokens=200_000,
        supports_vision=True,
    ),
    Model(
        name="frontier",
        capability=1.00,
        cost=1.00,
        latency=1.00,
        max_context_tokens=1_000_000,
        supports_vision=True,
    ),
]

THRESHOLDS = (0.30, 0.52, 0.74)

def _score_to_difficulty(score: float) -> Difficulty:
    if score < THRESHOLDS[0]:
        return Difficulty.LOW
    if score < THRESHOLDS[1]:
        return Difficulty.MEDIUM
    if score < THRESHOLDS[2]:
        return Difficulty.HIGH
    return Difficulty.ULTRA

# ---- Layer 1: Cheap Heuristic ----

class HeuristicClassifier:
    def classify(self, ctx: RequestContext) -> DifficultyResult:
        prompt_length_score = min(len(ctx.prompt) / 4000, 1.0)
        context_score = min(max(ctx.context_tokens, ctx.estimated_input_tokens) / 100_000, 1.0)
        file_score = min(ctx.file_count / 10, 1.0)
        tool_score = min(ctx.tool_count / 8, 1.0)

        score = (
            0.25 * prompt_length_score
            + 0.30 * context_score
            + 0.20 * file_score
            + 0.25 * tool_score
        )

        if ctx.security_sensitive:
            score += 0.20
        if ctx.requires_vision:
            score += 0.10
        score = min(score, 1.0)

        difficulty = _score_to_difficulty(score)

        # 閾値から離れているほどconfidenceが高い
        distance = min(abs(score - b) for b in THRESHOLDS)
        confidence = min(0.50 + distance * 3, 0.95)

        return DifficultyResult(difficulty=difficulty, score=score, confidence=confidence, source="heuristic")

# ---- Layer 2: LLM Classifier (Heuristicが曖昧な場合だけ呼ぶ) ----

LLMClassifierFn = Callable[[RequestContext], DifficultyResult]

# 記事末尾の「LLM Classifierのprompt」がベース。
# ただし claude CLI(Claude Code)は素のprompt文だとコーディングエージェント
# 人格のままPlanモード的な質問を返すことがあったため、
# 「実行はしない/分類だけ返す」を明示し、タスクは<task_to_classify>タグで
# 包んでデータとして渡す(下のclaude_cli_classifier参照)。
CLASSIFIER_SYSTEM_PROMPT = """You are a routing classifier for an AI inference gateway.
You do not execute, plan, or investigate the task. You only classify it.

Your job is to estimate the minimum reasoning capability required to
complete the user's task reliably.

Do NOT choose a specific model.
Do NOT consider model price.
Do NOT ask clarifying questions.

Classify required capability into: LOW, MEDIUM, HIGH, ULTRA.

Consider: reasoning complexity, ambiguity, amount of context, number of
dependent steps, tool usage, failure cost, domain expertise required.

Return JSON only, no markdown fences, no commentary:
{"difficulty": "low|medium|high|ultra", "confidence": 0.0, "reason": "short explanation"}
"""

# 実測スコアが無いため、閾値レンジの中央値を score として採用する。
_MIDPOINT = {
    Difficulty.LOW: 0.15,
    Difficulty.MEDIUM: 0.41,
    Difficulty.HIGH: 0.63,
    Difficulty.ULTRA: 0.87,
}

def example_llm_classifier(ctx: RequestContext) -> DifficultyResult:
    """実際にはOpenAI / Anthropic / Local LLMなどへ置換する疑似Classifier。"""

    prompt = ctx.prompt.lower()
    score = 0.45

    high_keywords = [
        "architecture", "migration", "security", "root cause", "refactor",
        "production", "distributed",
        "設計", "移行", "セキュリティ", "原因調査", "リファクタ",
    ]
    ultra_keywords = [
        "large-scale", "enterprise", "production incident", "threat model",
        "全体設計", "大規模",
    ]

    if any(x in prompt for x in high_keywords):
        score = 0.70
    if any(x in prompt for x in ultra_keywords):
        score = 0.85

    return DifficultyResult(
        difficulty=_score_to_difficulty(score),
        score=score,
        confidence=0.85,
        source="llm-classifier(pseudo)",
    )

def _extract_json_object(text: str) -> dict:
    """```json フェンス付き/生JSON どちらでも受け付ける。"""

    cleaned = re.sub(r"^```[a-zA-Z0-9_+-]*\n|\n```$", "", text.strip())
    match = re.search(r"\{.*\}", cleaned, re.DOTALL)
    if not match:
        raise ValueError(f"JSONが見つかりません: {text!r}")
    return json.loads(match.group(0))

def claude_cli_classifier(ctx: RequestContext) -> DifficultyResult:
    """claude CLI(サブスクリプション認証)をLLM Classifierとして呼ぶ。

    `claude` が無い/失敗する場合は疑似Classifierへフォールバックする。
    """

    if shutil.which("claude") is None:
        print("[claude_cli_classifier] claude CLI が見つかりません。疑似Classifierにフォールバックします。", file=sys.stderr)
        return example_llm_classifier(ctx)

    # 生のprompt文をそのまま渡すと「タスクとして実行しよう」としてしまうため、
    # 分類対象のデータであることが分かるようタグで包む。
    message = (
        f"<task_to_classify>\n{ctx.prompt.strip()}\n</task_to_classify>\n\n"
        "Output the JSON classification now."
    )

    last_error: Optional[Exception] = None
    for attempt in range(1, 3):
        try:
            proc = subprocess.run(
                [
                    "claude", "-p", message,
                    "--model", "haiku",
                    "--output-format", "json",
                    "--allowedTools", "",  # Workerにツールは使わせない
                    # --append-system-prompt だとデフォルトの
                    # コーディングアシスタント人格が残り分類を無視されることがあるため
                    # --system-prompt で完全に差し替える。
                    "--system-prompt", CLASSIFIER_SYSTEM_PROMPT,
                ],
                capture_output=True,
                text=True,
                timeout=60,
                check=False,
            )
            data = json.loads(proc.stdout)
            if data.get("is_error"):
                raise RuntimeError(f"claude CLI error: {data.get('result')}")

            payload = _extract_json_object(data["result"])
            difficulty = Difficulty(str(payload["difficulty"]).lower())
            return DifficultyResult(
                difficulty=difficulty,
                score=_MIDPOINT[difficulty],
                confidence=float(payload.get("confidence", 0.75)),
                source="claude-cli",
            )
        except Exception as exc:  # noqa: BLE001 - CLI呼び出しの失敗理由を一括で拾う
            last_error = exc
            print(f"[claude_cli_classifier] 試行{attempt}/2 失敗: {exc}", file=sys.stderr)

    print(f"[claude_cli_classifier] 疑似Classifierにフォールバックします(最終エラー: {last_error})", file=sys.stderr)
    return example_llm_classifier(ctx)

# ---- Layer 3: RouteLLM-style Strong-model probability ----

class StrongModelPredictor:
    def predict(self, ctx: RequestContext, difficulty: DifficultyResult) -> float:
        difficulty_value = {
            Difficulty.LOW: 0.15,
            Difficulty.MEDIUM: 0.40,
            Difficulty.HIGH: 0.70,
            Difficulty.ULTRA: 0.92,
        }[difficulty.difficulty]

        x = difficulty_value
        if ctx.context_tokens > 100_000:
            x += 0.08
        if ctx.requires_tools:
            x += 0.05
        if ctx.security_sensitive:
            x += 0.12

        probability = 1 / (1 + math.exp(-6 * (x - 0.5)))
        return min(max(probability, 0), 1)

# ---- Layer 4: Capability Mapper (Difficulty -> Required Capability) ----

class CapabilityMapper:
    def required_capability(self, difficulty: Difficulty, strong_probability: float) -> float:
        base = {
            Difficulty.LOW: 0.25,
            Difficulty.MEDIUM: 0.50,
            Difficulty.HIGH: 0.72,
            Difficulty.ULTRA: 0.90,
        }[difficulty]

        adjustment = (strong_probability - 0.5) * 0.20
        return min(max(base + adjustment, 0), 1)

# ---- Layer 5: Policy / Constraint Filter ----

class ModelSelector:
    def select(self, ctx: RequestContext, required_capability: float) -> Model:
        candidates = []
        for model in MODELS:
            if model.capability < required_capability:
                continue
            if model.max_context_tokens < ctx.context_tokens:
                continue
            if ctx.requires_tools and not model.supports_tools:
                continue
            if ctx.requires_vision and not model.supports_vision:
                continue
            candidates.append(model)

        if not candidates:
            # 最強モデルへfall-open
            return max(MODELS, key=lambda m: m.capability)

        # capabilityを満たした中でcost + latencyを最小化
        return min(candidates, key=lambda m: 0.7 * m.cost + 0.3 * m.latency)

# ---- Layer 6: Switchyard-style runtime escalation ----

class EscalationPolicy:
    def should_escalate(self, current_model: Model, signals: RuntimeSignals) -> tuple[bool, list[str]]:
        reasons = []

        if signals.tool_errors >= 2:
            reasons.append("multiple tool errors")
        if signals.repeated_failures >= 2:
            reasons.append("repeated failures")
        if signals.test_failures >= 2:
            reasons.append("repeated test failures")
        if signals.spinning_score >= 0.70:
            reasons.append("agent appears to be spinning")
        if signals.tool_calls >= 5 and signals.progress_score < 0.30:
            reasons.append("low progress despite tool usage")
        if signals.architecture_work and current_model.capability < 0.80:
            reasons.append("architecture work requires higher capability")

        return bool(reasons), reasons

# ---- Main Router ----

class IntelligentRouter:
    def __init__(self, llm_classifier: Optional[LLMClassifierFn] = None):
        self.heuristic = HeuristicClassifier()
        self.llm_classifier = llm_classifier or example_llm_classifier
        self.strong_predictor = StrongModelPredictor()
        self.capability_mapper = CapabilityMapper()
        self.model_selector = ModelSelector()
        self.escalation = EscalationPolicy()

    def route(self, ctx: RequestContext) -> RouteDecision:
        reasons = []

        # Step 1: Cheap heuristic
        difficulty = self.heuristic.classify(ctx)
        reasons.append(f"Heuristic score={difficulty.score:.2f}, confidence={difficulty.confidence:.2f}")

        # Step 2: 曖昧なときだけLLM Judge
        if difficulty.confidence < 0.70:
            llm_result = self.llm_classifier(ctx)
            reasons.append(f"LLM classifier used: {llm_result.difficulty.value} (source={llm_result.source})")
            difficulty = llm_result

        # Step 3: RouteLLM-style Strong Model必要確率
        strong_probability = self.strong_predictor.predict(ctx, difficulty)
        reasons.append(f"Strong-model probability={strong_probability:.2f}")

        # Step 4: Capabilityへ変換
        required_capability = self.capability_mapper.required_capability(difficulty.difficulty, strong_probability)
        reasons.append(f"Required capability={required_capability:.2f}")

        # Step 5: Cost / latency / constraints
        model = self.model_selector.select(ctx, required_capability)
        reasons.append(f"Selected model={model.name}")

        return RouteDecision(
            model=model,
            difficulty=difficulty.difficulty,
            required_capability=required_capability,
            strong_model_probability=strong_probability,
            reason=reasons,
        )

    def reevaluate(self, ctx: RequestContext, current: RouteDecision, runtime: RuntimeSignals) -> RouteDecision:
        escalate, reasons = self.escalation.should_escalate(current.model, runtime)

        if not escalate:
            current.reason.append("Runtime signals: continue current model")
            return current

        # Capabilityを引き上げる
        new_capability = min(current.required_capability + 0.20, 1.0)
        new_model = self.model_selector.select(ctx, new_capability)

        return RouteDecision(
            model=new_model,
            difficulty=current.difficulty,
            required_capability=new_capability,
            strong_model_probability=max(current.strong_model_probability, 0.80),
            reason=current.reason
            + [f"Escalation: {', '.join(reasons)}", f"Escalated to {new_model.name}"],
        )

# ---- Example ----

def _print_decision(title: str, decision: RouteDecision) -> None:
    print(f"\n=== {title} ===")
    print("difficulty:", decision.difficulty.value)
    print("strong probability:", round(decision.strong_model_probability, 2))
    print("required capability:", round(decision.required_capability, 2))
    print("model:", decision.model.name)
    for reason in decision.reason:
        print("-", reason)

def main() -> None:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument(
        "--classifier",
        choices=["pseudo", "claude"],
        default="pseudo",
        help="Layer2 LLM Classifierの実装 (既定: pseudo=キーワード疑似分類)",
    )
    args = parser.parse_args()

    classifier_fn = claude_cli_classifier if args.classifier == "claude" else example_llm_classifier
    router = IntelligentRouter(llm_classifier=classifier_fn)

    # --- 大規模・security-sensitiveな依頼: HIGH/ULTRA -> frontierへ ---
    big_request = RequestContext(
        prompt=(
            "大規模なNext.jsアプリの認証基盤をマルチテナント対応へ移行したい。"
            "現行コードを調査して、セキュリティを考慮した移行設計を作ってください。"
        ),
        context_tokens=45_000,
        estimated_input_tokens=8_000,
        file_count=12,
        tool_count=5,
        requires_tools=True,
        security_sensitive=True,
    )
    decision = router.route(big_request)
    _print_decision("INITIAL ROUTE (large / security-sensitive)", decision)

    runtime = RuntimeSignals(
        tool_calls=8, tool_errors=3, repeated_failures=2,
        test_failures=2, spinning_score=0.75, progress_score=0.20,
    )
    decision2 = router.reevaluate(big_request, decision, runtime)
    _print_decision("AFTER EXECUTION (runtime escalation)", decision2)

    # --- 小さな依頼: LOW -> smallモデルで足りる ---
    small_request = RequestContext(
        prompt="この関数のdocstringを直してください。",
        context_tokens=0,
        estimated_input_tokens=200,
        file_count=1,
        tool_count=1,
        requires_tools=False,
        security_sensitive=False,
    )
    decision3 = router.route(small_request)
    _print_decision("SMALL REQUEST (docstring fix)", decision3)

if __name__ == "__main__":
    main()

実行方法と実行ログ

実行はコマンド1つで完結します。--classifier オプションを付けなければ、キーワードマッチの疑似Classifierでデモが動きます。

$ python3 router.py

大規模でセキュリティ機微なリクエストと、docstring修正のような小さなリクエストを1回ずつ流すと、モデル選択が対照的になります。

=== INITIAL ROUTE (large / security-sensitive) ===
difficulty: ultra
strong probability: 0.97
required capability: 0.99
model: frontier
- Heuristic score=0.70, confidence=0.63
- LLM classifier used: ultra (source=llm-classifier(pseudo))
- Strong-model probability=0.97
- Required capability=0.99
- Selected model=frontier

=== AFTER EXECUTION (runtime escalation) ===
difficulty: ultra
strong probability: 0.97
required capability: 1.0
model: frontier
- Heuristic score=0.70, confidence=0.63
- LLM classifier used: ultra (source=llm-classifier(pseudo))
- Strong-model probability=0.97
- Required capability=0.99
- Selected model=frontier
- Escalation: multiple tool errors, repeated failures, repeated test failures, agent appears to be spinning, low progress despite tool usage
- Escalated to frontier

=== SMALL REQUEST (docstring fix) ===
difficulty: low
strong probability: 0.11
required capability: 0.17
model: small
- Heuristic score=0.05, confidence=0.95
- Strong-model probability=0.11
- Required capability=0.17
- Selected model=small

小さな依頼は model: small が選ばれ、大規模でセキュリティが絡む依頼は model: frontier が選ばれています。さらに、実行後にツールのエラーやテスト失敗、spinning（同じ行動の繰り返し）が観測されることがあります。そのときはEscalationが働き、より能力の高いモデルへ引き上がる様子も確認できました。

--classifier claude: 実際のLLM Judgeに差し替える

疑似Classifierの代わりに、claude CLIを実際のLLM Judgeとして呼び出すこともできます。Claude Codeのサブスクリプション認証があれば、claude -p コマンドをそのままサブプロセスとして呼べます。

$ python3 router.py --classifier claude

=== INITIAL ROUTE (large / security-sensitive) ===
difficulty: ultra
strong probability: 0.97
required capability: 0.99
model: frontier
- Heuristic score=0.70, confidence=0.63
- LLM classifier used: ultra (source=claude-cli)
- Strong-model probability=0.97
- Required capability=0.99
- Selected model=frontier
...(以下は疑似Classifier版と同じ)

source=claude-cli と表示されています。これは、疑似Classifierではなく実際に claude CLIを呼び出して分類していることを示しています。実行しているコマンドは次の形です。

claude -p --model haiku --output-format json --system-prompt <classifier prompt>

モデルにhaikuを指定し、--system-prompt で分類用のプロンプトを渡しています。

実装の過程では、システムプロンプトの渡し方でつまずきました。最初は --append-system-prompt で試しました。すると、Claude Codeの既定のコーディングエージェント人格が残ってしまいました。JSON分類の代わりに、「プロジェクトのパスを教えてください」のような対話的な応答が返ることがありました。既定のシステムプロンプトを完全に差し替える --system-prompt に切り替えました。渡すメッセージも <task_to_classify> タグで分類対象であることを明示しました。その結果、安定してJSON形式の分類が返るようになりました。

--classifier claude は今回の実測では5回連続で成功しました。ただし、CLI側の挙動が変われば失敗することもあり、100%を保証するものではありません。サンプルにはCLI呼び出しが失敗した場合に疑似Classifierへフォールバックする経路も実装してあります。

Classifier の Prompt

難易度を判定するClassifierのPromptは、pi-model-autoの「能力と価格を分離する」考え方をベースにしています。次のような内容で設計しました。

classifier-prompt.txt
You are a routing classifier for an AI inference gateway.

Your task is to estimate the minimum reasoning capability
required to complete the user's task reliably.

Do NOT choose a specific model.
Do NOT consider model price.

Classify required capability into:

LOW
- simple factual questions
- rewriting
- summarization
- formatting
- simple local code changes

MEDIUM
- normal coding
- debugging with clear symptoms
- small feature implementation
- several related files
- normal tool usage

HIGH
- multi-file debugging
- unclear root cause
- complex code changes
- API or system design
- substantial tool usage
- large context

ULTRA
- system architecture
- security-sensitive decisions
- production incidents
- large migrations
- large refactoring
- long-horizon investigation
- highly ambiguous problems

Consider:

1. reasoning complexity
2. ambiguity
3. amount of context
4. number of dependent steps
5. tool usage
6. failure cost
7. domain expertise required

Return JSON only.

{
    "difficulty": "low|medium|high|ultra",
    "confidence": 0.0,
    "reason": "short explanation"
}

このPromptで重要なのは2行です。「Do NOT choose a specific model」と「Do NOT consider model price」です。難易度の判定とモデル選択を同じPromptの中で一緒に考えさせると、Classifierの出力が不安定になったり、判定理由に価格の話が混ざったりします。難易度の判定はあくまで「このタスクにどれくらいの能力が要るか」だけに専念させます。コストやレイテンシを見てどのモデルを使うかは、Layer 5のPolicy/Constraint Filterという別の層に任せています。router.py内で実際に使っている CLASSIFIER_SYSTEM_PROMPT は、このPromptをCLI実行向けに簡略化したものです。実行や質問を禁止する一文も加えています。

本番に近づけるなら

このサンプルをそのまま使うのではなく、本番に近づけるならRouterを4つのコンポーネントに分けるのが良いと考えています。

  • Capability Estimator: リクエストから必要な能力値を見積もる（Layer 1〜4に相当）
  • Constraint Engine: コンテキスト長・ツール要件・セキュリティポリシー・データの所在といった制約を適用する
  • Model Optimizer: コスト・レイテンシ・Quota・キャッシュ状態・過去の成功率を見てモデルを決める（Layer 5に相当）
  • Runtime Evaluator: 実行中のエラーやテスト失敗、停滞を監視し、Escalationを判断する（Layer 6に相当）

この形にしておくと、pleaseai/shuntのようなLLM Gatewayの前段に置く構成にもできます。Intelligent Routerが「必要な能力レベル」を決めます。実際のProvider振り分けは、pleaseai/shuntのようなGatewayに任せる、という役割分担です。個人的には、次の段階としてこのサンプルをFastAPI化したいと考えています。/v1/messages のようなエンドポイントとして受けられるようにすると、実際のGatewayとして試せるレベルに近づきます。

まとめ

3つのshuntを調べて、「ルーティング」という言葉が指す対象が3つに分かれていることが分かりました。アカウント・モデル/Provider・作業そのもの、という3つです。自分が欲しかったのは「タスク難易度からモデルを選ぶ」仕組みでした。これはNVIDIA-NeMo/Switchyard・pi-model-auto・RouteLLMといった既存OSSの考え方を組み合わせたものです。依存ゼロのPythonファイル1本として動かせました。

設計で重要だったのは3点です。1つ目は、難易度とモデルを直接結びつけずに「必要な能力」という抽象レイヤーを挟むことです。2つ目は、Classifierに価格を判断させないことです。3つ目は、ルーティングを一発で決め切らず、実行中に再評価することです。モデルのカタログは今後も入れ替わりますが、この3つの考え方は比較的長く使えると考えています。

参考

0
1
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?