利用者データを『どのリージョンで処理したか』を後から言えるようにする — Gemini(Vertex AI) のエンドポイント固定とレジデンシ・ポリシーゲート

# Before: 既定クライアント。どこで処理されるか指定していない
from google import genai
 
client = genai.Client()  # Developer API 既定 or global 相当
 
def summarize(text: str) -> str:
    resp = client.models.generate_content(
        model="gemini-flash-latest",
        contents=text,
    )
    return resp.text

# After: 区分→リージョンを解決し、location を固定したクライアントを供給する
from google import genai
from functools import lru_cache
 
PROJECT = "my-gcp-project"
 
# データ区分ごとの「第一希望リージョン」と「許可リージョン」
REGION_POLICY = {
    "residency":     {"primary": "europe-west4", "allowed": {"europe-west4", "europe-west1"}, "global_ok": False},
    "regional-pref": {"primary": "europe-west4", "allowed": {"europe-west4", "europe-west1"}, "global_ok": False},
    "global-ok":     {"primary": "global",       "allowed": {"global"},                       "global_ok": True},
}
 
@lru_cache(maxsize=8)
def _client_for(location: str) -> genai.Client:
    # location ごとにクライアントを1つだけ生成しキャッシュする
    return genai.Client(vertexai=True, project=PROJECT, location=location)
 
def client_for_class(data_class: str, location: str | None = None) -> genai.Client:
    policy = REGION_POLICY[data_class]
    loc = location or policy["primary"]
    if loc not in policy["allowed"]:
        raise ResidencyError(f"{data_class} はリージョン {loc} を許可していません")
    return _client_for(loc)

class ResidencyError(RuntimeError):
    """residency 制約を破りうる呼び出しを止める"""
 
def generate(data_class: str, *, model: str, contents, location: str | None = None):
    policy = REGION_POLICY[data_class]
    client = client_for_class(data_class, location)
    used_location = location or policy["primary"]
    try:
        resp = client.models.generate_content(model=model, contents=contents)
        _ledger_write(data_class, used_location, model, status="ok")
        return resp
    except Exception as e:
        # ここで global へ逃がさない。許可リージョン内でのみ再試行する
        for alt in policy["allowed"]:
            if alt == used_location:
                continue
            try:
                resp = _client_for(alt).models.generate_content(model=model, contents=contents)
                _ledger_write(data_class, alt, model, status="ok-failover")
                return resp
            except Exception:
                continue
        _ledger_write(data_class, used_location, model, status="failed")
        if not policy["global_ok"]:
            # residency/regional-pref は global に落とさず、失敗として返す
            raise ResidencyError(f"{data_class}: 許可リージョン内で処理できませんでした") from e
        # global-ok のみ、ここで global への最終フォールバックが許される
        resp = _client_for("global").models.generate_content(model=model, contents=contents)
        _ledger_write(data_class, "global", model, status="ok-global-fallback")
        return resp

import sqlite3, time, uuid
from pathlib import Path
 
_DB = Path("data/residency_ledger.db")
 
def _ledger_init():
    con = sqlite3.connect(_DB)
    con.execute("""
      CREATE TABLE IF NOT EXISTS calls(
        id TEXT PRIMARY KEY, ts REAL, data_class TEXT,
        location TEXT, model TEXT, status TEXT
      )""")
    con.commit(); con.close()
 
def _ledger_write(data_class: str, location: str, model: str, status: str):
    con = sqlite3.connect(_DB)
    con.execute(
        "INSERT INTO calls VALUES(?,?,?,?,?,?)",
        (uuid.uuid4().hex, time.time(), data_class, location, model, status),
    )
    con.commit(); con.close()

SELECT location, status, COUNT(*) AS n
FROM calls
WHERE data_class = 'residency'
  AND ts >= strftime('%s', '2026-06-01')
GROUP BY location, status
ORDER BY n DESC;

RESIDENCY_MODEL_FALLBACK = {
    # 第一希望が許可リージョンに無い場合の、安定モデルへの縮退先
    "gemini-flash-latest": "gemini-2.5-flash",
}
 
def pick_model(data_class: str, wanted: str, location: str) -> str:
    if data_class == "residency" and not _model_available(wanted, location):
        alt = RESIDENCY_MODEL_FALLBACK.get(wanted)
        if alt and _model_available(alt, location):
            return alt
        raise ResidencyError(f"{wanted} も代替も {location} で利用できません")
    return wanted