Vertex AI Gemini プロダクション — エンタープライズ規模での本番デプロイ実装

from vertexai.generative_models import GenerativeModel
 
model = GenerativeModel("gemini-2.0-flash")
response = model.generate_content("Explain quantum computing in simple terms")
 
print(response.text)

response = model.generate_content(
    "Write a comprehensive guide on cloud architecture",
    stream=True
)
 
for chunk in response:
    print(chunk.text, end="", flush=True)

from vertexai.generative_models import GenerativeModel
 
model = GenerativeModel("gemini-2.0-flash")
chat = model.start_chat()
 
response1 = chat.send_message("What is machine learning?")
print(response1.text)
 
response2 = chat.send_message("How is it different from deep learning?")
print(response2.text)
 
# チャット履歴は自動的に管理されます

# 1,000 トークン/分の予約を作成
gcloud ai operations create-throughput \
  --provisioned-model-display-name="gemini-flash-prod" \
  --model-id="gemini-2.0-flash" \
  --input-token-rate=1000 \
  --project=$PROJECT_ID \
  --region=us-central1

from vertexai.generative_models import GenerativeModel
 
# プロビジョニングされたモデルを使用
model = GenerativeModel(
    "projects/{PROJECT_ID}/locations/us-central1/endpoints/{ENDPOINT_ID}"
)
 
response = model.generate_content("Your prompt here")

from vertexai.generative_models import (
    GenerativeModel,
    HarmCategory,
    HarmBlockThreshold,
)
 
model = GenerativeModel("gemini-2.0-flash")
 
# セーフティ設定を定義
safety_settings = [
    {
        "category": HarmCategory.HARM_CATEGORY_UNSPECIFIED,
        "threshold": HarmBlockThreshold.BLOCK_NONE,
    },
    {
        "category": HarmCategory.HARM_CATEGORY_HATE_SPEECH,
        "threshold": HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
    },
    {
        "category": HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
        "threshold": HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
    },
    {
        "category": HarmCategory.HARM_CATEGORY_HARASSMENT,
        "threshold": HarmBlockThreshold.BLOCK_ONLY_HIGH,
    },
    {
        "category": HarmCategory.HARM_CATEGORY_SEXUALLY_EXPLICIT,
        "threshold": HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
    },
]
 
response = model.generate_content(
    "Your prompt here",
    safety_settings=safety_settings
)

import re
from enum import Enum
 
class ContentPolicyViolation(Enum):
    """組織のポリシー違反タイプ"""
    COMPETITOR_MENTION = "competitor_mention"
    CONFIDENTIAL_DATA = "confidential_data"
    DISALLOWED_PRODUCT = "disallowed_product"
 
CONFIDENTIAL_PATTERNS = {
    ContentPolicyViolation.COMPETITOR_MENTION: [
        r"\b(competitor_a|competitor_b)\b",
    ],
    ContentPolicyViolation.DISALLOWED_PRODUCT: [
        r"\b(legacy_product_x|sunset_service)\b",
    ],
}
 
def check_policy_compliance(text: str) -> tuple[bool, list[str]]:
    """入力テキストのポリシー違反をチェック"""
    violations = []
 
    for violation_type, patterns in CONFIDENTIAL_PATTERNS.items():
        for pattern in patterns:
            if re.search(pattern, text, re.IGNORECASE):
                violations.append(violation_type.value)
 
    return len(violations) == 0, violations
 
# 使用例
user_prompt = "How do we compete with competitor_a on feature X?"
is_compliant, violations = check_policy_compliance(user_prompt)
 
if not is_compliant:
    print(f"ポリシー違反: {violations}")
    # リクエストを拒否またはログを記録
else:
    response = model.generate_content(user_prompt)

# main.py
import os
import logging
from datetime import datetime
from typing import Optional
 
from flask import Flask, request, jsonify, Response
import vertexai
from vertexai.generative_models import GenerativeModel, HarmCategory, HarmBlockThreshold
from google.cloud import logging as cloud_logging
 
