Gemini 2.5 Pro + Imagen 4 Content Automation Pipeline: Complete Build Guide

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

# .env
# GEMINI_API_KEY=YOUR_GEMINI_API_KEY

import os
import asyncio
import json
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 = """
You are a technical writer with direct experience as a solo developer.
Write from a first-person perspective when appropriate — "when I tested this," 
"in my experience" — rather than the generic third-person AI explanation style.
 
Writing rules:
- Start with a specific problem, discovery, or unexpected finding. Never with "In this article, we'll cover..."
- Include at least one Before/After code pattern showing a common mistake and the fix
- Explain *why* each implementation choice was made, not just what to do
- End with one concrete next action for the reader, not a summary of what was covered
 
Technical rules:
- All code must be complete and runnable (no incomplete snippets)
- Use YOUR_GEMINI_API_KEY as the placeholder, never a real key format
- Add inline comments explaining non-obvious choices
"""
 
async def generate_blog_article(topic: str, target_audience: str = "developers", tone: str = "practical") -> dict:
    """
    Generate a blog article using Gemini 2.5 Pro with structured JSON output.
    
    The key here is response_mime_type="application/json" — without this,
    Gemini wraps the JSON in markdown code fences which breaks parsing.
    """
    prompt = f"""
Create a blog article for the following:
 
Topic: {topic}
Target audience: {target_audience}
Tone: {tone}
 
Return JSON with these fields:
{{
  "title": "SEO-optimized title under 60 characters",
  "description": "Meta description under 160 characters",
  "content": "Full article in Markdown, 1500-2500 words",
  "tags": ["tag1", "tag2", "tag3"],
  "seo_keywords": ["keyword1", "keyword2"],
  "thumbnail_prompt": "English prompt for Imagen 4 to generate a relevant thumbnail"
}}
"""
    
    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",  # Critical — prevents markdown wrapping
        ),
        contents=prompt,
    )
    
    return json.loads(response.text)

# ❌ BEFORE — works locally, breaks in production
response = await client.aio.models.generate_content(
    model="gemini-2.5-pro-latest",
    contents=f"Write an article about {topic} in JSON format",
)
data = json.loads(response.text)  # Fails: json.JSONDecodeError
# Actual response starts with: ```json\n{ ... }\n```

# ✅ AFTER — reliable JSON output
response = await client.aio.models.generate_content(
    model="gemini-2.5-pro-latest",
    config=types.GenerateContentConfig(
        response_mime_type="application/json",  # This line is the fix
    ),
    contents=f"Write an article about {topic} in JSON format",
)
data = json.loads(response.text)  # Works reliably

async def generate_thumbnail(
    prompt: str,
    aspect_ratio: str = "16:9",
    output_path: str = "thumbnail.png"
) -> str:
    """
    Generate a thumbnail using Imagen 4.
    
    aspect_ratio options: "1:1", "9:16", "16:9", "4:3", "3:4"
    For blog thumbnails, "16:9" is standard.
    For Instagram square posts, use "1:1".
    """
    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("Image generation returned no results — likely blocked by Safety filter")
    
    image_bytes = response.generated_images[0].image.image_bytes
    with open(output_path, "wb") as f:
        f.write(image_bytes)
    
    print(f"✅ Thumbnail saved: {output_path} ({len(image_bytes) / 1024:.1f} KB)")
    return output_path

def build_thumbnail_prompt(scene: str, style: str = "tech") -> str:
    """
    Prompt structure that consistently produces usable blog thumbnails.
    
    The "no text overlay" instruction is essential — Imagen sometimes adds
    random text to images, which breaks your design if you want clean thumbnails.
    """
    styles = {
        "tech": "clean tech illustration, flat design, Google Material colors, subtle gradients",
        "minimal": "minimalist, white background, geometric shapes, modern",
        "dark": "dark background, glowing neon accents, cinematic lighting",
    }
    
    return (
        f"{scene}, {styles.get(style, styles['tech'])}, "
        "professional blog thumbnail, 16:9 ratio, "
        "no text overlay, no watermarks, no UI elements, "
        "sharp focus, high quality render"
    )

