記事一覧/連携・プラグイン

⬡ 連携・プラグイン/2026-04-17上級

Google Antigravity Python SDK 本番実装マスターガイド — マルチモーダル・エージェント・RAGパイプラインの設計から運用まで

Google Antigravity Python SDKを本番環境で使い倒すための完全ガイド。マルチモーダル入力・ツール呼び出し・RAGパイプライン・ストリーミング・コスト最適化・Cloud Runデプロイまで、実動作するコード付きで解説します。

python²⁶ antigravity⁴³⁵ sdk⁵ rag⁸ agents⁹⁰ multimodal⁶ production⁵⁹

✦ プレミアム記事

Antigravity のドキュメントを読んで「とりあえず動いた」段階から、実際に本番で使えるものを作るまでの間には、思いのほか大きな壁があります。

私がそれを実感したのは、最初に作ったチャットbotのログを見たときです。レート制限エラーが散発し、会話履歴がセッションをまたぐと消えて、ファイルアップロードは一定サイズを超えると無言で失敗していました。コードは動いているのに、使えるものになっていありません。そういう状態でした。

私自身、2014年からiOS/Androidの個人開発を続けてきて、壁紙系・癒し系のアプリを App Store・Google Play 経由で配信してきました。アプリそのものは AdMob 広告で運用が回っているのですが、そこに「ユーザーが画像を投げると壁紙の使い方アドバイスを返す」ような AI 機能を足そうとして、本番運用でつまずいたポイントが多々あります。このガイドの随所に、その実運用で見えた勘所を織り込んでいます。

このガイドでは、Antigravity Python SDK を使って本番に耐えるアプリを作るために必要なことを、実際に私がハマったポイントを中心にまとめています。マルチモーダル入力からエージェント設計、RAGパイプライン、そしてCloud Runへのデプロイまで、動くコードと一緒に解説します。

Python SDK の設計思想を理解する — なぜ直接APIを叩かないのか

まず「Python SDKを使うべき理由」から始めます。REST APIを直接叩けば同じことができるのに、なぜSDKを使うのかを理解していると、ドキュメントの読み方が変わります。

Antigravity Python SDK（google-generativeai パッケージ）が提供する最大の価値は、プロトコルの抽象化ではなく状態管理の自動化です。生のAPIリクエストでは、マルチターン会話の履歴管理・ストリーミングレスポンスのバッファリング・ファイルアップロードの一時URL管理・ツール呼び出しのループ処理などを自前で実装する必要があります。SDKはこれらを引き受けてくれます。

また、google-generativeai と vertexai という2つのパッケージがあることも混乱の原因になりがちです。

# パターン1: google-generativeai（個人・プロトタイプ向け）
import google.generativeai as genai
genai.configure(api_key="YOUR_API_KEY")
 
# パターン2: vertexai（Google Cloud・本番向け）
import vertexai
from vertexai.generative_models import GenerativeModel
vertexai.init(project="YOUR_PROJECT_ID", location="us-central1")

個人のAPIキーで素早く試すなら google-generativeai、Google Cloud プロジェクトに紐づけてスケールさせるなら vertexai です。本番運用では後者を選ぶことをお勧めします。Cloud Run との統合がシームレスで、IAM による権限管理が使えるからです。

このガイドでは主に google-generativeai を使いながら、本番デプロイのセクションで vertexai への移行パターンも示します。

環境構築と認証エラーを1時間で突破する

認証周りのエラーは最初のハードルです。よくあるパターンをまとめておきます。

# インストール
pip install google-generativeai python-dotenv
 
# .envファイルに書く（gitignore必須）
GOOGLE_API_KEY=AIza...（ここにあなたのAPIキー）

import os
from dotenv import load_dotenv
import google.generativeai as genai
 
load_dotenv()
genai.configure(api_key=os.environ["GOOGLE_API_KEY"])
 
# 接続テスト
model = genai.GenerativeModel("gemini-2.0-flash")
response = model.generate_content("テスト: 日本語で1文で自己紹介してください")
print(response.text)
# → 「私はGoogleが開発したAIアシスタントです。」

よくある認証エラーと解決策:

google.api_core.exceptions.PermissionDenied: 403 API key not valid → APIキーの先頭・末尾にスペースや改行が混入していることが多いです。.strip() を挟むか、.env ファイルを再確認してください。

google.api_core.exceptions.ResourceExhausted: 429 Quota exceeded → 無料枠のレート制限（1分あたりのリクエスト数）に達しています。後述のリトライ実装が必要です。

