Gemma 4 ローカル推論と Gemini API の使い分けで月額¥32,000を¥9,000台にした — ハイブリッドルーターの本番設計

# hybrid_router.py
import asyncio
import os
import time
from enum import Enum
from dataclasses import dataclass
from typing import Optional, AsyncGenerator
import google.generativeai as genai
from openai import AsyncOpenAI
from dotenv import load_dotenv
 
load_dotenv()
 
class ModelTarget(Enum):
    GEMINI = "gemini"
    GEMMA_LOCAL = "gemma_local"
 
@dataclass
class RoutingDecision:
    target: ModelTarget
    reason: str
    estimated_cost_usd: float
 
class HybridRouter:
    """
    Gemini API と Gemma 4（ローカル）を自動で使い分けるルーター。
    コスト・品質・レイテンシのバランスを自動最適化する。
    """
 
    # Gemini API の概算コスト（USD per 1M tokens）
    GEMINI_COSTS = {
        "gemini-2.5-pro": {"input": 1.25, "output": 10.0},
        "gemini-2.5-flash": {"input": 0.15, "output": 0.60},
        "gemini-2.0-flash": {"input": 0.10, "output": 0.40},
    }
 
    # ローカル推論の閾値（文字数）
    LOCAL_THRESHOLD_CHARS = 1500
    COMPLEX_TASK_KEYWORDS = [
        "コードを書いて", "実装して", "デバッグ", "分析して", "推論",
        "なぜ", "どのように", "比較して", "評価して",
        "write code", "implement", "debug", "analyze", "reasoning",
        "compare", "evaluate", "explain why"
    ]
 
    def __init__(
        self,
        gemini_model: str = "gemini-2.5-flash",
        gemma_model: str = None,
        force_local_threshold: float = 0.0,  # 0.0=判定しない
    ):
        # Gemini API 初期化
        genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
        self.gemini_client = genai.GenerativeModel(gemini_model)
        self.gemini_model_name = gemini_model
 
        # Gemma 4（Ollama）クライアント初期化
        self.gemma_client = AsyncOpenAI(
            base_url=f"{os.getenv('OLLAMA_BASE_URL', 'http://localhost:11434')}/v1",
            api_key="ollama",  # Ollama は認証不要
        )
        self.gemma_model = gemma_model or os.getenv("GEMMA_MODEL", "gemma4:27b")
        self.force_local_threshold = force_local_threshold
 
        # 統計収集
        self.stats = {"gemini_calls": 0, "local_calls": 0, "total_cost_usd": 0.0}
 
    def _decide_route(self, prompt: str, force_gemini: bool = False) -> RoutingDecision:
        """
        プロンプトを分析して最適なモデルを決定する。
        """
        if force_gemini:
            return RoutingDecision(
                target=ModelTarget.GEMINI,
                reason="caller explicitly requested Gemini",
                estimated_cost_usd=self._estimate_cost(prompt)
            )
 
        prompt_len = len(prompt)
 
        # マルチモーダルチェック（このルーターはテキストのみ対応）
        if "[IMAGE]" in prompt or "[AUDIO]" in prompt or "[VIDEO]" in prompt:
            return RoutingDecision(
                target=ModelTarget.GEMINI,
                reason="multimodal input detected",
                estimated_cost_usd=self._estimate_cost(prompt)
            )
 
        # 複雑なタスクキーワードチェック
        is_complex = any(kw in prompt for kw in self.COMPLEX_TASK_KEYWORDS)
        if is_complex and prompt_len > 300:
            return RoutingDecision(
                target=ModelTarget.GEMINI,
                reason=f"complex task keywords detected, length={prompt_len}",
                estimated_cost_usd=self._estimate_cost(prompt)
            )
 
        # 文字数閾値チェック
        if prompt_len >= self.LOCAL_THRESHOLD_CHARS:
            return RoutingDecision(
                target=ModelTarget.GEMINI,
                reason=f"prompt too long for local ({prompt_len} chars)",
                estimated_cost_usd=self._estimate_cost(prompt)
            )
 
        # 上記に該当しない → ローカルで処理
        return RoutingDecision(
            target=ModelTarget.GEMMA_LOCAL,
            reason=f"short, simple task (length={prompt_len})",
            estimated_cost_usd=0.0
        )
 
    def _estimate_cost(self, prompt: str) -> float:
        """Gemini API 使用時の概算コストを計算する（USD）"""
        model_costs = self.GEMINI_COSTS.get(self.gemini_model_name, self.GEMINI_COSTS["gemini-2.5-flash"])
        # おおよそ1文字=0.5トークン（日本語）
        input_tokens = len(prompt) * 0.5
        output_tokens = 500  # 平均出力トークン数（サービスに合わせて調整）
        cost = (input_tokens / 1_000_000) * model_costs["input"] + \
               (output_tokens / 1_000_000) * model_costs["output"]
        return round(cost, 6)
 
    async def generate(
        self,
        prompt: str,
        system_instruction: str = "",
        force_gemini: bool = False,
        temperature: float = 0.7,
    ) -> tuple[str, RoutingDecision]:
        """
        ルーティングを行い、適切なモデルで推論を実行する。
        
        Returns:
            (生成テキスト, RoutingDecision) のタプル
        """
        decision = self._decide_route(prompt, force_gemini)
        start_time = time.time()
 
        try:
            if decision.target == ModelTarget.GEMMA_LOCAL:
                # Gemma 4（Ollama）で推論
                messages = []
                if system_instruction:
                    messages.append({"role": "system", "content": system_instruction})
                messages.append({"role": "user", "content": prompt})
 
                response = await self.gemma_client.chat.completions.create(
                    model=self.gemma_model,
                    messages=messages,
                    temperature=temperature,
                    max_tokens=2048,
                )
                result = response.choices[0].message.content
                self.stats["local_calls"] += 1
 
            else:
                # Gemini API で推論
                full_prompt = f"{system_instruction}\n\n{prompt}" if system_instruction else prompt
                response = await asyncio.to_thread(
                    self.gemini_client.generate_content,
                    full_prompt,
                    generation_config=genai.types.GenerationConfig(
                        temperature=temperature,
                        max_output_tokens=2048,
                    )
                )
                result = response.text
                self.stats["gemini_calls"] += 1
                self.stats["total_cost_usd"] += decision.estimated_cost_usd
 
        except Exception as e:
            # Gemma 4 が失敗した場合、Gemini API にフォールバック
            if decision.target == ModelTarget.GEMMA_LOCAL:
                print(f"⚠️ Local model failed, falling back to Gemini: {e}")
                fallback_decision = RoutingDecision(
                    target=ModelTarget.GEMINI,
                    reason="local model failed, fallback",
                    estimated_cost_usd=self._estimate_cost(prompt)
                )
                full_prompt = f"{system_instruction}\n\n{prompt}" if system_instruction else prompt
                response = await asyncio.to_thread(
                    self.gemini_client.generate_content, full_prompt
                )
                result = response.text
                self.stats["gemini_calls"] += 1
                self.stats["total_cost_usd"] += fallback_decision.estimated_cost_usd
                return result, fallback_decision
            else:
                raise
 
        elapsed = time.time() - start_time
        print(f"[{decision.target.value}] {elapsed:.2f}s | reason: {decision.reason}")
        return result, decision
 
    def get_stats(self) -> dict:
        """コスト統計を返す"""
        total_calls = self.stats["gemini_calls"] + self.stats["local_calls"]
        local_ratio = self.stats["local_calls"] / total_calls if total_calls > 0 else 0
        return {
            **self.stats,
            "total_calls": total_calls,
            "local_ratio": f"{local_ratio:.1%}",
            "estimated_monthly_cost_jpy": self.stats["total_cost_usd"] * 150 * 30,  # 30日換算
        }
 