async def generate_thumbnail_with_fallback(
    prompt: str,
    fallback_prompts: list = None
) -> str:
    """
    Attempt thumbnail generation with automatic fallback on Safety blocks.
    
    In production, I keep a bank of 5-10 generic tech-themed fallback prompts
    that I know pass Safety filters reliably.
    """
    SAFE_FALLBACKS = [
        "abstract technology network visualization, blue nodes and connections, dark background, professional",
        "modern software development workspace, code editor on screen, clean minimal design",
        "geometric data visualization, colorful charts, white background, business professional",
    ]
    
    all_prompts = [prompt] + (fallback_prompts or []) + SAFE_FALLBACKS
    
    for i, current_prompt in enumerate(all_prompts):
        try:
            return await generate_thumbnail(current_prompt)
        except ValueError as e:
            if i < len(all_prompts) - 1:
                print(f"⚠️ Safety filter triggered (attempt {i+1}) — trying fallback")
                continue
            raise RuntimeError("All prompts exhausted — image generation failed") from e

PLATFORM_CONFIGS = {
    "x": {
        "max_chars": 140,
        "instruction": "Write a punchy single tweet. Start with a hook (question or surprising fact). End with 2-3 relevant hashtags. No promotional language.",
    },
    "instagram": {
        "max_chars": 2200,
        "instruction": "Write an Instagram caption with line breaks for readability. Use 3-5 relevant emojis placed naturally. List format for key points. End with a question to drive engagement. Add 10-15 hashtags at the end.",
    },
    "linkedin": {
        "max_chars": 1300,
        "instruction": "Write a professional LinkedIn post. Start with a concrete insight or result (not 'I'm excited to share'). Use short paragraphs. End with a takeaway for fellow developers.",
    },
}
 
async def generate_sns_content(article_data: dict) -> dict:
    """Generate platform-optimized SNS content from article data in parallel."""
    
    tasks = {
        platform: _generate_for_platform(article_data, platform, config)
        for platform, config in PLATFORM_CONFIGS.items()
    }
    
    results = await asyncio.gather(*tasks.values(), return_exceptions=True)
    
    return {
        platform: result if not isinstance(result, Exception) else f"Error: {result}"
        for platform, result in zip(tasks.keys(), results)
    }
 
async def _generate_for_platform(article_data: dict, platform: str, config: dict) -> str:
    response = await client.aio.models.generate_content(
        model="gemini-2.5-pro-latest",
        config=types.GenerateContentConfig(
            temperature=0.85,  # Slightly higher for social content
            max_output_tokens=800,
        ),
        contents=f"""
Article title: {article_data['title']}
Article summary: {article_data['description']}
Key topics: {', '.join(article_data.get('seo_keywords', []))}
 
Platform: {platform}
Instructions: {config['instruction']}
Character limit: {config['max_chars']} characters
 
Output only the post content, nothing else.
""",
    )
    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)
    cost_estimate_usd: float = 0.0
    elapsed_seconds: float = 0.0
 
async def run_content_pipeline(
    topic: str,
    target_audience: str = "developers",
    tone: str = "practical",
    output_dir: str = "./output",
) -> ContentResult:
    """
    Full pipeline execution.
    
    Execution order:
    1. Article generation (serial — thumbnail_prompt depends on this)
    2. Image + SNS generation (parallel — independent of each other)
    """
    import os
    os.makedirs(output_dir, exist_ok=True)
    
    result = ContentResult(topic=topic)
    t0 = time.time()
    
    # Phase 1: Article
    try:
        result.article = await generate_blog_article(topic, target_audience, tone)
        print(f"✅ Article: {result.article['title']}")
    except Exception as e:
        result.errors.append(f"Article generation failed: {e}")
        return result
    
    # Phase 2: Image + SNS in parallel
    thumb_path = os.path.join(output_dir, f"thumb_{int(t0)}.png")
    
    thumbnail_result, sns_result = await asyncio.gather(
        generate_thumbnail_with_fallback(
            result.article.get("thumbnail_prompt", f"{topic}, tech illustration"),
        ),
        generate_sns_content(result.article),
        return_exceptions=True,
    )
    
    if isinstance(thumbnail_result, Exception):
        result.errors.append(f"Thumbnail failed: {thumbnail_result}")
    else:
        result.thumbnail_path = thumb_path
    
    if isinstance(sns_result, Exception):
        result.errors.append(f"SNS generation failed: {sns_result}")
    else:
        result.sns_content = sns_result
    
    result.cost_estimate_usd = 0.08  # Article (~$0.03) + Image (~$0.04) + SNS (~$0.01)
    result.elapsed_seconds = time.time() - t0
    
    print(f"✅ Pipeline complete: {result.elapsed_seconds:.1f}s | ~${result.cost_estimate_usd:.3f}")
    return result