google.auth.exceptions.DefaultCredentialsError → vertexai パッケージ使用時に ADC（Application Default Credentials）が設定されていない場合に発生します。gcloud auth application-default login を実行してください。

✦

ここまでお読みいただきありがとうございます。

この記事の続きを読む

この先には、実装コードやベンチマーク結果など、実務でお役に立てる内容をご用意しています。このサイトは広告を掲載しておらず、サーバーや開発にかかる費用はメンバーの皆様のご支援で成り立っています。もしお役に立てていましたら、ご支援いただけますと大変ありがたいです。

この記事で得られること

✦5,000万DL規模の個人開発から学んだ、ユーザー1人あたり月10円以下に抑えるトークン予算試算とコード

✦AdMob × AI機能の相性問題を解消する『二段構え応答』『finish_reason ログ化』『カナリア比較』の具体実装

✦本番投入前に必ず仕込むべき3つのモニタリングライン（100%サンプリング・Slack通知・コスト閾値フォールバック）

Stripe による安全な決済 · いつでもキャンセル可能

✦

この記事を購入する

この先の内容をすべてお読みいただけます。一度のご購入で、いつでも何度でもアクセスできます。このサイトは広告を掲載しておらず、皆さまのご支援がサーバー費用などの運営を支えています。

または

メンバーシップなら全記事が読み放題 →

マルチモーダル入力の実装 — テキスト・画像・PDF・動画を扱う

Antigravity の最も強力な機能の一つが、テキスト以外のメディアを入力として受け取れることです。ただし、入力方式にはいくつかのパターンがあり、これを間違えると無言で失敗したり、コストが爆増したりします。

画像入力 — インラインデータ vs File API

import google.generativeai as genai
from PIL import Image
import io
 
model = genai.GenerativeModel("gemini-2.0-flash")
 
# 方法1: 小さい画像はbase64エンコードしてインラインで送る（4MB以下推奨）
with open("screenshot.png", "rb") as f:
    image_data = f.read()
 
response = model.generate_content([
    {"mime_type": "image/png", "data": image_data},
    "このスクリーンショットのUIの問題点を3つ挙げてください"
])
print(response.text)
 
# 方法2: 大きいファイルはFile APIを使う（最大2GB、48時間有効）
sample_file = genai.upload_file(
    path="large_video.mp4",
    display_name="分析対象動画"
)
# アップロード完了を待つ（動画は処理に時間がかかる）
import time
while sample_file.state.name == "PROCESSING":
    print(".", end="", flush=True)
    time.sleep(2)
    sample_file = genai.get_file(sample_file.name)
 
if sample_file.state.name == "FAILED":
    raise ValueError("ファイル処理に失敗しました")
 
response = model.generate_content([
    sample_file,
    "この動画の主要なポイントを要約してください"
])
print(response.text)
 
# ⚠️ 重要: 使い終わったらFile APIのファイルを削除する
# 削除しないと48時間後に自動削除されるが、料金が発生し続ける
genai.delete_file(sample_file.name)

よくある間違い: 動画を方法1（インライン）で送ろうとして invalid_argument エラーが出るケース。動画は必ずFile API経由で送る必要があります。PDFも同様です（20MBを超えるPDFはFile API必須）。

PDFドキュメントの解析

# PDFをアップロードして内容を解析する
pdf_file = genai.upload_file(
    path="technical_spec.pdf",
    display_name="技術仕様書"
)
 
# 処理完了を待つ
while pdf_file.state.name == "PROCESSING":
    time.sleep(1)
    pdf_file = genai.get_file(pdf_file.name)
 
response = model.generate_content([
    pdf_file,
    "この仕様書から実装が必要なAPIエンドポイントの一覧をJSON形式で出力してください。"
    "形式: [{\"endpoint\": \"/api/xxx\", \"method\": \"POST\", \"description\": \"...\"}]"
])
 
import json
# レスポンスからJSONを抽出
try:
    json_start = response.text.index("[")
    json_end = response.text.rindex("]") + 1
    endpoints = json.loads(response.text[json_start:json_end])
    for ep in endpoints:
        print(f"{ep['method']} {ep['endpoint']}: {ep['description']}")
except (ValueError, json.JSONDecodeError) as e:
    print(f"JSON解析エラー: {e}")
    print("Raw response:", response.text)

本番グレードのエージェント実装パターン

