ストーリー

モデルルーティングの設計

タスク複雑度に応じたモデル選定

インテリジェントルーティング

from enum import Enum

class TaskComplexity(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"

class ModelRouter:
    """タスク複雑度に基づくモデルルーティング"""

    MODEL_MAP = {
        TaskComplexity.LOW: {
            "model": "gpt-4o-mini",
            "cost_per_1m_input": 0.15,
            "cost_per_1m_output": 0.60,
        },
        TaskComplexity.MEDIUM: {
            "model": "gpt-4o-mini",
            "cost_per_1m_input": 0.15,
            "cost_per_1m_output": 0.60,
        },
        TaskComplexity.HIGH: {
            "model": "gpt-4o",
            "cost_per_1m_input": 2.50,
            "cost_per_1m_output": 10.00,
        },
        TaskComplexity.CRITICAL: {
            "model": "gpt-4o",
            "cost_per_1m_input": 2.50,
            "cost_per_1m_output": 10.00,
        },
    }

    def classify_task(self, task_type: str, input_length: int, requires_reasoning: bool) -> TaskComplexity:
        """タスクの複雑度を判定"""
        if task_type in ["classification", "sentiment", "extraction"]:
            return TaskComplexity.LOW
        elif task_type in ["summary", "translation", "faq"]:
            return TaskComplexity.MEDIUM
        elif requires_reasoning or input_length > 10000:
            return TaskComplexity.HIGH
        else:
            return TaskComplexity.MEDIUM

    def route(self, task_type: str, input_length: int, requires_reasoning: bool = False) -> dict:
        """最適なモデルにルーティング"""
        complexity = self.classify_task(task_type, input_length, requires_reasoning)
        return self.MODEL_MAP[complexity]

カスケードパターン

段階的モデル呼び出し

コストの低いモデルで最初に処理し、品質が不十分な場合のみ高性能モデルにエスカレーションするパターンです。

class CascadeRouter:
    """カスケード型モデルルーティング"""

    def __init__(self, quality_threshold: float = 0.8):
        self.quality_threshold = quality_threshold

    async def process(self, prompt: str) -> dict:
        """カスケード処理"""
        # Stage 1: 低コストモデルで処理
        result = await self._call_model("gpt-4o-mini", prompt)
        quality = await self._evaluate_quality(result)

        if quality >= self.quality_threshold:
            return {
                "response": result,
                "model_used": "gpt-4o-mini",
                "quality_score": quality,
                "cascaded": False
            }

        # Stage 2: 品質不足の場合、高性能モデルにエスカレーション
        result = await self._call_model("gpt-4o", prompt)
        quality = await self._evaluate_quality(result)

        return {
            "response": result,
            "model_used": "gpt-4o",
            "quality_score": quality,
            "cascaded": True
        }

    async def _call_model(self, model: str, prompt: str) -> str:
        """モデルを呼び出す（実装は省略）"""
        pass

    async def _evaluate_quality(self, response: str) -> float:
        """品質を簡易評価（0-1）"""
        # 実装例: 応答長、構造、キーワードの有無でスコアリング
        pass

カスケードの効果

トークン最適化

プロンプト圧縮テクニック

バッチ処理の活用

まとめ

チェックリスト

• タスク複雑度に応じたモデル選定基準を理解した
• カスケードパターンの仕組みと効果を把握した
• プロンプト圧縮テクニックを学んだ
• バッチ処理の適用場面を理解した

次のステップへ

次はToken最適化とFinOpsの実践を学びます。

推定読了時間: 30分