def quality_check(article: dict) -> tuple[bool, list]:
    issues = []
    content = article.get("content", "")
    title = article.get("title", "")
    
    if len(title) > 60:
        issues.append(f"Title too long: {len(title)} chars (max 60)")
    
    if len(content.split()) < 800:
        issues.append(f"Article too short: ~{len(content.split())} words (min 800)")
    
    # Detect template-style openers that indicate low-quality output
    bad_openers = ["in this article", "this article will", "we will cover", "in this guide"]
    first_50_words = " ".join(content.split()[:50]).lower()
    for pattern in bad_openers:
        if pattern in first_50_words:
            issues.append(f"Template opener detected: '{pattern}'")
            break
    
    description = article.get("description", "")
    if len(description) > 160:
        issues.append(f"Description too long: {len(description)} chars (max 160)")
    
    return len(issues) == 0, issues

# Sample pricing structure
PLANS = {
    "starter": {
        "monthly_jpy": 1980,
        "content_sets_per_month": 30,
    },
    "pro": {
        "monthly_jpy": 4980,
        "content_sets_per_month": 100,
        "custom_brand_voice": True,
    },
}
 
# Unit economics example:
# 100 Starter users: ¥198,000/month revenue
# API costs at 30 sets × 100 users × $0.10 = $300 (~¥45,000)
# Gross margin: ~77%

from dataclasses import dataclass
from typing import Optional
 
@dataclass
class BrandVoice:
    """Per-client brand voice configuration."""
    name: str
    tone: str                    # e.g., "authoritative and precise"
    vocabulary_avoid: list       # e.g., ["leverage", "synergy", "paradigm"]
    vocabulary_prefer: list      # e.g., ["we found", "our testing shows", "in practice"]
    example_opening: str         # A sentence in their voice to calibrate Gemini
    target_reading_level: str    # "general public", "technical professional", etc.
    custom_cta: Optional[str] = None
 
# Example client configurations
BRAND_VOICES = {
    "client_techblog": BrandVoice(
        name="TechBlog Co",
        tone="pragmatic and direct, never hype-driven",
        vocabulary_avoid=["revolutionary", "game-changing", "unlock the power"],
        vocabulary_prefer=["we measured", "the actual result", "in our codebase"],
        example_opening="After running this in production for three months, here's what the data shows.",
        target_reading_level="senior engineers",
    ),
    "client_startup": BrandVoice(
        name="GrowthStartup Inc",
        tone="energetic but evidence-based, founder voice",
        vocabulary_avoid=["utilize", "leverage", "paradigm shift"],
        vocabulary_prefer=["we shipped", "we tested", "here's what worked"],
        example_opening="We spent two weeks benchmarking six approaches so you don't have to.",
        target_reading_level="technical founders and lead developers",
    ),
}
 
def build_brand_system_instruction(voice: BrandVoice) -> str:
    """Build a system instruction tailored to a specific brand voice."""
    return f"""
You are a technical writer for {voice.name}.
 
Tone: {voice.tone}
Target reader: {voice.target_reading_level}
 
Words/phrases to AVOID: {', '.join(voice.vocabulary_avoid)}
Words/phrases to USE: {', '.join(voice.vocabulary_prefer)}
 
The following sentence captures the brand voice perfectly — write in this register:
"{voice.example_opening}"
 
Technical rules:
- Start with a specific finding, measurement, or problem. Never with "In this article..."
- Every claim needs evidence (a number, a test result, or a specific scenario)
- Code examples must be complete and runnable
- Use YOUR_GEMINI_API_KEY as the placeholder
"""
 
async def generate_article_for_client(
    topic: str,
    client_id: str,
    pipeline_client: genai.Client = None
) -> dict:
    """Generate an article using client-specific brand voice."""
    
    voice = BRAND_VOICES.get(client_id)
    if not voice:
        raise ValueError(f"Unknown client: {client_id}")
    
    _client = pipeline_client or client
    
    response = await _client.aio.models.generate_content(
        model="gemini-2.5-pro-latest",
        config=types.GenerateContentConfig(
            system_instruction=build_brand_system_instruction(voice),
            temperature=0.7,
            max_output_tokens=8192,
            response_mime_type="application/json",
        ),
        contents=f"Write an article about: {topic}. Return JSON with title, description, content, tags, seo_keywords, thumbnail_prompt fields.",
    )
    
    return json.loads(response.text)

import asyncio
from typing import AsyncIterator
 
