「返金に関するDMは優先キューへ、通報済みなら担当者へ回して」とAIに頼み、設定JSONを作らせるところまでは簡単になりました。
難しいのは、そのJSONを誰が正しいと判断し、誤分類やAIサービスの障害があってもDMを止めずに運用できる形へ変えることです。AIの説明が自然であるほど、設定自体も正しいように見えてしまいます。
本稿では、AIをルールの作成補助に限定し、実行時は決定的なルールエンジンを使います。候補JSONを検証用DMへ適用し、人が差分を確認した後だけ承認済み設定に昇格させます。
結論：AIには候補を書かせ、本番反映はテストと承認で決める
今回作る流れは次のとおりです。
- 人が自然言語でDMの振り分け要件を書く
- AIが制約付きの候補JSONを生成する
- Pythonでスキーマ、禁止操作、優先順位を検査する
- 合成したDMを再生し、期待する振り分けと比較する
- 人が結果を確認し、承認者名とハッシュを付ける
- 本番アプリは承認済みJSONだけを読み込む
AIが得意なのは、曖昧な要件を設定候補へ変換し、変更内容を説明することです。一方、次の判断はAIへ委ねません。
- 誤検知をどこまで許容するか
- どの話題をセンシティブとして扱うか
- 自動処理について利用者へ何を表示するか
- 削除、ブロック、返信送信などの不可逆操作を許すか
- 翻訳文だけを根拠にモデレーションしてよいか
この分離により、AIが利用できない時間でも、承認済みルールによるDM振り分けは継続できます。
前提：メッセージング、ルール作成、実行を分ける
Tencent RTCのSocial Messagingソリューションは、1対1チャット、グループディスカッション、コミュニティ、リッチメディア、ライブ配信ルーム内チャットなどのシナリオを扱います。
本稿では、そのうち1対1のDMを例にします。ただし、以下のPythonコードはTencent RTC SDKのAPIを再実装するものではありません。責任境界は次のように分けます。
| 層 | 担当 |
|---|---|
| Tencent RTC | DMの送受信とチャット体験 |
| アプリのAdapter | 受信イベントをルール入力へ正規化 |
| ルールエンジン | キュー、優先度、有人確認の判定 |
| AIアシスタント | 候補ルールの作成と説明 |
| 運用担当者 | テスト結果の確認、承認、ロールバック |
重要なのは、ルール作成用AIへ本番DMを直接読ませないことです。要件、項目定義、合成した例文だけでも候補は作れます。実メッセージを使った評価が必要な場合は、組織の同意、プライバシー、保持期間、アクセス権限を先に決める必要があります。
手順1：AIが出力できる範囲を固定する
今回は、実行可能な操作を次の3種類だけにします。
-
standard_queue: 通常キュー -
priority_queue: 優先キュー -
human_handoff: 自動判定を止めて担当者確認へ移す
AIには、返信、削除、ブロック、通報確定を表す操作を出力させません。プロンプトで禁止するだけでなく、後段のコードでも拒否します。
AIへ渡す依頼例は次のようになります。
DM振り分けポリシーの候補をJSONで作成してください。
要件:
- 通報済みDMは担当者確認へ送る
- 「返金」「請求」「詐欺」を含むDMは優先キューへ送る
- それ以外は通常キューへ送る
- 自動処理の案内に同意していない利用者は担当者確認へ送る
制約:
- actionはstandard_queue、priority_queue、human_handoffのみ
- 返信、削除、ブロック、通報確定は提案しない
- 各ルールに一意なidと整数のpriorityを付ける
- 自信がなくても制約外のactionを追加しない
- JSON以外を出力しない
ここでのAI出力は、あくまで次工程への入力です。「JSONが生成できた」ことは、「運用ポリシーが妥当である」ことを意味しません。
手順2：候補ポリシーと再生データを用意する
作業ディレクトリを作成します。
mkdir dm-policy-gate
cd dm-policy-gate
touch policy.candidate.json fixtures.json policy_gate.py
policy.candidate.jsonへ候補を保存します。
{
"revision": "2026-10-18-01",
"default_action": "standard_queue",
"rules": [
{
"id": "reported-message",
"priority": 10,
"when": {
"reported": true
},
"action": "human_handoff"
},
{
"id": "billing-keywords",
"priority": 20,
"when": {
"contains_any": ["返金", "請求", "詐欺"]
},
"action": "priority_queue"
}
]
}
続いて、fixtures.jsonへ合成メッセージと期待結果を保存します。
[
{
"name": "案内への同意がない",
"message": {
"id": "m-001",
"text": "配送予定を確認したいです",
"reported": false,
"automation_notice_accepted": false
},
"expected_action": "human_handoff",
"expected_reason": "consent_missing"
},
{
"name": "通報済みを最優先する",
"message": {
"id": "m-002",
"text": "通常の文章",
"reported": true,
"automation_notice_accepted": true
},
"expected_action": "human_handoff",
"expected_reason": "reported-message"
},
{
"name": "返金相談を優先キューへ送る",
"message": {
"id": "m-003",
"text": "返金の手続きを確認したいです",
"reported": false,
"automation_notice_accepted": true
},
"expected_action": "priority_queue",
"expected_reason": "billing-keywords"
},
{
"name": "一般的な問い合わせ",
"message": {
"id": "m-004",
"text": "プロフィール画像を変更したいです",
"reported": false,
"automation_notice_accepted": true
},
"expected_action": "standard_queue",
"expected_reason": "default"
}
]
automation_notice_acceptedは実装例のアプリポリシーであり、普遍的な法的要件を示すものではありません。実際に必要な通知や同意は、対象地域、サービス規約、処理目的に合わせて決めてください。
コード：禁止操作を拒否し、再生試験後だけ承認する
policy_gate.pyへ次のコードを保存します。外部パッケージは使いません。
from __future__ import annotations
import argparse
import hashlib
import json
import os
import re
import sys
from datetime import datetime, timezone
from pathlib import Path
from typing import Any
ALLOWED_ACTIONS = {
"standard_queue",
"priority_queue",
"human_handoff",
}
ALLOWED_CONDITIONS = {"reported", "contains_any"}
REVISION_PATTERN = re.compile(r"^[0-9A-Za-z._-]{1,64}$")
def load_json(path: str) -> Any:
return json.loads(Path(path).read_text(encoding="utf-8"))
def canonical_bytes(value: Any) -> bytes:
return json.dumps(
value,
ensure_ascii=False,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
def digest_policy(policy: dict[str, Any]) -> str:
return hashlib.sha256(canonical_bytes(policy)).hexdigest()
def require_exact_keys(
value: dict[str, Any],
allowed: set[str],
location: str,
) -> None:
unknown = set(value) - allowed
if unknown:
raise ValueError(
f"{location}: 未許可の項目があります: {sorted(unknown)}"
)
def validate_policy(policy: Any) -> None:
if not isinstance(policy, dict):
raise ValueError("policyはオブジェクトである必要があります")
require_exact_keys(
policy,
{"revision", "default_action", "rules"},
"policy",
)
revision = policy.get("revision")
if not isinstance(revision, str) or not REVISION_PATTERN.fullmatch(revision):
raise ValueError("revisionの形式が不正です")
if policy.get("default_action") not in ALLOWED_ACTIONS:
raise ValueError("default_actionが許可されていません")
rules = policy.get("rules")
if not isinstance(rules, list) or not rules:
raise ValueError("rulesには1件以上のルールが必要です")
ids: set[str] = set()
priorities: set[int] = set()
for index, rule in enumerate(rules):
location = f"rules[{index}]"
if not isinstance(rule, dict):
raise ValueError(f"{location}はオブジェクトである必要があります")
require_exact_keys(
rule,
{"id", "priority", "when", "action"},
location,
)
rule_id = rule.get("id")
if not isinstance(rule_id, str) or not REVISION_PATTERN.fullmatch(rule_id):
raise ValueError(f"{location}.idの形式が不正です")
if rule_id in ids:
raise ValueError(f"ルールIDが重複しています: {rule_id}")
ids.add(rule_id)
priority = rule.get("priority")
if not isinstance(priority, int) or isinstance(priority, bool):
raise ValueError(f"{location}.priorityは整数にしてください")
if priority in priorities:
raise ValueError(f"priorityが重複しています: {priority}")
priorities.add(priority)
if rule.get("action") not in ALLOWED_ACTIONS:
raise ValueError(
f"{location}.actionは許可されていません: {rule.get('action')}"
)
conditions = rule.get("when")
if not isinstance(conditions, dict) or not conditions:
raise ValueError(f"{location}.whenが空です")
unknown_conditions = set(conditions) - ALLOWED_CONDITIONS
if unknown_conditions:
raise ValueError(
f"{location}.whenに未許可の条件があります: "
f"{sorted(unknown_conditions)}"
)
if "reported" in conditions and not isinstance(
conditions["reported"], bool
):
raise ValueError(f"{location}.when.reportedは真偽値にしてください")
if "contains_any" in conditions:
terms = conditions["contains_any"]
if (
not isinstance(terms, list)
or not terms
or not all(isinstance(term, str) and term.strip() for term in terms)
):
raise ValueError(
f"{location}.when.contains_anyには空でない文字列配列が必要です"
)
def matches(rule: dict[str, Any], message: dict[str, Any]) -> bool:
conditions = rule["when"]
if "reported" in conditions:
if bool(message.get("reported", False)) != conditions["reported"]:
return False
if "contains_any" in conditions:
text = str(message.get("text", "")).casefold()
terms = [term.casefold() for term in conditions["contains_any"]]
if not any(term in text for term in terms):
return False
return True
def classify(
policy: dict[str, Any],
message: dict[str, Any],
) -> dict[str, str]:
if not message.get("automation_notice_accepted", False):
return {
"message_id": str(message.get("id", "unknown")),
"policy_revision": policy["revision"],
"action": "human_handoff",
"reason": "consent_missing",
"notice_key": "human_review_pending",
}
for rule in sorted(policy["rules"], key=lambda item: item["priority"]):
if matches(rule, message):
return {
"message_id": str(message.get("id", "unknown")),
"policy_revision": policy["revision"],
"action": rule["action"],
"reason": rule["id"],
"notice_key": (
"human_review_pending"
if rule["action"] == "human_handoff"
else "received"
),
}
return {
"message_id": str(message.get("id", "unknown")),
"policy_revision": policy["revision"],
"action": policy["default_action"],
"reason": "default",
"notice_key": "received",
}
def replay(policy: dict[str, Any], fixtures: Any) -> bool:
if not isinstance(fixtures, list) or not fixtures:
raise ValueError("fixturesには1件以上のケースが必要です")
passed = True
for fixture in fixtures:
result = classify(policy, fixture["message"])
ok = (
result["action"] == fixture["expected_action"]
and result["reason"] == fixture["expected_reason"]
)
mark = "PASS" if ok else "FAIL"
print(
f"[{mark}] {fixture['name']}: "
f"action={result['action']} reason={result['reason']}"
)
passed = passed and ok
return passed
def approve(policy: dict[str, Any], output_path: str) -> None:
approver = os.environ.get("APPROVER", "").strip()
if not approver:
raise ValueError("承認時は環境変数APPROVERを指定してください")
bundle = {
"policy": policy,
"approval": {
"approver": approver,
"approved_at": datetime.now(timezone.utc).isoformat(),
"sha256": digest_policy(policy),
},
}
Path(output_path).write_text(
json.dumps(bundle, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
print(f"承認済みポリシーを書き出しました: {output_path}")
def verify_bundle(path: str) -> None:
bundle = load_json(path)
policy = bundle["policy"]
approval = bundle["approval"]
validate_policy(policy)
actual = digest_policy(policy)
if actual != approval.get("sha256"):
raise ValueError("承認後にポリシーが変更されています")
print(
"承認済みポリシーを確認しました: "
f"revision={policy['revision']} approver={approval.get('approver')}"
)
def main() -> int:
parser = argparse.ArgumentParser()
parser.add_argument("--policy", default="policy.candidate.json")
parser.add_argument("--fixtures", default="fixtures.json")
parser.add_argument("--approve", action="store_true")
parser.add_argument("--output", default="policy.approved.json")
parser.add_argument("--verify-approved")
args = parser.parse_args()
try:
if args.verify_approved:
verify_bundle(args.verify_approved)
return 0
policy = load_json(args.policy)
fixtures = load_json(args.fixtures)
validate_policy(policy)
if not replay(policy, fixtures):
print("再生試験に失敗したため承認できません", file=sys.stderr)
return 1
if args.approve:
approve(policy, args.output)
else:
print("候補は有効です。まだ承認されていません。")
return 0
except (ValueError, KeyError, json.JSONDecodeError) as error:
print(f"ERROR: {error}", file=sys.stderr)
return 1
if __name__ == "__main__":
raise SystemExit(main())
確認方法：正常な候補より、危険な候補を先に試す
1. 候補を再生する
python policy_gate.py
すべてのケースがPASSになっても、まだ本番用ファイルは生成されません。
[PASS] 案内への同意がない: action=human_handoff reason=consent_missing
[PASS] 通報済みを最優先する: action=human_handoff reason=reported-message
[PASS] 返金相談を優先キューへ送る: action=priority_queue reason=billing-keywords
[PASS] 一般的な問い合わせ: action=standard_queue reason=default
候補は有効です。まだ承認されていません。
2. AIが禁止操作を出した場合
候補JSONのactionをauto_deleteへ変更して実行します。
python policy_gate.py
許可リストにないため、再生前に拒否されます。AIへの指示が守られなかった場合でも、実行境界で止められます。
3. 人が承認する
出力差分と再生結果を確認した担当者が、承認者名を指定します。
APPROVER=reviewer-a python policy_gate.py --approve
生成されたファイルを検証します。
python policy_gate.py --verify-approved policy.approved.json
policy.approved.json内のルールを後から書き換えると、ハッシュ不一致で拒否されます。ただし、SHA-256は認証や権限管理の代わりではありません。本番では、承認済みファイルを認証された設定ストアへ保存し、変更履歴とロールバック先も管理してください。
手順3：Tencent RTCのDMへ接続する
SDK固有の受信イベントを直接ルールエンジンへ渡さず、アプリ側で小さな入力形式へ変換します。
Tencent RTCでDMを受信
↓
アプリのAdapterで正規化
id / text / reported / automation_notice_accepted
↓
承認済みポリシーのハッシュを検証
↓
classify()を実行
↓
standard_queue / priority_queue / human_handoff
↓
アプリのキューを更新し、利用者へ状態を表示
接続時の原則は次の3点です。
- 候補ファイルを本番プロセスから読まない
-
human_handoffを「処理済み」ではなく「担当者確認待ち」と表示する - ルールの実行結果に
policy_revisionとreasonを残す
特に3点目が重要です。「なぜ優先キューへ入ったのか」を後から確認できないと、誤分類をルール改訂へ反映できません。DM本文を監査ログへ複製する必要があるとは限らないため、本文の保存範囲と保持期間は別途設計します。
翻訳を加える場合の判断
TUIChatにはオンデマンドのテキストメッセージ翻訳に関する公式手順があります。対応するコンテンツ形式、言語、エディション上の条件は変更され得るため、組み込み前に公式ドキュメントを確認してください。
翻訳表示とモデレーション判定は同じ判断にしない方が安全です。例えば、翻訳文に「詐欺」が現れたという理由だけで自動措置を行うと、文脈や翻訳誤差を取り込んでしまいます。
選択肢は次のように整理できます。
| 用途 | 推奨する扱い |
|---|---|
| 利用者がDMを読むための翻訳 | 原文へ戻れるUIを用意する |
| 担当者が内容を把握するための翻訳 | 原文と翻訳文を併記する |
| 翻訳文による優先度の補助 |
human_handoffへ寄せて人が確認する |
| 翻訳文だけを根拠に削除・ブロック | このルールエンジンでは許可しない |
多言語対応が必要なら、まず各言語の合成fixtureを追加します。単に日本語キーワードを増やすだけでは、否定、引用、皮肉、表記揺れを十分に扱えません。
AIを使うかどうかの判断基準
AIを使う価値があるのは、ルール候補の作成頻度が高く、人がJSONを手書きする負担を下げたい場合です。反対に、ルールが数個でほとんど変わらないなら、AI連携より手動編集とコードレビューの方が単純です。
| 判断対象 | AI利用 | 最終決定 |
|---|---|---|
| 自然言語要件から候補JSONを作る | 向いている | 人がレビュー |
| 既存ルールの説明文を作る | 向いている | 人が事実確認 |
| 未知のactionを追加する | 許可しない | スキーマ変更として設計レビュー |
| DMを削除・ブロックする | 任せない | 明示的な運用ポリシーと権限が必要 |
| 誤分類の許容範囲を決める | 任せない | プロダクト・運用担当者が決定 |
| AI障害時の本番処理 | 依存させない | 最後の承認済み設定を利用 |
見えにくいコストは、AIの利用料だけではありません。fixtureの保守、言語別のテスト、レビュー担当者の確保、誤分類の調査、ロールバック訓練まで含めて判断する必要があります。
注意点と失敗しやすいケース
キーワード一致は意味理解ではない
「詐欺ではありません」という否定文や、過去のDMを引用した文章にも一致します。最初から自動措置へ使わず、優先キューや有人確認の補助として始める方が扱いやすい設計です。
fixtureが少ないと、間違ったルールも通る
最低でも次のケースを追加します。
- 否定文
- 引用文
- 全角・半角や大文字・小文字の違い
- 空文字、添付のみ、削除済みメッセージ
- 複数ルールへ同時に一致する文章
- 同意状態が欠落したイベント
- 翻訳前後で意味が変わる文章
- 通報状態とキーワード条件が競合する文章
ハッシュだけでは承認者を証明できない
このサンプルのハッシュは、承認後の意図しない変更を検出するものです。悪意ある変更や承認者のなりすましを防ぐには、認証済みデプロイ経路、アクセス制御、署名、監査ログなどが別途必要です。
有人引き継ぎを行き止まりにしない
human_handoffへ振り分けるだけでなく、担当キュー、受付時刻、現在の状態、利用者に見せる案内を定義します。「AIでは判断できませんでした」だけを表示して会話を止めないことが重要です。
リリース前チェックリスト
- 本番プロセスは承認済みファイルだけを読む
- 許可されるactionが列挙されている
- 返信、削除、ブロックなどの不可逆操作が含まれていない
- AIが未知のフィールドを追加すると検証に失敗する
- ルールの優先順位が重複していない
- 否定、引用、多言語、空メッセージを再生した
- 判定結果へポリシーrevisionと理由を残す
- 有人確認中であることを利用者へ表示できる
- AIサービス停止中でも承認済みルールが動く
- 直前の承認済みrevisionへ戻す手順がある
チャットで設定を作れること自体は便利です。しかし、本番投入の信頼性を作るのは、自然な回答ではなく、許可リスト、再生可能なテスト、人の承認、改変検知です。AIを本番の判断者ではなく、検査可能な候補の作成者として置くと、便利さを残しながら責任の所在を明確にできます。
関係性の開示: 筆者はTencent RTCのコンテンツ制作に関わっています。本稿はTencent RTC公式ドキュメントを実装上の参照資料として使用しました。