# --- 使い方の例 ---
async def main():
    router = HybridRouter(gemini_model="gemini-2.5-flash")
 
    # シンプルなタスク → Gemma 4（ローカル）に自動ルーティング
    text, decision = await router.generate(
        prompt="次のテキストを「ポジティブ」「ネガティブ」「中立」に分類してください。\nテキスト: 今日の会議は思ったより早く終わりました。",
        system_instruction="あなたはテキスト分類の専門家です。分類結果のみ出力してください。",
    )
    print(f"分類結果: {text}")
    # 期待出力: 中立
    # 実行先: gemma_local（コスト: $0.000000）
 
    # 複雑なタスク → Gemini API に自動ルーティング
    text, decision = await router.generate(
        prompt="Pythonで非同期HTTPクライアントを実装してください。タイムアウト処理・リトライ・エラーハンドリングを含め、本番運用できるコードにしてください。",
    )
    print(f"コード生成完了: {len(text)}文字")
    # 実行先: gemini（complex task keywords detected）
 
    print(router.get_stats())
 
if __name__ == "__main__":
    asyncio.run(main())

# モデルをメモリに常駐させる（アンロード無効化）
OLLAMA_KEEP_ALIVE=-1 ollama serve

# warmup.py: 5分ごとにダミーリクエストを送ってモデルを常駐させる
import asyncio
import httpx
 
async def keep_warm():
    while True:
        try:
            async with httpx.AsyncClient() as client:
                await client.post(
                    "http://localhost:11434/api/generate",
                    json={"model": "gemma4:27b", "prompt": "ping", "stream": False},
                    timeout=30.0
                )
        except Exception as e:
            print(f"Warmup failed: {e}")
        await asyncio.sleep(270)  # 4分30秒ごと
 
asyncio.run(keep_warm())

# ユーザーに公開される重要な判断は常にGemini APIを使う
result, _ = await router.generate(
    prompt=user_input,
    force_gemini=True,  # このエンドポイントは品質優先
)

# context_cache_router.py
# Context Caching を活用した高度なルーター（抜粋）
import google.generativeai as genai
 
def create_cached_context(system_doc: str, expiry_minutes: int = 60):
    """
    大きなシステムドキュメントをキャッシュして
    繰り返しのトークン課金を回避する。
    1M tokens以上のドキュメントで特に効果的。
    """
    cache = genai.caching.CachedContent.create(
        model="models/gemini-2.5-flash-001",
        contents=[system_doc],
        ttl=f"{expiry_minutes * 60}s",
    )
    return cache
 
# キャッシュされたコンテキストを使うモデル
def get_model_with_cache(cache):
    return genai.GenerativeModel.from_cached_content(cache)

# ルーティング比率を定期ログに出力する例
import logging
import asyncio
 
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
 
async def log_routing_stats(router: HybridRouter, interval_seconds: int = 3600):
    """1時間ごとにルーティング統計をログに出力する"""
    while True:
        await asyncio.sleep(interval_seconds)
        stats = router.get_stats()
        logger.info(
            f"[HybridRouter Stats] "
            f"Gemini: {stats['gemini_calls']} calls | "
            f"Local: {stats['local_calls']} calls | "
            f"Local ratio: {stats['local_ratio']} | "
            f"Est. cost: ${stats['total_cost_usd']:.4f}"
        )