ツール呼び出し（Function Calling）は、Antigravity で実用的なエージェントを作る核心技術です。しかし、実際に本番で動かすには「ツール定義」だけでなく「エラーリカバリ」と「状態管理」が不可欠です。

ツール呼び出しの基本実装

import google.generativeai as genai
import requests
import json
 
# ツール定義（型情報を正確に書くことが重要）
tools = [
    {
        "function_declarations": [
            {
                "name": "get_weather",
                "description": "指定した都市の現在の天気情報を取得します",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "city": {
                            "type": "string",
                            "description": "都市名（例: 東京、大阪）"
                        },
                        "unit": {
                            "type": "string",
                            "enum": ["celsius", "fahrenheit"],
                            "description": "温度の単位"
                        }
                    },
                    "required": ["city"]
                }
            }
        ]
    }
]
 
def get_weather(city: str, unit: str = "celsius") -> dict:
    """実際の天気API呼び出し（エラーハンドリング付き）"""
    try:
        # 例: OpenWeatherMap API（実際のAPIキーが必要）
        response = requests.get(
            "https://api.openweathermap.org/data/2.5/weather",
            params={"q": city, "appid": "YOUR_OWM_KEY", "units": "metric"},
            timeout=5  # タイムアウト必須
        )
        response.raise_for_status()
        data = response.json()
        temp = data["main"]["temp"]
        if unit == "fahrenheit":
            temp = temp * 9/5 + 32
        return {
            "city": city,
            "temperature": round(temp, 1),
            "unit": unit,
            "description": data["weather"][0]["description"],
            "humidity": data["main"]["humidity"]
        }
    except requests.Timeout:
        return {"error": "天気APIがタイムアウトしました。しばらくしてから再試行してください。"}
    except requests.HTTPError as e:
        return {"error": f"天気の取得に失敗しました: {e.response.status_code}"}
    except Exception as e:
        return {"error": f"予期しないエラーが発生しました: {str(e)}"}
 
 
def run_agent(user_message: str) -> str:
    """ツール呼び出しループを管理するエージェント"""
    model = genai.GenerativeModel("gemini-2.0-flash", tools=tools)
    messages = [{"role": "user", "parts": [user_message]}]
    
    max_iterations = 5  # 無限ループ防止
    for iteration in range(max_iterations):
        response = model.generate_content(messages)
        candidate = response.candidates[0]
        
        # ツール呼び出しが含まれているか確認
        has_tool_call = False
        tool_results = []
        
        for part in candidate.content.parts:
            if hasattr(part, "function_call") and part.function_call:
                has_tool_call = True
                func_name = part.function_call.name
                func_args = dict(part.function_call.args)
                
                print(f"[Tool Call] {func_name}({func_args})")
                
                # ツールを実行
                if func_name == "get_weather":
                    result = get_weather(**func_args)
                else:
                    result = {"error": f"未知のツール: {func_name}"}
                
                print(f"[Tool Result] {result}")
                tool_results.append({
                    "function_response": {
                        "name": func_name,
                        "response": result
                    }
                })
        
        if not has_tool_call:
            # テキスト応答があれば完了
            return candidate.content.parts[0].text
        
        # ツール結果をコンテキストに追加してループ継続
        messages.append({"role": "model", "parts": candidate.content.parts})
        messages.append({"role": "user", "parts": tool_results})
    
    return "最大反復回数に達しました。処理を中断します。"
 
# 実行例
result = run_agent("東京の天気を摂氏で教えてください")
print(result)

会話履歴の永続化

セッションをまたいだ会話履歴の保存は、本番アプリでは不可欠です。インメモリだけでは、サーバー再起動や別インスタンスへのルーティングで履歴が消えます。

import json
import redis
import google.generativeai as genai
from typing import Optional
 