async def batch_generate(
    topics: list[str],
    max_concurrent: int = 5,
    delay_between_batches: float = 2.0,
    output_dir: str = "./batch_output",
) -> AsyncIterator[ContentResult]:
    """
    Generate content for multiple topics with controlled concurrency.
    
    max_concurrent=5 is conservative but reliable for most paid tier quotas.
    Increase to 10-20 if you're on a higher quota tier.
    """
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def run_with_semaphore(topic: str) -> ContentResult:
        async with semaphore:
            try:
                result = await run_content_pipeline(topic, output_dir=output_dir)
                # Small delay between requests to avoid burst rate limits
                await asyncio.sleep(delay_between_batches)
                return result
            except Exception as e:
                return ContentResult(topic=topic, errors=[str(e)])
    
    # Run all topics with controlled concurrency
    tasks = [run_with_semaphore(topic) for topic in topics]
    results = await asyncio.gather(*tasks)
    
    for result in results:
        yield result
 
# Usage example
async def main():
    topics = [
        "Gemini API rate limit best practices",
        "Imagen 4 prompt engineering for product photography",
        "Google ADK multi-agent architecture patterns",
        "Gemini 2.5 Pro context window optimization",
        "Firebase AI Logic iOS production guide",
    ]
    
    successful = 0
    failed = 0
    total_cost = 0.0
    
    async for result in batch_generate(topics, max_concurrent=3):
        if result.errors:
            failed += 1
            print(f"❌ {result.topic}: {result.errors[0]}")
        else:
            successful += 1
            total_cost += result.cost_estimate_usd
            print(f"✅ {result.topic[:40]}... ({result.elapsed_seconds:.1f}s)")
    
    print(f"\nBatch complete: {successful} success, {failed} failed")
    print(f"Total estimated cost: ${total_cost:.3f}")
 
if __name__ == "__main__":
    asyncio.run(main())

# run_daily_batch.py — triggered by Cloud Scheduler
import os
import json
import asyncio
from datetime import datetime
 
# Topics are loaded from a JSON file you maintain separately
# This keeps the code stable while you update the editorial calendar
TOPICS_FILE = os.environ.get("TOPICS_FILE", "topics_queue.json")
 
async def run_daily_batch():
    with open(TOPICS_FILE) as f:
        queue = json.load(f)
    
    today_str = datetime.now().strftime("%Y-%m-%d")
    today_topics = queue.get(today_str, [])
    
    if not today_topics:
        print(f"No topics scheduled for {today_str}")
        return
    
    print(f"Generating {len(today_topics)} articles for {today_str}")
    
    results = []
    async for result in batch_generate(today_topics, output_dir=f"./output/{today_str}"):
        results.append(result)
    
    # Write summary
    summary = {
        "date": today_str,
        "total": len(results),
        "success": sum(1 for r in results if not r.errors),
        "failed": sum(1 for r in results if r.errors),
        "total_cost_usd": sum(r.cost_estimate_usd for r in results),
    }
    
    with open(f"./output/{today_str}/summary.json", "w") as f:
        json.dump(summary, f, indent=2)
    
    print(f"Summary: {summary}")
 
if __name__ == "__main__":
    asyncio.run(run_daily_batch())

import logging
from datetime import datetime
 
# Configure structured logging
logging.basicConfig(
    level=logging.INFO,
    format='{"time": "%(asctime)s", "level": "%(levelname)s", "message": "%(message)s"}',
)
logger = logging.getLogger("content_pipeline")
 
class PipelineMetrics:
    """Track pipeline performance over time."""
    
    def __init__(self):
        self.runs = []
    
    def record(self, result: ContentResult):
        self.runs.append({
            "timestamp": datetime.now().isoformat(),
            "topic": result.topic[:50],
            "success": len(result.errors) == 0,
            "errors": result.errors,
            "elapsed_s": result.elapsed_seconds,
            "cost_usd": result.cost_estimate_usd,
            "has_thumbnail": result.thumbnail_path is not None,
            "sns_platforms": list(result.sns_content.keys()),
        })
        
        logger.info(f"Pipeline run: topic={result.topic[:30]}, "
                   f"success={len(result.errors)==0}, "
                   f"elapsed={result.elapsed_seconds:.1f}s, "
                   f"cost=${result.cost_estimate_usd:.3f}")
    
    def summary(self) -> dict:
        if not self.runs:
            return {}
        
        total = len(self.runs)
        successful = sum(1 for r in self.runs if r["success"])
        
        return {
            "total_runs": total,
            "success_rate": f"{successful/total*100:.1f}%",
            "avg_elapsed_s": sum(r["elapsed_s"] for r in self.runs) / total,
            "total_cost_usd": sum(r["cost_usd"] for r in self.runs),
            "most_common_error": self._most_common_error(),
        }
    
    def _most_common_error(self) -> str:
        errors = [e for r in self.runs for e in r["errors"]]
        if not errors:
            return "none"
        from collections import Counter
        return Counter(errors).most_common(1)[0][0]
 
