Vertex AI Gemini Production Guide— Enterprise-Scale Deployment Implementation

from vertexai.generative_models import GenerativeModel
 
model = GenerativeModel("gemini-2.0-flash")
response = model.generate_content(
    "Explain the fundamentals of distributed systems architecture"
)
 
print(response.text)
print(f"Finish reason: {response.candidates[0].finish_reason}")

# Streaming response for real-time user experience
response = model.generate_content(
    "Write a comprehensive guide to cloud architecture",
    stream=True
)
 
for chunk in response:
    if chunk.text:
        print(chunk.text, end="", flush=True)

from vertexai.generative_models import GenerativeModel
 
model = GenerativeModel("gemini-2.0-flash")
chat = model.start_chat(
    history=[]  # Empty history for new conversation
)
 
# First turn
response1 = chat.send_message("Explain machine learning")
print(response1.text)
 
# Second turn—context is automatically preserved
response2 = chat.send_message("How does supervised learning differ?")
print(response2.text)
 
# The chat object maintains conversation history internally

# Reserve 1,000 tokens/minute
gcloud ai operations create-throughput \
  --provisioned-model-display-name="gemini-flash-production" \
  --model-id="gemini-2.0-flash" \
  --input-token-rate=1000 \
  --project=$PROJECT_ID \
  --region=$LOCATION
 
# List existing throughput reservations
gcloud ai operations list --region=$LOCATION

# Query the endpoint ID from the creation output
PROVISIONED_ENDPOINT = "projects/{PROJECT_ID}/locations/us-central1/endpoints/{ENDPOINT_ID}"
 
model = GenerativeModel(PROVISIONED_ENDPOINT)
response = model.generate_content("Your prompt here")

from vertexai.generative_models import (
    GenerativeModel,
    HarmCategory,
    HarmBlockThreshold,
)
 
model = GenerativeModel("gemini-2.0-flash")
 
# Define safety settings for production
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,
    },
    {
        "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(
    prompt="Your prompt here",
    safety_settings=safety_settings
)
 
# Check if response was blocked
if response.candidates:
    print("Response generated successfully")
else:
    print("Response blocked by safety filters")

import re
from enum import Enum
from typing import Optional, List
 
class PolicyViolation(Enum):
    """Corporate policy violation types"""
    COMPETITOR_MENTION = "competitor_mention"
    CONFIDENTIAL_PRODUCT = "confidential_product"
    CUSTOMER_DATA = "customer_data"
    FORBIDDEN_TECHNOLOGY = "forbidden_technology"
 
class ContentPolicyManager:
    """Manages organization-specific content policies"""
 
    def __init__(self):
        self.policies = {
            PolicyViolation.COMPETITOR_MENTION: [
                r"\b(competitor_x|competitor_y|rival_corp)\b",
            ],
            PolicyViolation.CONFIDENTIAL_PRODUCT: [
                r"\b(project_classified|secret_initiative)\b",
            ],
            PolicyViolation.CUSTOMER_DATA: [
                r"(SSN|social security|credit card|CV|passport)",
            ],
        }
 
    def check_compliance(self, text: str) -> tuple[bool, List[str]]:
        """Check if text complies with corporate policies"""
        violations = []
 
        for violation_type, patterns in self.policies.items():
            for pattern in patterns:
                if re.search(pattern, text, re.IGNORECASE):
                    violations.append(violation_type.value)
 
        return len(violations) == 0, list(set(violations))
 
    def sanitize_prompt(self, prompt: str) -> str:
        """Remove or mask policy violations in prompts"""
        sanitized = prompt
 
        # Mask SSN-like patterns
        sanitized = re.sub(
            r"\b\d{3}-\d{2}-\d{4}\b",
            "[REDACTED_SSN]",
            sanitized
        )
 
        # Mask email addresses (optional)
        sanitized = re.sub(
            r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b",
            "[REDACTED_EMAIL]",
            sanitized
        )
 
        return sanitized
 
# Usage in production
policy_manager = ContentPolicyManager()
 