# Cloud Logging を設定
logging_client = cloud_logging.Client()
logging_client.setup_logging()
 
app = Flask(__name__)
 
# Vertex AI を初期化
PROJECT_ID = os.environ.get("GCP_PROJECT_ID")
LOCATION = os.environ.get("GCP_LOCATION", "us-central1")
vertexai.init(project=PROJECT_ID, location=LOCATION)
 
# モデルを初期化
model = GenerativeModel("gemini-2.0-flash")
 
# セーフティ設定
SAFETY_SETTINGS = [
    {
        "category": HarmCategory.HARM_CATEGORY_HATE_SPEECH,
        "threshold": HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
    },
    {
        "category": HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
        "threshold": HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
    },
]
 
@app.route("/generate", methods=["POST"])
def generate():
    """テキスト生成エンドポイント"""
    try:
        data = request.get_json()
        prompt = data.get("prompt")
        stream = data.get("stream", False)
 
        if not prompt:
            return jsonify({"error": "prompt is required"}), 400
 
        # リクエストをログ記録
        logging.info(
            "generate_request",
            extra={
                "prompt_length": len(prompt),
                "stream": stream,
                "timestamp": datetime.utcnow().isoformat(),
            },
        )
 
        # ストリーミング応答
        if stream:
            def generate_stream():
                try:
                    response = model.generate_content(
                        prompt,
                        stream=True,
                        safety_settings=SAFETY_SETTINGS,
                    )
 
                    for chunk in response:
                        yield f"data: {chunk.text}\n\n"
 
                except Exception as e:
                    logging.error(f"Streaming error: {str(e)}")
                    yield f"data: [ERROR] {str(e)}\n\n"
 
            return Response(generate_stream(), mimetype="text/event-stream")
 
        # 非ストリーミング応答
        response = model.generate_content(
            prompt,
            safety_settings=SAFETY_SETTINGS,
        )
 
        return jsonify({
            "response": response.text,
            "finish_reason": response.candidates[0].finish_reason.name,
        })
 
    except Exception as e:
        logging.error(f"Generation error: {str(e)}")
        return jsonify({"error": str(e)}), 500
 
@app.route("/health", methods=["GET"])
def health():
    """ヘルスチェックエンドポイント"""
    return jsonify({"status": "healthy"}), 200
 
if __name__ == "__main__":
    port = int(os.environ.get("PORT", 8080))
    app.run(host="0.0.0.0", port=port, debug=False)

FROM python:3.11-slim
 
WORKDIR /app
 
# 依存関係をインストール
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
# アプリケーションコードをコピー
COPY main.py .
 
# ヘルスチェック
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
  CMD curl -f http://localhost:8080/health || exit 1
 
CMD ["python", "main.py"]

# イメージをビルドしてArtifact Registryにプッシュ
gcloud builds submit --tag gcr.io/$PROJECT_ID/gemini-api:latest
 
# Cloud Run にデプロイ
gcloud run deploy gemini-api \
  --image gcr.io/$PROJECT_ID/gemini-api:latest \
  --platform managed \
  --region us-central1 \
  --service-account=${SERVICE_ACCOUNT_NAME}@${PROJECT_ID}.iam.gserviceaccount.com \
  --memory 4Gi \
  --cpu 4 \
  --timeout 900 \
  --max-instances 100 \
  --set-env-vars "GCP_PROJECT_ID=$PROJECT_ID,GCP_LOCATION=us-central1"
 
# サービスのURLを取得
gcloud run services describe gemini-api --region us-central1 --format 'value(status.url)'

