Gemini 2.5 Pro + Imagen 4 コンテンツ自動生成パイプライン構築ガイド

pip install google-genai>=0.8.0 Pillow python-dotenv asyncio

# .env ファイルに設定
# GEMINI_API_KEY=YOUR_GEMINI_API_KEY

import os
import asyncio
from google import genai
from google.genai import types
from dotenv import load_dotenv
 
load_dotenv()
 
client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
 
SYSTEM_INSTRUCTION = """
あなたは個人開発者・テクニカルライターとして、実体験に基づいた技術記事を書きます。
 
執筆スタイル:
- 敬体（です/ます調）で統一する
- 「ここでは〜を整理します」のようなテンプレ導入は絶対に使わない
- 具体的な問題提起または発見から記事を始める
- 公式ドキュメントの要約ではなく、実際に使ってみて気づいたことを書く
- よくある間違いや落とし穴を Before/After 形式で示す
- まとめは「読者の次のアクション」1つだけ示して締める
 
技術的な要件:
- コードは完全動作する形で記載し、期待する出力も示す
- コメントは日本語で記述する
- プレースホルダーには YOUR_GEMINI_API_KEY のような安全な形式を使う
"""
 
async def generate_blog_article(topic: str, target_audience: str, tone: str) -> dict:
    """
    Gemini 2.5 Pro でブログ記事を生成する
    
    Args:
        topic: 記事のトピック（例: "Gemini APIのレート制限対策"）
        target_audience: ターゲット読者（例: "個人開発者"）
        tone: 文体のトーン（例: "実践的・技術寄り"）
    
    Returns:
        記事データ（title, description, content, tags, seo_keywords）
    """
    prompt = f"""
以下の条件でブログ記事を作成してください。
 
トピック: {topic}
ターゲット読者: {target_audience}
トーン: {tone}
 
出力はJSON形式で以下のフィールドを含めること:
{{
  "title": "SEO最適化されたタイトル（60文字以内）",
  "description": "メタdescription（160文字以内）",
  "content": "記事本文（Markdown形式、3000〜5000文字）",
  "tags": ["タグ1", "タグ2", "タグ3"],
  "seo_keywords": ["キーワード1", "キーワード2"],
  "thumbnail_prompt": "サムネイル画像生成用の英語プロンプト（Imagen向け）"
}}
"""
    
    response = await client.aio.models.generate_content(
        model="gemini-2.5-pro-latest",
        config=types.GenerateContentConfig(
            system_instruction=SYSTEM_INSTRUCTION,
            temperature=0.7,
            max_output_tokens=8192,
            response_mime_type="application/json",
        ),
        contents=prompt,
    )
    
    import json
    return json.loads(response.text)

# ❌ これだとAI解説調の汎用記事が出てくる
prompt = f"{topic} について詳しく解説する記事を書いてください。"

# ✅ 具体的な読者の困りごとから始まる記事を要求する
prompt = f"""
{target_audience} が {topic} を実装しようとしたときに最初にぶつかる壁と、
その解決方法を中心に記事を構成してください。
「なぜこう実装するのか」の理由が伝わる内容にすること。
"""

async def generate_thumbnail(
    prompt: str, 
    aspect_ratio: str = "16:9",
    output_path: str = "thumbnail.png"
) -> str:
    """
    Imagen 4 でサムネイル画像を生成する
    
    Args:
        prompt: 画像生成プロンプト（英語推奨）
        aspect_ratio: アスペクト比（"1:1", "9:16", "16:9", "4:3", "3:4"）
        output_path: 保存先パス
    
    Returns:
        生成された画像の保存パス
    """
    response = await client.aio.models.generate_images(
        model="imagen-4.0-generate-preview-05-20",
        prompt=prompt,
        config=types.GenerateImagesConfig(
            number_of_images=1,
            aspect_ratio=aspect_ratio,
            safety_filter_level="BLOCK_MEDIUM_AND_ABOVE",
            include_rai_reason=True,
        ),
    )
    
    if not response.generated_images:
        raise ValueError("画像生成に失敗しました（Safety フィルターによりブロックされた可能性があります）")
    
    image_data = response.generated_images[0].image.image_bytes
    with open(output_path, "wb") as f:
        f.write(image_data)
    
    print(f"✅ サムネイル生成完了: {output_path} ({len(image_data) / 1024:.1f} KB)")
    return output_path

