ギフトを投げれば投げるほど神様が優しくなる。課金階層をアルゴリズムに埋め込み、TikTok LIVE上で全自動運用する「待ち行列エンタメ」を Gemini API だけで構築した記録。
作ったもの
AI-Fortune-Live は、TikTok LIVE の視聴者参加型・自動AI占いシステムです。
配信中に飛んでくるギフトとコメントをリアルタイムに解釈し、「イケメン神様」というAIキャラクターが占いを返します。ポイントは、ギフト額に応じて神様の態度が露骨に変わること。
| Class | 閾値（コイン） | モデル | キャラ口調 |
|---|---|---|---|
| C | 5〜 | gemini-2.0-flash-001 | ツンデレ神。塩対応で高速捌き |
| B | 99〜 | gemini-2.5-flash | 優しいお兄さんタイプ |
| A | 1,000〜 | gemini-2.5-pro | 独占的な彼氏モード・「神降ろし」演出 |
「順番抜かし」と「対応格差」を可視化することで承認欲求を刺激する、エンタメとしてのアルゴリズム設計が裏テーマでした。
アーキテクチャ
[TikTok LIVE] ─► TikTokLiveClient
│
├─(Gift)──► QueueManager ──► SessionRunner
│ │ │
└─(Comment)────┘ ├─► ColdReading（神の眼）
├─► Gemini Text（階級別モデル）
├─► Gemini TTS（音声生成）
├─► Moderation（NGワード）
├─► ProfitGuard（収支ガード）
└─► SQLite（アカシックレコード）
│
▼
OBS WebSocket ──► OBS Browser Source（1080x1920）コア言語は Python + asyncio、AI は Google Gemini API に完全統一（Text/TTS どちらも）、フロントエンドは OBS のブラウザソースとして表示する HTML/JS です。サービスごとにクラスを切り、main.py で asyncio.gather して並列実行するシンプルな構成。
設計のキモ：1000000000 対 1000 という桁違いの base_points
視聴者の優先順位を決めるキューは heapq で管理しているのですが、ここで面白い設計をしました。
queue:
base_points:
A: 1000000000 # 10億
B: 1000000 # 100万
C: 1000 # 千優先度スコアは次の式で計算されます。
priority_score = base_points[tier] + (total_gift_coins * 100) + streak_bonus - wait_penaltybase_points に桁差を入れることで、Class A は Class B/C に「数学的に」絶対に負けない 設計。待たせ時間による救済（wait_penalty）を入れても、この桁差は埋まりません。
なぜこうしたか？
エンタメ的に「格差」を明確に見せたかったからです。大量に Class C ギフトを投げても Class A ひとりに勝てない、という演出。実質的に課金圧力になります。heapq は最小値を取り出すため、タプルは (-priority_score, enqueued_at, user_id) で管理。
ギフトの「蓄積バッファ」——閾値未満のギフトを捨てない
シンプルな実装なら「閾値未満のギフトは無視」でいいのですが、それだと薔薇1個（1コイン）を何十回投げるユーザーが報われません。
そこで accumulation_buffer を実装しました。
- 閾値未満のギフトは バッファに蓄積
- 合算して閾値を超えたら 昇格 してキューに投入
- 同一ユーザーの追撃ギフトは、セッション開始前なら同一枠に合算
- セッション開始後のギフトは アクティブセッションを吸収 してティアを動的にアップグレード
この「追撃でティア昇格」の演出が意外とウケます。Class C で呼ばれた人が喋っている最中に仲間が Class A 供物を叩き込む、みたいな乱入劇が発生する。
「神の眼」——AIじゃなく正規表現で観察させる
占いのリアリティを上げるために、AIに投げる前にルールベースで特徴抽出します。これをプロジェクト内では「神の眼（Cold Reading）」と呼んでいます。
# core/cold_reading.py
if length <= 15:
tags.append("TAG:WORDS_MISSING") # 言葉足らず・恐れ
if ellipsis_count >= 2:
tags.append("TAG:SWALLOWING_TRUTH") # 言い淀み・隠し事
if kanji_ratio >= 0.35:
tags.append("TAG:SUPPRESSING_EMOTION_WITH_LOGIC") # 抑圧
elif kanji_ratio < 0.1 and length > 10:
tags.append("TAG:CHILDISH_OR_EMOTIONAL")このタグ群をプロンプトに [OBSERVATION] として注入することで、AIが「相談者を見抜いている風」の回答をしてくれる。
これ、AIでやらなかったのが正解でした。
- 決定論的で挙動が安定
- 無料（APIコストゼロ）
- 高速（ミリ秒未満）
- そして「神様の能力」として演出するなら、実はこれで十分
AIに全部やらせる前に「ルールで十分なもの」を切り出すのは、コスト最適化の基本だと再確認しました。
ProfitGuard——赤字なら安価モデルに切り替える
Class A (gemini-2.5-pro) は高いので、収益がプラスになる範囲でのみ高性能モデルを使うガードを実装しました。
# core/profit_guard.py
# 収益: coins * coin_usd_value * revenue_share
# コスト: Text/TTS の入出力トークンから推定
if not profit_guard.can_afford_text(model_id, input_chars, output_chars_est):
# 安価モデル (gemini-2.5-flash) にフォールバック
# それも無理なら定型文＋TTS省略仕組みはこう：
- ギフトが来たら
balance_usdにクレジット - Text / TTS を呼ぶ前に残高をチェック
- 足りなければ fallback_model で再試行
- それも足りなければ固定文（
fallback_texts）を返し、TTSもスキップ
数字がマイナスになりそうならソフトに品質を落として事業としての収支ラインを死守する設計。コスト管理ロジックだけ単体テストしやすいのもポイント。
階級別プロンプトとキャラクター演出
プロンプトは Class によって persona 部分だけ差し替える固定テンプレートです。
# core/gemini_text.py
if tier == "A":
persona = "階級: A (Obsessive Male God). 口調: 独占的、支配的、甘美。"
elif tier == "B":
persona = "階級: B (Guardian of Charity). 口調: 優しく、包容力がある。"
else:
persona = "階級: C (Tsundere God). 『勘違いするなよ』『今回だけだ』が口癖。"加えて、Class A にだけ 人工的な「溜め」を入れています。
performance:
min_think_time_sec:
class_c: 0.1 # 即レス
class_b: 0.6
class_a: 3.0 # わざと待たせる＝「神降ろし」演出実際には gemini-2.5-pro の thinking_budget: 4096 で思考時間が伸びるので、そこにさらに人工遅延を足して「特別感」を出す。待つことで価値が上がる、という心理を使っています。
Gemini Native TTS の罠
TTSは Gemini Native TTS (gemini-2.5-flash-preview-tts) を使っています。Gemini API だけで Text も TTS も完結するのが一番大きい利点でした（サービスアカウント不要、APIキー1本）。
ただレスポンスフォーマットに PCM生データが返ってくるケースがあり、ブラウザでそのまま再生できません。_normalize_audio で mime を見て必要なら WAV ヘッダを被せる実装を入れました。
# core/gemini_tts.py
if "pcm" in normalized_mime or "raw" in normalized_mime or not normalized_mime:
sample_rate = self._parse_sample_rate(mime_type) or self.sample_rate_hz
return self._pcm_to_wav(audio_bytes, sample_rate_hz=sample_rate), "audio/wav"mime_type のクエリ文字列から rate=24000 を正規表現で抜く、みたいな泥臭い処理も入っています。プレビュー系APIはこういうところで時間を溶かしがち。
OBS連携——WebSocketで音声もBase64で流す
フロントエンドは OBS のブラウザソースとして 1080x1920 で表示します（縦配信 / TikTok LIVE 向け）。
[Python Backend] ──WebSocket(8765)──► [OBS Browser Source (HTML/JS)]
├─ 待機列リスト
├─ キャラ画像（idle/speaking切替＋口パク）
├─ 吹き出し
└─ 音声再生（Base64 Data URL）面白いのは、音声データを WebSocket に Base64 で流して、ブラウザで new Audio("data:audio/wav;base64,...") で再生していること。ローカルで完結し、OBS側のキャッシュや配信ポリシーに引っかかりにくい。
// assets/obs_overlay_v2/script.js
function playAudio(base64Audio, mimeType) {
const audio = new Audio(`data:${mimeType};base64,${base64Audio}`);
audio.onplay = () => startLipSync();
audio.onended = () => stopLipSync();
audio.play();
}再生開始時に 150ms 間隔で CSS クラスを付け替え、口パク（lip sync）を簡易再現しています。音声解析は入れていない割に、それっぽく見えるのが実装効率としてよかった点。
モデレーションは「全文置換」で安全側へ
NGワードが入力に含まれていたら、一部マスクではなく全文を固定の警告文に置換します。
# core/moderation.py
def check_input(self, text):
for word in self.ng_words:
if word in text:
return False, self.safety_message # 全文置換
return True, textなぜか？ 一部置換だとコンテキスト的に文意を保ったまま問題のある内容がAIに渡ってしまう（プロンプトインジェクション的なリスク）。占いというエンタメ用途なので、安全側に倒すのが正解です。
一方、AIからの出力については該当ワードだけ記号に置換する部分伏せ字。文意を残しつつ BAN リスクを避ける。入出力で非対称な戦略にしています。
アカシックレコード（SQLite）
ユーザーデータは aiosqlite でSQLiteに保存しています。
users (user_id, display_name, nickname_by_god, total_gift_coins, last_seen_at, summary)
sessions (session_id, user_id, tier, enqueued_at, started_at, ended_at, result_text)過去の占い結果を summary として次回のプロンプトに差し込めるので、配信をまたいだ文脈継続ができる。「お前、前回も同じこと聞いていたな？」みたいな返しが出るとエモい。
マイグレーションは PRAGMA table_info で既存列を確認して足りないカラムを ALTER TABLE する素朴な実装。小規模なら十分です。
LangFuse で Gemini API の使用量を追跡
Gemini の各リクエストに langfuse.start_as_current_observation でトレースを仕込んでいます。
# core/gemini_text.py
generation_ctx = langfuse.start_as_current_observation(
as_type="generation",
name=f"gemini_text_tier_{tier}",
model=model_id,
metadata={"tier": tier, "tags": tags}
)
# 完了後に usage_metadata から token_count を拾って updateresponse.usage_metadata.prompt_token_count / candidates_token_count を拾って LangFuse に送るので、ティア別・モデル別のコスト可視化ができます。ProfitGuard は推定値ですが、LangFuse 側の実測と突き合わせることで「推定が外れすぎていないか」を検証できる。
環境変数が未設定ならサイレントに無効化する設計で、開発/本番で差し替え不要なのも良いところ。
アイドル時の独り言（Auto-Babble）
誰もギフトを投げていない無音時間は配信の死です。そこで 20 秒キューが空ならランダムに一言喋らせる機構を入れました。
idle_samples = [
{"id": "idle_01", "text": "ただ見ているだけか？ ギフトでも投げてみるんだな。"},
{"id": "idle_02", "text": "今は手が空いている。今投げれば待たずに視てやれるぞ？"},
]ポイントは、事前録音した wav を使い回していること。TTS を毎回呼ぶとコストが嵩むので、「視聴者に向けたあおり文句」だけは録り置きしておく。コスト最適化とレイテンシ最適化を同時に達成できる。
得た学び
- ルールで十分な処理をAIに投げない ── Cold Reading は正規表現で6行書けば完成
- AIコストを配信収益と連動させる ── ProfitGuard でソフトに品質を落とす設計は精神衛生に良い
- エンタメの「待ち時間」は資産 ── Class A だけ3秒遅延させる人工溜めが一番効いた演出
- OBS×WebSocket は開発体験が良い ── HTML/CSS/JSだけでオーバーレイが組める。
?demo=1でモックデータを流せるのも楽 - Gemini API 完全統一のシンプルさ ── Text/TTS同じキー、同じSDK、同じ障害ドメイン。運用が1つで済む
動かしてみる
git clone <this-repo>
cd AI-Fortune-Live
python -m venv .venv
source .venv/bin/activate # or .venv\Scripts\activate.bat
pip install -r requirements.txt
cp .env.example .env # GOOGLE_API_KEY を設定
# config.yaml の tiktok.unique_id を自分のIDに
python main.pyOBS の「ブラウザ」ソースに assets/obs_overlay_v2/index.html の絶対パスを指定して、サイズを 1080x1920 に設定すれば、縦配信向けのオーバーレイが出ます。
ディレクトリ構成
AI-Fortune-Live/
├── main.py # asyncio.gather のエントリポイント
├── config.yaml # 階級閾値・価格・音声設定
├── core/
│ ├── tiktok_client.py # TikTokLive イベント受信
│ ├── queue_manager.py # heapq + 蓄積バッファ
│ ├── session_runner.py # 1セッションの状態遷移
│ ├── gemini_text.py # 階級別 Text 生成
│ ├── gemini_tts.py # Native TTS + PCM正規化
│ ├── cold_reading.py # 神の眼（正規表現）
│ ├── moderation.py # NGワード
│ ├── profit_guard.py # 収支ガード
│ ├── database.py # SQLite（aiosqlite）
│ └── langfuse_setup.py # トレース
├── services/
│ ├── obs_ws_server.py # WebSocket Server
│ └── audio_player.py # 音声ブロードキャスト
└── assets/
├── images/ # キャラ画像（idle / speaking）
├── sounds/idle/ # 独り言の録音
└── obs_overlay_v2/ # フロントエンド (HTML/CSS/JS)おわりに
「AI占い」は題材として人気ですが、この作品は占いそのものの精度ではなく、課金階層をアルゴリズムに埋め込んだエンタメを作ったのが軸です。base_points の桁差、人工溜め、蓄積バッファ、全部が「視聴者に課金行動を取らせる仕掛け」として設計されています。
技術的にはシンプルな Python + asyncio + Gemini API ですが、それらをエンタメ設計と経済設計に結びつけるのが一番難しくて楽しかった部分でした。
同じような「リアルタイムAIキャラクター配信」を考えている人の参考になれば嬉しいです。
コメント