AIチャットUIは短いコードで作れます。しかし実運用で難しいのは、入力欄や吹き出しではありません。
- ユーザーは既存チャットとAI画面のどちらへ投稿すべきか
- AIが誤ったとき、誰が返信を引き取るのか
- グループ内の発言を、どこまでAIへ渡してよいのか
- 担当者が確認していないAI回答を、誰が送信したことにするのか
ここを決めずにStreamlit製チャットを追加すると、便利なデモが「誰も所有していない新しい窓口」になりがちです。
本稿では発想を反転し、ユーザー向けチャットはTencent RTC Social Messaging側に残し、Streamlitは運営者専用の承認キューにする構成を作ります。
結論：AIには返信案を作らせ、送信権と案件所有権は人に残す
今回の実装では、次のルールを固定します。
| 状況 | AIへ渡す範囲 | 最終送信者 |
|---|---|---|
| 通常メッセージ | 渡さない | なし |
/ai 質問 |
現在の発言だけ | 担当者 |
/ai+ 質問 |
現在の発言＋同じ送信者の直近発言 | 担当者 |
/human 相談 |
渡さない | 担当者 |
| 秘密情報らしき文字列を含む | 渡さない | 担当者 |
| 画像などテキスト以外 | 渡さない | 担当者 |
AIが実証できるのは、渡された文脈から返信案を生成できることまでです。回答の正しさ、投稿者の意図、組織として実行してよい操作までは保証しません。
そのため、LLMの性能評価より先に以下を実装します。
- ユーザーによる明示的なAI利用
- AIへ渡す文脈の制限
- 秘密情報の簡易ゲート
- 重複イベントの排除
- 担当者による案件の取得
- 編集・承認後だけ送信する経路
前提：ユーザーの入口と運営画面を分離する
Tencent RTCのSocial Messagingソリューションは、1対1チャット、グループ、コミュニティ、リッチメディア、ライブ配信ルーム内チャットなどのシナリオを扱います。
本稿では製品固有の未確認API名を仮定せず、Social Messagingを利用するアプリケーションサーバーから、受信イベントを次の共通形式へ正規化する境界を置きます。
{
"event_id": "evt-001",
"conversation_id": "group-42",
"sender_id": "user-7",
"content_type": "text",
"text": "/ai 配送先を変更できますか"
}
実装する全体フローは次のとおりです。
Social Messagingのチャット
↓ 受信イベントを正規化
FastAPIゲートウェイ
├─ 通常投稿 → AI処理なし
├─ 秘密情報・非テキスト → 有人キュー
└─ /ai、/ai+ → AI返信案を生成
↓
SQLite承認キュー
↓
Streamlit運営画面
↓ 取得・編集・承認
送信アダプター
↓
Social Messagingの元の会話へ返信
Streamlitは新しい問い合わせ窓口ではなく、既存チャットの裏側にある運営ツールです。
手順1：検証環境を作る
Python 3.11以降を前提にします。
mkdir chat-review-queue
cd chat-review-queue
python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn streamlit pydantic
作成するファイルは2つです。
chat-review-queue/
├── app.py # 受信、ポリシー判定、返信案生成
└── console.py # 担当者向け承認キュー
コード1：チャット受信とAI返信案を分離する
app.pyを作成します。
from datetime import datetime, timezone
import json
import sqlite3
import uuid
from fastapi import FastAPI
from pydantic import BaseModel
DB = "chat.db"
app = FastAPI()
def now() -> str:
return datetime.now(timezone.utc).isoformat()
def connect() -> sqlite3.Connection:
con = sqlite3.connect(DB)
con.row_factory = sqlite3.Row
return con
def init_db() -> None:
with connect() as con:
con.executescript("""
CREATE TABLE IF NOT EXISTS messages (
seq INTEGER PRIMARY KEY AUTOINCREMENT,
event_id TEXT NOT NULL UNIQUE,
conversation_id TEXT NOT NULL,
sender_id TEXT NOT NULL,
content_type TEXT NOT NULL,
text TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS cases (
id TEXT PRIMARY KEY,
event_id TEXT NOT NULL UNIQUE,
conversation_id TEXT NOT NULL,
status TEXT NOT NULL,
reason TEXT NOT NULL,
owner TEXT,
context_json TEXT NOT NULL,
draft TEXT,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS outbox (
id TEXT PRIMARY KEY,
conversation_id TEXT NOT NULL,
body TEXT NOT NULL,
state TEXT NOT NULL,
created_at TEXT NOT NULL
);
""")
init_db()
class IncomingMessage(BaseModel):
event_id: str
conversation_id: str
sender_id: str
content_type: str
text: str = ""
# 本番のモデレーション製品ではなく、AIへ送らないための最小ゲート。
SENSITIVE_MARKERS = (
"パスワード",
"秘密鍵",
"クレジットカード",
"access_token",
"BEGIN PRIVATE KEY",
)
def has_sensitive_marker(text: str) -> bool:
lowered = text.lower()
return any(marker.lower() in lowered for marker in SENSITIVE_MARKERS)
def create_draft(context: list[dict]) -> str:
"""再現用の決定的なダミー。ここだけを任意のLLM呼び出しに交換する。"""
question = context[-1]["text"]
return (
f"【AI返信案】「{question}」について確認しました。"
"確定情報が不足しているため、担当者が条件を確認して回答します。"
)
def enqueue_notice(con: sqlite3.Connection, conversation_id: str, body: str) -> None:
con.execute(
"INSERT INTO outbox VALUES (?, ?, ?, 'pending', ?)",
(str(uuid.uuid4()), conversation_id, body, now()),
)
@app.post("/events/message")
def receive(msg: IncomingMessage):
with connect() as con:
inserted = con.execute(
"""
INSERT OR IGNORE INTO messages
(event_id, conversation_id, sender_id, content_type, text, created_at)
VALUES (?, ?, ?, ?, ?, ?)
""",
(
msg.event_id,
msg.conversation_id,
msg.sender_id,
msg.content_type,
msg.text,
now(),
),
)
if inserted.rowcount == 0:
return {"result": "duplicate"}
current_seq = con.execute(
"SELECT seq FROM messages WHERE event_id = ?",
(msg.event_id,),
).fetchone()["seq"]
if msg.content_type != "text":
mode, question, reason = "human", "", "non_text"
elif msg.text.startswith("/human "):
mode, question, reason = "human", msg.text[7:].strip(), "requested_human"
elif msg.text.startswith("/ai+ "):
mode, question, reason = "ai_context", msg.text[5:].strip(), "ai_opt_in_context"
elif msg.text.startswith("/ai "):
mode, question, reason = "ai_single", msg.text[4:].strip(), "ai_opt_in_single"
else:
return {"result": "ignored"}
if has_sensitive_marker(question):
mode, reason = "human", "sensitive_marker"
context: list[dict] = []
if mode == "ai_context":
# グループ内の他ユーザーの発言は含めない。
rows = con.execute(
"""
SELECT sender_id, text
FROM messages
WHERE conversation_id = ?
AND sender_id = ?
AND content_type = 'text'
AND seq < ?
ORDER BY seq DESC
LIMIT 6
""",
(msg.conversation_id, msg.sender_id, current_seq),
).fetchall()
context.extend(
{"sender_id": row["sender_id"], "text": row["text"]}
for row in reversed(rows)
)
if question:
context.append({"sender_id": msg.sender_id, "text": question})
draft = create_draft(context) if mode.startswith("ai") and context else None
status = "awaiting_review" if draft else "needs_human"
case_id = str(uuid.uuid4())
con.execute(
"""
INSERT INTO cases
(id, event_id, conversation_id, status, reason, owner,
context_json, draft, created_at)
VALUES (?, ?, ?, ?, ?, NULL, ?, ?, ?)
""",
(
case_id,
msg.event_id,
msg.conversation_id,
status,
reason,
json.dumps(context, ensure_ascii=False),
draft,
now(),
),
)
enqueue_notice(
con,
msg.conversation_id,
"担当者が内容を確認します。AIの返信案は、承認されるまで送信されません。",
)
return {"result": "queued", "case_id": case_id, "status": status}
@app.get("/outbox")
def list_outbox():
with connect() as con:
rows = con.execute(
"SELECT * FROM outbox ORDER BY created_at"
).fetchall()
return [dict(row) for row in rows]
この例のcreate_draft()は、ワークフローを再現するためのダミーです。LangChainや任意のLLM SDKへ差し替える場合も、モデルにチャット送信権を渡さず、文字列の返信案を返すだけの関数として閉じ込めます。
また、SENSITIVE_MARKERSは本格的なモデレーションではありません。秘密情報らしき入力をLLMへ送らないための、保守的な検証用ゲートです。
コード2：Streamlitを担当者の承認キューにする
console.pyを作成します。
import os
import sqlite3
import uuid
from datetime import datetime, timezone
import streamlit as st
DB = "chat.db"
OPERATOR = os.getenv("OPERATOR", "operator@example.com")
def now() -> str:
return datetime.now(timezone.utc).isoformat()
def connect() -> sqlite3.Connection:
con = sqlite3.connect(DB)
con.row_factory = sqlite3.Row
return con
st.set_page_config(page_title="Chat review queue", layout="wide")
st.title("チャット返信・有人引き継ぎキュー")
st.caption(f"担当者: {OPERATOR}")
with connect() as con:
cases = con.execute(
"""
SELECT * FROM cases
WHERE status IN ('awaiting_review', 'needs_human', 'claimed')
ORDER BY created_at
"""
).fetchall()
if not cases:
st.success("未処理案件はありません")
for case in cases:
with st.container(border=True):
st.write({
"case_id": case["id"],
"conversation_id": case["conversation_id"],
"status": case["status"],
"reason": case["reason"],
"owner": case["owner"],
})
st.code(case["context_json"], language="json")
if case["owner"] is None:
if st.button("この案件を取得", key=f"claim-{case['id']}"):
with connect() as con:
con.execute(
"""
UPDATE cases
SET owner = ?, status = 'claimed'
WHERE id = ? AND owner IS NULL
""",
(OPERATOR, case["id"]),
)
st.rerun()
continue
if case["owner"] != OPERATOR:
st.info(f"{case['owner']} が対応中です")
continue
reply = st.text_area(
"送信内容",
value=case["draft"] or "",
key=f"reply-{case['id']}",
)
col1, col2 = st.columns(2)
with col1:
if st.button("承認して送信待ちへ", key=f"send-{case['id']}"):
if not reply.strip():
st.error("返信本文が空です")
else:
with connect() as con:
con.execute(
"""
INSERT INTO outbox
(id, conversation_id, body, state, created_at)
VALUES (?, ?, ?, 'pending', ?)
""",
(str(uuid.uuid4()), case["conversation_id"], reply, now()),
)
con.execute(
"UPDATE cases SET status = 'approved' WHERE id = ?",
(case["id"],),
)
st.rerun()
with col2:
if st.button("回答せず終了", key=f"close-{case['id']}"):
with connect() as con:
con.execute(
"UPDATE cases SET status = 'closed' WHERE id = ?",
(case["id"],),
)
st.rerun()
起動します。
uvicorn app:app --reload --port 8000
別のターミナルで運営画面を起動します。
source .venv/bin/activate
OPERATOR=alice@example.com streamlit run console.py
手順3：Tencent RTCとの接続境界を決める
検証コードでは、送信予定メッセージをoutboxへ保存しています。本番では次の2か所だけをSocial Messagingのアプリケーション統合へ置き換えます。
- 受信イベントを
IncomingMessage形式へ正規化し、/events/messageへ渡す -
outbox.state = 'pending'のレコードを、元のconversation_idへ送信する
認証情報や送信処理をStreamlitへ置かないことが重要です。Streamlitは取得・編集・承認だけを担当し、実際の送信はサーバー側アダプターに限定します。
送信に成功したらoutbox.stateをsentへ、失敗したらfailedへ更新します。再試行時はoutbox.idを冪等キーとして扱い、同じ返信を二重送信しない設計にします。
確認方法：5つの入力で制御経路を検証する
1. 通常発言はAIへ渡らない
curl -s http://localhost:8000/events/message \
-H 'content-type: application/json' \
-d '{
"event_id":"evt-001",
"conversation_id":"group-42",
"sender_id":"user-7",
"content_type":"text",
"text":"こんにちは"
}'
期待結果はignoredです。AIを会話へ追加したからといって、全発言を自動的に収集しません。
2. 明示された1件だけAI返信案になる
curl -s http://localhost:8000/events/message \
-H 'content-type: application/json' \
-d '{
"event_id":"evt-002",
"conversation_id":"group-42",
"sender_id":"user-7",
"content_type":"text",
"text":"/ai 配送先を変更できますか"
}'
Streamlitにawaiting_reviewの案件が表示されます。返信案はまだユーザーへ送られていません。
3. 同一イベントを再送しても案件を増やさない
同じevt-002をもう一度POSTします。期待結果は次です。
{"result":"duplicate"}
チャット基盤やアプリケーションサーバーの再送を、AIへの二重依頼や二重返信に変換しないための確認です。
4. 秘密情報らしき入力はAIへ渡さない
curl -s http://localhost:8000/events/message \
-H 'content-type: application/json' \
-d '{
"event_id":"evt-003",
"conversation_id":"group-42",
"sender_id":"user-7",
"content_type":"text",
"text":"/ai access_tokenを貼ったので調べて"
}'
statusはneeds_human、reasonはsensitive_markerとなり、draftは生成されません。
5. 案件を取得した担当者だけが承認する
Streamlitで案件を取得し、返信文を編集して「承認して送信待ちへ」を押します。その後、次を確認します。
curl -s http://localhost:8000/outbox
担当者が確定した本文だけがpendingとして追加されていれば成功です。
翻訳を入れるなら、AI文脈とは別の操作にする
多言語チャットでは、翻訳文をそのままAIの入力へ使いたくなります。しかし、自動翻訳とAI回答を一体化すると、誤訳なのか回答ミスなのかを切り分けにくくなります。
TUIChatにはテキストメッセージをオンデマンド翻訳する方法が用意されています。対応するコンテンツ形式、言語、エディション上の条件は、実装時点の公式ドキュメントを確認してください。
運用上は次の分離が安全です。
- ユーザー向け翻訳：TUIChat上で明示操作する
- AIへの入力：原文を保存し、必要なら翻訳文を別フィールドにする
- 担当者画面：原文と翻訳文を併記する
- 最終返信：担当者が送信言語を確認する
翻訳文だけを保存すると、後から判断根拠を追跡できません。
注意点とトレードオフ
Streamlit自体を認証境界にしない
サンプルでは環境変数で担当者名を指定していますが、本番の本人確認にはなりません。社内SSO、リバースプロキシ、アクセス制御を組み合わせ、運営画面を一般公開しないでください。
人の承認は回答速度と引き換えになる
全件承認は安全側ですが、担当者数を超えるとキューが滞留します。自動送信へ急いで移行するのではなく、まず以下を計測します。
- 未取得案件の件数と最古作成時刻
- AI返信案がそのまま承認された割合
- 編集された項目
- 有人へ切り替えた理由
- 送信失敗と再試行回数
十分に限定された質問だけを後から自動化候補にする方が、最初から万能ボットを目指すより判断しやすくなります。
簡易キーワード判定を安全機能と呼ばない
表記揺れ、画像内文字列、難読化、文脈依存の攻撃は検出できません。本番では利用目的に合ったモデレーション、添付ファイル処理、監査、保持期間、削除手続きを別途設計してください。
/ai+でもグループ全体を渡さない
今回のコードは同じ送信者の直近6件だけを取得します。他の参加者がAI利用へ同意したとは限らないためです。グループ全体の文脈が必要なら、参加者への表示、同意単位、保持期間、退出後の扱いを先に決めます。
最終チェックリスト
- AIを使う操作がユーザーから見える
- 通常メッセージを無条件でLLMへ送っていない
- 非テキストと秘密情報候補を有人経路へ送れる
- 同じ受信イベントから案件が重複生成されない
- AIモデルは返信案しか返せない
- 案件に担当者が記録される
- 承認前の返信案がチャットへ送信されない
- 送信処理がStreamlitの外にある
- ユーザーに有人確認中であることが見える
- 翻訳文と原文を区別して追跡できる
短いAIチャットUIは、アイデアを触るためには有効です。一方、実運用へ進める際に必要なのは画面の豪華さではなく、誰がAIを呼び、何を渡し、誰が案件を所有し、誰が送信を確定したかを追跡できることです。
Streamlitをユーザー向けの新しい窓口ではなく、人が制御を取り戻すための運営キューとして使うと、デモから運用への境界を明確にできます。
関係性の開示: 筆者はTencent RTCのコンテンツ制作に関係しています。本稿の製品情報と実装上の境界確認には、Tencent RTCの公式ドキュメントを参照しました。