class PersistentConversation:
    """Redisを使った会話履歴の永続化"""
    
    def __init__(self, session_id: str, redis_url: str = "redis://localhost:6379"):
        self.session_id = session_id
        self.redis = redis.from_url(redis_url, decode_responses=True)
        self.model = genai.GenerativeModel("gemini-2.0-flash")
        self.ttl = 24 * 60 * 60  # 24時間でセッション自動削除
    
    def _load_history(self) -> list:
        """Redisから会話履歴を取得"""
        raw = self.redis.get(f"chat:{self.session_id}")
        if raw:
            return json.loads(raw)
        return []
    
    def _save_history(self, history: list) -> None:
        """会話履歴をRedisに保存"""
        self.redis.setex(
            f"chat:{self.session_id}",
            self.ttl,
            json.dumps(history, ensure_ascii=False)
        )
    
    def send_message(self, user_message: str) -> str:
        """メッセージを送信し、応答を返す"""
        history = self._load_history()
        
        # ChatSessionに履歴を渡して続きから開始
        chat = self.model.start_chat(history=history)
        response = chat.send_message(user_message)
        
        # 更新された履歴を保存（ChatSessionのhistoryはAPIが管理）
        updated_history = []
        for message in chat.history:
            updated_history.append({
                "role": message.role,
                "parts": [part.text for part in message.parts if hasattr(part, "text")]
            })
        self._save_history(updated_history)
        
        return response.text
    
    def clear(self) -> None:
        """会話履歴をリセット"""
        self.redis.delete(f"chat:{self.session_id}")
 
# 使用例
conv = PersistentConversation(session_id="user_12345")
print(conv.send_message("こんにちは！私の名前は政樹です"))
print(conv.send_message("私の名前を覚えていますか？"))  # → 政樹と答えるはず

RAGパイプラインの構築 — ベクター検索から応答生成まで

Retrieval-Augmented Generation（RAG）は、自社データやドキュメントをAntigravityと組み合わせる最も実用的な手法です。実装の核心は「どこで精度が落ちるか」を把握することです。

エンベディングとベクトル検索

import google.generativeai as genai
import numpy as np
from typing import List, Tuple
 
# エンベディングの生成
def embed_text(text: str) -> list[float]:
    """テキストをベクトルに変換"""
    result = genai.embed_content(
        model="models/text-embedding-004",
        content=text,
        task_type="retrieval_document"  # ドキュメント用
    )
    return result["embedding"]
 
def embed_query(query: str) -> list[float]:
    """クエリをベクトルに変換（ドキュメントと異なるtask_type）"""
    result = genai.embed_content(
        model="models/text-embedding-004",
        content=query,
        task_type="retrieval_query"  # クエリ用（これを間違えると精度が下がる）
    )
    return result["embedding"]
 
def cosine_similarity(a: list, b: list) -> float:
    """コサイン類似度の計算"""
    a_np = np.array(a)
    b_np = np.array(b)
    return float(np.dot(a_np, b_np) / (np.linalg.norm(a_np) * np.linalg.norm(b_np)))
 
# シンプルなインメモリベクターストア（プロトタイプ用）
class SimpleVectorStore:
    def __init__(self):
        self.documents: List[dict] = []
    
    def add(self, text: str, metadata: dict = None) -> None:
        """ドキュメントを追加"""
        embedding = embed_text(text)
        self.documents.append({
            "text": text,
            "embedding": embedding,
            "metadata": metadata or {}
        })
    
    def search(self, query: str, top_k: int = 3) -> List[Tuple[str, float]]:
        """類似度の高いドキュメントを返す"""
        query_embedding = embed_query(query)
        results = []
        for doc in self.documents:
            score = cosine_similarity(query_embedding, doc["embedding"])
            results.append((doc["text"], score, doc["metadata"]))
        
        # スコア降順でソートしてtop_kを返す
        results.sort(key=lambda x: x[1], reverse=True)
        return results[:top_k]
 
# RAGパイプライン全体
def rag_query(store: SimpleVectorStore, question: str) -> str:
    """質問に対してRAGで回答を生成"""
    # 1. 関連ドキュメントを検索
    results = store.search(question, top_k=3)
    
    # 閾値以下の低品質な結果を除外（これを忘れると幻覚が増える）
    relevant = [(text, score) for text, score, _ in results if score > 0.6]
    
    if not relevant:
        return "関連する情報が見つかりませんでした。より具体的な質問をお試しください。"
    
    # 2. コンテキストを構築
    context_parts = []
    for i, (text, score) in enumerate(relevant, 1):
        context_parts.append(f"[参考資料{i}（関連度: {score:.2f}）]\n{text}")
    context = "\n\n".join(context_parts)
    
    # 3. Antigravityで回答を生成
    model = genai.GenerativeModel("gemini-2.0-flash")
    prompt = f"""以下の参考資料を基に、質問に回答してください。
参考資料にない情報は「この資料には記載がありません」と答えてください。
 
参考資料:
{context}
 
質問: {question}
 
回答:"""
    
    response = model.generate_content(prompt)
    return response.text
 
