Gemini API で構築する自動収益化インフラ — マルチモーダル×Function Callingで実現する6つの収益エンジン

# api/agent.py
# Gemini Function Calling で外部サービスを自動操作
 
import google.generativeai as genai
 
# 利用可能なツール定義
tools = [
    genai.protos.Tool(function_declarations=[
        genai.protos.FunctionDeclaration(
            name="search_products",
            description="ECサイトで商品を検索する",
            parameters=genai.protos.Schema(
                type=genai.protos.Type.OBJECT,
                properties={
                    "query": genai.protos.Schema(type=genai.protos.Type.STRING),
                    "max_price": genai.protos.Schema(type=genai.protos.Type.NUMBER),
                    "category": genai.protos.Schema(type=genai.protos.Type.STRING),
                },
                required=["query"],
            ),
        ),
        genai.protos.FunctionDeclaration(
            name="create_invoice",
            description="請求書を自動作成する",
            parameters=genai.protos.Schema(
                type=genai.protos.Type.OBJECT,
                properties={
                    "client_name": genai.protos.Schema(type=genai.protos.Type.STRING),
                    "items": genai.protos.Schema(type=genai.protos.Type.STRING),
                    "due_date": genai.protos.Schema(type=genai.protos.Type.STRING),
                },
                required=["client_name", "items"],
            ),
        ),
        genai.protos.FunctionDeclaration(
            name="schedule_meeting",
            description="カレンダーにミーティングを追加する",
            parameters=genai.protos.Schema(
                type=genai.protos.Type.OBJECT,
                properties={
                    "title": genai.protos.Schema(type=genai.protos.Type.STRING),
                    "datetime": genai.protos.Schema(type=genai.protos.Type.STRING),
                    "attendees": genai.protos.Schema(type=genai.protos.Type.STRING),
                },
                required=["title", "datetime"],
            ),
        ),
    ]),
]
 
model = genai.GenerativeModel("gemini-2.5-pro", tools=tools)
 
async def process_business_request(user_request: str):
    """自然言語の指示からビジネスタスクを自動実行"""
    chat = model.start_chat()
    response = chat.send_message(user_request)
 
    # Function Call の結果を実行
    for part in response.parts:
        if fn := part.function_call:
            result = await execute_function(fn.name, dict(fn.args))
            # 結果をGeminiに返して次のアクションを決定
            response = chat.send_message(
                genai.protos.Content(parts=[
                    genai.protos.Part(function_response=genai.protos.FunctionResponse(
                        name=fn.name,
                        response={"result": result},
                    ))
                ])
            )
 
    return response.text

# scripts/content_pipeline.py
# 画像 + テキストからリッチコンテンツを自動生成
 
async def generate_rich_content(
    topic: str,
    reference_images: list[bytes],
    platform: str
) -> dict:
    """画像を参照しながらプラットフォーム最適化されたコンテンツを生成"""
 
    model = genai.GenerativeModel("gemini-2.5-pro")
 
    # 画像を参照資料として入力
    content_parts = [
        f"""以下のトピックについて、{platform}向けの記事を生成してください。
添付画像を参照資料として活用し、画像の内容を記事に組み込んでください。
 
トピック: {topic}
プラットフォーム: {platform}
目標文字数: 3000文字
 
SEO最適化された見出し構造で、読者がすぐに実践できる内容にしてください。""",
    ]
 
    for img in reference_images:
        content_parts.append({"mime_type": "image/jpeg", "data": img})
 
    response = model.generate_content(content_parts)
    return {"content": response.text, "platform": platform}

# api/data_analysis.py
# コンテキストキャッシュで繰り返し分析のコストを75%削減
 
from google.generativeai import caching
import datetime
 
async def create_analysis_cache(dataset: str) -> str:
    """大規模データセットをキャッシュし、複数クエリで再利用"""
 
    cache = caching.CachedContent.create(
        model="models/gemini-2.5-pro",
        display_name="sales_data_2026",
        system_instruction="あなたはデータアナリストです。提供されたデータに基づいて分析してください。",
        contents=[dataset],
        ttl=datetime.timedelta(hours=24),  # 24時間キャッシュ
    )
 
    return cache.name
 
async def query_cached_data(cache_name: str, question: str) -> str:
    """キャッシュされたデータに対してクエリ（通常の1/4のコスト）"""
    cached_content = caching.CachedContent.get(cache_name)
    model = genai.GenerativeModel.from_cached_content(cached_content)
 
    response = model.generate_content(question)
    return response.text
 
# 使用例:
# 1回目: データセット全体をキャッシュ（通常料金）
# 2回目以降: キャッシュに対するクエリ（入力コストの1/4）
# → 10回の分析クエリで約75%のコスト削減