EP.12 は リリース後のプレイログを に集約し、 で「言葉」 で解釈するパイプライン。「 が下がった」 で止まるのではなく、「序盤チュートリアルの 3 番目でつまづく初心者層が増えている」 まで踏み込む話。 / GPT で自動分類・クラスタ命名・SQL 生成まで。
1. プレイログの設計
イベントの JSON スキーマ例
JSON
{ "event_name": "stage_died", "event_timestamp": "2026-05-26T14:23:11.534Z", "user_id": "u_abc123", "session_id": "s_xyz789", "client_version": "1.2.3", "platform": "iOS", "params": { "stage_id": "stage_3_boss", "death_cause": "boss_attack", "death_location_x": 245, "death_location_y": 120, "player_hp": 0, "elapsed_seconds": 187, "items_used": ["potion_red", "potion_blue"] }}2. Firebase Analytics → BigQuery エクスポート
- Firebase Console → プロジェクト設定 → 統合 → BigQuery: 自動エクスポート設定 (無料枠あり)
- Daily export: 翌日にイベントログが BigQuery 内に events_YYYYMMDD テーブルとして作成される
- Streaming export: リアルタイム (有料、Firebase Spark/Blaze プラン要)
- スキーマ: event_name / event_timestamp / user_pseudo_id / event_params (RECORD) / user_properties / device / geo / app_info
3. 基本クエリ: 日次 DAU / 継続率
DAU と D1 継続率
SQL
-- DAU (日次アクティブユーザー数)SELECT event_date, COUNT(DISTINCT user_pseudo_id) AS dauFROM `project.analytics_123456.events_*`WHERE _TABLE_SUFFIX BETWEEN '20260501' AND '20260526' AND event_name = 'session_start'GROUP BY event_dateORDER BY event_date;
-- D1 継続率 (新規ユーザーが翌日も戻ってきた率)WITH new_users AS ( SELECT user_pseudo_id, MIN(event_date) AS first_date FROM `project.analytics_123456.events_*` WHERE event_name = 'first_open' GROUP BY user_pseudo_id),day1_active AS ( SELECT DISTINCT n.user_pseudo_id, n.first_date FROM new_users n JOIN `project.analytics_123456.events_*` e ON n.user_pseudo_id = e.user_pseudo_id AND e.event_date = FORMAT_DATE('%Y%m%d', DATE_ADD(PARSE_DATE('%Y%m%d', n.first_date), INTERVAL 1 DAY)))SELECT n.first_date, COUNT(DISTINCT n.user_pseudo_id) AS new_count, COUNT(DISTINCT d1.user_pseudo_id) AS returned_count, SAFE_DIVIDE(COUNT(DISTINCT d1.user_pseudo_id), COUNT(DISTINCT n.user_pseudo_id)) AS d1_retentionFROM new_users nLEFT JOIN day1_active d1 USING (user_pseudo_id)GROUP BY n.first_dateORDER BY n.first_date;分析でステージ別の離脱率を見るのが定番。各ステージの開始 / クリアイベントを並べ、どこで脱落が集中するかを で集計します。
4. 離脱ポイント分析 (Funnel)
ステージ別の完了率
SQL
WITH stage_events AS ( SELECT user_pseudo_id, event_name, (SELECT value.string_value FROM UNNEST(event_params) WHERE key = 'stage_id') AS stage_id, event_timestamp FROM `project.analytics_123456.events_*` WHERE event_name IN ('stage_start', 'stage_clear') AND _TABLE_SUFFIX BETWEEN '20260501' AND '20260526')SELECT stage_id, COUNT(DISTINCT IF(event_name = 'stage_start', user_pseudo_id, NULL)) AS starters, COUNT(DISTINCT IF(event_name = 'stage_clear', user_pseudo_id, NULL)) AS clearers, SAFE_DIVIDE( COUNT(DISTINCT IF(event_name = 'stage_clear', user_pseudo_id, NULL)), COUNT(DISTINCT IF(event_name = 'stage_start', user_pseudo_id, NULL)) ) AS clear_rateFROM stage_eventsWHERE stage_id IS NOT NULLGROUP BY stage_idORDER BY clear_rate ASC;-- ↑ クリア率の低い順、難所候補が浮かぶ5. LLM で離脱ユーザーの行動を「物語化」
ユーザー 1 人の行動ログを LLM に要約させる
Python
from google.cloud import bigqueryfrom anthropic import Anthropicimport os
bq = bigquery.Client()ai = Anthropic(api_key=os.environ["ANTHROPIC_API_KEY"])
def fetch_user_events(user_id: str) -> list[dict]: query = f""" SELECT event_timestamp, event_name, ARRAY_AGG(STRUCT(key, value.string_value, value.int_value)) AS params FROM `project.analytics_123456.events_*` e, UNNEST(e.event_params) p WHERE user_pseudo_id = '{user_id}' AND _TABLE_SUFFIX BETWEEN '20260501' AND '20260526' GROUP BY event_timestamp, event_name ORDER BY event_timestamp LIMIT 200 """ return [dict(row) for row in bq.query(query).result()]
def summarize_journey(user_id: str) -> str: events = fetch_user_events(user_id) events_text = "\n".join( f"{e['event_timestamp']} {e['event_name']} {e['params']}" for e in events ) prompt = f"""以下はゲームプレイヤーの行動ログ ({len(events)} 件) です。プレイヤーの旅路を 5-7 文で要約してください。特に: 詰まったポイント、離脱の決定打、楽しんでいた要素 を含めて。
{events_text[:8000]}""" response = ai.messages.create( model="claude-opus-4-7", max_tokens=400, messages=[{"role": "user", "content": prompt}], ) return response.content[0].text
# 直近離脱したユーザー 10 人を分析churned_users = ["u_abc", "u_def", "u_ghi"] # 別クエリで抽出for uid in churned_users: print(f"\n=== {uid} ===") print(summarize_journey(uid))6. クラスタリング + LLM 命名
K-Means でユーザーを分類、LLM でクラスタ名生成
Python
import pandas as pdfrom sklearn.cluster import KMeansfrom sklearn.preprocessing import StandardScaler
# BigQuery から特徴量を取得 (例: session_count, total_minutes, stage_max, item_count, paid)df = bq.query("""SELECT user_pseudo_id, COUNT(DISTINCT (SELECT value.string_value FROM UNNEST(event_params) WHERE key = 'session_id')) AS session_count, SUM(CASE WHEN event_name = 'session_end' THEN (SELECT value.int_value FROM UNNEST(event_params) WHERE key = 'duration_minutes') ELSE 0 END) AS total_minutes, MAX(CASE WHEN event_name = 'stage_clear' THEN (SELECT value.int_value FROM UNNEST(event_params) WHERE key = 'stage_number') ELSE 0 END) AS stage_max, SUM(CASE WHEN event_name = 'item_get' THEN 1 ELSE 0 END) AS item_count, MAX(CASE WHEN event_name = 'purchase' THEN 1 ELSE 0 END) AS paidFROM `project.analytics_123456.events_*`WHERE _TABLE_SUFFIX BETWEEN '20260501' AND '20260526'GROUP BY user_pseudo_id""").to_dataframe()
features = ["session_count", "total_minutes", "stage_max", "item_count", "paid"]X = StandardScaler().fit_transform(df[features])km = KMeans(n_clusters=5, random_state=42, n_init=10).fit(X)df["cluster"] = km.labels_
# 各クラスタの統計を取って LLM に名付けさせるfor c in range(5): sub = df[df["cluster"] == c] stats = sub[features].mean().to_dict() prompt = f"""ゲームユーザーのクラスタ統計:{stats}
このクラスタを 「ライトユーザー」「コアプレイヤー」 のように短く (5-15 文字) 命名し、特徴を 2 文で説明してください。""" response = ai.messages.create( model="claude-haiku-4-5-20251001", max_tokens=150, messages=[{"role": "user", "content": prompt}] ) print(f"\nCluster {c} ({len(sub)} users): {response.content[0].text}")7. LLM SQL 生成 (Text-to-SQL)
- スキーマを LLM に渡す: 「テーブル events のスキーマは...」 を system prompt に
- 自然言語クエリ → BigQuery SQL: 「先週新規ユーザの 3 日後継続率を出して」
- 生成 SQL のレビュー必須: コスト爆発しないか dry run で確認、`bq query --dry_run`
- ダッシュボード化: Looker Studio / Hex / Streamlit に組込み
8. リアルタイム監視 (Slack 通知)
DAU が前週比 20% 下落で Slack 通知
Python
import requestsimport os
def check_dau_drop(): query = """ SELECT DATE_DIFF(CURRENT_DATE(), PARSE_DATE('%Y%m%d', event_date), DAY) AS days_ago, COUNT(DISTINCT user_pseudo_id) AS dau FROM `project.analytics_123456.events_*` WHERE _TABLE_SUFFIX BETWEEN FORMAT_DATE('%Y%m%d', DATE_SUB(CURRENT_DATE(), INTERVAL 14 DAY)) AND FORMAT_DATE('%Y%m%d', CURRENT_DATE()) AND event_name = 'session_start' GROUP BY event_date ORDER BY days_ago ASC """ rows = list(bq.query(query).result()) if len(rows) < 8: return
yesterday = rows[1].dau week_ago = rows[8].dau if yesterday < week_ago * 0.8: msg = f"⚠️ DAU 下落検知: 昨日 {yesterday}, 1 週間前 {week_ago} ({yesterday/week_ago*100:.1f}%)" requests.post(os.environ["SLACK_WEBHOOK"], json={"text": msg})
# Cloud Functions / Cloud Run / GitHub Actions で日次実行9. ユーザーフィードバック (フォーラム / レビュー) の自動分類
Steam レビュー / App Store レビューを LLM で分類
Python
def classify_review(text: str) -> dict: prompt = f"""以下のゲームレビューを分析:"{text}"
JSON で返答:{{ "sentiment": "positive" | "negative" | "neutral", "category": ["bug" | "request" | "praise" | "balance" | "graphics" | "story" | "performance"], "summary": "1 文で要約", "priority": 1-5 (1=即対応、5=参考程度)}}""" response = ai.messages.create( model="claude-haiku-4-5-20251001", max_tokens=300, messages=[{"role": "user", "content": prompt}], ) import json text = response.content[0].text return json.loads(text[text.index("{"):text.rindex("}")+1])
# 1000 件のレビューを並列分類 → BigQuery に保存 → ダッシュボードで可視化10. シリーズの到達点と今後
本シリーズはここまでで 12 回。EP.01 のゲームエンジン地図から、EP.12 のプレイログ解析まで、LLM とゲーム開発の接点を全工程に渡って具体例で扱ってきました。今後は新エンジン / 新アセット生成ツール / 実行時 LLM の発展に合わせ、随時新 EP を追加します。Multiplayer / Live Service / VR-AR との組合せなど未踏領域も。
この記事の感想を教えてください
あなたの 1 クリックで、本当にこの記事は更新されます。「もっと詳しく」「続編希望」が一定数集まった記事は、 ふくふくが 実際に内容を拡充したり続編記事を公開 します。 送信したリアクションはお使いのブラウザに記録され、再カウントされません。