新しいTTSモデルを見ると、まず試したくなるのは声質や表現力です。しかしリアルタイム音声AIを作る側の悩みは、別の場所にあります。
LLMの回答には、見出し、URL、コード、数式、箇条書きが混ざります。これをそのままTTSへ渡すと、声が自然でも「シャープを3回」「エイチティーティーピーエス」と読み始めます。逆に、音声用の文章をLLMへ別生成させると、今度は画面と発話の内容がずれます。
つまり、人が管理すべきなのは声の演技だけではありません。何を画面に出し、何を耳へ届け、どこから再開できるかという会話の編集規則です。
この記事では、Geminiなど任意のLLM／TTSを交換できるようにしつつ、回答を一度「意味単位のデータ」に変換し、そこから画面表示と読み上げを生成する方法をPythonで再現します。
結論：回答文を2本作らず、1つの意味データを2通りに描画する
今回の構成は次のとおりです。
ユーザー音声
↓
音声認識
↓
LLM ──→ AnswerAtom[]
├─→ 画面用Renderer ──→ Markdown
└─→ 音声用Compiler ──→ SpeechSegment[] ──→ TTS
↓
RTC音声
重要なルールは4つです。
- LLMに画面用回答と音声用回答を別々に書かせない
- URL、コード、専門用語を型で区別する
- 読み方辞書と発話スタイルはアプリ側で管理する
- TTSへ渡す文章を短いセグメントに分け、完了位置を記録する
この方法で防げるのは、音声AIの幻覚そのものではありません。防げるのは、同じ回答から派生した画面と音声が食い違うこと、コードやURLを誤って読み上げること、再接続時に長文を最初から読み直すことです。
前提：RTC、LLM、TTSを一つの処理として扱わない
Tencent Conversational AIは、リアルタイム音声対話と複数のLLMプロバイダーを組み合わせるためのシナリオを提供しています。
公式ドキュメントでは、OpenAI互換モデルやDify、Cozeなどのエージェント基盤との接続、およびリクエスト識別子を使ったルーティングや観測について説明されています。
本記事のコンパイラは、Tencent Conversational AIの製品APIを置き換えるものではありません。LLM出力とTTS入力の間に置くアプリケーション層です。RTCのメディア転送、音声認識、LLM、TTS、再生状態を分離したまま接続します。
動作確認にはPython 3.11以降を使います。外部パッケージは不要です。
mkdir speech-plan-demo
cd speech-plan-demo
python -m venv .venv
source .venv/bin/activate # Windowsでは .venv\Scripts\activate
まず再現する失敗
LLMが次のような回答を返したとします。
### 実行方法
`python app.py` を実行してください。
詳細は https://example.com/docs を参照してください。
GDPは国内総生産を表します。
これをそのままTTSへ送る設計には、次の問題があります。
| 入力 | 起きやすい問題 |
|---|---|
### |
記号を読む、または不自然な間が入る |
python app.py |
コードを一文字ずつ読む可能性がある |
| URL | 長い文字列を読み続ける |
GDP |
意図した日本語読みにならない可能性がある |
| 長いMarkdown全体 | 割り込みや再接続時の再開位置が曖昧になる |
正規表現でMarkdownを除去するだけでは不十分です。コードを消すと、利用者には「何を実行するのか」が伝わらなくなります。
そこで、回答を最初から意味単位へ分解します。
手順1：回答をAnswerAtomの配列にする
今回使う種類は5つだけです。
| 種類 | 画面 | 音声 |
|---|---|---|
text |
通常の文章 | そのまま読む |
term |
専門用語 | 承認済み辞書の読みを使う |
number |
数値 | 承認済みの読みがある場合だけ読む |
code |
コード表示 | 「コードは画面に表示します」と案内 |
url |
リンク表示 | 「リンクは画面に表示します」と案内 |
例えば先ほどの回答は、次のデータになります。
atoms = [
AnswerAtom(Kind.TEXT, "実行方法を説明します。"),
AnswerAtom(Kind.CODE, "python app.py"),
AnswerAtom(Kind.TEXT, "を実行してください。"),
AnswerAtom(Kind.URL, "https://example.com/docs"),
AnswerAtom(Kind.TEXT, "に詳細があります。"),
AnswerAtom(Kind.TERM, "GDP", key="gdp"),
AnswerAtom(Kind.TEXT, "は国内総生産を表します。"),
]
LLMには自由な読み上げ文ではなく、このJSON相当の構造を返させます。ただし、LLMが返したJSONは必ずアプリ側で検証します。
コード：画面表示と音声セグメントを同時に生成する
speech_plan.pyを作成します。
from __future__ import annotations
import hashlib
import re
from dataclasses import dataclass
from enum import Enum
from typing import Iterable
class Kind(str, Enum):
TEXT = "text"
TERM = "term"
NUMBER = "number"
CODE = "code"
URL = "url"
@dataclass(frozen=True)
class AnswerAtom:
kind: Kind
value: str
key: str | None = None
@dataclass(frozen=True)
class SpeechSegment:
turn_id: str
index: int
segment_id: str
text: str
profile: str
class CompileError(ValueError):
pass
URL_PATTERN = re.compile(r"https?://\S+", re.IGNORECASE)
MARKDOWN_PATTERN = re.compile(r"[`#*_]")
SENTENCE_PATTERN = re.compile(r".*?(?:[。！？]|$)")
class SpeechPlanCompiler:
def __init__(
self,
pronunciations: dict[str, str],
max_chars: int = 42,
) -> None:
self.pronunciations = pronunciations
self.max_chars = max_chars
def validate(self, atoms: Iterable[AnswerAtom]) -> list[AnswerAtom]:
validated = list(atoms)
if not validated:
raise CompileError("回答が空です")
for atom in validated:
if not atom.value.strip():
raise CompileError(f"空の値は使用できません: {atom.kind}")
# textへURLやMarkdownを埋め込むと型分けが無意味になる。
if atom.kind == Kind.TEXT:
if URL_PATTERN.search(atom.value):
raise CompileError("URLはurl型へ分離してください")
if MARKDOWN_PATTERN.search(atom.value):
raise CompileError("Markdown記号はtext型へ含めないでください")
if atom.kind in {Kind.TERM, Kind.NUMBER}:
if atom.key is None:
raise CompileError(
f"{atom.kind.value}型には読み辞書のkeyが必要です"
)
if atom.key not in self.pronunciations:
raise CompileError(
f"未承認の読み方です: {atom.key}"
)
if atom.kind == Kind.URL and not URL_PATTERN.fullmatch(atom.value):
raise CompileError(f"URL形式が不正です: {atom.value}")
return validated
def render_markdown(self, atoms: Iterable[AnswerAtom]) -> str:
parts: list[str] = []
for atom in self.validate(atoms):
match atom.kind:
case Kind.TEXT | Kind.TERM | Kind.NUMBER:
parts.append(atom.value)
case Kind.CODE:
parts.append(f"`{atom.value}`")
case Kind.URL:
parts.append(f"[{atom.value}]({atom.value})")
return "".join(parts)
def render_spoken_text(self, atoms: Iterable[AnswerAtom]) -> str:
parts: list[str] = []
previous_cue: str | None = None
for atom in self.validate(atoms):
match atom.kind:
case Kind.TEXT:
spoken = atom.value
case Kind.TERM | Kind.NUMBER:
assert atom.key is not None
spoken = self.pronunciations[atom.key]
case Kind.CODE:
spoken = "コードは画面に表示します。"
case Kind.URL:
spoken = "リンクは画面に表示します。"
case _:
raise CompileError(f"未対応の型です: {atom.kind}")
# コードが連続しても同じ案内を何度も読まない。
if spoken == previous_cue and atom.kind in {Kind.CODE, Kind.URL}:
continue
parts.append(spoken)
previous_cue = spoken if atom.kind in {Kind.CODE, Kind.URL} else None
return "".join(parts)
def compile(
self,
turn_id: str,
atoms: Iterable[AnswerAtom],
profile: str = "live",
) -> list[SpeechSegment]:
spoken = self.render_spoken_text(atoms)
chunks = self._split(spoken)
segments: list[SpeechSegment] = []
for index, text in enumerate(chunks):
digest = hashlib.sha256(
f"{turn_id}:{index}:{text}".encode("utf-8")
).hexdigest()[:16]
segments.append(
SpeechSegment(
turn_id=turn_id,
index=index,
segment_id=digest,
text=text,
profile=profile,
)
)
return segments
def _split(self, text: str) -> list[str]:
sentences = [
match.group(0).strip()
for match in SENTENCE_PATTERN.finditer(text)
if match.group(0).strip()
]
chunks: list[str] = []
buffer = ""
for sentence in sentences:
if len(buffer) + len(sentence) <= self.max_chars:
buffer += sentence
continue
if buffer:
chunks.append(buffer)
buffer = ""
# 長すぎる一文は読点を優先して分割する。
pieces = self._split_long_sentence(sentence)
chunks.extend(pieces[:-1])
buffer = pieces[-1]
if buffer:
chunks.append(buffer)
return chunks
def _split_long_sentence(self, sentence: str) -> list[str]:
pieces = re.split(r"(?<=、)", sentence)
chunks: list[str] = []
buffer = ""
for piece in pieces:
if len(buffer) + len(piece) <= self.max_chars:
buffer += piece
else:
if buffer:
chunks.append(buffer)
# 読点のない長文に対する最終フォールバック。
while len(piece) > self.max_chars:
chunks.append(piece[: self.max_chars])
piece = piece[self.max_chars :]
buffer = piece
if buffer:
chunks.append(buffer)
return chunks
if __name__ == "__main__":
compiler = SpeechPlanCompiler(
pronunciations={
"gdp": "ジーディーピー",
"python_311": "パイソン、さんてんいちいち",
}
)
atoms = [
AnswerAtom(Kind.TEXT, "実行方法を説明します。"),
AnswerAtom(Kind.CODE, "python app.py"),
AnswerAtom(Kind.TEXT, "を実行してください。"),
AnswerAtom(Kind.URL, "https://example.com/docs"),
AnswerAtom(Kind.TEXT, "に詳細があります。"),
AnswerAtom(Kind.TERM, "GDP", key="gdp"),
AnswerAtom(Kind.TEXT, "は国内総生産を表します。"),
]
print("--- 画面 ---")
print(compiler.render_markdown(atoms))
print("\n--- TTSへ渡すセグメント ---")
for segment in compiler.compile("turn-001", atoms):
print(segment)
実行します。
python speech_plan.py
出力例は次の形になります。
--- 画面 ---
実行方法を説明します。`python app.py`を実行してください。[https://example.com/docs](https://example.com/docs)に詳細があります。GDPは国内総生産を表します。
--- TTSへ渡すセグメント ---
SpeechSegment(..., text='実行方法を説明します。コードは画面に表示します。', ...)
SpeechSegment(..., text='を実行してください。リンクは画面に表示します。', ...)
SpeechSegment(..., text='に詳細があります。ジーディーピーは国内総生産を表します。', ...)
TTSにはSpeechSegment.textだけを渡します。profileはアプリ内部の抽象名であり、特定製品のAPI名ではありません。実際のTTSアダプターで、利用するモデルの設定へ対応付けてください。
手順2：LLMのJSONを境界で検査する
LLMには、例えば次の形式だけを返すよう依頼します。
{
"atoms": [
{"kind": "text", "value": "実行方法を説明します。"},
{"kind": "code", "value": "python app.py"},
{"kind": "text", "value": "を実行してください。"}
]
}
ただし、構造化出力を指定できることと、内容が正しいことは別問題です。アプリ側では最低限、次を検査します。
- 未知の
kindを拒否する -
textにURLやMarkdown記号が入っていたら拒否する -
termとnumberは承認済み辞書のkeyを必須にする - URL形式を検証する
- Atom数と全体文字数に上限を設ける
- LLMが指定した任意の声名や演技指示を、そのままTTSへ渡さない
読み方をLLMに毎回推測させないのがポイントです。製品名、人名、略語、数式の読みには業務上の判断が含まれます。辞書の変更はコードレビュー対象にすると、モデル変更後も発音方針を維持できます。
手順3：セグメント単位で再生完了を記録する
長文を1回のTTS入力にまとめると、再接続後にどこから再開するか決めにくくなります。セグメント単位で次の状態を持ちます。
from dataclasses import dataclass, field
@dataclass
class PlaybackLedger:
turn_id: str
completed_indexes: set[int] = field(default_factory=set)
def mark_completed(self, index: int) -> None:
self.completed_indexes.add(index)
def next_index(self, total: int) -> int | None:
for index in range(total):
if index not in self.completed_indexes:
return index
return None
再生開始ではなく、再生完了を確認した時点でmark_completed()を呼びます。
ネットワーク切断がセグメントの途中で起きた場合、この例では未完了セグメントを先頭から再生します。少し重複して聞こえる可能性はありますが、内容の欠落を避けられます。重複を避けたいサービスでは、未完了セグメントを破棄して「続きから話します」という固定音声を入れる設計もあります。
新しいユーザー発話による割り込みは、ネットワーク復旧と同じ扱いにしません。割り込み時は古いターンの未送信セグメントを破棄し、新しいturn_idを発行します。そうしないと、新しい質問へ答えた後に古い説明が再開されます。
TTSプロファイルを選ぶ判断表
表現力の高いTTSが利用できても、すべての文を同じ方法で生成する必要はありません。アプリ内部では、用途だけを表すプロファイルを定義します。
| 用途 | プロファイル例 | 方針 |
|---|---|---|
| 通常の会話応答 | live |
短いセグメントを順次処理する |
| 固定の案内 | prompt |
承認済み文面を使用し、必要なら事前生成を検討する |
| 感情表現が必要な台詞 | expressive |
許可済みスタイルだけを使う |
| コードやURL | TTSへ送らない | 画面へ誘導する短い案内に置換する |
実際のモデル名、声、利用可能なスタイルはTTSプロバイダーの公式仕様を確認してアダプター側へ設定します。LLMにはモデル名を選ばせず、「通常応答」「案内」「演出的な台詞」といった用途だけを提案させる方が、プロバイダー交換時の影響を限定できます。
セグメントを細かくすると、最初の音声を処理しやすくなる一方、TTS呼び出し回数、接続管理、文脈をまたぐイントネーション調整の負担が増えます。サンプルのmax_chars=42は性能保証値ではありません。使用言語、TTS、端末、ネットワークごとに計測して変更してください。
Tencent Conversational AIへ接続する位置
接続時の責任分界は次のようにします。
Tencent RTC側のリアルタイム音声経路
│
├─ 音声認識結果
│ ↓
├─ LLM設定・ルーティング
│ ↓
├─ アプリ層: JSON検証
│ ↓
├─ SpeechPlanCompiler
│ ├─ 画面表示
│ └─ TTS入力セグメント
│
└─ 合成音声を会話へ返す
公式のLLM設定ドキュメントで扱われるリクエスト識別子は、ルーティングや観測へ利用できます。本記事のturn_idやsegment_idは、それとは別にアプリ内の会話単位と再生単位を追跡するための識別子です。
ログには少なくとも次を残します。
{
"request_id": "provider-or-platform-request-id",
"turn_id": "turn-001",
"segment_id": "a1b2c3d4e5f6g7h8",
"segment_index": 1,
"profile": "live",
"event": "playback_completed"
}
生の音声、全文、個人情報を無条件にログへ保存する必要はありません。障害解析に必要な識別子と状態を選び、保持期間、アクセス権、削除方法を決めてください。
確認方法：TTSを接続する前にコンパイラをテストする
test_speech_plan.pyを作成します。
import unittest
from speech_plan import (
AnswerAtom,
CompileError,
Kind,
SpeechPlanCompiler,
)
class SpeechPlanCompilerTest(unittest.TestCase):
def setUp(self) -> None:
self.compiler = SpeechPlanCompiler(
pronunciations={"gdp": "ジーディーピー"},
max_chars=24,
)
def test_code_body_is_not_spoken(self) -> None:
atoms = [
AnswerAtom(Kind.CODE, "rm -rf /tmp/example"),
AnswerAtom(Kind.TEXT, "を実行します。"),
]
spoken = self.compiler.render_spoken_text(atoms)
self.assertNotIn("rm -rf", spoken)
self.assertIn("コードは画面に表示します", spoken)
def test_url_inside_text_is_rejected(self) -> None:
atoms = [
AnswerAtom(Kind.TEXT, "https://example.com を見てください。")
]
with self.assertRaises(CompileError):
self.compiler.compile("turn-001", atoms)
def test_unknown_pronunciation_is_rejected(self) -> None:
atoms = [
AnswerAtom(Kind.TERM, "LLM", key="llm")
]
with self.assertRaises(CompileError):
self.compiler.compile("turn-001", atoms)
def test_segment_ids_are_stable(self) -> None:
atoms = [
AnswerAtom(Kind.TEXT, "最初の説明です。次の説明です。")
]
first = self.compiler.compile("turn-001", atoms)
second = self.compiler.compile("turn-001", atoms)
self.assertEqual(
[x.segment_id for x in first],
[x.segment_id for x in second],
)
def test_different_turn_has_different_ids(self) -> None:
atoms = [AnswerAtom(Kind.TEXT, "同じ回答です。")]
first = self.compiler.compile("turn-001", atoms)
second = self.compiler.compile("turn-002", atoms)
self.assertNotEqual(first[0].segment_id, second[0].segment_id)
if __name__ == "__main__":
unittest.main()
実行します。
python -m unittest -v
実際のTTSとRTCを接続した後は、次の順に確認します。
- Markdown記号を読まない
- URL本文を読まない
- コードを一文字ずつ読まない
- 未登録の略語が検証エラーになる
- 画面にはコードとURLが残る
- 再接続後は未完了セグメントから再開する
- 新しいユーザー発話後に古いターンを再開しない
-
同じ
turn_idと内容から同じsegment_idが生成される - TTS失敗時にテキスト表示だけは継続できる
- 音声を止める操作がLLMやTTSの応答待ちにならない
AIで改善できる範囲と、人が残すべき判断
新しいTTSの表現力は、音声コンパニオンの聞きやすさを改善する可能性があります。しかし、モデルが高性能になることと、会話体験が自動的に良くなることは同義ではありません。
AIへ任せやすいのは次の範囲です。
- 回答を意味単位へ分類する
- 長い説明を短い文章へ整える
- 通常応答か演出的な台詞かを提案する
- 画面へ出すべきコードやURLを抽出する
一方、人が管理した方がよいのは次の範囲です。
- 固有名詞や専門用語の正式な読み方
- 利用を許可する声とスタイル
- 誤解を招く感情表現の禁止ルール
- 声の複製や本人に似せる場合の同意
- ログへ保存する情報と保持期間
- 割り込み、停止、再接続時の最終動作
音声AIに対する不安は、「自分の話し方までAIに決められるのではないか」という感覚にもつながります。実装上は逆で、TTSの選択肢が増えるほど、発音辞書、スタイル許可、停止方法といった人間側の編集能力が重要になります。
注意点とトレードオフ
1. 画面と音声が同じでも、内容の正しさは保証されない
本実装は二重生成による食い違いを減らしますが、LLMが誤った説明を生成する問題は別途残ります。重要な領域では検索、根拠表示、有人確認などを追加してください。
2. 一般的な日本語数値変換を安易に自作しない
3.11はバージョンなら「さんてんいちいち」、数量なら別の読みになるかもしれません。数値だけで読み方を確定せず、用途を表す辞書キーを使う方が安全です。
3. セグメント分割は音質にも影響する
短く切りすぎると、文をまたぐ抑揚が不自然になる可能性があります。最初の発話までの時間、割り込み後に残る音声量、TTS失敗率、主観評価を一緒に計測して調整します。
4. 固定音声のキャッシュには変更管理が必要
固定案内を事前生成する場合、文章、声、モデル、言語、スタイルをキャッシュキーへ含めます。文面だけ変更して古い音声が残る事故を避けてください。
5. 音声が失敗しても会話全体を止めない
TTSが利用できない場合は、画面表示へ縮退し、再試行ボタンや読み上げ再開操作を出します。リアルタイム会話の継続を、単一のTTSモデルだけに依存させない設計が必要です。
まとめ
TTSモデルを比較するとき、声質だけを見ると実装上の難所を見落とします。リアルタイム音声AIで先に固定すべきなのは、LLM回答をどの単位で受け取り、画面と音声へどう変換し、どこから再開するかです。
- 回答を
AnswerAtomへ構造化する - 画面と音声を同じAtom列から生成する
- URLとコードを読み上げ対象から外す
- 発音辞書とスタイル許可を人が管理する
- TTS入力を再開可能なセグメントへ分割する
- RTC、LLM、TTS、再生状態を別の責務として観測する
Geminiを含む新しいモデルへ交換するときも、この境界を維持すれば、変えるのは主にLLM／TTSアダプターです。会話の編集規則と人間の制御は、モデルの外側へ残せます。
関係開示: 本稿はTencent RTCの開発者向けコミュニティコンテンツとして作成しており、実装上の事実確認にはTencent RTC公式ドキュメントを参照しました。