連載：AIに仕事を奪われる不安から始めるハーネス作成入門
第12回 / 全24回（週2回・12週間）

モデルルーターの疑似コードでLLM切替を設計する

前回、ローカルLLMと外部LLMの使い分け基準を整理しました。「なるほど、タスクで使い分ければいいのか」と理解した次の疑問は、「で、実際のコードでどう切り替えるの？」でしょう。

今回は、Pythonの疑似コードで「モデルルーター」を設計します。タスク種別に応じてローカルLLMとAPI LLMを自動的に切り替える仕組みを、一緒に読み解いていきましょう。

📖 連載目次

回	タイトル	状態
1〜4	基礎編（不安の言語化〜ハーネスの概念）	✅
5〜9	実践編（プロンプト設計〜MCP連携）	✅
10	設計メモをKnowledge MCPへ登録する前提のMarkdownテンプレート	✅
11	ローカルLLMを使うべき場面と外部LLMに任せる場面	✅
12	モデルルーターの疑似コードでLLM切替を設計する	📖
13	AIハーネスにログ設計が必要な理由	─
14〜24	運用設計・テスト・発展編	─

1. モデルルーターとは

モデルルーターは、タスクの特性に応じて適切なLLMを自動選択するコンポーネントです。前回の判断基準表をコードで表現したもの、と考えてください。

タスク受付
  └─ ModelRouter.route(task)
       ├─ ルールマッチ → ローカルLLM
       ├─ ルールマッチ → 外部LLM
       └─ フォールバック → デフォルトLLM

2. 設計方針

モデルルーターの設計で重視するポイントは3つです。

ルールベース: if/elseや設定ファイルで明示的にルーティング
フォールバック付き: ローカルが失敗したら外部に自動切替
設定駆動: ルーティング条件をコード外に切り出す

3. タスク分類の定義

まず、タスクを分類するためのデータ構造を定義します。

from dataclasses import dataclass
from enum import Enum
from typing import Optional

class TaskComplexity(Enum):
    """task complexity レベル"""
    LOW = "low"       # 分類・抽出など単純なタスク
    MEDIUM = "medium" # 要約・変換などの中程度のタスク
    HIGH = "high"     # 複雑な推論・生成タスク

class PrivacyLevel(Enum):
    """データの機密度"""
    PUBLIC = "public"         # 公開情報
    INTERNAL = "internal"     # 社内情報
    CONFIDENTIAL = "confidential"  # 機密情報

@dataclass
class TaskRequest:
    """LLMに渡すタスクの定義"""
    task_type: str              # タスク種別（"classify", "summarize", "generate"等）
    complexity: TaskComplexity  # 複雑度
    privacy: PrivacyLevel      # 機密度
    prompt: str                # プロンプト本文
    max_tokens: int = 1000     # 最大トークン数
    metadata: Optional[dict] = None

4. ルーティング設定ファイル

ルーティング条件をYAMLで外出しします。コードを変えずにルールを調整できるのがメリットです。

# config/routing_rules.yaml
default_model: "openai:gpt-4o-mini"

rules:
  - name: "confidential_local"
    description: "機密データはローカルで処理"
    conditions:
      privacy: "confidential"
    model: "ollama:llama3.1"
    priority: 100

  - name: "simple_task_local"
    description: "単純タスクはローカルでコスト削減"
    conditions:
      complexity: "low"
      privacy: ["public", "internal"]
    model: "ollama:llama3.1"
    priority: 50

  - name: "complex_task_api"
    description: "複雑タスクはAPIで高精度"
    conditions:
      complexity: "high"
    model: "openai:gpt-4o"
    priority: 30

fallback:
  model: "openai:gpt-4o-mini"
  enabled: true

5. ModelRouterクラスの実装

ルーティングの中核となるクラスです。

import yaml
from typing import Optional

class ModelRouter:
    """タスクに応じてLLMを自動選択するルーター"""
    
    def __init__(self, config_path: str = "config/routing_rules.yaml"):
        with open(config_path, "r", encoding="utf-8") as f:
            self.config = yaml.safe_load(f)
        self.rules = self.config.get("rules", [])
        self.default_model = self.config.get("default_model", "openai:gpt-4o-mini")
        self.fallback = self.config.get("fallback", {})
    
    def route(self, task: TaskRequest) -> str:
        """タスクに対して最適なモデルを返す"""
        # プライオリティ順にルールを評価
        sorted_rules = sorted(
            self.rules,
            key=lambda r: r.get("priority", 0),
            reverse=True
        )
        
        for rule in sorted_rules:
            if self._match(rule, task):
                return rule["model"]
        
        return self.default_model
    
    def _match(self, rule: dict, task: TaskRequest) -> bool:
        """ルールの条件がタスクにマッチするか判定"""
        conditions = rule.get("conditions", {})
        
        for key, expected in conditions.items():
            actual = self._get_task_attr(task, key)
            if actual is None:
                return False
            
            # リストの場合はいずれかにマッチすればOK
            if isinstance(expected, list):
                if actual not in expected:
                    return False
            elif actual != expected:
                return False
        
        return True
    
    def _get_task_attr(self, task: TaskRequest, key: str) -> Optional[str]:
        """タスクから属性値を取得"""
        value = getattr(task, key, None)
        if hasattr(value, "value"):  # Enumの場合
            return value.value
        return value

# 使用例
router = ModelRouter()
task = TaskRequest(
    task_type="classify",
    complexity=TaskComplexity.LOW,
    privacy=PrivacyLevel.INTERNAL,
    prompt="このメールをカテゴリ分類してください"
)
model = router.route(task)
print(f"選択されたモデル: {model}")
# => 選択されたモデル: ollama:llama3.1