user_prompt = "How do we compete with competitor_x?"
is_compliant, violations = policy_manager.check_compliance(user_prompt)
 
if not is_compliant:
    print(f"Policy violations detected: {violations}")
    # Log for audit and reject request
    return {"error": f"Content violates policies: {violations}"}, 403
 
# Safe to proceed
response = model.generate_content(user_prompt)

# main.py
import os
import json
import logging
from datetime import datetime
from typing import Optional, Tuple
from functools import wraps
 
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
 
# Initialize Cloud Logging
logging_client = cloud_logging.Client()
logging_client.setup_logging()
 
app = Flask(__name__)
 
# Initialize 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)
 
# Initialize model
MODEL_NAME = os.environ.get("MODEL_NAME", "gemini-2.0-flash")
model = GenerativeModel(MODEL_NAME)
 
# Safety settings for production
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,
    },
    {
        "category": HarmCategory.HARM_CATEGORY_HARASSMENT,
        "threshold": HarmBlockThreshold.BLOCK_ONLY_HIGH,
    },
]
 
def require_auth(f):
    """Decorator to validate authorization header"""
    @wraps(f)
    def decorated_function(*args, **kwargs):
        auth_header = request.headers.get("Authorization")
 
        if not auth_header:
            return jsonify({"error": "Missing Authorization header"}), 401
 
        # Validate bearer token (implement your auth logic)
        # For this example, we'll use a simple API key
        api_key = os.environ.get("API_KEY")
 
        if not auth_header.startswith("Bearer "):
            return jsonify({"error": "Invalid Authorization format"}), 401
 
        token = auth_header.split("Bearer ")[1]
 
        if token != api_key:
            return jsonify({"error": "Invalid API key"}), 401
 
        return f(*args, **kwargs)
 
    return decorated_function
 