# 使用例
store = SimpleVectorStore()
store.add("Antigravityは2025年にGoogleが公開したAI開発環境です。", {"source": "intro.md"})
store.add("Antigravityのプレミアムプランは月額$19.99です。", {"source": "pricing.md"})
store.add("AntigravityはGemini 2.0をベースとしたコード補完を提供します。", {"source": "features.md"})
 
answer = rag_query(store, "Antigravityの料金はいくらですか？")
print(answer)

重要な設計判断: task_type を retrieval_document と retrieval_query で使い分けることは必須です。同じ retrieval_document でクエリもエンベディングすると、検索精度が目に見えて低下します。これは公式ドキュメントに書いてあることですが、見落としがちなポイントです。

本番では ChromaDB（ローカル・軽量）や pgvector（PostgreSQL統合）、Vertex AI Search（Google Cloud フルマネージド）から要件に合わせて選択します。詳しくはRAGパイプラインとベクター検索の実践ガイドを参照してください。

ストリーミングレスポンスとリアルタイム処理

チャットUIやログ表示で、生成された文字がリアルタイムに流れる体験を実現するにはストリーミングが必要です。FastAPIとWebSocketを組み合わせた実装例を示します。

import asyncio
import google.generativeai as genai
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.responses import HTMLResponse
 
app = FastAPI()
model = genai.GenerativeModel("gemini-2.0-flash")
 
@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await websocket.accept()
    
    try:
        while True:
            # ユーザーメッセージを受信
            message = await websocket.receive_text()
            
            # ストリーミング生成
            response_stream = await asyncio.to_thread(
                model.generate_content,
                message,
                stream=True  # ストリーミングを有効化
            )
            
            # チャンクを順次送信
            full_response = ""
            for chunk in response_stream:
                if chunk.text:
                    full_response += chunk.text
                    await websocket.send_json({
                        "type": "chunk",
                        "text": chunk.text
                    })
            
            # 完了シグナルを送信
            await websocket.send_json({
                "type": "done",
                "full_text": full_response
            })
    
    except WebSocketDisconnect:
        print("クライアントが切断しました")
    except Exception as e:
        await websocket.send_json({"type": "error", "message": str(e)})
        await websocket.close()

レート制限と指数バックオフリトライ

import time
import random
from functools import wraps
from google.api_core.exceptions import ResourceExhausted, ServiceUnavailable
 