def build_thumbnail_prompt(topic: str, style: str = "tech") -> str:
    """
    ブログサムネイル向けプロンプトを構築する
    
    私が試した中で最も安定して品質の高い画像が出るパターン:
    1. 被写体・シーンの説明
    2. スタイル指定（realistic / flat design / isometric 等）
    3. 色調指定（Google カラーや Tech ブランド系）
    4. 不要要素の明示的な除外
    """
    style_map = {
        "tech": "clean tech illustration, flat design, Google Material Design colors",
        "minimal": "minimalist, white background, subtle gradients, modern typography",
        "cinematic": "cinematic lighting, dark background, glowing neon accents",
    }
    
    base_style = style_map.get(style, style_map["tech"])
    
    return (
        f"{topic}, {base_style}, "
        "16:9 aspect ratio for blog thumbnail, "
        "no text overlay, no watermarks, high quality, "
        "professional design, sharp focus"
    )
 
# 使用例
prompt = build_thumbnail_prompt(
    "Python code running on Google Cloud, API connections visualized as glowing lines",
    style="tech"
)
# 期待する出力: Googleカラーを基調とした
# クリーンなテック系イラスト（テキストなし）

async def generate_thumbnail_with_retry(
    prompt: str, 
    max_retries: int = 3,
    fallback_prompts: list = None
) -> str:
    """Safety フィルター対応のリトライ付き画像生成"""
    
    prompts_to_try = [prompt] + (fallback_prompts or [])
    
    for i, current_prompt in enumerate(prompts_to_try):
        try:
            return await generate_thumbnail(current_prompt, output_path=f"thumb_{i}.png")
        except ValueError as e:
            if "Safety" in str(e) and i < len(prompts_to_try) - 1:
                # より安全なプロンプトにフォールバック
                print(f"⚠️ Safety フィルター（試行 {i+1}/{len(prompts_to_try)}）")
                continue
            raise
    
    raise RuntimeError("全プロンプトで画像生成に失敗しました")

PLATFORM_CONFIGS = {
    "x": {
        "max_chars": 140,
        "style": "簡潔でインパクトのある文体。問いかけか発見から始める。ハッシュタグ2〜3個。",
        "example": "Gemini APIのレート制限、実は回避策があります。指数バックオフより効果的な方法を試したので共有します。 #GeminiAPI #個人開発",
    },
    "instagram": {
        "max_chars": 2200,
        "style": "絵文字を適度に使用。改行多め。学びや気づきをリスト形式で。最後にCTA。",
        "example": "今日の発見 ✨\n\nGemini + Imagen 4 で...",
    },
    "note": {
        "max_chars": 800,
        "style": "ブログの導入文として機能する文量。読者の共感を得てから本題へ誘導する形式。",
        "example": "コンテンツ制作で一番時間がかかるのはどこですか？...",
    },
}
 
async def generate_sns_content(article_data: dict) -> dict:
    """
    記事データからSNS用コンテンツを一括生成する
    """
    sns_results = {}
    
    # 各プラットフォームを並列処理（後の Step 4 で詳説）
    tasks = []
    for platform, config in PLATFORM_CONFIGS.items():
        task = _generate_for_platform(article_data, platform, config)
        tasks.append((platform, task))
    
    results = await asyncio.gather(*[t for _, t in tasks])
    
    for (platform, _), result in zip(tasks, results):
        sns_results[platform] = result
    
    return sns_results
 
async def _generate_for_platform(article_data: dict, platform: str, config: dict) -> str:
    prompt = f"""
以下のブログ記事を元に、{platform} 用の投稿文を生成してください。
 
記事タイトル: {article_data['title']}
記事概要: {article_data['description']}
主なキーワード: {', '.join(article_data.get('seo_keywords', []))}
 
スタイル指定: {config['style']}
文字数上限: {config['max_chars']}文字以内
 
参考例（このスタイルで書く）:
{config['example']}
 
投稿文のみ出力してください（余計な説明は不要）。
"""
    
    response = await client.aio.models.generate_content(
        model="gemini-2.5-pro-latest",
        config=types.GenerateContentConfig(
            temperature=0.8,  # SNS文は少し創造性を上げる
            max_output_tokens=1024,
        ),
        contents=prompt,
    )
    return response.text.strip()