@app.route("/generate", methods=["POST"])
@require_auth
def generate():
    """Generate text using Gemini"""
    try:
        data = request.get_json()
 
        if not data:
            return jsonify({"error": "Request body must be JSON"}), 400
 
        prompt = data.get("prompt")
        stream = data.get("stream", False)
 
        if not prompt or not isinstance(prompt, str):
            return jsonify({"error": "prompt (string) is required"}), 400
 
        # Log generation request
        logging.info(
            json.dumps({
                "event": "generation_request",
                "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:
                        if chunk.text:
                            yield f"data: {json.dumps({'text': chunk.text})}\n\n"
 
                except Exception as e:
                    logging.error(f"Streaming error: {str(e)}")
                    yield f"data: {json.dumps({'error': str(e)})}\n\n"
 
            return Response(generate_stream(), mimetype="text/event-stream")
 
        # Non-streaming response
        response = model.generate_content(
            prompt,
            safety_settings=SAFETY_SETTINGS,
        )
 
        # Log successful generation
        logging.info(
            json.dumps({
                "event": "generation_success",
                "response_length": len(response.text) if response.text else 0,
                "finish_reason": str(response.candidates[0].finish_reason) if response.candidates else "unknown",
            })
        )
 
        return jsonify({
            "response": response.text,
            "finish_reason": (
                response.candidates[0].finish_reason.name
                if response.candidates
                else "unknown"
            ),
        })
 
    except Exception as e:
        logging.error(f"Generation error: {str(e)}", exc_info=True)
        return jsonify({"error": "Internal server error"}), 500
 
@app.route("/health", methods=["GET"])
def health():
    """Health check endpoint for Cloud Run"""
    return jsonify({"status": "healthy", "timestamp": datetime.utcnow().isoformat()}), 200
 
@app.route("/ready", methods=["GET"])
def readiness():
    """Readiness check—verify model can be loaded"""
    try:
        # Quick test to ensure model is accessible
        test_response = model.generate_content("test")
        return jsonify({"ready": True}), 200
    except Exception as e:
        logging.error(f"Readiness check failed: {str(e)}")
        return jsonify({"ready": False, "error": str(e)}), 503
 
if __name__ == "__main__":
    port = int(os.environ.get("PORT", 8080))
    app.run(host="0.0.0.0", port=port, debug=False, threaded=True)

FROM python:3.11-slim
 
WORKDIR /app
 
# Install system dependencies
RUN apt-get update && apt-get install -y curl && rm -rf /var/lib/apt/lists/*
 
# Copy and install Python dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
 
# Copy application code
COPY main.py .
 
# Use default 8080 port expected by Cloud Run
ENV PORT=8080
 
# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
  CMD curl -f http://localhost:8080/health || exit 1
 
CMD ["python", "main.py"]

# Build and push Docker image
gcloud builds submit --tag gcr.io/$PROJECT_ID/gemini-api:latest \
  --project=$PROJECT_ID
 
# Deploy to Cloud Run
gcloud run deploy gemini-api \
  --image gcr.io/$PROJECT_ID/gemini-api:latest \
  --platform managed \
  --region $LOCATION \
  --service-account=${SERVICE_ACCOUNT_NAME}@${PROJECT_ID}.iam.gserviceaccount.com \
  --memory 4Gi \
  --cpu 4 \
  --timeout 900 \
  --max-instances 100 \
  --min-instances 0 \
  --set-env-vars "GCP_PROJECT_ID=$PROJECT_ID,GCP_LOCATION=$LOCATION,API_KEY=your-secret-key" \
  --project=$PROJECT_ID
 
# Get the service URL
SERVICE_URL=$(gcloud run services describe gemini-api --region $LOCATION --format 'value(status.url)' --project=$PROJECT_ID)
echo "Service deployed at: $SERVICE_URL"
 
# Test the deployment
curl -X POST $SERVICE_URL/generate \
  -H "Content-Type: application/json" \
  -H "Authorization: Bearer your-secret-key" \
  -d '{"prompt": "Hello, world!"}'

# Create monitoring dashboard for production metrics
cat > dashboard.json <<'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 (5xx)",
          "xyChart": {
            "dataSets": [
              {
                "timeSeriesQuery": {
                  "timeSeriesFilter": {
                    "filter": "resource.type=\"cloud_run_revision\" AND metric.type=\"run.googleapis.com/request_count\" AND response_code_class=\"5xx\" AND resource.label.service_name=\"gemini-api\""
                  }
                }
              }
            ]
          }
        }
      },
      {
        "yPos": 4,
        "width": 6,
        "height": 4,
        "widget": {
          "title": "Requests per Minute",
          "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\""
                  }
                }
              }
            ]
          }
        }
      },
      {
        "xPos": 6,
        "yPos": 4,
        "width": 6,
        "height": 4,
        "widget": {
          "title": "Container Memory Usage",
          "xyChart": {
            "dataSets": [
              {
                "timeSeriesQuery": {
                  "timeSeriesFilter": {
                    "filter": "resource.type=\"cloud_run_revision\" AND metric.type=\"run.googleapis.com/container_memory_utilization\" AND resource.label.service_name=\"gemini-api\""
                  }
                }
              }
            ]
          }
        }
      }
    ]
  }
}
EOF
 
gcloud monitoring dashboards create --config-from-file=dashboard.json

import logging
import json
import time
from datetime import datetime
from contextlib import contextmanager
from google.cloud import logging as cloud_logging
 
# Setup Cloud Logging
logging_client = cloud_logging.Client()
handler = cloud_logging.handlers.CloudLoggingHandler(logging_client)
logger = logging.getLogger(__name__)
logger.addHandler(handler)
 
class ProductionLogger:
    """Production-grade structured logging for Gemini API"""
 
    @staticmethod
    def log_generation_event(
        prompt_tokens: int,
        completion_tokens: int,
        latency_ms: float,
        model: str,
        success: bool,
        error: Optional[str] = None,
        user_id: Optional[str] = None,
    ):
        """Log generation event with all metrics"""
        log_entry = {
            "event": "gemini_generation",
            "model": model,
            "prompt_tokens": prompt_tokens,
            "completion_tokens": completion_tokens,
            "total_tokens": prompt_tokens + completion_tokens,
            "latency_ms": round(latency_ms, 2),
            "success": success,
            "timestamp": datetime.utcnow().isoformat(),
            "user_id": user_id,
        }
 
        if error:
            log_entry["error"] = error
            log_entry["severity"] = "ERROR"
        else:
            log_entry["severity"] = "INFO"
 
        logger.info(json.dumps(log_entry))
 
    @contextmanager
    @staticmethod
    def track_generation(model: str, user_id: Optional[str] = None):
        """Context manager to track generation metrics"""
        start_time = time.time()
        try:
            yield
        except Exception as e:
            latency_ms = (time.time() - start_time) * 1000
            ProductionLogger.log_generation_event(
                prompt_tokens=0,
                completion_tokens=0,
                latency_ms=latency_ms,
                model=model,
                success=False,
                error=str(e),
                user_id=user_id,
            )
            raise
        else:
            latency_ms = (time.time() - start_time) * 1000
            # Calculate tokens (this would come from the actual response)
            ProductionLogger.log_generation_event(
                prompt_tokens=100,  # Replace with actual
                completion_tokens=150,  # Replace with actual
                latency_ms=latency_ms,
                model=model,
                success=True,
                user_id=user_id,
            )
 
# Usage in Flask app
@app.route("/generate", methods=["POST"])
def generate():
    with ProductionLogger.track_generation("gemini-2.0-flash", user_id=request.headers.get("X-User-ID")):
        # Generation logic here
        pass

# Create alert for high error rate (>5%)
gcloud alpha monitoring policies create \
  --notification-channels=YOUR_CHANNEL_ID \
  --display-name="Gemini API High Error Rate Alert" \
  --condition-display-name="Error Rate exceeds 5%" \
  --condition-threshold-value=0.05 \
  --condition-threshold-duration=300s \
  --condition-threshold-filter='resource.type="cloud_run_revision" AND metric.type="run.googleapis.com/request_count"' \
  --project=$PROJECT_ID
 
# Create alert for high latency (p95 > 2s)
gcloud alpha monitoring policies create \
  --notification-channels=YOUR_CHANNEL_ID \
  --display-name="Gemini API High Latency Alert" \
  --condition-display-name="Request latency p95 exceeds 2 seconds" \
  --condition-threshold-value=2000 \
  --condition-threshold-duration=300s \
  --project=$PROJECT_ID

# Get your billing account ID
BILLING_ACCOUNT=$(gcloud billing accounts list --format="value(name)" | head -1)
 
# Create monthly budget of $2,000 with alerts
gcloud billing budgets create \
  --billing-account=$BILLING_ACCOUNT \
  --display-name="Gemini Production Budget" \
  --budget-amount=2000 \
  --threshold-rule=percent=50 \
  --threshold-rule=percent=90 \
  --threshold-rule=percent=100 \
  --notifications-rule=pubsub-topic=projects/$PROJECT_ID/topics/billing-alerts

# View current quota usage
gcloud compute project-info describe \
  --format="table(quotas[].{metric:metric,limit:limit,usage:usage})"
 
# Request quota increase (if needed)
gcloud compute project-info update \
  --update-quotas=METRIC=NEW_LIMIT

# Development project
gcloud config set project dev-gemini-api
gcloud projects add-iam-policy-binding dev-gemini-api \
  --member=user:dev-team@company.com \
  --role=roles/editor
 
# Production project
gcloud config set project prod-gemini-api
gcloud projects add-iam-policy-binding prod-gemini-api \
  --member=serviceAccount:prod-ops@company.iam.gserviceaccount.com \
  --role=roles/aiplatform.user

import {VertexAI, HarmCategory, HarmBlockThreshold} from "@google-cloud/vertexai";
import {Request, Response} from "express";
 
const vertexAI = new VertexAI({
  project: process.env.GCP_PROJECT_ID || "your-project",
  location: process.env.GCP_LOCATION || "us-central1",
});
 
async function generateContent(prompt: string): Promise<string> {
  const model = vertexAI.getGenerativeModel({
    model: "gemini-2.0-flash",
  });
 
  const safetySettings = [
    {
      category: HarmCategory.HARM_CATEGORY_HATE_SPEECH,
      threshold: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
    },
    {
      category: HarmCategory.HARM_CATEGORY_DANGEROUS_CONTENT,
      threshold: HarmBlockThreshold.BLOCK_MEDIUM_AND_ABOVE,
    },
  ];
 
  const response = await model.generateContent({
    contents: [
      {
        role: "user",
        parts: [
          {
            text: prompt,
          },
        ],
      },
    ],
    safetySettings: safetySettings,
  });
 
  const textContent = response.response.candidates?.[0].content.parts[0];
  return textContent && "text" in textContent ? textContent.text : "";
}
 
// Usage
(async () => {
  const result = await generateContent("Explain blockchain technology");
  console.log(result);
})();

import express, {Request, Response} from "express";
import {VertexAI, HarmBlockThreshold} 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,
      level: "info",
    }),
    new winston.transports.Console({
      format: winston.format.simple(),
    }),
  ],
});
 
app.post("/generate", async (req: Request, res: Response) => {
  const startTime = Date.now();
 
  try {
    const {prompt, stream} = req.body;
 
    if (!prompt || typeof prompt !== "string") {
      res.status(400).json({error: "prompt (string) is required"});
      return;
    }
 
    const model = vertexAI.getGenerativeModel({
      model: "gemini-2.0-flash",
    });
 
    if (stream) {
      res.setHeader("Content-Type", "text/event-stream");
      res.setHeader("Cache-Control", "no-cache");
      res.setHeader("Connection", "keep-alive");
 
      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];
        if (text && "text" in text) {
          res.write(`data: ${JSON.stringify({text: text.text})}\n\n`);
        }
      }
 
      res.end();
    } else {
      const response = await model.generateContent({
        contents: [{role: "user", parts: [{text: prompt}]}],
      });
 
      const latency = Date.now() - startTime;
      logger.info("generation_success", {
        latency_ms: latency,
        response_length: response.response.candidates?.[0].content.parts[0] ? 1 : 0,
      });
 
      res.json({
        response: response.response.candidates?.[0].content.parts[0] && "text" in response.response.candidates[0].content.parts[0]
          ? response.response.candidates[0].content.parts[0].text
          : "",
      });
    }
  } catch (error) {
    const latency = Date.now() - startTime;
    logger.error("generation_error", {
      error: String(error),
      latency_ms: latency,
    });
 
    res.status(500).json({error: "Internal server error"});
  }
});
 
app.get("/health", (_req: Request, res: Response) => {
  res.json({status: "healthy"});
});
 
const PORT = process.env.PORT || 8080;
app.listen(PORT, () => {
  console.log(`Server listening on port ${PORT}`);
});

gcloud projects create my-gemini-project
gcloud config set project my-gemini-project
 
# Enable Vertex AI API
gcloud services enable aiplatform.googleapis.com
 
# Authenticate
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("Explain how async/await works in Python")
print(response.text)

model = GenerativeModel("gemini-3-flash-preview")
chat = model.start_chat()
 
responses = [
    chat.send_message("I want to learn React"),
    chat.send_message("Can you explain useState?"),
    chat.send_message("Show me a real counter app example"),
]
 
for response in responses:
    print(response.text)
    print("---")

response = model.generate_content(
    "Write a comprehensive overview of distributed systems",
    stream=True
)
 
for chunk in response:
    print(chunk.text, end="", flush=True)

model = GenerativeModel(
    "gemini-3-flash-preview",
    system_instruction="""You are an experienced software engineer.
    Always provide concrete code examples.
    Include security best practices when relevant."""
)
 
response = model.generate_content("How do I prevent SQL injection?")
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
            time.sleep(wait_time)

gcloud billing budgets create \
  --billing-account=BILLING_ACCOUNT_ID \
  --display-name="Gemini Budget Alert" \
  --budget-amount=100USD \
  --threshold-rules=percent=50,percent=90,percent=100