def with_retry(max_retries: int = 3, base_delay: float = 1.0):
    """指数バックオフ付きリトライデコレータ"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except (ResourceExhausted, ServiceUnavailable) as e:
                    if attempt == max_retries - 1:
                        raise  # 最後の試行でも失敗したら例外を上げる
                    
                    # ジッターを加えた指数バックオフ
                    delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                    print(f"APIエラー（{type(e).__name__}）: {delay:.1f}秒後に再試行... ({attempt + 1}/{max_retries})")
                    time.sleep(delay)
        return wrapper
    return decorator
 
model = genai.GenerativeModel("gemini-2.0-flash")
 
@with_retry(max_retries=3)
def safe_generate(prompt: str) -> str:
    """リトライ付きのコンテンツ生成"""
    response = model.generate_content(prompt)
    return response.text
 
# 使用例
result = safe_generate("Pythonでクイックソートを実装してください")
print(result)

コスト最適化とモニタリング

個人開発でAPIコストを制御するには、使用量の可視化と意図的なキャッシング戦略が必要です。

トークン使用量のトラッキング

from dataclasses import dataclass, field
from typing import Optional
import time
 
@dataclass
class TokenUsageTracker:
    """トークン使用量を追跡するクラス"""
    input_tokens: int = 0
    output_tokens: int = 0
    total_requests: int = 0
    start_time: float = field(default_factory=time.time)
    
    # Gemini 2.0 Flash の料金（2026年4月時点）
    INPUT_COST_PER_1M = 0.075   # USD / 1M tokens
    OUTPUT_COST_PER_1M = 0.30   # USD / 1M tokens
    
    def record(self, response) -> None:
        """APIレスポンスからトークン数を記録"""
        if hasattr(response, "usage_metadata"):
            self.input_tokens += response.usage_metadata.prompt_token_count
            self.output_tokens += response.usage_metadata.candidates_token_count
        self.total_requests += 1
    
    def estimated_cost_usd(self) -> float:
        """推定コスト（USD）を計算"""
        input_cost = (self.input_tokens / 1_000_000) * self.INPUT_COST_PER_1M
        output_cost = (self.output_tokens / 1_000_000) * self.OUTPUT_COST_PER_1M
        return input_cost + output_cost
    
    def report(self) -> str:
        elapsed = time.time() - self.start_time
        return (
            f"=== トークン使用量レポート ===\n"
            f"入力トークン: {self.input_tokens:,}\n"
            f"出力トークン: {self.output_tokens:,}\n"
            f"総リクエスト数: {self.total_requests}\n"
            f"推定コスト: ${self.estimated_cost_usd():.4f} USD\n"
            f"経過時間: {elapsed:.1f}秒\n"
            f"平均コスト/リクエスト: ${self.estimated_cost_usd() / max(1, self.total_requests):.5f}"
        )
 
tracker = TokenUsageTracker()
model = genai.GenerativeModel("gemini-2.0-flash")
 
response = model.generate_content("Pythonのリスト内包表記を説明してください")
tracker.record(response)
print(response.text)
print("\n" + tracker.report())

よくある落とし穴: 同じシステムプロンプトを毎回リクエストに含めると、入力トークンが無駄に消費されます。Context Caching APIを使うと、頻繁に使うシステムプロンプトやドキュメントをキャッシュして、2回目以降のコストを大幅に削減できます。詳細はGemini API 高度統合マスタークラスを参照してください。

本番デプロイ — Cloud Run への移行

ローカルで動いたアプリをCloud Runに乗せるときの、実際の手順とハマりポイントをまとめます。

Dockerfile と requirements.txt

# Dockerfile
FROM python:3.12-slim
 
WORKDIR /app
 
# 依存関係のインストール（キャッシュを活用するため先にコピー）
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
# アプリのコピー
COPY . .
 
# Cloud Runはポート8080を期待する
ENV PORT=8080
EXPOSE 8080
 
# gunicornで本番起動
CMD ["gunicorn", "main:app", "-w", "2", "-k", "uvicorn.workers.UvicornWorker", "--bind", "0.0.0.0:8080"]

# requirements.txt
google-generativeai==0.8.3
fastapi==0.115.0
uvicorn==0.30.0
gunicorn==22.0.0
python-dotenv==1.0.1
redis==5.0.1
httpx==0.27.0

APIキーをSecret Managerで管理

# APIキーをSecret Managerに登録
echo -n "YOUR_GOOGLE_API_KEY" | gcloud secrets create GOOGLE_API_KEY --data-file=-
 
# Cloud Run にシークレットをマウント
gcloud run deploy my-ai-app \
  --image gcr.io/YOUR_PROJECT_ID/my-ai-app \
  --region asia-northeast1 \
  --set-secrets "GOOGLE_API_KEY=GOOGLE_API_KEY:latest" \
  --memory 1Gi \
  --cpu 1 \
  --concurrency 80 \
  --max-instances 10 \
  --allow-unauthenticated

# Cloud Run環境でのAPIキー読み込み（環境変数から取得）
import os
import google.generativeai as genai
 
# Secret ManagerからマウントされたAPIキーはenv varとして利用可能
api_key = os.environ.get("GOOGLE_API_KEY")
if not api_key:
    raise EnvironmentError("GOOGLE_API_KEY環境変数が設定されていません")
 
genai.configure(api_key=api_key)

費用感: Cloud Run は「リクエストがない時間はゼロコスト」のサーバーレスです。個人開発の初期段階では月額数百円以下で運用できることが多いです。スケールしてきたら vertexai パッケージに移行し、IAM認証を使ってAPIキーを排除するとセキュリティが向上します。

Cloud Runへのデプロイ詳細はAntigravityと外部サービス連携のカスタムMCPサーバーガイド、監視設定はOpenTelemetry AI可観測性パイプラインも合わせて参照すると全体像がつかみやすいです。

よくある間違いと落とし穴 — 本番で遭遇するエラーパターン

実際にAntigravity Python SDKを本番で使ってきて、繰り返し遭遇するエラーパターンをまとめておきます。

1. StopCandidateException: SAFETY — 安全フィルターによる中断

# ❌ 安全フィルターを無視した場合の問題
try:
    response = model.generate_content(user_message)
    return response.text  # フィルターが発動するとここで例外
except Exception as e:
    return "エラーが発生しました"  # 原因が不明
 
# ✅ 安全フィルターを適切に処理する
from google.generativeai.types import StopCandidateException
 
try:
    response = model.generate_content(user_message)
    if response.candidates and response.candidates[0].finish_reason == "STOP":
        return response.text
    elif response.candidates and response.candidates[0].finish_reason == "SAFETY":
        return "このコンテンツはAntigravityの安全ガイドラインに基づいて処理できませんでした。"
    else:
        return "予期しない終了理由: " + str(response.candidates[0].finish_reason if response.candidates else "no candidates")
except StopCandidateException as e:
    return "コンテンツ安全フィルターによって処理が中断されました。"

2. 非同期コード内でのブロッキング呼び出し

# ❌ FastAPI の async ハンドラ内で同期SDKを直接呼ぶ（イベントループをブロック）
@app.get("/generate")
async def generate_bad(prompt: str):
    response = model.generate_content(prompt)  # ← これはブロッキング
    return {"text": response.text}
 
# ✅ asyncio.to_thread でスレッドプールに逃がす
import asyncio
 
@app.get("/generate")
async def generate_good(prompt: str):
    response = await asyncio.to_thread(model.generate_content, prompt)
    return {"text": response.text}

3. 長いドキュメントのコンテキストウィンドウ超過

Gemini 2.0 Flash は 1M トークンのコンテキストウィンドウを持ちますが、RAGで大量のドキュメントを詰め込んでも精度が上がるわけではありません。「lost in the middle」問題（コンテキストの中央部分の情報を見落とす）が発生します。

トークン数を確認してから処理する実装が堅牢です：

def count_tokens(text: str) -> int:
    """テキストのトークン数を確認する"""
    model = genai.GenerativeModel("gemini-2.0-flash")
    response = model.count_tokens(text)
    return response.total_tokens
 
# コンテキストが長すぎる場合はチャンクに分割
MAX_CONTEXT_TOKENS = 100_000  # 安全マージンを持った上限
document = "非常に長いドキュメント..."
token_count = count_tokens(document)
 
if token_count > MAX_CONTEXT_TOKENS:
    print(f"⚠️ トークン数 {token_count:,} が上限を超えます。RAGで分割処理します。")
    # → RAGパイプラインで処理
else:
    response = model.generate_content([document, "要約してください"])

個人開発の現場から見えた、ドキュメントに書かれていない勘所

ここまで紹介したコードはすべて私が実運用に通している実装パターンですが、本やドキュメントを読むだけでは見えてこない実務的な落とし穴がいくつかあります。2014年からの個人アプリ開発で、AdMob を中心に運用してきた経験を踏まえて、本番投入前に必ず確認しておきたいポイントをまとめます。

1. 「ユーザー1人あたりの実コスト」を最初に出しておく

API コストは「総額」ではなく「ユーザー1人あたり」「リクエスト1本あたり」で考えないと判断を誤ります。私の壁紙アプリは累計5,000万DL規模で、月間アクティブも数百万単位ですが、ここに無計画に Antigravity API を組み込むと、AdMob の eCPM では到底吸収できないコストになります。

実用的な目安として、私は「1ユーザーあたり月10円以下」をAI機能の上限予算にしています。Gemini 2.0 Flash の現行価格（入力 $0.075 / 1M tokens、出力 $0.30 / 1M tokens）に当てはめると、ざっくり次のような試算になります。

# 1ユーザー月10円（=約$0.067）に収めるためのトークン予算
BUDGET_USD_PER_USER_MONTH = 0.067
INPUT_COST_PER_TOKEN = 0.075 / 1_000_000
OUTPUT_COST_PER_TOKEN = 0.30 / 1_000_000
 
# 入力:出力 = 3:1 と仮定
def budget_tokens(budget_usd: float) -> dict:
    # x = 入力トークン、出力 = x/3 と仮定
    # x * INPUT + (x/3) * OUTPUT = budget
    denom = INPUT_COST_PER_TOKEN + OUTPUT_COST_PER_TOKEN / 3
    input_tokens = budget_usd / denom
    output_tokens = input_tokens / 3
    return {
        "input_tokens": int(input_tokens),
        "output_tokens": int(output_tokens),
        "approx_requests_at_2k_each": int(input_tokens / 2_000),
    }
 
print(budget_tokens(BUDGET_USD_PER_USER_MONTH))
# → {'input_tokens': 446666, 'output_tokens': 148888, 'approx_requests_at_2k_each': 223}

ユーザー1人あたり月223リクエスト（=1日約7回）が実質的な上限になります。これを超えると AdMob の広告収益では赤字になり、メンバーシップなどの別収益モデルが必須になります。本番投入前に必ずこの試算を出しておくと、後から「コストで畳むしかない」という事態を避けられます。

2. 「無言の失敗」をログから消し去る

私が最初に作ったプロトタイプで一番痛かったのは、安全フィルター・タイムアウト・トークン超過などが「例外を出さずに空文字を返す」ケースが多発したことです。ユーザーには「何も返ってこなかった」という体験になり、ストアレビューで ★1 を連打されました。これは AdMob 連携のアプリでは致命的で、ユーザー単価が大きく落ちます。

そこで本番には必ず「終了理由ログ」を仕込んでいます:

import logging
from google.generativeai.types import HarmCategory
 
logger = logging.getLogger("antigravity")
 
def generate_with_finish_log(model, prompt, user_id: str):
    """終了理由を必ずログに残す。空応答を放置しない。"""
    response = model.generate_content(prompt)
    candidates = response.candidates or []
    if not candidates:
        logger.warning(
            "empty_candidates user=%s prompt_hash=%s",
            user_id, hash(prompt) % 10_000_000,
        )
        return None, "no_candidates"
    finish = str(candidates[0].finish_reason)
    if finish != "STOP":
        # SAFETY, RECITATION, OTHER などをすべてカウント
        logger.warning(
            "non_stop_finish user=%s finish=%s prompt_hash=%s",
            user_id, finish, hash(prompt) % 10_000_000,
        )
    text = response.text if finish == "STOP" else None
    return text, finish

この finish_reason を BigQuery などに流して、SAFETY が全体の何%を占めているかを毎日見ています。私のアプリの場合、SAFETY が 0.5% を超えたらプロンプトを見直すというルールにしています。ユーザーが投げる画像によって SAFETY が連発する場合、プロンプトの言い回しではなく入力前バリデーション（解像度・暴力的内容のクライアント側フィルタ）で減らすほうが効果的です。

3. 「Cloud Run のコールドスタート」は、AdMob 広告との相性が悪い

Cloud Run はリクエストがゼロの時間はコストゼロで魅力的ですが、コールドスタートで応答が3〜5秒遅れるケースがあります。AdMob のインタースティシャル広告と同時に AI 応答を待たせると、ユーザーは広告を閉じた直後に「まだローディング中…」となり、離脱率が跳ね上がります。

そこで本番では、AI 応答を待つUIの直前に「軽い処理（例: テンプレート応答の表示）」を挟む二段構えにしています:

# サーバ側: 「即座に返せる仮レスポンス」と「本物のレスポンス」を分離
@app.post("/chat")
async def chat(req: ChatRequest):
    # まず250ms以内に「処理中ヒント」を返す
    quick_hint = generate_quick_hint(req.message)  # 軽量な分類モデル
    # 本物のレスポンスは Server-Sent Events で後追い
    return StreamingResponse(
        stream_full_response(req.message, quick_hint),
        media_type="text/event-stream",
    )

クライアントは quick_hint を見て「あなたの質問はXに関するもののようですね、回答を生成中です…」と即座にUIを更新します。これだけで体感速度は明らかに改善され、AdMob の表示タイミングとの相性も良くなります。本来 SDK の責務ではないのですが、本番アプリでは UI 設計と一緒に考えないと意味がないというのが正直なところです。

4. モデル切替は「環境変数 + フラグ」で必ず外出しにする

Gemini 系のモデルは半年に1度くらいの頻度で新世代が出ます。本番に直接モデル名を埋め込むと、新モデル評価のために何度もデプロイし直すことになります。私は次のような切替レイヤーを必ず挟んでいます:

import os
 
# モデル選択を環境変数で切り替え可能にする
MODEL_NAME = os.environ.get("ANTIGRAVITY_MODEL", "gemini-2.0-flash")
MODEL_FALLBACK = os.environ.get("ANTIGRAVITY_MODEL_FALLBACK", "gemini-1.5-flash")
 
def get_model(name: str | None = None):
    name = name or MODEL_NAME
    try:
        return genai.GenerativeModel(name)
    except Exception as e:
        logger.warning("model_init_failed name=%s err=%s; falling back", name, e)
        return genai.GenerativeModel(MODEL_FALLBACK)

新モデルが出たら、トラフィックの1〜5%だけ環境変数で切り替えて、finish_reason 分布・平均レイテンシ・1リクエストあたりコストを並列計測します。私はこれを「カナリア比較ノート」として Google Sheets で記録しており、最低でも48時間は観察してから全体ロールアウトを判断するようにしています。