Gemini 2.5 Flash の低コスト特性でコンテンツ量産サービスを収益化する

# gemini_router.py
import google.generativeai as genai
from enum import Enum
from dataclasses import dataclass
 
genai.configure(api_key="YOUR_GEMINI_API_KEY")
 
class ContentTask(Enum):
    # Flash 向け（高速・低コスト）
    PRODUCT_DESCRIPTION = "product_description"
    SHORT_SUMMARY = "short_summary"
    SNS_POST = "sns_post"
    EMAIL_TEMPLATE = "email_template"
    TRANSLATION = "translation"
    FAQ_ANSWER = "faq_answer"
    # Pro 向け（高品質・高コスト）
    LONG_FORM_ARTICLE = "long_form_article"
    LEGAL_ANALYSIS = "legal_analysis"
    TECHNICAL_DOCUMENTATION = "technical_documentation"
    COMPLEX_RESEARCH = "complex_research"
 
FLASH_TASKS = {
    ContentTask.PRODUCT_DESCRIPTION,
    ContentTask.SHORT_SUMMARY,
    ContentTask.SNS_POST,
    ContentTask.EMAIL_TEMPLATE,
    ContentTask.TRANSLATION,
    ContentTask.FAQ_ANSWER,
}
 
@dataclass
class GenerationResult:
    text: str
    model_used: str
    input_tokens: int
    output_tokens: int
    cost_usd: float
 
COST_PER_MILLION = {
    "gemini-2.5-flash": {"input": 0.15, "output": 0.60},
    "gemini-2.5-pro":   {"input": 1.25, "output": 10.00},
}
 
def calculate_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    rates = COST_PER_MILLION[model]
    return (input_tokens / 1_000_000) * rates["input"] + \
           (output_tokens / 1_000_000) * rates["output"]
 
def generate_content(
    prompt: str,
    task: ContentTask,
    system_instruction: str = None,
) -> GenerationResult:
    """タスク種別に応じてFlashまたはProを自動選択して生成する"""
    
    model_name = "gemini-2.5-flash" if task in FLASH_TASKS else "gemini-2.5-pro"
    
    model = genai.GenerativeModel(
        model_name=model_name,
        system_instruction=system_instruction,
    )
    
    response = model.generate_content(prompt)
    
    # トークン使用量を取得
    usage = response.usage_metadata
    input_tokens = usage.prompt_token_count
    output_tokens = usage.candidates_token_count
    cost = calculate_cost(model_name, input_tokens, output_tokens)
    
    return GenerationResult(
        text=response.text,
        model_used=model_name,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        cost_usd=cost,
    )
 
# 使用例
result = generate_content(
    prompt="以下の商品スペックから、ECサイト用の商品説明文を200文字で生成してください:\n\n商品名: ワイヤレスイヤホン X200\nバッテリー: 8時間再生\n防水: IPX5\n重量: 5g/片耳",
    task=ContentTask.PRODUCT_DESCRIPTION,
    system_instruction="あなたはECサイトのコピーライターです。読者が買いたくなる魅力的な文章を書いてください。",
)
 
print(f"生成結果: {result.text}")
print(f"使用モデル: {result.model_used}")
print(f"コスト: ${result.cost_usd:.6f}")
# 出力例:
# 生成結果: 8時間連続再生で一日中快適に使えるワイヤレスイヤホン。わずか5gの超軽量設計で長時間着けていても耳が疲れません...
# 使用モデル: gemini-2.5-flash
# コスト: $0.000298

# batch_generator.py
import asyncio
import google.generativeai as genai
from typing import List
 
async def generate_batch_async(
    prompts: List[str],
    task: ContentTask,
    max_concurrent: int = 10,
) -> List[GenerationResult]:
    """非同期で並列生成（同時実行数を制限してレート制限を回避）"""
    
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def generate_single(prompt: str) -> GenerationResult:
        async with semaphore:
            # 非同期生成（google-generativeai の async API を使用）
            model_name = "gemini-2.5-flash" if task in FLASH_TASKS else "gemini-2.5-pro"
            model = genai.GenerativeModel(model_name)
            response = await model.generate_content_async(prompt)
            usage = response.usage_metadata
            cost = calculate_cost(
                model_name,
                usage.prompt_token_count,
                usage.candidates_token_count,
            )
            return GenerationResult(
                text=response.text,
                model_used=model_name,
                input_tokens=usage.prompt_token_count,
                output_tokens=usage.candidates_token_count,
                cost_usd=cost,
            )
    
    tasks = [generate_single(p) for p in prompts]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    # エラー結果をフィルタリング
    valid_results = []
    for i, result in enumerate(results):
        if isinstance(result, Exception):
            print(f"[Error] prompt[{i}] failed: {result}")
            # エラーはスキップまたはリトライ
        else:
            valid_results.append(result)
    
    return valid_results
 
# 商品説明文を100件一括生成する例
async def bulk_generate_product_descriptions(products: List[dict]) -> List[str]:
    prompts = [
        f"商品名: {p['name']}\nスペック: {p['spec']}\n\n"
        f"ECサイト用の商品説明文を150文字で生成してください。"
        for p in products
    ]
    
    results = await generate_batch_async(
        prompts=prompts,
        task=ContentTask.PRODUCT_DESCRIPTION,
        max_concurrent=20,  # 同時20リクエスト
    )
    
    total_cost = sum(r.cost_usd for r in results)
    print(f"100件生成コスト: ${total_cost:.4f}")  # 例: $0.0270
    
    return [r.text for r in results]

# 品質を上げるプロンプトエンジニアリングの例
POOR_PROMPT = "商品説明を書いてください"
 
BETTER_PROMPT = """
以下の商品について、ECサイト掲載用の商品説明文を作成してください。
 
【条件】
- 文字数: 150〜200文字
- 冒頭に商品の最大の魅力を入れる
- 具体的な数値・スペックを1〜2つ含める
- 読者が「これほしい」と感じるような感情的な表現を使う
- 最後に行動を促す一文で締める
 
【商品情報】
{product_info}
"""