リアルタイムAI音声で構造化JSONを使うと、次の緊張関係が生まれます。
- 自由文だけでは、発話・聞き返し・終了をアプリから安全に制御しにくい
- JSON全体を待つと、ユーザーが「無視された」と感じる待ち時間が増える
- 生成途中のJSONを読み上げると、検証前の内容を止められない
つまり必要なのは、単にLLMへ「JSONで返して」と頼むことではありません。LLMの出力を、リアルタイム会話で実行できる命令へ変換する境界が必要です。
この記事では、AI音声コンパニオンを対象に、LLMの返答を speak、clarify、handoff、silent の4種類に限定し、不正JSONやタイムアウトが起きても会話セッション自体を壊さない構成をPythonで作ります。
結論
実装方針は次の5点です。
- LLMが返したJSONを、そのままTTSへ渡さない
- プロンプト上の型指定とは別に、Pydanticで実行時検証する
- JSONの部分抽出や推測修復はせず、安全な定型応答へ戻す
- LLMのストリーミング断片ではなく、検証済みアクションだけを読み上げる
-
request_id、プロンプト版、検証結果、キャンセル理由を記録する
LLMが実際に得意なのは、曖昧な発話から返答候補や聞き返し候補を作ることです。一方、JSONの完全性、発話してよい内容、現在も有効なターンかどうかまでは保証しません。そこはアプリケーションと運営者が責任を持つ領域です。
先に整理する失敗パターン
| 失敗 | 音声体験への影響 | アプリ側の処理 |
|---|---|---|
| JSONの前後に説明文が付く | パースできず無音になる | 不正出力として復旧文を返す |
| 必須キーが欠ける | TTSへ渡す本文を決められない | スキーマ検証で拒否する |
| 未定義アクションが返る | 意図しない処理が走る | 列挙型以外を拒否する |
| 古いリクエストの結果が遅れて届く | 割り込み後に前の話題を読み上げる |
request_id 不一致なら破棄する |
| JSON生成が長引く | 無音が続く | 期限超過として短い復旧文へ切り替える |
confidence: high と自己申告する |
誤答が自動承認される | 自己評価を実行条件に使わない |
特に confidence は、LLM自身の申告にすぎません。人間確認へ回す判断には、根拠の有無、禁止トピック、失敗回数、ユーザーの明示要求など、アプリ側で観測できる条件を使います。
前提：音声経路と判断経路を分ける
構成は以下を想定します。
ユーザー音声
↓
RTC / 音声認識
↓ transcript
会話オーケストレーター
↓ request_id + prompt
LLM
↓ 未検証JSON
Pydantic検証 + アプリ方針
↓ 検証済み発話
音声合成
↓
RTCで再生
Tencent Conversational AIは、複数のLLMプロバイダーと組み合わせたリアルタイム音声対話を扱うための構成を提供しています。全体像は公式概要を参照してください。
また、LLM設定ではOpenAI互換モデルやエージェント基盤との接続、リクエスト識別を含むルーティング・観測の考え方が案内されています。
RTC、音声認識、LLM、音声合成は別々の責任を持ちます。以下のコードは特定SDKのイベント名を仮定せず、その間に置くアプリケーション層を実装します。
手順1：検証環境を作る
Python 3.11以降を想定します。
mkdir voice-json-contract
cd voice-json-contract
python -m venv .venv
source .venv/bin/activate
pip install 'pydantic>=2,<3' 'httpx>=0.27,<1' 'pytest>=8,<9'
構成は次のようにします。
voice-json-contract/
├── app/
│ ├── contract.py
│ ├── llm_client.py
│ └── orchestrator.py
└── tests/
└── test_orchestrator.py
手順2：発話可能なアクションを型で限定する
app/contract.py を作成します。
from typing import Annotated, Literal, Union
from pydantic import BaseModel, ConfigDict, Field
class StrictModel(BaseModel):
model_config = ConfigDict(extra="forbid")
class SpeakAction(StrictModel):
action: Literal["speak"]
request_id: str
utterance: str = Field(min_length=1, max_length=500)
class ClarifyAction(StrictModel):
action: Literal["clarify"]
request_id: str
question: str = Field(min_length=1, max_length=200)
reason_code: Literal["ambiguous", "missing_context"]
class HandoffAction(StrictModel):
action: Literal["handoff"]
request_id: str
summary: str = Field(min_length=1, max_length=300)
reason_code: Literal["user_requested", "policy_boundary", "repeated_failure"]
class SilentAction(StrictModel):
action: Literal["silent"]
request_id: str
reason_code: Literal["interrupted", "no_response_needed"]
VoiceAction = Annotated[
Union[SpeakAction, ClarifyAction, HandoffAction, SilentAction],
Field(discriminator="action"),
]
class VoiceEnvelope(StrictModel):
contract_version: Literal["1"]
result: VoiceAction
重要なのは、すべての文字列を任意入力にしないことです。
-
actionは4種類だけ -
reason_codeも列挙型 - 余分なキーは
extra="forbid"で拒否 - 読み上げ文には長さ上限を設定
-
request_idを必須にする
型ヒントを書くだけでは実行時の値は検証されません。Pydanticの model_validate_json() を通して初めて、外部から届いたJSONに対する実行時の境界になります。
手順3：プロンプトとJSON Schemaを同じ型から作る
モデルによって構造化出力機能の有無や指定方法は異なります。ここでは特定モデル固有の機能へ依存せず、Pydanticから生成したJSON Schemaをプロンプトへ含めます。
app/llm_client.py を作成します。
import json
from typing import Protocol
import httpx
from pydantic import TypeAdapter
from app.contract import VoiceEnvelope
VOICE_SCHEMA = TypeAdapter(VoiceEnvelope).json_schema()
class LLMClient(Protocol):
async def generate(self, *, request_id: str, transcript: str) -> str:
...
class OpenAICompatibleClient:
def __init__(
self,
*,
base_url: str,
api_key: str,
model: str,
timeout_seconds: float,
) -> None:
self.base_url = base_url.rstrip("/")
self.api_key = api_key
self.model = model
self.timeout_seconds = timeout_seconds
async def generate(self, *, request_id: str, transcript: str) -> str:
system_prompt = f"""
あなたはリアルタイム音声コンパニオンの応答候補を作ります。
出力はJSONオブジェクト1個だけにしてください。
Markdown、コードフェンス、前置き、後書きは禁止です。
request_idには必ず {request_id} をそのまま入れてください。
ユーザーの意図が曖昧なら推測せずclarifyを選んでください。
契約バージョンは1です。
JSON Schema:
{json.dumps(VOICE_SCHEMA, ensure_ascii=False)}
""".strip()
payload = {
"model": self.model,
"messages": [
{"role": "system", "content": system_prompt},
{"role": "user", "content": transcript},
],
"temperature": 0,
}
async with httpx.AsyncClient(timeout=self.timeout_seconds) as client:
response = await client.post(
f"{self.base_url}/chat/completions",
headers={"Authorization": f"Bearer {self.api_key}"},
json=payload,
)
response.raise_for_status()
body = response.json()
return body["choices"][0]["message"]["content"]
これはOpenAI互換エンドポイント向けのアダプター例です。Geminiを含め、接続先がこの互換形式を実際に受け付けるか、構造化出力をネイティブに指定できるかは、選択したモデルと接続先の現行仕様を確認してください。
モデルを交換しても、LLMClient のインターフェースと VoiceEnvelope は変更しないのがポイントです。
手順4：不正JSONを「うまく直そう」としない
次に、検証、タイムアウト、古い結果の破棄をまとめます。
app/orchestrator.py を作成します。
import asyncio
import json
import time
import uuid
from dataclasses import asdict, dataclass
from pathlib import Path
from pydantic import TypeAdapter, ValidationError
from app.contract import (
ClarifyAction,
HandoffAction,
SilentAction,
SpeakAction,
VoiceEnvelope,
)
from app.llm_client import LLMClient
ENVELOPE_ADAPTER = TypeAdapter(VoiceEnvelope)
@dataclass(frozen=True)
class TurnOutcome:
request_id: str
status: str
spoken_text: str | None
elapsed_ms: int
class VoiceOrchestrator:
def __init__(
self,
llm: LLMClient,
*,
deadline_seconds: float,
log_path: Path,
) -> None:
self.llm = llm
self.deadline_seconds = deadline_seconds
self.log_path = log_path
self.active_request_id: str | None = None
self.cancelled: set[str] = set()
async def respond(self, transcript: str) -> TurnOutcome:
request_id = str(uuid.uuid4())
self.active_request_id = request_id
started = time.monotonic()
try:
raw = await asyncio.wait_for(
self.llm.generate(
request_id=request_id,
transcript=transcript,
),
timeout=self.deadline_seconds,
)
except TimeoutError:
return self._finish(
request_id,
"timeout",
"少し時間がかかっています。短く言い換えてもらえますか？",
started,
)
except Exception:
return self._finish(
request_id,
"upstream_error",
"いま返答を作れません。もう一度試すか、会話を終了できます。",
started,
)
if not self._is_current(request_id):
return self._finish(request_id, "cancelled", None, started)
try:
envelope = ENVELOPE_ADAPTER.validate_json(raw)
except ValidationError:
return self._finish(
request_id,
"invalid_contract",
"うまく整理できませんでした。質問を一つに絞ってもらえますか？",
started,
)
action = envelope.result
if action.request_id != request_id:
return self._finish(request_id, "request_id_mismatch", None, started)
if not self._is_current(request_id):
return self._finish(request_id, "cancelled", None, started)
if isinstance(action, SpeakAction):
return self._finish(request_id, "speak", action.utterance, started)
if isinstance(action, ClarifyAction):
return self._finish(request_id, "clarify", action.question, started)
if isinstance(action, HandoffAction):
# 実運用では、ここで有人対応キューへの登録や画面表示を行う。
text = "この内容はAIだけで決めず、確認できる窓口へ引き継ぎます。"
return self._finish(request_id, "handoff", text, started)
if isinstance(action, SilentAction):
return self._finish(request_id, "silent", None, started)
return self._finish(request_id, "unsupported", None, started)
def interrupt(self) -> None:
if self.active_request_id is not None:
self.cancelled.add(self.active_request_id)
self.active_request_id = None
def _is_current(self, request_id: str) -> bool:
return (
self.active_request_id == request_id
and request_id not in self.cancelled
)
def _finish(
self,
request_id: str,
status: str,
spoken_text: str | None,
started: float,
) -> TurnOutcome:
elapsed_ms = int((time.monotonic() - started) * 1000)
outcome = TurnOutcome(
request_id=request_id,
status=status,
spoken_text=spoken_text,
elapsed_ms=elapsed_ms,
)
self._append_log(outcome)
return outcome
def _append_log(self, outcome: TurnOutcome) -> None:
self.log_path.parent.mkdir(parents=True, exist_ok=True)
with self.log_path.open("a", encoding="utf-8") as f:
f.write(json.dumps(asdict(outcome), ensure_ascii=False) + "\n")
ここではコードフェンスを正規表現で取り除いたり、最初の { から最後の } までを抽出したりしていません。
そのような修復は一見便利ですが、次の問題があります。
- 2個のJSONが返ったとき、どちらを採用するか曖昧
- 説明文中のJSON例を誤採用する可能性がある
- 欠けた引用符などを補うと、元の意味を変え得る
- モデル変更後の契約違反が見えなくなる
音声では、不完全な返答を流すより、短い復旧文を返して再入力できるほうが制御しやすくなります。
手順5：RTCの割り込みと接続する
RTCまたは音声認識側で新しいユーザー発話を検知したら、次の2つを同時に行います。
- 再生中の音声合成を停止する
-
orchestrator.interrupt()を呼ぶ
async def on_user_speech_started() -> None:
await tts_output.stop()
orchestrator.interrupt()
async def on_transcript_finalized(text: str) -> None:
outcome = await orchestrator.respond(text)
if outcome.spoken_text is None:
return
await tts_output.speak(outcome.spoken_text)
tts_output は利用する音声合成・再生実装に合わせたアダプターです。特定SDKのAPI名ではありません。
注意すべき競合は、JSON検証後からTTS開始までの間にも割り込みが起こり得ることです。実運用では、speak() の直前にも request_id が有効か確認できるインターフェースにしてください。
確認方法
偽LLMを使えば、実際のモデルへ課金せず異常系を固定できます。
tests/test_orchestrator.py を作成します。
import asyncio
import pytest
from app.orchestrator import VoiceOrchestrator
class FakeLLM:
def __init__(self, response_factory, delay: float = 0) -> None:
self.response_factory = response_factory
self.delay = delay
async def generate(self, *, request_id: str, transcript: str) -> str:
await asyncio.sleep(self.delay)
return self.response_factory(request_id)
@pytest.mark.asyncio
async def test_valid_speak(tmp_path):
llm = FakeLLM(lambda request_id: f'''{{
"contract_version": "1",
"result": {{
"action": "speak",
"request_id": "{request_id}",
"utterance": "こんばんは。今日は何を話しますか？"
}}
}}''')
app = VoiceOrchestrator(
llm,
deadline_seconds=1,
log_path=tmp_path / "turns.jsonl",
)
result = await app.respond("こんばんは")
assert result.status == "speak"
assert result.spoken_text is not None
@pytest.mark.asyncio
async def test_markdown_fence_is_rejected(tmp_path):
llm = FakeLLM(lambda _: '```json\n{"action":"speak"}\n```')
app = VoiceOrchestrator(
llm,
deadline_seconds=1,
log_path=tmp_path / "turns.jsonl",
)
result = await app.respond("話して")
assert result.status == "invalid_contract"
assert result.spoken_text is not None
@pytest.mark.asyncio
async def test_extra_field_is_rejected(tmp_path):
def response(request_id: str) -> str:
return f'''{{
"contract_version": "1",
"result": {{
"action": "speak",
"request_id": "{request_id}",
"utterance": "返答です",
"confidence": "high"
}}
}}'''
app = VoiceOrchestrator(
FakeLLM(response),
deadline_seconds=1,
log_path=tmp_path / "turns.jsonl",
)
result = await app.respond("質問")
assert result.status == "invalid_contract"
@pytest.mark.asyncio
async def test_timeout_has_recovery_message(tmp_path):
llm = FakeLLM(lambda _: "{}", delay=0.2)
app = VoiceOrchestrator(
llm,
deadline_seconds=0.01,
log_path=tmp_path / "turns.jsonl",
)
result = await app.respond("長い質問")
assert result.status == "timeout"
assert result.spoken_text is not None
@pytest.mark.asyncio
async def test_interrupted_result_is_not_spoken(tmp_path):
llm = FakeLLM(
lambda request_id: f'''{{
"contract_version": "1",
"result": {{
"action": "speak",
"request_id": "{request_id}",
"utterance": "古い返答"
}}
}}''',
delay=0.05,
)
app = VoiceOrchestrator(
llm,
deadline_seconds=1,
log_path=tmp_path / "turns.jsonl",
)
task = asyncio.create_task(app.respond("最初の質問"))
await asyncio.sleep(0.01)
app.interrupt()
result = await task
assert result.status == "cancelled"
assert result.spoken_text is None
実行します。
pytest -q
実機で確認する項目
自動テストに加えて、端末とネットワーク条件を変えて次を確認します。
- LLM待機中に話し始めると、古い結果が再生されない
- TTS再生中の割り込みで、再生停止とリクエスト無効化の両方が行われる
- 不正JSONでもセッションが終了せず、再入力できる
- タイムアウト値を短くしたとき、復旧文が連続再生されない
-
handoffを選んだ後に、AIが勝手に通常会話へ戻らない - ミュート、終了、AI利用停止などの操作が画面から分かる
レイテンシは一つの数値にまとめず、少なくとも以下を別々に記録します。
音声認識確定まで
LLM開始まで
LLM完了まで
JSON検証時間
TTS開始まで
割り込み検知から再生停止まで
適切な期限値は、モデル、ネットワーク、発話長、端末によって変わります。記事中のテスト値を本番の性能目標として流用せず、実際の分布を測って決めてください。
判断フレームワーク：どこまで自動化するか
| 処理 | LLMに任せる | アプリで決める | 人間が決める |
|---|---|---|---|
| 自然な返答案の作成 | ○ | ||
| 聞き返し文の作成 | ○ | ||
| JSONの妥当性 | ○ | ||
| 現在も有効なターンか | ○ | ||
| 発話の停止 | ○ | ユーザー操作も許可 | |
| センシティブな相談への対応方針 | 方針を強制 | ○ | |
| 会話ログの保存期間 | ○ |
AI音声コンパニオンが自然に話せることと、ユーザーの代わりに判断してよいことは別問題です。特に感情的・個人的な会話では、AIであること、会話を停止できること、データの扱いをユーザーへ見える形で示す必要があります。
注意点とトレードオフ
JSON全体を待つと初動は遅くなる
完全なJSONを検証してからTTSへ渡すため、生成途中から読み上げる方式より開始は遅くなります。ただし、未検証の断片を読み上げないという安全性が得られます。
待ち時間を埋めたい場合は、LLMが生成した途中文ではなく、アプリ側で管理する短い効果音や待機表示を使います。毎回同じ相づちを音声再生すると、ユーザーがAIの回答開始と誤解するため、用途を明示したほうがよいでしょう。
JSONLへ会話本文を無条件に残さない
サンプルログには状態と所要時間だけを保存しています。全文トランスクリプトを保存すると調査しやすくなる一方、個人情報やセンシティブな会話の保管対象が増えます。
本文を記録する場合は、少なくとも以下を先に決めます。
- 保存目的
- 保存期間
- 閲覧権限
- 削除方法
- ユーザーへの表示と同意
復旧文も無限ループさせない
LLM失敗のたびに「もう一度お願いします」と返すだけでは、ユーザーへ再試行の負担を押し付けます。同一セッションで失敗が続いたら、AI会話を一時停止する、テキスト入力へ切り替える、終了する、有人窓口を案内するなど、別経路を提示します。
プロンプト版を記録する
モデル名だけでなく、contract_version とプロンプト版もログへ残します。同じモデルでも、指示やスキーマが変われば失敗傾向は変わるためです。モデル交換前には、保存した匿名化入力または合成入力を使い、正常系と異常系を再生します。
まとめ
リアルタイムAI音声の構造化出力は、JSONを返させるプロンプトだけでは完成しません。
- 発話可能なアクションを少数に限定する
- Pydanticで実行時に検証する
- 不正JSONを推測修復しない
- 検証済みの全文だけをTTSへ渡す
- 割り込み、期限超過、有人対応を契約に含める
- リクエストIDとプロンプト版で追跡する
最初の一歩としては、現在の音声プロンプトへJSON例を追加するより先に、不正出力時に何を話し、何を絶対に実行しないかを1枚の判断表にしてください。その境界が決まれば、LLMやGeminiなどのモデルを交換しても、会話アプリ側の責任を保ちやすくなります。
関係性の開示: 筆者はTencent RTCのコンテンツ制作に関与しています。本稿はTencent RTC公式ドキュメントを実装上の参照資料として使用しました。