AIエージェント同士がメッセージを送り合えるようになると、実装速度は上がります。一方、ユーザー向けDMまで同じ勢いで自動化すると、別の問題が現れます。
- AIがどの会話を読んだのか分からない
- 誤った返信を誰が止めるのか決まっていない
- 有人対応へ切り替えたはずなのに、担当者がいない
- AI用キューとサポート用キューが分かれ、どちらも所有されない
ここでの不安は「AIが十分に賢いか」だけではありません。送信されたメッセージの責任主体が消えることが本質です。
この記事では、1対1のソーシャルメッセージングを題材に、AIを返信の下書きに限定し、同意・担当者・承認・配送を明示的な状態として管理します。LLMやチャットSDK固有のAPIには依存せず、Python標準ライブラリだけで境界を再現します。
結論：AIの能力ではなく、送信権と担当者を先に固定する
今回のルールは次の5つです。
- AI利用への同意がないスレッドをLLMへ渡さない
- センシティブまたは判断困難な内容は、下書きせず人へ渡す
- AIは下書きを作れるが、ユーザーへ送信できない
- 担当者はスレッドごとに1人だけにする
- 下書き後にユーザーから新着メッセージが来たら、古い下書きを承認できない
有人引き継ぎは別の管理チャネルへ投稿するのではなく、元のDMスレッドへ状態を表示します。これにより、利用者にも運営者にも「現在は誰が対応しているか」が見える構成になります。
前提：Tencent RTCが担うメッセージングと、アプリが担う判断を分ける
Tencent RTCのSocial Messagingソリューションは、1対1チャット、グループディスカッション、コミュニティ、リッチメディアなどのソーシャルメッセージング用途を扱います。
本実装では、そのメッセージ配送部分とアプリケーション判断を分けます。
ユーザーのDM
↓
Tencent RTCを利用するチャットUI／メッセージ経路
↓
アプリケーションサーバー
├─ 同意確認
├─ モデレーション結果による振り分け
├─ AIへ渡す文脈の選択
├─ 担当者の割り当て
├─ 下書きの承認
└─ 送信Outbox
↓
Tencent RTC接続アダプター
LLMが実際にできるのは、与えられた文脈から返信候補を生成することです。次の判断まで保証するわけではありません。
- ユーザーがAI利用へ同意したか
- メッセージが緊急・法務・決済などの領域に該当するか
- 生成後に会話の前提が変わっていないか
- 誰が最終責任を持つか
- メッセージが実際に配送されたか
これらはLLMのプロンプトではなく、アプリケーション状態として管理します。
手順1：振り分けルールを文章ではなく表にする
まず、AIへ渡す前の判断を固定します。
| AI利用への同意 | モデレーション分類 | 動作 |
|---|---|---|
| なし | 任意 | AIへ渡さず有人対応 |
| あり | normal |
AI下書きを作り、担当者が確認 |
| あり | sensitive |
本文をAIへ渡さず有人対応 |
| あり | unknown |
推測せず有人対応 |
normal、sensitive、unknownの分類方法は、サービスのポリシーや利用するモデレーション手段によって異なります。以下のコードは分類器そのものではなく、分類結果を受け取った後の制御を実装します。
AIによる完全自動送信を最初から追加しないのは、精度だけが理由ではありません。自動送信を許すと、誤送信、権限、監査、再送、苦情対応を同時に設計する必要があるためです。まず下書き支援で運用上の失敗を観測し、それから自動化範囲を判断する方が安全です。
手順2：検証用プロジェクトを作る
Python 3.11以降を想定します。外部パッケージは不要です。
mkdir ai-dm-handoff
cd ai-dm-handoff
python3 -m venv .venv
source .venv/bin/activate
cat > demo.py
続けて、次のコードをdemo.pyへ保存します。
コード：下書き、単一担当者、古い文脈の拒否を実装する
import sqlite3
from contextlib import contextmanager
from dataclasses import dataclass
from typing import Protocol
SCHEMA = """
CREATE TABLE threads (
id TEXT PRIMARY KEY,
ai_consent INTEGER NOT NULL CHECK (ai_consent IN (0, 1)),
status TEXT NOT NULL CHECK (status IN ('OPEN', 'WAITING_HUMAN')),
owner_id TEXT
);
CREATE TABLE messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
thread_id TEXT NOT NULL,
sender_type TEXT NOT NULL CHECK (
sender_type IN ('user', 'staff', 'system')
),
body TEXT NOT NULL,
FOREIGN KEY (thread_id) REFERENCES threads(id)
);
CREATE TABLE drafts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
thread_id TEXT NOT NULL,
context_upto_message_id INTEGER NOT NULL,
body TEXT NOT NULL,
state TEXT NOT NULL CHECK (
state IN ('PROPOSED', 'APPROVED', 'SENT', 'CANCELLED')
),
FOREIGN KEY (thread_id) REFERENCES threads(id)
);
CREATE TABLE outbox (
id INTEGER PRIMARY KEY AUTOINCREMENT,
draft_id INTEGER NOT NULL UNIQUE,
thread_id TEXT NOT NULL,
payload TEXT NOT NULL,
state TEXT NOT NULL CHECK (state IN ('PENDING', 'SENT')),
FOREIGN KEY (draft_id) REFERENCES drafts(id)
);
"""
class DraftModel(Protocol):
def create_draft(self, context: list[dict[str, str]]) -> str:
...
class MessageAdapter(Protocol):
def send_text(
self, thread_id: str, text: str, idempotency_key: str
) -> None:
...
@dataclass(frozen=True)
class RouteDecision:
action: str
reason: str
def decide_route(ai_consent: bool, moderation: str) -> RouteDecision:
if not ai_consent:
return RouteDecision('HANDOFF', 'AI利用への同意なし')
if moderation != 'normal':
return RouteDecision('HANDOFF', '担当者による確認が必要')
return RouteDecision('DRAFT', 'AI下書きを作成可能')
class DmService:
def __init__(self, db: sqlite3.Connection):
self.db = db
@contextmanager
def tx(self):
self.db.execute('BEGIN IMMEDIATE')
try:
yield
self.db.execute('COMMIT')
except Exception:
self.db.execute('ROLLBACK')
raise
def create_thread(self, thread_id: str, ai_consent: bool):
self.db.execute(
"INSERT INTO threads VALUES (?, ?, 'OPEN', NULL)",
(thread_id, int(ai_consent)),
)
def add_message(self, thread_id: str, sender_type: str, body: str):
self.db.execute(
"""
INSERT INTO messages(thread_id, sender_type, body)
VALUES (?, ?, ?)
""",
(thread_id, sender_type, body),
)
def handoff(self, thread_id: str, reason: str):
# reasonには生の個人情報やモデレーション詳細を入れない
with self.tx():
self.db.execute(
"UPDATE threads SET status = 'WAITING_HUMAN' WHERE id = ?",
(thread_id,),
)
self.add_message(
thread_id,
'system',
f'有人対応へ切り替えました。理由: {reason}',
)
def claim(self, thread_id: str, staff_id: str):
with self.tx():
row = self.db.execute(
"SELECT owner_id FROM threads WHERE id = ?",
(thread_id,),
).fetchone()
if row is None:
raise ValueError('thread not found')
if row[0] not in (None, staff_id):
raise ValueError('another staff member already owns this thread')
self.db.execute(
"""
UPDATE threads
SET owner_id = ?, status = 'WAITING_HUMAN'
WHERE id = ?
""",
(staff_id, thread_id),
)
self.add_message(
thread_id,
'system',
f'担当者 {staff_id} が対応を開始しました。',
)
def request_draft(self, thread_id: str, model: DraftModel) -> int:
thread = self.db.execute(
"SELECT ai_consent FROM threads WHERE id = ?",
(thread_id,),
).fetchone()
if thread is None:
raise ValueError('thread not found')
if thread[0] != 1:
raise PermissionError('AI consent is required')
# systemメッセージはLLM文脈へ入れない。
# user/staffの直近8件だけを対象にする。
rows = self.db.execute(
"""
SELECT id, sender_type, body
FROM messages
WHERE thread_id = ?
AND sender_type IN ('user', 'staff')
ORDER BY id DESC
LIMIT 8
""",
(thread_id,),
).fetchall()
if not rows:
raise ValueError('no conversational message')
rows.reverse()
watermark = rows[-1][0]
context = [
{'role': sender_type, 'content': body}
for _, sender_type, body in rows
]
draft_body = model.create_draft(context)
cursor = self.db.execute(
"""
INSERT INTO drafts(
thread_id, context_upto_message_id, body, state
) VALUES (?, ?, ?, 'PROPOSED')
""",
(thread_id, watermark, draft_body),
)
return int(cursor.lastrowid)
def approve(self, draft_id: int, staff_id: str, edited_body: str):
if not edited_body.strip():
raise ValueError('empty message cannot be approved')
with self.tx():
draft = self.db.execute(
"""
SELECT thread_id, context_upto_message_id, state
FROM drafts WHERE id = ?
""",
(draft_id,),
).fetchone()
if draft is None or draft[2] != 'PROPOSED':
raise ValueError('draft is not approvable')
thread_id, watermark, _ = draft
owner = self.db.execute(
"SELECT owner_id FROM threads WHERE id = ?",
(thread_id,),
).fetchone()[0]
if owner != staff_id:
raise PermissionError('only the thread owner can approve')
latest = self.db.execute(
"""
SELECT COALESCE(MAX(id), 0)
FROM messages
WHERE thread_id = ?
AND sender_type IN ('user', 'staff')
""",
(thread_id,),
).fetchone()[0]
if latest != watermark:
self.db.execute(
"UPDATE drafts SET state = 'CANCELLED' WHERE id = ?",
(draft_id,),
)
raise ValueError('conversation changed; regenerate the draft')
self.db.execute(
"UPDATE drafts SET body = ?, state = 'APPROVED' WHERE id = ?",
(edited_body, draft_id),
)
self.db.execute(
"""
INSERT INTO outbox(draft_id, thread_id, payload, state)
VALUES (?, ?, ?, 'PENDING')
""",
(draft_id, thread_id, edited_body),
)
def deliver_pending(self, adapter: MessageAdapter):
rows = self.db.execute(
"""
SELECT id, draft_id, thread_id, payload
FROM outbox WHERE state = 'PENDING'
ORDER BY id
"""
).fetchall()
for outbox_id, draft_id, thread_id, payload in rows:
key = f'draft:{draft_id}'
adapter.send_text(thread_id, payload, key)
with self.tx():
self.db.execute(
"UPDATE outbox SET state = 'SENT' WHERE id = ?",
(outbox_id,),
)
self.db.execute(
"UPDATE drafts SET state = 'SENT' WHERE id = ?",
(draft_id,),
)
self.add_message(thread_id, 'staff', payload)
class FakeModel:
def __init__(self):
self.calls = 0
def create_draft(self, context):
self.calls += 1
latest = context[-1]['content']
return f'お問い合わせを確認しました: {latest}'
class FakeMessageAdapter:
def __init__(self):
self.sent = {}
def send_text(self, thread_id, text, idempotency_key):
# 検証用。実運用では配送先との重複制御と照合が必要。
self.sent.setdefault(idempotency_key, (thread_id, text))
def new_db():
db = sqlite3.connect(':memory:', isolation_level=None)
db.execute('PRAGMA foreign_keys = ON')
db.executescript(SCHEMA)
return db
def run_checks():
# 1. 同意なしではLLMを呼ばない
db = new_db()
service = DmService(db)
model = FakeModel()
service.create_thread('dm-1', ai_consent=False)
service.add_message('dm-1', 'user', '契約について質問があります')
try:
service.request_draft('dm-1', model)
raise AssertionError('must reject without consent')
except PermissionError:
pass
assert model.calls == 0
# 2. 新着メッセージが来たら古い下書きを拒否する
db = new_db()
service = DmService(db)
model = FakeModel()
service.create_thread('dm-2', ai_consent=True)
service.add_message('dm-2', 'user', '青を希望します')
draft_id = service.request_draft('dm-2', model)
service.handoff('dm-2', '返信内容の確認')
service.claim('dm-2', 'staff-a')
service.add_message('dm-2', 'user', '訂正します。赤を希望します')
try:
service.approve(draft_id, 'staff-a', '青で承りました')
raise AssertionError('stale draft must be rejected')
except ValueError as error:
assert 'conversation changed' in str(error)
# 3. 担当者が確認した下書きだけを配送する
db = new_db()
service = DmService(db)
adapter = FakeMessageAdapter()
service.create_thread('dm-3', ai_consent=True)
service.add_message('dm-3', 'user', '営業時間を教えてください')
draft_id = service.request_draft('dm-3', FakeModel())
service.handoff('dm-3', 'AI下書きの有人確認')
service.claim('dm-3', 'staff-b')
service.approve(
draft_id,
'staff-b',
'担当者が確認しました。営業時間は案内ページをご確認ください。',
)
service.deliver_pending(adapter)
service.deliver_pending(adapter)
assert len(adapter.sent) == 1
print('all checks passed')
if __name__ == '__main__':
run_checks()
実行します。
python demo.py
成功時は次のように表示されます。
all checks passed
手順3：実際のDMへ接続する位置を限定する
FakeMessageAdapterは検証用です。本番ではMessageAdapterを実装し、send_textの内部だけをTencent RTCを利用するメッセージ送信処理へ置き換えます。
重要なのは、チャットイベントを受信したクライアントからLLMを直接呼ばないことです。サーバー側で次を確認してからrequest_draftへ進めます。
- 認証済みのユーザーとスレッドが一致している
- AI利用への同意が現在も有効である
- モデレーション結果が
normalである - LLMへ渡してよいメッセージ種別である
- 保持件数や保持期間のポリシーを満たしている
配送時も、LLMからMessageAdapterを直接呼ばせません。draftsとoutboxを経由することで、「生成された」「承認された」「配送対象になった」を別々に確認できます。
手順4：翻訳を加える場合も、原文と責任を消さない
多言語DMでは、TUIChatにオンデマンドのテキストメッセージ翻訳を追加できます。ただし、対応するコンテンツ種別、言語、エディション上の条件は公式ドキュメントで確認してください。
翻訳をAI下書きと組み合わせる場合は、次のデータを分けて保持します。
{
"messageId": "m-42",
"originalText": "ユーザーが送った原文",
"displayTranslation": "画面表示用の翻訳",
"draftSource": "original",
"translationRequestedByUser": true
}
画面上の翻訳文だけを原文として保存すると、誤訳が発生した際に担当者が検証できません。また、翻訳を表示することへの操作と、翻訳文をLLMへ渡す同意も同一とは限りません。サービスの同意画面で対象を分けて説明します。
確認方法：正常な返信より「責任が消える境界」を試す
実装後は、文章の自然さだけでなく次の項目を確認します。
1. AI利用への同意を撤回する
- 撤回後の新着メッセージがLLMへ送られない
- 既存の未承認下書きを破棄できる
- 有人対応は同じDMスレッドで継続できる
2. 下書き生成直後にユーザーが訂正する
- 古い
context_upto_message_idの下書きを承認できない - 担当者へ再生成または手動返信を促す
- 「青」から「赤」への訂正を古い返信で上書きしない
3. 2人の担当者が同時に引き受ける
- 先に確定した担当者だけが
owner_idになる - 後から操作した担当者には競合を表示する
- AI用キューとは別に担当者不明のキューを作らない
4. センシティブ分類を入力する
- 本文をLLMへ送る前に処理が止まる
- スレッドには有人切り替えが表示される
- 切り替え理由へ生の個人情報をコピーしない
5. 配送処理をタイムアウトさせる
- 承認済みデータが失われない
- 配送結果が不明な場合、無条件に再送しない
- メッセージ基盤側の配送結果と照合できる運用を用意する
最後のケースは特に重要です。サンプルのFakeMessageAdapterは冪等キーで重複を抑えますが、これはTencent RTCの特定API仕様を示すものではありません。本番では、採用する送信方法の公式仕様に合わせて、重複制御、配送結果、再試行方法を設計してください。
判断フレームワーク：どこまでAIへ任せるか
導入段階ごとの目安は次の通りです。
| 段階 | AIの役割 | 人の役割 | 主なコスト |
|---|---|---|---|
| 0 | 利用しない | 全文を作成・送信 | 応答負荷が高い |
| 1 | 要約・下書き | 全件確認して送信 | レビュー待ちが発生 |
| 2 | 低リスク分類の下書き | 例外確認と監査 | 分類ルールの保守が必要 |
| 3 | 限定条件で自動送信 | 事後監査と即時停止 | 誤送信、説明責任、復旧設計が重い |
最初に確認すべき指標は「AIが何件返信したか」ではありません。
- 同意なしでLLMが呼ばれた件数
- 新着後に古い下書きが拒否された件数
- 担当者未設定のまま滞留しているスレッド数
- AI下書きから有人が大幅修正した理由
- 配送結果を確定できないOutboxの件数
これらが観測できなければ、自動化を増やしても、失敗が速く流れるだけです。
注意点とトレードオフ
文脈を減らすと精度が落ち、増やすとプライバシー負担が増える
サンプルでは直近8件に限定しましたが、8件が普遍的な正解という意味ではありません。会話の長さ、必要な参照範囲、データ保持方針に応じて決め、LLMへ渡したメッセージIDを監査可能にします。
単一担当者は競合を減らすが、休止時の引き継ぎが必要
owner_idを固定するだけでは、担当者の離席時に滞留します。実運用では、担当期限、明示的な担当解除、再割り当て履歴を追加してください。無言で所有権を奪う実装は避けます。
AIの下書き表示を隠さない
運営画面では、AI生成文と担当者の編集後テキストを区別します。ユーザー側にも、サービスの方針に応じてAI支援の利用と有人確認の有無を分かる形で提示します。
モデレーションと有人判断は代替関係ではない
分類器がnormalと返しても、事実確認や例外判断が必要な場合があります。分類結果は送信許可ではなく、ワークフロー選択の入力として扱います。
AIエージェントが高速にメッセージを作れることと、ユーザーとのDMを安心して任せられることは別です。速度を抑えるのではなく、同意、文脈、担当者、送信権を分離することで、速さの責任を人が取り戻すのがこの設計の狙いです。
関係性の開示: 筆者はTencent RTCのコンテンツ制作に関与しています。本記事の製品仕様に関する記述は、Tencent RTCの公式Social MessagingソリューションおよびTUIChat公式ドキュメントを実装上の参照資料として使用しています。