Antigravity の並列エージェントで AdMob メディエーションを A/B 検証する実装ノート

# orchestrator.py
from antigravity import Agent, AgentRunner
from admob import fetch_weekly_report
import json
 
ORCHESTRATOR_PROMPT = """あなたは AdMob メディエーションの最適化担当です。
先週の数字を見て、改善余地のある仮説を 4 つ立ててください。
 
入力データ:
{report_summary}
 
評価軸:
- eCPM (実効 CPM): $0.50 単位で意味のある差
- ARPU (Average Revenue Per User): セグメント別に評価
- fill rate: 90% を下回るネットワークは要注意
 
仮説の出力形式 (JSON 配列):
[
  {
    "id": "exp_001",
    "hypothesis": "Banner 広告で Meta Audience Network を最上位に上げると eCPM 改善",
    "format": "banner",
    "change": {"network_order": ["meta", "applovin", "admob", "unity"]},
    "target_segment": "iOS / 日本 / 30 代以上",
    "expected_outcome": "eCPM +5% / fill rate 維持",
  },
  ...
]
"""
 
def orchestrate(week_start: str) -> list[dict]:
    report = fetch_weekly_report(week_start)
    report_summary = json.dumps(report, ensure_ascii=False, indent=2)
    
    agent = Agent(
        model="gemini-3-pro",  # 仮説立案は Pro 推奨
        system_prompt=ORCHESTRATOR_PROMPT.format(report_summary=report_summary),
    )
    
    response = agent.run("先週のレポートから 4 つの仮説を JSON で出力してください。")
    hypotheses = json.loads(response.text)
    
    assert len(hypotheses) == 4, "Orchestrator は必ず 4 仮説を返すこと"
    return hypotheses

# experimenter.py
from firebase_admin import remote_config
from antigravity import Agent
 
EXPERIMENTER_PROMPT = """あなたは AdMob メディエーション実験の実行担当です。
与えられた仮説を Firebase Remote Config の A/B テストとして起動してください。
 
仮説:
{hypothesis}
 
以下を生成:
1. Remote Config の parameter キーと値
2. Conditions (5% ロールアウト)
3. ロールバック判定の閾値 (eCPM 悪化 -20% / ARPU 悪化 -10%)
"""
 
def launch_experiment(hypothesis: dict) -> dict:
    agent = Agent(
        model="gemini-3-flash",
        system_prompt=EXPERIMENTER_PROMPT.format(
            hypothesis=json.dumps(hypothesis, ensure_ascii=False)
        ),
    )
    
    config = json.loads(agent.run("Remote Config の設定を JSON で出力").text)
    
    # Remote Config に書き込み
    template = remote_config.get_template()
    template.parameters[f"network_order_{hypothesis['format']}"] = \
        remote_config.Parameter(
            default_value=remote_config.ParameterValue(
                value=json.dumps(config["control_value"])
            ),
            conditional_values={
                "experiment_5pct": remote_config.ParameterValue(
                    value=json.dumps(config["variant_value"])
                ),
            },
        )
    
    template.conditions.append(remote_config.Condition(
        name="experiment_5pct",
        expression=f"percent({hypothesis['id']}) <= 5",
        tag_color="GREEN",
    ))
    
    remote_config.publish_template(template)
    return config

// iOS (Swift) — AdMobMediator.swift
import GoogleMobileAds
import FirebaseRemoteConfig
 
final class AdMobMediator {
    static let shared = AdMobMediator()
    private let remoteConfig = RemoteConfig.remoteConfig()
    
    func setupBanner(rootViewController: UIViewController) -> GADBannerView {
        let order = networkOrder(for: "banner")
        // メディエーションは AdMob 管理画面側の設定が本体だが、
        // アプリ側で UnitID を切り替える形で実験対応
        let unitId = adUnitId(for: "banner", order: order)
        
        let bannerView = GADBannerView(adSize: GADAdSizeBanner)
        bannerView.adUnitID = unitId
        bannerView.rootViewController = rootViewController
        bannerView.load(GADRequest())
        return bannerView
    }
    
    private func networkOrder(for format: String) -> [String] {
        let key = "network_order_\(format)"
        guard let json = remoteConfig.configValue(forKey: key).stringValue,
              let data = json.data(using: .utf8),
              let order = try? JSONDecoder().decode([String].self, from: data)
        else {
            return ["admob"]  // フォールバック
        }
        return order
    }
    
    private func adUnitId(for format: String, order: [String]) -> String {
        // 実験ごとに事前に作っておいた UnitID をマッピング
        let key = "\(format)_\(order.joined(separator: "_"))"
        return adUnitMapping[key] ?? defaultUnitId(format: format)
    }
}

# evaluator.py
EVALUATOR_PROMPT = """あなたは AdMob A/B 実験の結果評価担当です。
以下の 4 実験の 7 日間データを評価してください。
 
各実験データ:
{experiments_data}
 
評価ルール (重要):
- eCPM が +5% でも ARPU が -5% 以上ならロールバック
- fill rate が 90% を切ったら eCPM がいくら良くてもロールバック
- 統計的有意性: サンプル 16,000 人 × 7 日 = 11.2 万人で p < 0.05 を要求
- 改善が確認できた実験のみ 100% ロールアウト推奨
 
出力形式 (JSON):
{
  "results": [
    {
      "experiment_id": "exp_001",
      "outcome": "rollout" | "rollback" | "extend",
      "rationale": "eCPM +8%, ARPU +6%, fill rate 92% で全体改善",
      "next_action": "Remote Config の experiment_5pct を 100% に拡大",
    },
    ...
  ]
}
"""
 
def evaluate(experiments: list[dict]) -> list[dict]:
    data = []
    for exp in experiments:
        metrics = fetch_experiment_metrics(exp["id"], days=7)
        data.append({"experiment": exp, "metrics": metrics})
    
    agent = Agent(
        model="gemini-3-pro",
        system_prompt=EVALUATOR_PROMPT.format(
            experiments_data=json.dumps(data, ensure_ascii=False, indent=2)
        ),
    )
    
    decisions = json.loads(agent.run("各実験の判定を JSON で出力").text)
    return decisions["results"]

def rollback(experiment_id: str):
    template = remote_config.get_template()
    
    # 該当 experiment の condition を削除
    template.conditions = [
        c for c in template.conditions if c.name != f"experiment_{experiment_id}"
    ]
    
    # 該当 parameter の conditional_value を削除
    for key, param in template.parameters.items():
        if f"experiment_{experiment_id}" in param.conditional_values:
            del param.conditional_values[f"experiment_{experiment_id}"]
    
    remote_config.publish_template(template)
    
    # 強制 Refresh をトリガー
    # アプリ側は次回起動時に新しい設定を取得
    print(f"⛔ Rolled back {experiment_id}")