# Cloud Run メトリクスのダッシュボードを作成
gcloud monitoring dashboards create --config-from-file=- <<EOF
{
  "displayName": "Gemini API Production Dashboard",
  "mosaicLayout": {
    "columns": 12,
    "tiles": [
      {
        "width": 6,
        "height": 4,
        "widget": {
          "title": "Request Latency (p95)",
          "xyChart": {
            "dataSets": [
              {
                "timeSeriesQuery": {
                  "timeSeriesFilter": {
                    "filter": "resource.type=\"cloud_run_revision\" AND metric.type=\"run.googleapis.com/request_latencies\" AND resource.label.service_name=\"gemini-api\""
                  }
                }
              }
            ]
          }
        }
      },
      {
        "xPos": 6,
        "width": 6,
        "height": 4,
        "widget": {
          "title": "Error Rate",
          "xyChart": {
            "dataSets": [
              {
                "timeSeriesQuery": {
                  "timeSeriesFilter": {
                    "filter": "resource.type=\"cloud_run_revision\" AND metric.type=\"run.googleapis.com/request_count\" AND resource.label.service_name=\"gemini-api\" AND metric.response_code_class=\"5xx\""
                  }
                }
              }
            ]
          }
        }
      }
    ]
  }
}
EOF

import logging
import json
from google.cloud import logging as cloud_logging
 
# Cloud Logging クライアントをセットアップ
logging_client = cloud_logging.Client()
handler = cloud_logging.handlers.CloudLoggingHandler(logging_client)
logging.getLogger().addHandler(handler)
 
# 構造化ログの書き込み
def log_generation_event(
    prompt_length: int,
    response_length: int,
    latency_ms: float,
    success: bool,
    error: Optional[str] = None,
):
    """生成イベントを構造化ログで記録"""
    log_entry = {
        "event": "gemini_generation",
        "prompt_tokens": estimate_tokens(prompt_length),
        "response_tokens": estimate_tokens(response_length),
        "latency_ms": latency_ms,
        "success": success,
        "timestamp": datetime.utcnow().isoformat(),
    }
 
    if error:
        log_entry["error"] = error
 
    logging.info(json.dumps(log_entry))
 
