音声AIのデモは、質問に滑らかに答えられるだけで成立します。しかし問い合わせ窓口として公開すると、すぐに別の難しさが出てきます。
- ユーザーが遮ったのに、古い回答を読み上げ続ける
- STTの誤認識を前提にLLMが回答してしまう
- 課金やアカウント固有の質問までAIが断定する
- 「担当者に確認します」と言った後、誰の受信箱にも残らない
個人開発者にとっての緊張点は、AIをどこまで賢くするかではありません。リアルタイム会話の気持ちよさを保ちながら、回答責任を人間へ戻せるかです。
この記事では、Tencent RTC Conversational AIを音声経路として使い、Python側で会話状態と問い合わせチケットを管理します。ゴールは、音声受付から人間の返信がアプリ内へ戻るところまでを再現可能にすることです。
結論
実運用のAI音声問い合わせでは、次の4点をアプリケーション側の責務として固定します。
- ユーザーが話し始めたら、再生中の音声を止める
- 各発話に世代番号を付け、古いLLM結果を読み上げない
- アカウント固有・課金・安全性に関わる質問はチケット化する
- チケットを唯一の受信箱とし、人間の返信を既存アプリへ戻す
RTC、STT、LLM、TTSはそれぞれ交換可能な処理です。会話の継続可否や人間への引き継ぎ判断まで、LLMに委ねないことが重要です。
ユーザー音声
↓
RTC ─→ STT ─→ 会話制御 ─→ LLM ─→ TTS ─→ RTC
│
├─ 古い世代の結果を破棄
├─ 明示的な引き継ぎ規則
↓
問い合わせDB
↓
運営者が確認・返信
↓
既存アプリの受信箱
前提
この記事では次を前提にします。
- Python 3.11以上
- ログイン済みユーザーを識別できるアプリ
- SQLiteを使った最小構成
- Tencent RTC Conversational AIの音声セッションを利用する
- 人間の返信先はメールではなく、既存アプリ内の問い合わせ画面とする
Tencent RTC Conversational AIは、RTCによる音声経路とSTT、LLM、TTSを組み合わせたリアルタイム音声対話を扱います。対応する構成要素と導入の全体像は公式ドキュメントを参照してください。
LLM設定ではOpenAI互換モデルやエージェント基盤との接続、リクエスト識別子を使ったルーティング・追跡が説明されています。本稿では、識別子を音声セッションIDと発話世代から組み立てます。
なお、以下に登場する stop_playback、generate、on_transcript は、責務を説明するために定義するアプリ側のアダプター名です。Tencent RTCの実在API名を表すものではありません。実際の音声セッション接続は上記の公式ドキュメントに従ってください。
最初に決める「AIが答えない条件」
LLMの精度評価より先に、人間へ渡す条件を決めます。
| 条件 | 処理 | 理由 |
|---|---|---|
| 公開FAQで回答できる | AIが回答候補を生成 | 個別判断が不要 |
| ユーザーが担当者を希望 | 即時チケット化 | AIが説得して引き留めない |
| 課金、返金、退会、本人確認 | 即時チケット化 | アカウント固有の判断が必要 |
| STT失敗が連続 | チケット化または文字入力へ切り替え | 誤認識をLLMで補完しない |
| LLMがタイムアウト、不正形式 | チケット化 | 無言や推測回答を避ける |
| 確認質問が上限を超えた | チケット化 | 会話ループを止める |
ここでのAIは、公開情報の説明と問い合わせ要約を補助できます。一方、「このユーザーへ返金してよいか」「本人の契約がどうなっているか」は、人間または権限を持つ業務システムの判断です。
データモデル
会話ログを別の受信箱にしないため、対応が必要になった時点で必ず tickets に集約します。
-- schema.sql
CREATE TABLE IF NOT EXISTS tickets (
id TEXT PRIMARY KEY,
user_id TEXT NOT NULL,
rtc_session_id TEXT NOT NULL,
status TEXT NOT NULL CHECK (
status IN ('waiting_human', 'replied', 'closed')
),
summary TEXT NOT NULL,
llm_request_id TEXT,
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS messages (
id INTEGER PRIMARY KEY AUTOINCREMENT,
ticket_id TEXT NOT NULL REFERENCES tickets(id),
sender TEXT NOT NULL CHECK (sender IN ('user', 'assistant', 'operator')),
body TEXT NOT NULL,
created_at TEXT NOT NULL
);
CREATE TABLE IF NOT EXISTS voice_events (
event_id TEXT PRIMARY KEY,
rtc_session_id TEXT NOT NULL,
turn_generation INTEGER NOT NULL,
event_type TEXT NOT NULL,
created_at TEXT NOT NULL
);
voice_events.event_id を主キーにするのは、ネットワーク再接続などで同じイベントを再受信しても二重処理しないためです。
全文の音声認識結果を長期間保存する必要はありません。問い合わせとして必要な発話だけを messages に残し、デバッグ用イベントには本文を入れない構成にしています。
手順1：プロジェクトを作る
mkdir voice-support
cd voice-support
python -m venv .venv
source .venv/bin/activate
pip install fastapi uvicorn pytest pytest-asyncio
sqlite3 support.db < schema.sql
構成は次のようにします。
voice-support/
├── schema.sql
├── voice_session.py
├── ticket_store.py
├── api.py
└── tests/
└── test_voice_session.py
手順2：割り込みを世代番号で扱う
ユーザーが話し始めた瞬間に generation を進めます。LLMやTTSの処理が完了しても、開始時と世代が違えば結果を破棄します。
# voice_session.py
import asyncio
from dataclasses import dataclass
from typing import Protocol
class LLMAdapter(Protocol):
async def generate(self, text: str, request_id: str) -> dict: ...
class AudioAdapter(Protocol):
async def stop_playback(self) -> None: ...
async def speak(self, text: str) -> None: ...
class TicketStore(Protocol):
def create(
self,
*,
user_id: str,
rtc_session_id: str,
summary: str,
request_id: str | None,
) -> str: ...
@dataclass
class VoiceSession:
user_id: str
session_id: str
llm: LLMAdapter
audio: AudioAdapter
tickets: TicketStore
generation: int = 0
recognition_failures: int = 0
clarification_count: int = 0
HUMAN_KEYWORDS = (
"担当者",
"人に代わって",
"返金",
"課金",
"退会",
"本人確認",
"不正利用",
)
async def on_user_speech_started(self) -> None:
# 進行中のLLM/TTS結果を論理的に無効化する
self.generation += 1
await self.audio.stop_playback()
async def on_recognition_failed(self) -> str | None:
self.recognition_failures += 1
if self.recognition_failures < 2:
await self.audio.speak(
"うまく聞き取れませんでした。もう一度お願いします。"
)
return None
return await self._handoff(
"音声を繰り返し認識できなかったため、確認が必要です。",
request_id=None,
)
async def on_transcript(self, text: str) -> str | None:
current_generation = self.generation
normalized = text.strip()
if not normalized:
return await self.on_recognition_failed()
self.recognition_failures = 0
request_id = f"{self.session_id}:{current_generation}"
if any(word in normalized for word in self.HUMAN_KEYWORDS):
return await self._handoff(normalized, request_id)
try:
result = await asyncio.wait_for(
self.llm.generate(normalized, request_id),
timeout=8.0,
)
except (TimeoutError, asyncio.TimeoutError, ValueError):
return await self._handoff(normalized, request_id)
# 待機中にユーザーが次の発話を始めた場合、古い結果は読まない
if current_generation != self.generation:
return None
action = result.get("action")
message = result.get("message")
if action == "handoff":
summary = result.get("summary") or normalized
return await self._handoff(summary, request_id)
if action not in {"answer", "clarify"} or not isinstance(message, str):
return await self._handoff(normalized, request_id)
if action == "clarify":
self.clarification_count += 1
if self.clarification_count > 2:
return await self._handoff(normalized, request_id)
# speak直前にも世代を再確認する
if current_generation == self.generation:
await self.audio.speak(message)
return None
async def _handoff(
self,
summary: str,
request_id: str | None,
) -> str:
# 音声案内より先に保存する。TTS失敗でも問い合わせを失わない。
ticket_id = self.tickets.create(
user_id=self.user_id,
rtc_session_id=self.session_id,
summary=summary,
request_id=request_id,
)
try:
await self.audio.speak(
"担当者への問い合わせとして受け付けました。"
"返信はアプリの問い合わせ画面に表示します。"
)
except Exception:
pass
return ticket_id
割り込み時に実行中タスクをキャンセルするだけでは不十分です。外部LLMへのリクエストがすでに送信済みの場合、キャンセル後も結果が戻ることがあります。世代番号を確認すれば、遅れて届いた結果を読み上げずに済みます。
手順3：チケットを先に保存する
# ticket_store.py
import sqlite3
import uuid
from datetime import datetime, timezone
def now() -> str:
return datetime.now(timezone.utc).isoformat()
class SQLiteTicketStore:
def __init__(self, path: str = "support.db"):
self.path = path
def connect(self):
return sqlite3.connect(self.path)
def create(
self,
*,
user_id: str,
rtc_session_id: str,
summary: str,
request_id: str | None,
) -> str:
ticket_id = str(uuid.uuid4())
timestamp = now()
with self.connect() as db:
db.execute(
"""
INSERT INTO tickets (
id, user_id, rtc_session_id, status, summary,
llm_request_id, created_at, updated_at
) VALUES (?, ?, ?, 'waiting_human', ?, ?, ?, ?)
""",
(
ticket_id,
user_id,
rtc_session_id,
summary,
request_id,
timestamp,
timestamp,
),
)
db.execute(
"""
INSERT INTO messages (ticket_id, sender, body, created_at)
VALUES (?, 'user', ?, ?)
""",
(ticket_id, summary, timestamp),
)
return ticket_id
def reply(self, ticket_id: str, body: str) -> None:
timestamp = now()
with self.connect() as db:
ticket = db.execute(
"SELECT status FROM tickets WHERE id = ?",
(ticket_id,),
).fetchone()
if ticket is None:
raise KeyError(ticket_id)
if ticket[0] == "closed":
raise ValueError("closed ticket cannot be replied")
db.execute(
"""
INSERT INTO messages (ticket_id, sender, body, created_at)
VALUES (?, 'operator', ?, ?)
""",
(ticket_id, body, timestamp),
)
db.execute(
"""
UPDATE tickets
SET status = 'replied', updated_at = ?
WHERE id = ?
""",
(timestamp, ticket_id),
)
保存と音声案内の順番を逆にすると、TTS障害時に「担当者へ渡す予定だった問い合わせ」が消えます。先にチケットを確定し、案内は失敗してもよい副作用として扱います。
手順4：人間の返信を既存アプリへ戻す
最小APIでは、運営者の返信とユーザーの取得経路を分けます。
# api.py
import os
import secrets
import sqlite3
from fastapi import FastAPI, Header, HTTPException
from pydantic import BaseModel, Field
from ticket_store import SQLiteTicketStore
app = FastAPI()
store = SQLiteTicketStore()
ADMIN_TOKEN = os.environ.get("ADMIN_TOKEN", "")
class ReplyBody(BaseModel):
body: str = Field(min_length=1, max_length=4000)
def require_admin(authorization: str | None) -> None:
expected = f"Bearer {ADMIN_TOKEN}"
if not ADMIN_TOKEN or not authorization:
raise HTTPException(status_code=401)
if not secrets.compare_digest(authorization, expected):
raise HTTPException(status_code=403)
@app.post("/operator/tickets/{ticket_id}/reply")
def reply_ticket(
ticket_id: str,
payload: ReplyBody,
authorization: str | None = Header(default=None),
):
require_admin(authorization)
try:
store.reply(ticket_id, payload.body)
except KeyError:
raise HTTPException(status_code=404)
except ValueError as exc:
raise HTTPException(status_code=409, detail=str(exc))
return {"status": "replied"}
@app.get("/me/tickets/{ticket_id}")
def get_ticket(
ticket_id: str,
x_user_id: str = Header(),
):
# X-User-IDはローカル再現用。本番では既存アプリの認証結果を使う。
with sqlite3.connect("support.db") as db:
ticket = db.execute(
"""
SELECT id, status, summary, created_at, updated_at
FROM tickets
WHERE id = ? AND user_id = ?
""",
(ticket_id, x_user_id),
).fetchone()
if ticket is None:
raise HTTPException(status_code=404)
messages = db.execute(
"""
SELECT sender, body, created_at
FROM messages
WHERE ticket_id = ?
ORDER BY id
""",
(ticket_id,),
).fetchall()
return {
"id": ticket[0],
"status": ticket[1],
"summary": ticket[2],
"created_at": ticket[3],
"updated_at": ticket[4],
"messages": [
{"sender": row[0], "body": row[1], "created_at": row[2]}
for row in messages
],
}
起動します。
export ADMIN_TOKEN="$(python -c 'import secrets; print(secrets.token_urlsafe(32))')"
uvicorn api:app --reload
運営者は、問い合わせの要約だけで判断せず、必要ならユーザーへ追加確認してから返信します。
curl -X POST http://localhost:8000/operator/tickets/TICKET_ID/reply \
-H "Authorization: Bearer $ADMIN_TOKEN" \
-H "Content-Type: application/json" \
-d '{"body":"状況を確認しました。設定画面から再度お試しください。"}'
ユーザー側は既存の認証済み画面から取得します。
curl http://localhost:8000/me/tickets/TICKET_ID \
-H "X-User-ID: user-123"
これにより、音声AIの管理画面を新しい受信箱にせず、waiting_human のチケットだけを運営対象にできます。
手順5：LLMへ渡す出力契約を限定する
LLMには自由文だけでなく、次の3種類の処理結果を要求します。
{
"action": "answer | clarify | handoff",
"message": "ユーザーへ読み上げる文。handoffでは空でもよい",
"summary": "人間へ渡す短い要約"
}
ただし、LLMが answer を返したから読み上げるのではありません。処理順は次のとおりです。
- アプリ側の禁止条件を検査する
- LLMを呼ぶ
- JSON形式を検査する
- 発話世代が変わっていないか確認する
- 許可された公開FAQだけを読み上げる
request_id には rtc_session_id:generation を使います。これにより、音声セッション、STT結果、LLM処理、チケットを後から関連付けられます。LLMプロバイダーの具体的な接続設定は、公式のLarge Language Model configurationに合わせて行います。
確認方法
1. 古いLLM回答が読み上げられないこと
遅い偽LLMを使って、結果が返る前に次の発話を開始します。
# tests/test_voice_session.py
import asyncio
import pytest
from voice_session import VoiceSession
class SlowLLM:
async def generate(self, text, request_id):
await asyncio.sleep(0.1)
return {"action": "answer", "message": "古い回答です"}
class FakeAudio:
def __init__(self):
self.spoken = []
self.stop_count = 0
async def stop_playback(self):
self.stop_count += 1
async def speak(self, text):
self.spoken.append(text)
class FakeTickets:
def create(self, **kwargs):
return "ticket-1"
@pytest.mark.asyncio
async def test_stale_llm_result_is_not_spoken():
audio = FakeAudio()
session = VoiceSession(
user_id="user-123",
session_id="rtc-abc",
llm=SlowLLM(),
audio=audio,
tickets=FakeTickets(),
)
await session.on_user_speech_started()
task = asyncio.create_task(session.on_transcript("使い方を教えて"))
await asyncio.sleep(0.02)
await session.on_user_speech_started()
await task
assert "古い回答です" not in audio.spoken
assert audio.stop_count == 2
pytest -q
2. TTSが失敗しても問い合わせが残ること
AudioAdapter.speak() が例外を投げる偽物へ差し替え、tickets.create() が呼ばれたことを確認します。期待する順番は次です。
チケット保存 → TTS案内を試行 → TTS失敗を無視 → ticket_idを返す
3. 人間への明示的な依頼をLLMへ送らないこと
「担当者に代わってください」を入力し、LLMアダプターの呼び出し回数が0、チケット件数が1になることを確認します。
4. 別ユーザーが返信を読めないこと
curl -i http://localhost:8000/me/tickets/TICKET_ID \
-H "X-User-ID: another-user"
404 になることを確認します。本番では X-User-ID を信用せず、既存アプリのセッションや署名済みトークンからユーザーIDを取得してください。
5. 遅延を工程別に測ること
「会話が遅い」という一つの値だけでは原因を判断できません。少なくとも以下を別々に記録します。
speech_end_at
transcript_ready_at
llm_completed_at
tts_first_audio_at
playback_stopped_at
確認する値は次です。
- STT時間:
transcript_ready_at - speech_end_at - LLM時間:
llm_completed_at - transcript_ready_at - TTS開始時間:
tts_first_audio_at - llm_completed_at - 割り込み停止時間:
playback_stopped_at - user_speech_started_at
端末、ネットワーク、言語、接続するモデルで結果が変わるため、普遍的な目標値を置くのではなく、実際の対象環境で中央値と上位パーセンタイルを測ります。上限を超えた場合は、短い待機音声、文字入力、人間への引き継ぎのいずれかへ切り替えます。
実装判断の目安
AI音声を使いやすい問い合わせ
- 操作場所の案内
- 公開FAQの検索
- 問い合わせ内容の聞き取り
- 必要項目の不足確認
- 人間向け要約の作成
最初から人間へ渡した方がよい問い合わせ
- 返金、契約、本人確認
- 不正利用や安全性の報告
- 個別データを調査しないと答えられない内容
- 強い不満や担当者希望が明示された場合
- 音声認識が安定しない環境
AIコンパニオンやキャラクター会話でも境界は同じです。Tencent RTCのSocial Entertainment向け公式ページでは、AI virtual companionsやcharacter dialogueなどのシナリオが示されていますが、親しみやすいキャラクター表現と、問い合わせ回答の権限は分けて設計する必要があります。
注意点
- 音声を録音・保存する場合は、開始前に目的と保存期間を表示する
- STT全文をデバッグログへ無期限に残さない
- LLMへ送ってよい個人情報の範囲を決める
- 割り込み後に古いTTS音声が再開しないことを端末ごとに確認する
- モデルのタイムアウトとRTC切断を別の障害として記録する
- 人間対応の担当者と確認頻度を決め、
waiting_humanを放置しない - AIが作った要約だけで、返金やアカウント操作を実行しない
- 本番のユーザー認証、運営者認証、レート制限は既存アプリの認証基盤へ統合する
音声AIが自然に話せることと、問い合わせ窓口として信頼できることは別です。割り込み、古い結果の破棄、失敗時の保存、人間の返信先までを一つの導線として実装すると、会話デモから運用可能な機能へ進めます。
関係性の開示: 筆者はTencent RTCに関する技術コンテンツの制作に関わっており、本稿では実装上の参照資料としてTencent RTCの公式ドキュメントを使用しています。