metrics = PipelineMetrics()
 
# Usage: wrap every pipeline run
async def tracked_pipeline(topic: str, **kwargs) -> ContentResult:
    result = await run_content_pipeline_with_quality(topic, **kwargs)
    metrics.record(result)
    return result

# Example: "cybersecurity dashboard with lock icon" sometimes triggers
# Solution: Use more abstract descriptions
# ❌ "security breach visualization with hacker and server"
# ✅ "network security monitoring dashboard, abstract data flow visualization"

# Detection
try:
    data = json.loads(response.text)
except json.JSONDecodeError:
    # Check if the response was truncated
    if response.candidates[0].finish_reason.name == "MAX_TOKENS":
        # Retry with a request for shorter content
        # or increase max_output_tokens
        logger.warning(f"Article truncated for topic: {topic}")

def enforce_character_limits(sns_content: dict) -> dict:
    """Trim SNS content to platform limits if Gemini exceeded them."""
    limits = {"x": 140, "instagram": 2200, "linkedin": 1300}
    
    result = {}
    for platform, content in sns_content.items():
        limit = limits.get(platform, 5000)
        if len(content) > limit:
            # Trim at the last complete sentence before the limit
            trimmed = content[:limit]
            last_period = trimmed.rfind(".")
            result[platform] = trimmed[:last_period + 1] if last_period > limit * 0.7 else trimmed
            logger.info(f"Trimmed {platform} content: {len(content)} → {len(result[platform])} chars")
        else:
            result[platform] = content
    
    return result

response = await client.aio.models.generate_content(...)
 
# Check actual token usage
if hasattr(response, "usage_metadata"):
    input_tokens = response.usage_metadata.prompt_token_count
    output_tokens = response.usage_metadata.candidates_token_count
    
    # Gemini 2.5 Pro pricing reference (verify current rates at ai.google.dev)
    estimated_cost = (input_tokens * 0.00000125) + (output_tokens * 0.000010)
    
    if output_tokens > 6000:
        logger.warning(f"Unexpectedly long output: {output_tokens} tokens (~${estimated_cost:.4f})")

import asyncio
 
async def smoke_test_pipeline():
    """
    Quick validation of all pipeline components.
    Run this before any production deployment or client demo.
    """
    print("🔍 Running pipeline smoke test...")
    
    test_topic = "How to handle async errors in Python"
    
    # Test 1: Article generation
    print("  Test 1: Article generation...")
    try:
        article = await generate_blog_article(test_topic, "Python developers", "practical")
        assert "title" in article and "content" in article
        assert len(article["content"]) > 500
        passed, issues = quality_check(article)
        print(f"    ✅ Article generated: {len(article['content'])} chars")
        if not passed:
            print(f"    ⚠️  Quality issues: {issues}")
    except Exception as e:
        print(f"    ❌ Article generation failed: {e}")
        return False
    
    # Test 2: Thumbnail generation
    print("  Test 2: Thumbnail generation...")
    try:
        prompt = build_thumbnail_prompt("Python async error handling, code editor screenshot")
        path = await generate_thumbnail(prompt, output_path="./test_thumbnail.png")
        print(f"    ✅ Thumbnail generated: {path}")
    except Exception as e:
        print(f"    ❌ Thumbnail failed: {e}")
        # Non-fatal: continue to SNS test
    
    # Test 3: SNS content
    print("  Test 3: SNS content generation...")
    try:
        sns = await generate_sns_content(article)
        for platform, content in sns.items():
            if isinstance(content, str) and not content.startswith("Error:"):
                print(f"    ✅ {platform}: {len(content)} chars")
            else:
                print(f"    ❌ {platform}: {content}")
    except Exception as e:
        print(f"    ❌ SNS generation failed: {e}")
        return False
    
    print("✅ Smoke test complete")
    return True
 
if __name__ == "__main__":
    asyncio.run(smoke_test_pipeline())