Gemini Flash で段階公開の continue / pause / rollback を 3 指標から自動判定する個人開発の運用メモ

from pydantic import BaseModel
from typing import Literal
 
class StagedRolloutDecision(BaseModel):
    verdict: Literal["continue", "pause", "rollback"]
    confidence: Literal["high", "medium", "low"]
    primary_signal: Literal["crash", "review", "revenue", "composite"]
    reason: str  # 80〜200 字程度
    next_action: str  # 「2026-05-28 12:00 に再評価」など
    flagged_devices: list[str]  # ["Pixel 7 Pro (Android 14)", "iPhone 13 mini (iOS 17.5)"]
    flagged_review_keywords: list[str]  # ["起動しない", "consent ループ"]

import google.generativeai as genai
from pydantic import BaseModel
 
SYSTEM_PROMPT = """\
あなたは個人開発の iOS / Android アプリの段階公開を監視する SRE です。
新バージョンのリリース後 24〜72 時間における Crashlytics・レビュー・AdMob 売上の
3 指標を見て、ロールアウトを継続するか、一時停止するか、巻き戻すかを判断します。
 
判断方針:
- クラッシュフリー率が前バージョン比 -0.5% 未満なら原則 pause、-1.0% 未満なら rollback
- ただし、低下が特定機種・特定 OS バージョンに集中している場合は flagged_devices に記録
- レビューで「起動しない / 落ちる / 課金できない / consent ループ」等のキーワードが
  新バージョン全体の 10% を超えたら原則 pause
- AdMob ARPU が前比 70% 以下かつ DAU が前比 90% 以上なら、収益構造の事故と判定して pause
- 上記の単独基準を満たさなくても、複数指標が同じ機種に集中していたら composite として pause
- 判断に迷う場合は confidence=low を返し、reason に「人間レビュー推奨」と書く
"""
 
def build_user_prompt(app_id: str, snapshot: dict) -> str:
    return f"""\
アプリ: {app_id}
新バージョン: {snapshot['new_version']}（リリース {snapshot['hours_since_release']}h 経過）
ロールアウト割合: {snapshot['rollout_percent']}%
 
[Crashlytics]
- 新バージョンのクラッシュフリー率: {snapshot['crash_free_new']}%
- 前バージョンのクラッシュフリー率: {snapshot['crash_free_prev']}%
- 新規 issue 数: {snapshot['new_issues']}
- 機種別クラッシュ分布: {snapshot['crash_by_device']}
 
[Reviews]
- 新バージョン以降のレビュー件数: {snapshot['review_count']}
- ★1〜★2 の比率: {snapshot['low_rating_ratio']}%
- 抽出されたネガティブキーワード: {snapshot['negative_keywords']}
- 主要レビュー本文（抜粋 5 件）: {snapshot['review_excerpts']}
 
[AdMob]
- 新バージョンの ARPU（24h）: {snapshot['arpu_new']}
- 前バージョンの ARPU（24h）: {snapshot['arpu_prev']}
- 新バージョンの DAU: {snapshot['dau_new']}（前比 {snapshot['dau_ratio']}%）
- 新バージョンの fill rate: {snapshot['fill_rate_new']}%
 
上記から、verdict / confidence / primary_signal / reason / next_action /
flagged_devices / flagged_review_keywords を構造化 JSON で返してください。
"""

def evaluate(app_id: str, snapshot: dict) -> StagedRolloutDecision:
    model = genai.GenerativeModel(
        "gemini-3.0-flash",
        system_instruction=SYSTEM_PROMPT,
        generation_config={
            "response_mime_type": "application/json",
            "response_schema": StagedRolloutDecision,
            "temperature": 0.2,  # 安定性を優先
        },
    )
    response = model.generate_content(build_user_prompt(app_id, snapshot))
    return StagedRolloutDecision.model_validate_json(response.text)

def get_baseline_version(app_id: str) -> str:
    doc = firestore.collection("apps").document(app_id).get()
    return doc.to_dict()["stable_baseline"]

import re
 
def normalize_version(s: str) -> str:
    m = re.match(r"^v?(\d+\.\d+\.\d+)", s)
    return m.group(1) if m else s

# Cloud Run Jobs の entrypoint
import os, json
from firebase_admin import firestore
 
def main():
    app_id = os.environ["APP_ID"]
    snapshot = collect_snapshot(app_id)  # Crashlytics + Reviews + AdMob を集める
    decision = evaluate(app_id, snapshot)
    firestore.collection("rollout_decisions").add({
        "app_id": app_id,
        "ts": firestore.SERVER_TIMESTAMP,
        "snapshot": snapshot,
        "decision": decision.model_dump(),
    })
    if decision.verdict == "rollback":
        send_urgent_notification(app_id, decision)
    elif decision.verdict == "pause":
        send_review_request(app_id, decision)