Gemini API の価格優位性を活かした SaaS 設計 — 競合の半額で利益を出すコスト戦略

import anthropic
from google import genai
from enum import Enum
 
class ModelChoice(Enum):
    LITE = "gemini-2.5-lite"
    FLASH = "gemini-2.5-flash"
    PRO = "gemini-2.5-pro"
 
def estimate_complexity(prompt: str, context_length: int) -> ModelChoice:
    """
    プロンプト複雑度を推定してモデルを選択する
    
    複雑度スコア：
    - キーワード数（推論キーワード: "analyze", "compare", "reasoning"）
    - コンテキスト長（長いほど複雑）
    - 要求タスク種別（タブルな分類タスク vs 創作・推論）
    """
    complexity_score = 0
    
    # 推論キーワードの検出
    reasoning_keywords = ["analyze", "compare", "reasoning", "strategy", "trade-off", 
                          "design", "optimize", "justify", "explain why"]
    complexity_score += sum(1 for kw in reasoning_keywords if kw.lower() in prompt.lower()) * 2
    
    # コンテキスト長による複雑度
    if context_length > 20000:
        complexity_score += 3
    elif context_length > 5000:
        complexity_score += 2
    elif context_length > 1000:
        complexity_score += 1
    
    # 簡単なタスク（翻訳、フォーマット）の検出
    simple_keywords = ["translate", "format", "convert", "extract", "rewrite as"]
    if any(kw in prompt.lower() for kw in simple_keywords) and complexity_score < 2:
        return ModelChoice.LITE
    
    # ルーティング判定
    if complexity_score >= 5:
        return ModelChoice.PRO
    elif complexity_score >= 2:
        return ModelChoice.FLASH
    else:
        return ModelChoice.LITE
 
async def route_and_call(prompt: str, context: str = "", system_prompt: str = "") -> dict:
    """
    リクエストを複雑度に応じてルーティングし、Gemini API を呼び出す
    """
    client = genai.Client()
    
    # 複雑度推定とモデル選択
    model_choice = estimate_complexity(prompt, len(context))
    
    # API 呼び出し
    response = await client.aio.models.generate_content(
        model=model_choice.value,
        contents=[
            {
                "role": "user",
                "parts": [
                    {"text": system_prompt + "\n\n" + context + "\n\n" + prompt}
                ]
            }
        ],
        config=genai.types.GenerateContentConfig(
            temperature=0.7,
        )
    )
    
    return {
        "model_used": model_choice.value,
        "response": response.text,
        "usage": {
            "input_tokens": response.usage_metadata.prompt_token_count,
            "output_tokens": response.usage_metadata.candidates_token_count,
        }
    }

async def analyze_document_with_cache(
    user_id: str,
    document_content: str,
    user_question: str,
    system_instructions: str
) -> dict:
    """
    ユーザーが繰り返し同じドキュメントを分析する場合、
    キャッシュを活用してコストを削減
    """
    client = genai.Client()
    
    # キャッシュキー（ユーザーごとに一意）
    cache_key = f"doc_cache_{user_id}"
    
    # システムプロンプト（キャッシュ対象）
    system_content = [
        {
            "type": "text",
            "text": system_instructions,
            "cache_control": {"type": "ephemeral"}  # 一時キャッシュ（5分）
        },
        {
            "type": "text",
            "text": f"# Document for Analysis\n\n{document_content}",
            "cache_control": {"type": "ephemeral"}
        }
    ]
    
    # キャッシュ付きリクエスト
    response = await client.aio.models.generate_content(
        model="gemini-2.5-flash",
        contents=[
            {
                "role": "user",
                "parts": [
                    *system_content,
                    {"text": user_question}
                ]
            }
        ]
    )
    
    # キャッシュ統計情報を取得
    usage = response.usage_metadata
    cache_hit = getattr(usage, 'cache_read_input_token_count', 0) > 0
    
    return {
        "response": response.text,
        "cost_breakdown": {
            "input_tokens": usage.prompt_token_count,
            "cache_read_tokens": getattr(usage, 'cache_read_input_token_count', 0),
            "output_tokens": usage.candidates_token_count,
            "cache_hit": cache_hit,
            # キャッシュ利用時: 1,000 tokens = $0.025 (Lite rate)
            # キャッシュなし: 1,000 tokens = $0.30 (通常rate)
            "estimated_cost": (
                usage.prompt_token_count * 0.000075 +
                getattr(usage, 'cache_read_input_token_count', 0) * 0.0000075 +
                usage.candidates_token_count * 0.00030
            )
        }
    }

import json
import asyncio
from datetime import datetime
 