import time
from dataclasses import dataclass, field
from typing import Optional
 
@dataclass
class ContentResult:
    topic: str
    article: dict = field(default_factory=dict)
    thumbnail_path: Optional[str] = None
    sns_content: dict = field(default_factory=dict)
    errors: list = field(default_factory=list)
    total_cost_estimate_usd: float = 0.0
    elapsed_seconds: float = 0.0
 
async def run_content_pipeline(
    topic: str,
    target_audience: str = "個人開発者",
    tone: str = "実践的・技術寄り",
    output_dir: str = "./output",
) -> ContentResult:
    """
    コンテンツ自動生成パイプラインの完全実装
    
    実行フロー:
    1. 記事生成（直列 — thumbnail_prompt が必要なため）
    2. 画像生成 + SNS生成（並列 — 互いに依存しない）
    """
    import os
    os.makedirs(output_dir, exist_ok=True)
    
    result = ContentResult(topic=topic)
    start_time = time.time()
    
    # Phase 1: 記事生成
    print(f"📝 記事生成開始: {topic}")
    try:
        result.article = await generate_blog_article(topic, target_audience, tone)
        print(f"✅ 記事生成完了: {result.article['title']}")
    except Exception as e:
        result.errors.append(f"記事生成エラー: {e}")
        return result  # 記事がなければ後続処理も中断
    
    # Phase 2: 画像生成とSNS生成を並列実行
    thumbnail_path = os.path.join(output_dir, f"{topic[:20]}_thumbnail.png")
    
    thumbnail_task = generate_thumbnail_with_retry(
        prompt=result.article.get("thumbnail_prompt", f"{topic}, tech illustration"),
        fallback_prompts=[
            "abstract technology background, blue gradients, clean design",
            "minimal geometric shapes, Google colors, professional",
        ],
    )
    sns_task = generate_sns_content(result.article)
    
    print("🎨 画像生成・SNS文生成を並列実行中...")
    thumbnail_result, sns_result = await asyncio.gather(
        thumbnail_task,
        sns_task,
        return_exceptions=True,  # どちらかが失敗しても両方の結果を取得
    )
    
    if isinstance(thumbnail_result, Exception):
        result.errors.append(f"画像生成エラー: {thumbnail_result}")
    else:
        result.thumbnail_path = thumbnail_path
    
    if isinstance(sns_result, Exception):
        result.errors.append(f"SNS生成エラー: {sns_result}")
    else:
        result.sns_content = sns_result
    
    # コスト概算（参考値）
    result.total_cost_estimate_usd = 0.03 + 0.04 + 0.01  # 記事 + 画像 + SNS
    result.elapsed_seconds = time.time() - start_time
    
    print(f"✅ パイプライン完了: {result.elapsed_seconds:.1f}秒 / 推定コスト ${result.total_cost_estimate_usd:.3f}")
    return result

def quality_check(article: dict) -> tuple[bool, list]:
    """
    生成記事の品質チェック
    
    Returns:
        (通過/失敗, 問題点のリスト)
    """
    issues = []
    
    content = article.get("content", "")
    title = article.get("title", "")
    description = article.get("description", "")
    
    # タイトルチェック
    if len(title) > 60:
        issues.append(f"タイトルが長すぎます: {len(title)}文字（上限60文字）")
    if any(ng in title for ng in ["ここでは", "完全ガイド", "徹底解説"]):
        issues.append("テンプレート的なタイトルパターンを検出")
    
    # 本文チェック
    if len(content) < 2000:
        issues.append(f"本文が短すぎます: {len(content)}文字（下限2000文字）")
    
    # テンプレート導入文のチェック
    template_intros = [
        "ここでは", "ここでは", "本ガイドでは",
        "〜を整理します", "〜を紹介します",
    ]
    first_200 = content[:200]
    for pattern in template_intros:
        if pattern in first_200:
            issues.append(f"テンプレート導入文を検出: 「{pattern}」")
            break
    
    # descriptionチェック
    if len(description) > 160:
        issues.append(f"descriptionが長すぎます: {len(description)}文字（上限160文字）")
    
    return len(issues) == 0, issues
 