6. フォールバックの実装

ローカルLLMがダウンしている場合に外部APIに自動切替する仕組みです。

class ModelExecutor:
    """モデル実行 + フォールバック"""
    
    def __init__(self, router: ModelRouter):
        self.router = router
    
    def execute(self, task: TaskRequest) -> dict:
        """タスクを実行し、失敗時はフォールバック"""
        primary_model = self.router.route(task)
        
        try:
            result = self._call_model(primary_model, task)
            return {
                "model_used": primary_model,
                "fallback": False,
                "result": result,
            }
        except ModelUnavailableError:
            # フォールバックが有効なら代替モデルでリトライ
            fallback_config = self.router.fallback
            if fallback_config.get("enabled"):
                fallback_model = fallback_config["model"]
                result = self._call_model(fallback_model, task)
                return {
                    "model_used": fallback_model,
                    "fallback": True,
                    "result": result,
                }
            raise
    
    def _call_model(self, model_id: str, task: TaskRequest) -> str:
        """モデルIDに応じて適切なクライアントを呼ぶ（疑似コード）"""
        provider, model_name = model_id.split(":", 1)
        
        if provider == "ollama":
            return self._call_ollama(model_name, task)
        elif provider == "openai":
            return self._call_openai(model_name, task)
        else:
            raise ValueError(f"Unknown provider: {provider}")
    
    def _call_ollama(self, model: str, task: TaskRequest) -> str:
        """Ollama呼び出し（疑似コード）"""
        # 実際はOllamaのPythonライブラリを使用
        print(f"  Ollama({model}): {task.prompt[:30]}...")
        return "Ollamaからの応答"
    
    def _call_openai(self, model: str, task: TaskRequest) -> str:
        """OpenAI呼び出し（疑似コード）"""
        # 実際はOpenAI SDKを使用
        print(f"  OpenAI({model}): {task.prompt[:30]}...")
        return "OpenAIからの応答"

class ModelUnavailableError(Exception):
    pass

7. コードの解説

主要なポイントを整理します。

コンポーネント	役割	ポイント
`TaskRequest`	タスクの構造化	判断に必要な情報を明示的に持たせる
`ModelRouter`	ルーティング判断	YAML設定でルール変更が容易
`ModelExecutor`	実行 + フォールバック	失敗時の自動切替で信頼性向上
`routing_rules.yaml`	ルール定義	コードを変えずにルール調整

8. 拡張ポイント

この疑似コードをベースに、以下の拡張が考えられます。

コストベースルーティング: 月額予算を超えたらローカルに切替
ロードバランシング: 複数のローカルモデルに分散
A/Bテスト: モデルごとの精度比較を自動化
キャッシュレイヤー: 同じプロンプトの結果をキャッシュ

ただし、最初から全部作る必要はありません。まずは基本のルーティング + フォールバックだけで十分です。

9. テストの考え方

モデルルーターのテストは、主に3つの観点で行います。

# テスト例（pytest）
def test_confidential_routes_to_local():
    """機密データはローカルにルーティングされる"""
    router = ModelRouter("config/routing_rules.yaml")
    task = TaskRequest(
        task_type="summarize",
        complexity=TaskComplexity.LOW,
        privacy=PrivacyLevel.CONFIDENTIAL,
        prompt="機密文書を要約",
    )
    assert router.route(task) == "ollama:llama3.1"

def test_complex_routes_to_api():
    """複雑タスクはAPIにルーティングされる"""
    router = ModelRouter("config/routing_rules.yaml")
    task = TaskRequest(
        task_type="generate",
        complexity=TaskComplexity.HIGH,
        privacy=PrivacyLevel.PUBLIC,
        prompt="複雑な分析レポートを生成",
    )
    assert router.route(task) == "openai:gpt-4o"

def test_fallback_on_failure():
    """ローカル失敗時にフォールバックが動作する"""
    # ModelExecutorのexecuteをテスト
    pass

10. 設計メモとして記録する

モデルルーターの設計判断も、第10回のテンプレートでKnowledge MCPに登録しておきましょう。「なぜルールベースにしたのか」「なぜプライオリティをこの順にしたのか」——こうした判断の経緯が、後からのメンテナンスで役立ちます。

11. まとめ

今回のポイントを整理します。

モデルルーターは「タスクに応じたLLM自動選択」の仕組み
TaskRequest でタスクを構造化し、YAMLでルーティングルールを定義
ModelRouter.route() でプライオリティ順にルールを評価
ModelExecutor でフォールバック付き実行を実現
テストは「ルーティング正確性」「フォールバック動作」「設定変更反映」の3観点

🔜 次回予告

第13回：AIハーネスにログ設計が必要な理由

モデルルーターで自動切替ができるようになったら、次に必要なのは「何が起きたかを記録する」仕組みです。AIエージェントが業務で使われるとき、「失敗したらどうするのか」「暴走しないか」という不安があります。その不安を「ログで可視化する」ことで解消するアプローチを解説します。

どんな項目をログに残すべきか、ログレベルの設計、そして人間承認（Human-in-the-loop）との連携——実務で使えるログ項目表を提示します。

連載情報
連載名：AIに仕事を奪われる不安から始めるハーネス作成入門
全24回（12週間・週2回）
著者: @singula00991
投稿曜日：火曜日（キャリア・設計・概念）/ 金曜日（実装・検証・テンプレート）