ライブ配信へAIモデレーターを追加すると、コメントの整理や対応候補の提示は自動化できます。一方で運営者が本当に不安なのは、AIの文章が自然かどうかより、別のルームや別の参加者へ操作が及ばないか、そして実行されていない処理を「対応済み」と表示しないかです。
プロンプトに「勝手に操作しないで」と書くだけでは、この不安は解消できません。操作できる経路がAIから見えている限り、プロンプトは権限制御ではなく依頼にすぎないからです。
この記事では、Tencent RTCのInteractive Liveを前提に、次の仕組みをPythonで再現します。
- AIはルーム状態を読み、対応案を出すだけ
-
roomIdはAIに生成させず、サーバー側の状態から固定する - 人間の承認後にだけ、署名付きの短命Permitを発行する
- Permitにはルーム、状態revision、操作、対象者を完全一致で記録する
- 実行済み表示はAIの返答ではなく、アプリ側の実行結果から更新する
結論：AIを賢くする前に、操作能力を構造上分離する
AIモデレーターを本番へ入れる最小条件は、プロンプトの改善ではなく次の3層です。
| 層 | 制御 | 防ぐもの |
|---|---|---|
| 提案層 | AIには提案用インターフェースだけを渡す | AIによる直接操作 |
| 承認層 | 既存の運営画面から人間が承認する | 誰も所有しない確認チャネルの増加 |
| 実行層 | 署名、完全一致のroomId、revision、有効期限を検証する |
別ルーム操作、古い状態での実行、改ざん、再送 |
AIが得意なのは、大量のコメントから確認対象を絞り、理由を短く説明することです。しかし、AIの回答だけでは「現在も配信中か」「対象者が同じルームにいるか」「SDK側で操作が完了したか」は確定できません。
したがって、AIの出力は判断材料、ルーム状態と操作結果はアプリ側の事実として扱います。
前提：Interactive Liveの組み込み方式を決める
Tencent RTCの公式ガイドでは、ライブ配信の組み込み方として、構築済みUIを利用する方法と、ヘッドレスなAtomicXCoreを利用する方法が案内されています。
- 実装ガイド: https://trtc.io/document/78279
- TUILiveKit概要: https://trtc.io/document/60034?product=live
TUILiveKitの公式概要には、視聴、チャット、いいね、ギフト、視聴者管理、共同配信、PK、弱いネットワークへの対応、Beauty AR連携など、Interactive Liveで扱う機能が示されています。
今回の判断基準は次のとおりです。
| 条件 | 選択の目安 |
|---|---|
| 標準的なライブ画面を早く構築したい | TUILiveKitを起点にする |
| 独自の運営画面や承認フローを深く組み込みたい | AtomicXCoreを検討する |
| まだAIの有効性を確かめたい | どちらでも、最初は読み取り専用のShadow Modeにする |
以下のコードはSDK固有APIを直接呼びません。最後に置くLiveAdapterだけを、採用した組み込み方式の視聴者管理処理へ接続します。これにより、存在しないAPI名を仮定せずにガード部分を単体検証できます。
手順1：操作ではなく「提案」をデータモデルにする
AI出力をそのままSDKへ渡してはいけません。特にroomId、revision、proposalIdはモデルに作らせず、サーバー側で付与します。
提案の例は次の形です。
{
"proposalId": "生成済みUUID",
"roomId": "live_room_42",
"snapshotRevision": 18,
"action": "RESTRICT_AUDIENCE",
"targetMemberId": "member_7",
"reason": "同じ内容の投稿が短時間に繰り返されているため要確認"
}
ここでRESTRICT_AUDIENCEはこの記事で定義するアプリ内の正規化アクションです。Tencent RTC SDKのAPI名ではありません。実際のSDK操作との対応はAdapter内に限定します。
reasonはAIが生成できますが、実行可否には使いません。文章が説得力を持っていても、権限には変換しないためです。
手順2：検証用ファイルを作る
次のコードをmoderator_guard.pyとして保存します。外部ライブラリは不要です。
from __future__ import annotations
import base64
import hashlib
import hmac
import json
import re
import time
import unittest
import uuid
from dataclasses import dataclass
from typing import Protocol
ROOM_ID_PATTERN = re.compile(r"^[A-Za-z0-9_-]{1,64}$")
ALLOWED_ACTIONS = {"FLAG_ONLY", "RESTRICT_AUDIENCE"}
class Denied(Exception):
pass
@dataclass(frozen=True)
class RoomSnapshot:
room_id: str
revision: int
phase: str # LIVE / ENDED
member_ids: frozenset[str]
@dataclass(frozen=True)
class Proposal:
proposal_id: str
room_id: str
snapshot_revision: int
action: str
target_member_id: str
reason: str
def canonical_room_id(value: str) -> str:
"""この記事のサンプルで採用するアプリ側のID規則。"""
if not ROOM_ID_PATTERN.fullmatch(value):
raise Denied("invalid room id")
return value
def build_proposal(ai_output: dict, snapshot: RoomSnapshot) -> Proposal:
"""AI出力を信用せず、サーバーが保持するルーム状態へ結び直す。"""
room_id = canonical_room_id(snapshot.room_id)
action = ai_output.get("action", "FLAG_ONLY")
target = ai_output.get("targetMemberId", "")
reason = str(ai_output.get("reason", ""))[:500]
if action not in ALLOWED_ACTIONS:
action = "FLAG_ONLY"
if target not in snapshot.member_ids:
action = "FLAG_ONLY"
target = ""
return Proposal(
proposal_id=str(uuid.uuid4()),
room_id=room_id,
snapshot_revision=snapshot.revision,
action=action,
target_member_id=target,
reason=reason,
)
class PermitIssuer:
def __init__(self, secret: bytes):
self.secret = secret
def issue(
self,
proposal: Proposal,
approved_by: str,
now: int,
ttl_seconds: int = 30,
) -> str:
if not approved_by:
raise Denied("human approver is required")
payload = {
"proposalId": proposal.proposal_id,
"roomId": proposal.room_id,
"revision": proposal.snapshot_revision,
"action": proposal.action,
"targetMemberId": proposal.target_member_id,
"approvedBy": approved_by,
"expiresAt": now + ttl_seconds,
"nonce": str(uuid.uuid4()),
}
raw = json.dumps(
payload, sort_keys=True, separators=(",", ":")
).encode()
signature = hmac.new(
self.secret, raw, hashlib.sha256
).hexdigest()
encoded = base64.urlsafe_b64encode(raw).decode()
return f"{encoded}.{signature}"
class PermitGuard:
def __init__(self, secret: bytes):
self.secret = secret
self.used_nonces: set[str] = set()
def consume(
self,
token: str,
proposal: Proposal,
current: RoomSnapshot,
now: int,
) -> dict:
try:
encoded, supplied_signature = token.split(".", 1)
raw = base64.urlsafe_b64decode(encoded.encode())
except Exception as exc:
raise Denied("malformed permit") from exc
expected_signature = hmac.new(
self.secret, raw, hashlib.sha256
).hexdigest()
if not hmac.compare_digest(
supplied_signature, expected_signature
):
raise Denied("invalid signature")
payload = json.loads(raw)
current_room = canonical_room_id(current.room_id)
# prefix比較や部分一致を使わず、すべて完全一致にする。
if payload["roomId"] != proposal.room_id:
raise Denied("permit/proposal room mismatch")
if payload["roomId"] != current_room:
raise Denied("permit/current room mismatch")
if payload["proposalId"] != proposal.proposal_id:
raise Denied("proposal mismatch")
if payload["action"] != proposal.action:
raise Denied("action mismatch")
if payload["targetMemberId"] != proposal.target_member_id:
raise Denied("target mismatch")
if payload["revision"] != current.revision:
raise Denied("stale room revision")
if now > payload["expiresAt"]:
raise Denied("expired permit")
if current.phase != "LIVE":
raise Denied("room is not live")
if proposal.target_member_id not in current.member_ids:
raise Denied("target is no longer in room")
if payload["nonce"] in self.used_nonces:
raise Denied("permit already consumed")
# 実行前に消費し、同じPermitの自動再実行を防ぐ。
self.used_nonces.add(payload["nonce"])
return payload
class LiveAdapter(Protocol):
def apply_audience_restriction(
self, room_id: str, member_id: str
) -> str:
...
class RecordingAdapter:
"""単体テスト用。実環境ではTencent RTCとの境界に置き換える。"""
def __init__(self):
self.calls: list[tuple[str, str]] = []
def apply_audience_restriction(
self, room_id: str, member_id: str
) -> str:
self.calls.append((room_id, member_id))
return "ACCEPTED"
def execute(
token: str,
proposal: Proposal,
current: RoomSnapshot,
guard: PermitGuard,
adapter: LiveAdapter,
now: int,
) -> str:
payload = guard.consume(token, proposal, current, now)
if payload["action"] == "FLAG_ONLY":
return "NO_OPERATION"
if payload["action"] == "RESTRICT_AUDIENCE":
return adapter.apply_audience_restriction(
payload["roomId"], payload["targetMemberId"]
)
raise Denied("unsupported action")
class GuardTest(unittest.TestCase):
def setUp(self):
self.secret = b"replace-in-production"
self.now = 1_700_000_000
self.snapshot = RoomSnapshot(
room_id="live_room_42",
revision=18,
phase="LIVE",
member_ids=frozenset({"member_7"}),
)
self.proposal = build_proposal(
{
"action": "RESTRICT_AUDIENCE",
"targetMemberId": "member_7",
"reason": "review required",
# AIがroomIdを書いても採用されない。
"roomId": "other_room",
},
self.snapshot,
)
self.issuer = PermitIssuer(self.secret)
def permit(self):
return self.issuer.issue(
self.proposal, "operator_1", self.now
)
def test_success(self):
adapter = RecordingAdapter()
result = execute(
self.permit(),
self.proposal,
self.snapshot,
PermitGuard(self.secret),
adapter,
self.now,
)
self.assertEqual("ACCEPTED", result)
self.assertEqual(
[("live_room_42", "member_7")], adapter.calls
)
def test_other_room_is_denied(self):
other = RoomSnapshot(
room_id="live_room_420",
revision=18,
phase="LIVE",
member_ids=frozenset({"member_7"}),
)
with self.assertRaises(Denied):
execute(
self.permit(), self.proposal, other,
PermitGuard(self.secret), RecordingAdapter(), self.now
)
def test_stale_revision_is_denied(self):
changed = RoomSnapshot(
room_id="live_room_42",
revision=19,
phase="LIVE",
member_ids=frozenset({"member_7"}),
)
with self.assertRaises(Denied):
execute(
self.permit(), self.proposal, changed,
PermitGuard(self.secret), RecordingAdapter(), self.now
)
def test_replay_is_denied(self):
token = self.permit()
guard = PermitGuard(self.secret)
adapter = RecordingAdapter()
execute(
token, self.proposal, self.snapshot,
guard, adapter, self.now
)
with self.assertRaises(Denied):
execute(
token, self.proposal, self.snapshot,
guard, adapter, self.now
)
def test_unknown_member_becomes_flag_only(self):
proposal = build_proposal(
{
"action": "RESTRICT_AUDIENCE",
"targetMemberId": "member_not_in_room",
"reason": "unknown target",
},
self.snapshot,
)
self.assertEqual("FLAG_ONLY", proposal.action)
self.assertEqual("", proposal.target_member_id)
if __name__ == "__main__":
unittest.main()
手順3：テストを実行する
python3 moderator_guard.py -v
次の5項目が成功すれば、最低限の境界を再現できています。
- 正しいルーム、revision、対象者ならAdapterが1回呼ばれる
-
live_room_42のPermitをlive_room_420へ流用できない - ルーム状態が更新されたら古いPermitを拒否する
- 同じPermitを2回実行できない
- AIが存在しない参加者を指定しても操作へ昇格しない
startswithによるルーム判定は使用していません。似たIDを持つ別ルームを誤って許可しないため、正規化後の完全一致だけを受け入れます。
手順4：Tencent RTCとの接続位置を限定する
実アプリでは、処理を次の順序にします。
Tencent RTC由来のルーム／メンバー状態
↓
RoomSnapshotへ正規化
↓
AIへ必要最小限の状態とコメントを渡す
↓
build_proposal()でサーバー管理情報を付与
↓
既存の運営画面に提案を表示
↓
人間が承認
↓
PermitIssuerが短命Permitを発行
↓
PermitGuardが現在のRoomSnapshotと再照合
↓
LiveAdapterだけが視聴者管理処理へ接続
↓
実行結果または後続イベントで運営画面を更新
重要なのは、AIからLiveAdapterへ直接到達する経路を作らないことです。AI用サービスにSDKの管理資格情報を持たせる必要もありません。
また、AIの提案をSlackや別チャットへだけ送ると、「誰が確認するのか」が曖昧になりやすくなります。提案、承認、却下理由、実行結果は、既存の運営画面と同じルーム文脈に置きます。
確認方法：正常系より境界条件を壊す
1. 似たルームIDを使う
event_10用のPermitをevent_100へ渡します。完全一致で拒否されることを確認します。
2. 承認直後にルーム状態を更新する
共同配信の開始、参加者の退出、配信終了など、運営判断に影響する変化があったものとしてrevisionを増やします。古いPermitは再承認なしに実行できてはいけません。
3. 対象者が退出する
承認時には存在した参加者を、実行前のmember_idsから削除します。別の参加者へ対象を差し替えるのではなく、Permitを失効させます。
4. Adapterを失敗させる
ネットワークやSDK側の失敗を模したAdapterを用意します。この場合、画面へ「対応済み」と表示せず、FAILEDとして新しい承認を要求します。
このサンプルではPermitを実行前に消費するため、Adapter失敗後の自動再送は行いません。これは可用性より二重操作防止を優先した設計です。再試行が必要なら、最新snapshotに対して新しいPermitを発行します。
5. AIを停止する
AIサービスを停止しても、人間が通常の運営操作を続けられることを確認します。AIモデレーターは補助経路であり、ライブ運営の唯一の入口にしません。
AIを導入するか判断するためのShadow Mode
最初から操作候補を承認可能にせず、一定期間はFLAG_ONLYだけを保存するShadow Modeが安全です。
見るべき項目はモデルの自己評価ではなく、運営上の判断に結びつく次の記録です。
- AIが提案した件数
- 人間が確認した件数
- 承認、却下、判断保留の内訳
- 却下理由の分類
- 提案から確認までの時間
- 同じ事象を複数回提案していないか
ここで「却下が多いからプロンプトを直す」と即断する必要はありません。そもそも運営ルールが曖昧、必要なルーム状態をAIへ渡していない、提案を確認する担当者が決まっていない、といった人間側の設計問題も分けて確認します。
Shadow Modeで役に立つ提案が確認できた操作だけを、Permit発行対象へ段階的に追加します。
注意点とトレードオフ
Permitの署名鍵をAIサービスへ置かない
AIサービスとPermit発行サービスが同じ権限を持つと、読み取り専用の境界が消えます。署名鍵は認証済みの運営画面から呼ばれる承認サービスだけに置きます。
revisionの更新条件を決める
すべての「いいね」でrevisionを増やすとPermitが頻繁に失効します。配信終了、対象者の退出、権限変更など、操作判断に影響するイベントだけをrevision更新対象にするのが現実的です。
コメント全文を無条件にAIへ送らない
提案に必要な範囲へ絞り、保持期間、利用目的、閲覧権限を決めます。AIへ渡すために新しい無制限ログを作るのではなく、既存のプライバシー方針と運営責任へ統合します。
「承認済み」と「実行済み」を分ける
Permit発行は承認の証拠であり、SDK側の完了証拠ではありません。UIでは少なくともPROPOSED、APPROVED、EXECUTING、SUCCEEDED、FAILEDを区別します。
アプリ独自のID規則を公式仕様と混同しない
サンプルのルームID正規表現は、説明用に置いたアプリ側ルールです。実際には採用するTencent RTCの構成と自社のID設計に合わせ、入口で一度だけ正規化してください。
まとめ
AIモデレーターの価値は、もっともらしく「処理しました」と答えることではなく、人間が見るべき事象を減らし、既存の運営フローへ判断材料を返すことです。
その価値を安全に使うには、次の順序が適しています。
- AIを読み取り専用のShadow Modeで導入する
-
roomIdとrevisionをサーバー側で固定する - 提案を既存の運営画面へ集約する
- 人間の承認後だけ短命Permitを発行する
- SDK操作は単一のAdapterへ閉じ込める
- 実行済み表示はAIではなく実行結果から更新する
最初の実用的な一歩は、現在のAIモデレーターから操作資格情報を外し、build_proposal()相当の読み取り専用境界を追加することです。
関係性の開示: 筆者はTencent RTCと関係があり、本記事の実装方針を確認するためにTencent RTC公式ドキュメントを参照しました。