Routing Between Local Gemma 4 and the Gemini API Cut My Bill from ¥32,000 to ¥9,000 — A Production Hybrid Router Design

# hybrid_router.py
import asyncio
import os
import time
from enum import Enum
from dataclasses import dataclass
from typing import Optional
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:
    """
    Automatically routes requests between Gemini API and Gemma 4 local inference.
    Optimizes cost, quality, and latency based on request characteristics.
    """
 
    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", "step by step",
        "コードを書いて", "実装して", "デバッグ", "分析して", "推論",
    ]
 
    def __init__(
        self,
        gemini_model: str = "gemini-2.5-flash",
        gemma_model: str = None,
    ):
        genai.configure(api_key=os.getenv("GEMINI_API_KEY"))
        self.gemini_client = genai.GenerativeModel(gemini_model)
        self.gemini_model_name = gemini_model
 
        self.gemma_client = AsyncOpenAI(
            base_url=f"{os.getenv('OLLAMA_BASE_URL', 'http://localhost:11434')}/v1",
            api_key="ollama",
        )
        self.gemma_model = gemma_model or os.getenv("GEMMA_MODEL", "gemma4:27b")
 
        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)
            )
 
        # Multimodal signals
        if any(tag in prompt for tag in ["[IMAGE]", "[AUDIO]", "[VIDEO]"]):
            return RoutingDecision(
                target=ModelTarget.GEMINI,
                reason="multimodal input detected",
                estimated_cost_usd=self._estimate_cost(prompt)
            )
 
        # Complexity check
        prompt_len = len(prompt)
        is_complex = any(kw in prompt.lower() 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 (len={prompt_len})",
                estimated_cost_usd=self._estimate_cost(prompt)
            )
 
        # Length threshold
        if prompt_len >= self.LOCAL_THRESHOLD_CHARS:
            return RoutingDecision(
                target=ModelTarget.GEMINI,
                reason=f"prompt exceeds local threshold ({prompt_len} chars)",
                estimated_cost_usd=self._estimate_cost(prompt)
            )
 
        return RoutingDecision(
            target=ModelTarget.GEMMA_LOCAL,
            reason=f"short, simple task (len={prompt_len})",
            estimated_cost_usd=0.0
        )
 
    def _estimate_cost(self, prompt: str) -> float:
        model_costs = self.GEMINI_COSTS.get(
            self.gemini_model_name, self.GEMINI_COSTS["gemini-2.5-flash"]
        )
        # Rough estimate: 0.5 tokens per character (Japanese-heavy apps)
        input_tokens = len(prompt) * 0.5
        output_tokens = 500
        return round(
            (input_tokens / 1_000_000) * model_costs["input"] +
            (output_tokens / 1_000_000) * model_costs["output"],
            6
        )
 
    async def generate(
        self,
        prompt: str,
        system_instruction: str = "",
        force_gemini: bool = False,
        temperature: float = 0.7,
    ) -> tuple[str, RoutingDecision]:
        """
        Routes the request and generates a response.
 
        Returns:
            Tuple of (generated text, RoutingDecision)
        """
        decision = self._decide_route(prompt, force_gemini)
        start_time = time.time()
 
        try:
            if decision.target == ModelTarget.GEMMA_LOCAL:
                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:
                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:
            # Automatic fallback: local failure → Gemini API
            if decision.target == ModelTarget.GEMMA_LOCAL:
                print(f"⚠️ Local model failed, falling back to Gemini API: {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 | {decision.reason}")
        return result, decision
 
    def get_stats(self) -> dict:
        total = self.stats["gemini_calls"] + self.stats["local_calls"]
        local_ratio = self.stats["local_calls"] / total if total > 0 else 0
        return {
            **self.stats,
            "total_calls": total,
            "local_ratio": f"{local_ratio:.1%}",
        }
 
# Usage example
async def main():
    router = HybridRouter(gemini_model="gemini-2.5-flash")
 
    # Simple task → automatically routes to Gemma 4 (local)
    text, decision = await router.generate(
        prompt="Classify this text as Positive, Negative, or Neutral: 'The meeting ended earlier than expected.'",
        system_instruction="You are a text classifier. Output only the classification label.",
    )
    print(f"Classification: {text}")
    # Expected output: Neutral
    # Route: gemma_local (cost: $0.000000)
 
    # Complex task → automatically routes to Gemini API
    text, decision = await router.generate(
        prompt="Implement an async HTTP client in Python with timeout handling, retry logic with exponential backoff, and structured error handling suitable for production use.",
    )
    print(f"Code generated: {len(text)} chars | Route: {decision.target.value}")
    # Route: gemini (complex task keywords detected)
 
    print(router.get_stats())
 
if __name__ == "__main__":
    asyncio.run(main())

OLLAMA_KEEP_ALIVE=-1 ollama serve

# warmup.py — Keeps Gemma 4 resident in memory
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
                )
            print("Warmup ping sent")
        except Exception as e:
            print(f"Warmup failed: {e}")
        await asyncio.sleep(270)  # Every 4.5 minutes
 
asyncio.run(keep_warm())

# For customer-facing responses, always use Gemini API
result, _ = await router.generate(
    prompt=user_input,
    force_gemini=True,  # Quality-critical endpoint
)

# context_cache_router.py (excerpt)
import google.generativeai as genai
 
def create_cached_context(system_doc: str, expiry_minutes: int = 60):
    """
    Cache a large system document to avoid re-billing input tokens
    on every request. Most effective for documents over 32K 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 a model instance that uses the cached context."""
    return genai.GenerativeModel.from_cached_content(cache)

# routing_monitor.py — Log routing statistics hourly
import logging
import asyncio
 
logger = logging.getLogger(__name__)
 
async def log_routing_stats(router: HybridRouter, interval_seconds: int = 3600):
    """Emit routing statistics every hour for cost monitoring."""
    while True:
        await asyncio.sleep(interval_seconds)
        stats = router.get_stats()
        logger.info(
            f"[HybridRouter] "
            f"Gemini: {stats['gemini_calls']} | "
            f"Local: {stats['local_calls']} | "
            f"Local ratio: {stats['local_ratio']} | "
            f"Cost: ${stats['total_cost_usd']:.4f}"
        )
        # Alert if local ratio drops below 50% (possible misconfiguration)
        if stats["total_calls"] > 100:
            local_ratio_num = stats["local_calls"] / stats["total_calls"]
            if local_ratio_num < 0.5:
                logger.warning(
                    f"⚠️ Local inference ratio below 50% — review routing thresholds"
                )