# パイプライン内での使用
async def run_content_pipeline_with_quality(topic: str, **kwargs) -> ContentResult:
    result = await run_content_pipeline(topic, **kwargs)
    
    if result.article:
        passed, issues = quality_check(result.article)
        if not passed:
            print(f"⚠️ 品質チェック: {len(issues)}件の問題を検出")
            for issue in issues:
                print(f"  - {issue}")
            # リトライ（最大1回）
            result.article["_quality_issues"] = issues
    
    return result

# 簡易的な料金設計例（月額制）
 
PLANS = {
    "starter": {
        "monthly_price_jpy": 1980,
        "articles_per_month": 30,
        "includes_images": True,
        "includes_sns": True,
    },
    "pro": {
        "monthly_price_jpy": 4980,
        "articles_per_month": 100,
        "includes_images": True,
        "includes_sns": True,
        "custom_brand_voice": True,
    },
}
 
# 月100人の starter ユーザーで月収 198,000円
# 原価: 100人 × 30記事 × $0.10 = $300（約45,000円）
# 粗利: 約77%

import os
from functools import lru_cache
 
@lru_cache(maxsize=None)
def get_client() -> genai.Client:
    """
    API クライアントのシングルトン生成
    
    環境変数から読み込み、コードに直接書かない。
    本番環境では Google Cloud Secret Manager の使用を推奨。
    """
    api_key = os.environ.get("GEMINI_API_KEY")
    if not api_key:
        raise EnvironmentError(
            "GEMINI_API_KEY が設定されていません。"
            ".env ファイルまたは環境変数を確認してください。"
        )
    return genai.Client(api_key=api_key)

import asyncio
from collections import deque
from datetime import datetime, timedelta
 
class RateLimitedPipeline:
    """
    レート制限を考慮したパイプライン実行クラス
    
    Gemini 2.5 Pro の無料枠: 5 RPM（有料: 1000 RPM）
    Imagen 4: 別のクォータ制限あり
    """
    
    def __init__(self, max_rpm: int = 50):
        self.max_rpm = max_rpm
        self.request_times = deque()
    
    async def wait_if_needed(self):
        """1分間のリクエスト数が上限を超えたら待機"""
        now = datetime.now()
        one_minute_ago = now - timedelta(minutes=1)
        
        # 1分より古いタイムスタンプを削除
        while self.request_times and self.request_times[0] < one_minute_ago:
            self.request_times.popleft()
        
        if len(self.request_times) >= self.max_rpm:
            wait_seconds = (self.request_times[0] - one_minute_ago).total_seconds() + 1
            print(f"⏳ レート制限: {wait_seconds:.1f}秒待機")
            await asyncio.sleep(wait_seconds)
        
        self.request_times.append(now)
    
    async def run(self, topic: str, **kwargs) -> ContentResult:
        await self.wait_if_needed()
        return await run_content_pipeline_with_quality(topic, **kwargs)

class CostTracker:
    def __init__(self, monthly_budget_usd: float = 10.0):
        self.monthly_budget = monthly_budget_usd
        self.current_month_cost = 0.0
    
    def add_cost(self, cost_usd: float):
        self.current_month_cost += cost_usd
        usage_rate = self.current_month_cost / self.monthly_budget
        
        if usage_rate >= 0.9:
            print(f"🚨 コスト警告: 月次予算の{usage_rate*100:.0f}%を消費 (${self.current_month_cost:.2f}/{self.monthly_budget})")
        elif usage_rate >= 0.7:
            print(f"⚠️ コスト通知: 月次予算の{usage_rate*100:.0f}%を消費")
    
    def is_budget_exceeded(self) -> bool:
        return self.current_month_cost >= self.monthly_budget