Monetizing Content Production Services with Gemini 2.5 Flash's Low-Cost Advantage

# 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 tasks (fast, low-cost)
    PRODUCT_DESCRIPTION = "product_description"
    SHORT_SUMMARY = "short_summary"
    SNS_POST = "sns_post"
    EMAIL_TEMPLATE = "email_template"
    TRANSLATION = "translation"
    FAQ_ANSWER = "faq_answer"
    # Pro tasks (higher quality, higher cost)
    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:
    """Auto-selects Flash or Pro based on task type and generates content."""
    
    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,
    )
 
# Example usage
result = generate_content(
    prompt=(
        "Generate a 150-word product description for an e-commerce listing:\n\n"
        "Product: Wireless Earbuds X200\nBattery: 8-hour playback\n"
        "Water resistance: IPX5\nWeight: 5g per earbud"
    ),
    task=ContentTask.PRODUCT_DESCRIPTION,
    system_instruction="You are an e-commerce copywriter. Write compelling descriptions that make people want to buy.",
)
 
print(f"Output: {result.text}")
print(f"Model: {result.model_used}")
print(f"Cost: ${result.cost_usd:.6f}")
# Output:
# Model: gemini-2.5-flash
# Cost: $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]:
    """Async parallel generation with concurrency limit to avoid rate limiting."""
    
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def generate_single(prompt: str) -> GenerationResult:
        async with semaphore:
            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_list = [generate_single(p) for p in prompts]
    results = await asyncio.gather(*tasks_list, 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
 
# Generate 100 product descriptions in bulk
async def bulk_generate_product_descriptions(products: List[dict]) -> List[str]:
    prompts = [
        f"Product: {p['name']}\nSpecs: {p['spec']}\n\n"
        f"Write a 100-word e-commerce product description."
        for p in products
    ]
    
    results = await generate_batch_async(
        prompts=prompts,
        task=ContentTask.PRODUCT_DESCRIPTION,
        max_concurrent=20,
    )
    
    total_cost = sum(r.cost_usd for r in results)
    print(f"100-item generation cost: ${total_cost:.4f}")  # e.g., $0.0270
    
    return [r.text for r in results]

# Before: vague prompt
POOR_PROMPT = "Write a product description"
 
# After: structured prompt with explicit constraints
BETTER_PROMPT = """
Write a product description for an e-commerce listing.
 
Requirements:
- Length: 150–200 words
- Open with the product's strongest selling point
- Include 1–2 specific specs or numbers
- Use emotionally resonant language that makes readers want to buy
- End with a clear call to action
 
Product details:
{product_info}
"""