Gradual rollout of Gemini 3 Pro behind a feature flag

A gradual rollout Gemini 3 Pro feature flag lets you shift a small percentage of production traffic to the new model while keeping a proven fallback in place. This tutorial builds a minimal Python proxy that routes requests based on a weighted, sticky flag, so you can watch latency and quality before committing 100% of calls to an unproven endpoint.

Prerequisites

• Python 3.11 or newer
• openai, fastapi, and uvicorn installed (pip install openai fastapi uvicorn)
• An API key for an OpenAI-compatible gateway. We point the client at a gateway that exposes /v1/chat/completions.
• redis if you run more than one worker process (pip install redis)
• Basic comfort with environment variables and curl

Set your configuration:

export LLM_API_KEY="sk-..."
export LLM_BASE_URL="https://api.n4n.ai/v1"  # OpenAI-compatible endpoint
export OLD_MODEL="gemini-1.5-pro"
export NEW_MODEL="gemini-3-pro"

Step 1: OpenAI-compatible client

The openai package works against any compliant endpoint. Configure it once at module load.

from openai import OpenAI
import os

client = OpenAI(
    api_key=os.environ["LLM_API_KEY"],
    base_url=os.environ["LLM_BASE_URL"],
)

def complete(model: str, prompt: str) -> str:
    resp = client.chat.completions.create(
        model=model,
        messages=[{"role": "user", "content": prompt}],
        temperature=0.2,
    )
    return resp.choices[0].message.content

This call signature is identical regardless of which provider sits behind the gateway.

Step 2: Define the flag schema

A flag needs a name, a rollout percentage, and a salt for deterministic hashing. Store it in a JSON file so you can edit without a full redeploy.

{
  "gemini_3_pro_rollout": {
    "enabled": true,
    "percentage": 10,
    "salt": "prod-2025-04",
    "old_model": "gemini-1.5-pro",
    "new_model": "gemini-3-pro"
  }
}

Load it at startup and expose a refresh hook:

import json, threading, time

FLAGS = {}

def load_flags():
    global FLAGS
    with open("flags.json") as f:
        FLAGS = json.load(f)

load_flags()

def refresh_loop():
    while True:
        time.sleep(30)
        load_flags()

threading.Thread(target=refresh_loop, daemon=True).start()

Step 3: Sticky percentage evaluation

Random assignment per request causes a single user to see both models across calls, which corrupts evaluation. Hash a stable ID (user ID or tenant ID) with the salt.

import hashlib

def assigned_to_flag(flag_key: str, bucket: str) -> bool:
    flag = FLAGS.get(flag_key)
    if not flag or not flag["enabled"]:
        return False
    h = hashlib.sha256(f"{flag['salt']}:{bucket}".encode()).hexdigest()
    value = int(h[:8], 16) % 100
    return value < flag["percentage"]

For percentage: 10, roughly 10% of distinct buckets map to True. The same bucket always receives the same decision until you change the salt or percentage.

Step 4: Wire into a FastAPI app

We expose a /chat endpoint that reads an X-User-Id header, evaluates the gradual rollout Gemini 3 Pro feature flag, and calls the appropriate model.

from fastapi import FastAPI, Request

app = FastAPI()

@app.post("/chat")
async def chat(request: Request):
    body = await request.json()
    user_id = request.headers.get("X-User-Id", "anon")
    prompt = body.get("prompt", "")

    flag_key = "gemini_3_pro_rollout"
    if assigned_to_flag(flag_key, user_id):
        model = FLAGS[flag_key]["new_model"]
    else:
        model = FLAGS[flag_key]["old_model"]

    result = complete(model, prompt)
    return {"model": model, "response": result}

Run it:

uvicorn main:app --port 8080

Step 5: Verify routing

Send two requests with different user IDs. With percentage: 10, most users hit the old model.

curl -s -X POST localhost:8080/chat \
  -H "X-User-Id: user-123" \
  -H "Content-Type: application/json" \
  -d '{"prompt":"Summarize distributed tracing"}'

Expected output (model may vary):

{"model":"gemini-1.5-pro","response":"Distributed tracing tracks requests across services using a correlation ID..."}

Now a user that hashes into the cohort:

curl -s -X POST localhost:8080/chat \
  -H "X-User-Id: beta-tester-99" \
  -H "Content-Type: application/json" \
  -d '{"prompt":"Summarize distributed tracing"}'

Expected output if assigned:

{"model":"gemini-3-pro","response":"Distributed tracing is a diagnostic method that annotates logs with a span tree..."}

Check your flag coverage by logging the model field. After 1,000 distinct users, count how many returned gemini-3-pro; it should be near 100.

Step 6: Promote gradually

Bump percentage in flags.json and wait for the 30-second refresh. Move from 10% → 25% → 50% → 100% over several days. The gradual rollout Gemini 3 Pro feature flag should be evaluated per request but stay sticky per user, so promotion does not reshuffle everyone.

Watch error rates and p95 latency per model. If the new model throws a higher rate of malformed JSON, roll back to 0% by setting enabled: false.

Step 7: Multi-worker consistency

If you run multiple uvicorn workers behind a load balancer, the in-process refresh is fine because the flag file is shared on disk. For dynamic flag stores, use Redis:

import redis

r = redis.Redis(host="localhost", port=6379, db=0)

def load_flags_redis():
    global FLAGS
    raw = r.get("feature_flags")
    if raw:
        FLAGS = json.loads(raw)

Write the JSON to Redis from your CI deploy step. The hashing remains local; no cross-call state needed.

Safe fallback and metering

When you route through n4n.ai, its OpenAI-compatible endpoint automatically falls back when a provider is rate-limited or degraded, so a flag misconfiguration or a Gemini 3 Pro outage won’t spike 500s. Per-token usage metering lets you compare cost between the old and new model directly from the usage object in the response.

If you self-host, wrap the call:

def complete_safe(model: str, prompt: str) -> tuple[str, str]:
    try:
        return complete(model, prompt), model
    except Exception:
        fallback = FLAGS["gemini_3_pro_rollout"]["old_model"]
        return complete(fallback, prompt), fallback

Swap complete for complete_safe in the endpoint to guarantee a response even when the canary model fails.

Observing output quality

Latency and errors are necessary but not sufficient. Log a sample of prompts and responses to a warehouse, tagged with the model field. Run an offline eval (exact match, BLEU, or LLM-as-judge) on the two cohorts. Only promote the gradual rollout Gemini 3 Pro feature flag when the new model meets your quality bar on real traffic, not just on a test set.

import logging
logging.basicConfig(level=logging.INFO)

@app.post("/chat")
async def chat(request: Request):
    ...
    result, used_model = complete_safe(model, prompt)
    logging.info("model=%s user=%s prompt_len=%d", used_model, user_id, len(prompt))
    return {"model": used_model, "response": result}

Cleanup

Once percentage hits 100 and stability holds for a week, flip enabled to false and hardcode NEW_MODEL as the default. Delete the flag code or keep the hashing utility for the next gradual rollout Gemini 3 Pro feature flag cycle with the next model generation.

Keep the sticky hash; it is reusable for any canary model shift and avoids the thundering herd of random per-request toggles.

Takeaways

• Sticky assignment is non-negotiable for clean evaluation.
• A JSON file and a SHA-256 hash beat a full flag SaaS for simple model rollouts.
• Automatic provider fallback turns a risky swap into a boring config change.
• Promote on data, not on calendar days.

Build the flag, ship at 5%, and let the traffic tell you when to flip to 100%.