def estimate_tokens(text_length: int) -> int:
    """テキスト長からトークン数を推定（約4文字=1トークン）"""
    return max(1, text_length // 4)

# エラー率が5%を超えた場合にアラート
gcloud alpha monitoring policies create \
  --notification-channels=CHANNEL_ID \
  --display-name="Gemini API High Error Rate" \
  --condition-display-name="Error Rate > 5%" \
  --condition-threshold-value=0.05 \
  --condition-threshold-duration=300s

# 月額予算を$1,000に設定
gcloud billing budgets create \
  --billing-account=BILLING_ACCOUNT_ID \
  --display-name="Gemini API Budget" \
  --budget-amount=1000 \
  --threshold-rule=percent=50 \
  --threshold-rule=percent=90 \
  --threshold-rule=percent=100 \
  --notifications-rule=pubsub-topic=projects/$PROJECT_ID/topics/billing-alerts

# Vertex AI API クォータを確認
gcloud compute project-info describe \
  --format='value(quotas[name="aiplatform.googleapis.com/requests"].usage)'

# 開発環境
gcloud config set project dev-project
# ...開発作業...
 
# 本番環境
gcloud config set project prod-project
# ...本番運用...

import {VertexAI} from "@google-cloud/vertexai";
 
const vertexAI = new VertexAI({
  project: process.env.GCP_PROJECT_ID,
  location: process.env.GCP_LOCATION || "us-central1",
});
 
const model = vertexAI.getGenerativeModel({
  model: "gemini-2.0-flash",
});
 
async function generateContent(prompt: string): Promise<string> {
  const response = await model.generateContent({
    contents: [
      {
        role: "user",
        parts: [
          {
            text: prompt,
          },
        ],
      },
    ],
  });
 
  return response.response.candidates?.[0].content.parts[0].text || "";
}
 
// 使用例
(async () => {
  const result = await generateContent(
    "Explain quantum computing in 100 words"
  );
  console.log(result);
})();

import express, { Request, Response } from "express";
import { VertexAI } from "@google-cloud/vertexai";
import { CloudLoggingTransport } from "@google-cloud/logging-winston";
import winston from "winston";
 
const app = express();
app.use(express.json());
 
const vertexAI = new VertexAI({
  project: process.env.GCP_PROJECT_ID,
  location: "us-central1",
});
 
const logger = winston.createLogger({
  transports: [
    new CloudLoggingTransport({
      projectId: process.env.GCP_PROJECT_ID,
    }),
    new winston.transports.Console(),
  ],
});
 
app.post("/generate", async (req: Request, res: Response) => {
  try {
    const { prompt, stream } = req.body;
 
    if (!prompt) {
      return res.status(400).json({ error: "prompt is required" });
    }
 
    const model = vertexAI.getGenerativeModel({
      model: "gemini-2.0-flash",
    });
 
    if (stream) {
      res.setHeader("Content-Type", "text/event-stream");
      res.setHeader("Cache-Control", "no-cache");
 
      const streamResponse = await model.generateContentStream({
        contents: [{ role: "user", parts: [{ text: prompt }] }],
      });
 
      for await (const chunk of streamResponse.stream) {
        const text =
          chunk.candidates?.[0].content.parts[0].text || "";
        res.write(`data: ${text}\n\n`);
      }
 
      res.end();
    } else {
      const response = await model.generateContent({
        contents: [{ role: "user", parts: [{ text: prompt }] }],
      });
 
      res.json({
        response: response.response.candidates?.[0].content.parts[0].text,
      });
    }
  } catch (error) {
    logger.error("Generation error", error);
    res.status(500).json({ error: String(error) });
  }
});
 
const port = process.env.PORT || 8080;
app.listen(port, () => {
  console.log(`Server listening on port ${port}`);
});

# gcloud CLI をインストール後
gcloud projects create my-gemini-project
gcloud config set project my-gemini-project
 
# Vertex AI API を有効化
gcloud services enable aiplatform.googleapis.com
 
# 認証設定
gcloud auth application-default login

pip install google-cloud-aiplatform

import vertexai
from vertexai.generative_models import GenerativeModel
 
# 初期化（プロジェクト・リージョンを指定）
vertexai.init(
    project="my-gemini-project",
    location="us-central1"
)
 
model = GenerativeModel("gemini-3-flash-preview")
response = model.generate_content("Pythonでフィボナッチ数列を実装してください")
print(response.text)

model = GenerativeModel("gemini-3-flash-preview")
chat = model.start_chat()
 
responses = [
    chat.send_message("こんにちは！React を始めたいのですが"),
    chat.send_message("useState の使い方を教えてください"),
    chat.send_message("実際のカウンターアプリのコード例を見せてください"),
]
 
for response in responses:
    print(response.text)
    print("---")

response = model.generate_content(
    "日本の歴史について長文で教えてください",
    stream=True
)
 
for chunk in response:
    print(chunk.text, end="", flush=True)

model = GenerativeModel(
    "gemini-3-flash-preview",
    system_instruction="""あなたは経験豊富なソフトウェアエンジニアです。
    コードの説明は常に日本語で行い、具体的な例を示してください。
    セキュリティのベストプラクティスも合わせて説明してください。"""
)
 
response = model.generate_content("SQLインジェクションの防ぎ方を教えてください")
print(response.text)

from google.api_core import exceptions
import time
 
def generate_with_retry(model, prompt, max_retries=3):
    for attempt in range(max_retries):
        try:
            return model.generate_content(prompt)
        except exceptions.ResourceExhausted:
            if attempt == max_retries - 1:
                raise
            wait_time = 2 ** attempt  # Exponential backoff
            print(f"Rate limit hit. Waiting {wait_time}s...")
            time.sleep(wait_time)

# 予算アラートの設定（Google Cloud Console でも可能）
gcloud billing budgets create \
  --billing-account=BILLING_ACCOUNT_ID \
  --display-name="Gemini Budget Alert" \
  --budget-amount=10000JPY \
  --threshold-rules=percent=50,percent=90,percent=100