async def batch_analyze_documents(documents: list[dict]) -> str:
    """
    複数の文書をバッチAPIで処理
    （結果は数秒から数分で帰ってくる）
    """
    client = genai.Client()
    
    # バッチリクエストを構築
    requests = []
    for i, doc in enumerate(documents):
        requests.append({
            "custom_id": f"doc_{i}",
            "request_body": {
                "contents": [
                    {
                        "role": "user",
                        "parts": [
                            {
                                "text": f"Analyze this document for key insights:\n\n{doc['content']}"
                            }
                        ]
                    }
                ],
                "generationConfig": {
                    "maxOutputTokens": 500
                }
            }
        })
    
    # バッチ投入
    batch_request = {
        "requests": requests
    }
    
    # API経由でバッチを投入
    # 実際のAPI呼び出しはGoogle Batch APIドキュメント参照
    batch_response = await client.aio.batches.process(
        requests=requests,
        model="gemini-2.5-flash"
    )
    
    # 結果を待機（通常は数秒）
    results = []
    while batch_response.state != "COMPLETED":
        await asyncio.sleep(1)
        batch_response = await client.aio.batches.get(batch_response.name)
    
    # 結果を取得
    for result in batch_response.results:
        results.append({
            "doc_id": result.custom_id,
            "analysis": result.response.candidates[0].content.parts[0].text,
            "cost_savings": "50%"  # Batch割引適用
        })
    
    return results

import json
from datetime import datetime
from typing import Optional
from dataclasses import dataclass
 
@dataclass
class APIUsageRecord:
    user_id: str
    timestamp: datetime
    model_used: str
    input_tokens: int
    output_tokens: int
    cache_read_tokens: int = 0
    is_batch: bool = False
    
    def calculate_cost(self) -> float:
        """モデルとキャッシュ状況に応じたコスト計算"""
        # ベースレート（ドル）
        rates = {
            "gemini-2.5-lite": {"input": 0.0375 / 1_000_000, "output": 0.15 / 1_000_000},
            "gemini-2.5-flash": {"input": 0.075 / 1_000_000, "output": 0.30 / 1_000_000},
            "gemini-2.5-pro": {"input": 1.25 / 1_000_000, "output": 5.00 / 1_000_000},
        }
        
        rate = rates.get(self.model_used, rates["gemini-2.5-flash"])
        
        # 入力コスト（キャッシュは 90% 割引）
        input_cost = self.input_tokens * rate["input"]
        cache_cost = self.cache_read_tokens * rate["input"] * 0.1
        
        # 出力コスト
        output_cost = self.output_tokens * rate["output"]
        
        # バッチは 50% 割引
        total = input_cost + cache_cost + output_cost
        if self.is_batch:
            total *= 0.5
        
        return total
 
class CostTracker:
    """ユーザーごと・モデルごとのコスト追跡"""
    
    def __init__(self, db_connection):
        self.db = db_connection
    
    async def log_usage(self, record: APIUsageRecord) -> None:
        """API利用をログに記録"""
        cost = record.calculate_cost()
        
        await self.db.execute("""
            INSERT INTO api_usage_log 
            (user_id, timestamp, model, input_tokens, output_tokens, 
             cache_read_tokens, is_batch, cost_usd)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            record.user_id, record.timestamp, record.model_used,
            record.input_tokens, record.output_tokens,
            record.cache_read_tokens, record.is_batch, cost
        ))
    
    async def get_monthly_summary(self, year: int, month: int) -> dict:
        """月間集計レポート"""
        result = await self.db.fetch("""
            SELECT 
                model,
                COUNT(*) as request_count,
                SUM(input_tokens) as total_input,
                SUM(output_tokens) as total_output,
                SUM(cache_read_tokens) as total_cache_read,
                SUM(cost_usd) as total_cost
            FROM api_usage_log
            WHERE YEAR(timestamp) = ? AND MONTH(timestamp) = ?
            GROUP BY model
        """, (year, month))
        
        return {
            "month": f"{year}-{month:02d}",
            "by_model": [dict(row) for row in result],
            "total_cost_usd": sum(row["total_cost"] for row in result),
            "request_count": sum(row["request_count"] for row in result),
        }
    
    async def get_user_profitability(self, year: int, month: int) -> dict:
        """ユーザーごとの利益性分析"""
        result = await self.db.fetch("""
            SELECT 
                u.user_id,
                u.plan,
                SUM(aul.cost_usd) as api_cost,
                COUNT(DISTINCT DATE(aul.timestamp)) as usage_days,
                CASE 
                    WHEN u.plan = 'pro' THEN 9
                    WHEN u.plan = 'standard' THEN 5
                    ELSE 0
                END * (SELECT DATEDIFF(LAST_DAY(?), DATE_TRUNC('month', ?))) as monthly_revenue
            FROM users u
            LEFT JOIN api_usage_log aul ON u.user_id = aul.user_id
            WHERE YEAR(aul.timestamp) = ? AND MONTH(aul.timestamp) = ?
            GROUP BY u.user_id
        """, (
            f"{year}-{month:02d}-01", f"{year}-{month:02d}-01",
            year, month
        ))
        
        profitable = [
            r for r in result 
            if r["monthly_revenue"] and r["api_cost"] < r["monthly_revenue"] * 0.4
        ]
        
        return {
            "total_users": len(result),
            "profitable_users": len(profitable),
            "avg_api_cost_per_user": sum(r["api_cost"] for r in result) / len(result),
        }