0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@susie000720

音声AIの人格設定をプロンプトへ直書きしない：会話契約をJSON化して割り込み・遅延復旧まで検証する

0
Last updated at Posted at 2026-09-20

リアルタイム音声AIのデモは、自然な返答が1回できるだけでも成立します。しかし運用では、別の不安が残ります。

  • ユーザーが話し始めたら、AIは本当に黙るのか
  • LLMが遅いとき、無言のまま固まらないか
  • モデルを変更しても、名乗り方や禁止事項が維持されるか
  • 復旧メッセージまでLLMへ依存していないか

これは「もっと良いプロンプトを書けるか」という問題ではありません。会話中に必ず守る契約を、人間がレビューでき、機械でも検証できる形にしたかという設計問題です。

Webサイトの情報を機械可読にする発想と同じように、音声AIでも人格、根拠、割り込み、遅延時の処理を構造化できます。本稿では conversation-manifest.json を単一の入力とし、LLM用プロンプトとアプリ用復旧ポリシーを生成します。

結論：文章と制御ルールを同じプロンプトへ混ぜない

会話仕様を次の3層へ分けます。

  1. LLMへ渡す情報：役割、口調、回答根拠、回答してはいけない範囲
  2. アプリが決定論的に実行する制御：TTS停止、進行中ターンの破棄、再入力受付、ローカル復旧文
  3. 人間が決める運用方針：待機を許容する時間、有人対応へ切り替える条件、保存するログ

LLMは自然な文章生成には向いていますが、「割り込みを受けたら必ず音声を止める」といった保証主体には向きません。停止や復旧は、モデルの返答ではなくアプリ側の規則として実装します。

Tencent Conversational AIは、リアルタイム音声対話と複数のLLMプロバイダーを組み合わせるための構成を提供しています。公式概要は以下です。

公式のLLM設定資料では、OpenAI互換モデルやDify、Cozeなどとの接続、およびルーティングや観測に利用できるリクエスト識別子が案内されています。ただし、接続できることと、会話契約が守られることは別です。後者を今回のJSONとテストで補います。

前提：音声経路を6つの責務に分ける

実装前に、次の境界を混ぜないようにします。

ユーザー音声
  ↓
RTC／メディア伝送
  ↓
音声認識（STT）
  ↓
会話制御 ─── conversation-manifest.json
  ↓
LLM
  ↓
音声合成（TTS）
  ↓
再生

これとは別に、モデレーション、同意管理、ログ保存、有人停止を配置します。

Tencent RTCはリアルタイム音声の経路を担当します。一方、今回実装するマニフェストはアプリケーション層の設定です。後述する stop_tts などの名称も説明用のアプリ内部アクションであり、Tencent RTC SDKのAPI名ではありません。

検証環境はPython 3.11以降とします。外部パッケージは使いません。

mkdir voice-contract
cd voice-contract
touch conversation-manifest.json build_contract.py test_contract.py

手順1：会話契約をJSONで定義する

次の内容を conversation-manifest.json として保存します。

待ち時間の値は製品性能を示すものではなく、このサンプルで使うアプリ側の仮の判断値です。本番では実測値とユーザーテストから決めてください。

{
  "schemaVersion": "1.0",
  "agent": {
    "id": "reading-companion",
    "displayName": "読書コンパニオン",
    "disclosure": "私はAI音声アシスタントです。",
    "style": [
      "日本語で簡潔に話す",
      "分からないことを推測で断定しない",
      "一度に一つの質問だけを返す"
    ]
  },
  "knowledge": {
    "facts": [
      {
        "id": "service-purpose",
        "text": "このサービスは読書メモの整理を支援する。",
        "source": "https://example.com/about"
      }
    ],
    "unknownAnswer": "確認できる情報がないため、ここでは断定できません。"
  },
  "turnPolicy": {
    "llmDeadlineMs": 2500,
    "maximumRetryCount": 1,
    "acceptNewSpeechDuringTts": true
  },
  "localPhrases": {
    "llmDelayed": "回答の準備に時間がかかっています。質問を短く言い換えることもできます。",
    "providerUnavailable": "現在、回答を生成できません。少し待ってからもう一度お試しください。",
    "sttUncertain": "うまく聞き取れませんでした。もう一度お願いします。"
  },
  "recoveryMatrix": {
    "USER_SPEECH_DURING_TTS": [
      "stop_tts",
      "cancel_active_turn",
      "open_input"
    ],
    "LLM_DEADLINE": [
      "cancel_active_turn",
      "play_local:llmDelayed",
      "open_input"
    ],
    "LLM_PROVIDER_ERROR": [
      "cancel_active_turn",
      "play_local:providerUnavailable",
      "open_input"
    ],
    "STT_UNCERTAIN": [
      "play_local:sttUncertain",
      "open_input"
    ]
  },
  "privacy": {
    "sendToLlm": [
      "final_transcript",
      "approved_conversation_context"
    ],
    "neverSendToLlm": [
      "raw_audio_unless_separately_consented",
      "authentication_token"
    ]
  }
}

ここで重要なのは、recoveryMatrix をプロンプトへ埋め込まないことです。

たとえば「ユーザーが話したら読み上げをやめてください」とLLMへ指示しても、LLM自身は端末の音声再生を直接止められません。USER_SPEECH_DURING_TTS を検知したアプリが、TTS停止と進行中ターンの無効化を実行する必要があります。

手順2：マニフェストを検証して成果物を生成する

次のスクリプトは、必須イベントの不足、存在しないローカル文言への参照、危険な復旧順序を検査します。検証後、以下を生成します。

  • dist/system-prompt.txt：LLMへ渡す人格・知識境界
  • dist/runtime-policy.json：アプリが読む制御規則

build_contract.py:

from __future__ import annotations

import json
from pathlib import Path
from typing import Any

REQUIRED_EVENTS = {
    "USER_SPEECH_DURING_TTS",
    "LLM_DEADLINE",
    "LLM_PROVIDER_ERROR",
    "STT_UNCERTAIN",
}

ALLOWED_ACTIONS = {
    "stop_tts",
    "cancel_active_turn",
    "open_input",
}


def load_manifest(path: str) -> dict[str, Any]:
    with open(path, encoding="utf-8") as f:
        return json.load(f)


def validate(manifest: dict[str, Any]) -> None:
    errors: list[str] = []

    if manifest.get("schemaVersion") != "1.0":
        errors.append("schemaVersion must be 1.0")

    agent = manifest.get("agent", {})
    if not agent.get("id"):
        errors.append("agent.id is required")
    if not agent.get("disclosure"):
        errors.append("agent.disclosure is required")

    policy = manifest.get("turnPolicy", {})
    deadline = policy.get("llmDeadlineMs")
    if not isinstance(deadline, int) or deadline <= 0:
        errors.append("turnPolicy.llmDeadlineMs must be a positive integer")

    matrix = manifest.get("recoveryMatrix", {})
    missing = REQUIRED_EVENTS - set(matrix)
    if missing:
        errors.append(f"missing recovery events: {sorted(missing)}")

    local_phrases = manifest.get("localPhrases", {})

    for event, actions in matrix.items():
        if not isinstance(actions, list) or not actions:
            errors.append(f"{event}: actions must be a non-empty list")
            continue

        for action in actions:
            if action.startswith("play_local:"):
                phrase_key = action.split(":", 1)[1]
                if phrase_key not in local_phrases:
                    errors.append(
                        f"{event}: unknown local phrase '{phrase_key}'"
                    )
            elif action not in ALLOWED_ACTIONS:
                errors.append(f"{event}: unknown action '{action}'")

    interruption = matrix.get("USER_SPEECH_DURING_TTS", [])
    required_interruption_actions = {
        "stop_tts",
        "cancel_active_turn",
        "open_input",
    }
    if not required_interruption_actions.issubset(interruption):
        errors.append(
            "USER_SPEECH_DURING_TTS must stop TTS, "
            "cancel the active turn, and open input"
        )

    for event in ("LLM_DEADLINE", "LLM_PROVIDER_ERROR"):
        actions = matrix.get(event, [])
        if actions and not any(a.startswith("play_local:") for a in actions):
            errors.append(f"{event}: must have an LLM-independent local phrase")

    if errors:
        raise ValueError("Invalid conversation manifest:\n- " + "\n- ".join(errors))


def build_system_prompt(manifest: dict[str, Any]) -> str:
    agent = manifest["agent"]
    knowledge = manifest["knowledge"]

    style_lines = "\n".join(f"- {item}" for item in agent["style"])
    fact_lines = "\n".join(
        f"- [{fact['id']}] {fact['text']} (source: {fact['source']})"
        for fact in knowledge["facts"]
    )

    return f"""# Role
You are {agent['displayName']}.

# Required disclosure
{agent['disclosure']}

# Speaking style
{style_lines}

# Approved facts
{fact_lines}

# Unknown information
If the approved facts do not support an answer, say:
{knowledge['unknownAnswer']}

# Boundary
Do not claim that you stopped audio playback, cancelled a request,
or changed the application state. Those operations are controlled by the application.
"""


def build_runtime_policy(manifest: dict[str, Any]) -> dict[str, Any]:
    return {
        "schemaVersion": manifest["schemaVersion"],
        "agentId": manifest["agent"]["id"],
        "turnPolicy": manifest["turnPolicy"],
        "localPhrases": manifest["localPhrases"],
        "recoveryMatrix": manifest["recoveryMatrix"],
        "privacy": manifest["privacy"],
    }


def main() -> None:
    manifest = load_manifest("conversation-manifest.json")
    validate(manifest)

    output_dir = Path("dist")
    output_dir.mkdir(exist_ok=True)

    (output_dir / "system-prompt.txt").write_text(
        build_system_prompt(manifest),
        encoding="utf-8",
    )
    (output_dir / "runtime-policy.json").write_text(
        json.dumps(
            build_runtime_policy(manifest),
            ensure_ascii=False,
            indent=2,
        ),
        encoding="utf-8",
    )

    print("conversation contract is valid")


if __name__ == "__main__":
    main()

実行します。

python build_contract.py

成功すれば次のように表示されます。

conversation contract is valid

この生成処理をCIへ入れることで、復旧イベントを削除したまま設定を公開する事故を防げます。

手順3：復旧アクションをアプリ側で実行する

生成した runtime-policy.json を読み、イベントに対応する処理を順番に実行します。

以下はSDK非依存の最小例です。AudioPort、TurnPort、InputPort の中で、実際のRTC、TTS、会話セッション管理へ接続します。

import json
from dataclasses import dataclass, field
from typing import Protocol


class AudioPort(Protocol):
    def stop_tts(self) -> None: ...
    def play_local(self, text: str) -> None: ...


class TurnPort(Protocol):
    def cancel_active_turn(self) -> None: ...


class InputPort(Protocol):
    def open_input(self) -> None: ...


@dataclass
class RecoveryExecutor:
    audio: AudioPort
    turns: TurnPort
    input: InputPort
    policy: dict

    def handle(self, event: str) -> None:
        actions = self.policy["recoveryMatrix"].get(event)
        if actions is None:
            raise KeyError(f"unhandled recovery event: {event}")

        for action in actions:
            if action == "stop_tts":
                self.audio.stop_tts()
            elif action == "cancel_active_turn":
                self.turns.cancel_active_turn()
            elif action == "open_input":
                self.input.open_input()
            elif action.startswith("play_local:"):
                key = action.split(":", 1)[1]
                self.audio.play_local(self.policy["localPhrases"][key])
            else:
                raise ValueError(f"unsupported action: {action}")


@dataclass
class FakePorts:
    calls: list[str] = field(default_factory=list)

    def stop_tts(self) -> None:
        self.calls.append("stop_tts")

    def play_local(self, text: str) -> None:
        self.calls.append(f"play_local:{text}")

    def cancel_active_turn(self) -> None:
        self.calls.append("cancel_active_turn")

    def open_input(self) -> None:
        self.calls.append("open_input")


with open("dist/runtime-policy.json", encoding="utf-8") as f:
    policy = json.load(f)

ports = FakePorts()
executor = RecoveryExecutor(ports, ports, ports, policy)
executor.handle("USER_SPEECH_DURING_TTS")

print(ports.calls)

期待する出力は次のとおりです。

['stop_tts', 'cancel_active_turn', 'open_input']

ここで cancel_active_turn は、ネットワーク上のLLM処理を必ず停止できるという意味ではありません。重要なのは、そのターンが後から完了しても、TTSへ渡さない状態にすることです。プロバイダー側のキャンセル可否とは分けて考えます。

手順4：Tencent Conversational AIとの接続位置を固定する

Tencent Conversational AIへ組み込む際は、生成物を次の位置で利用します。

system-prompt.txt
  └─ 設定したLLM／エージェントへ渡すシステム指示

runtime-policy.json
  ├─ ユーザー発話開始イベント → 割り込み処理
  ├─ アプリ計測の期限超過 → ローカル復旧
  ├─ LLMエラー → ローカル復旧
  └─ STT不確実判定 → 再発話依頼

モデル接続の具体的な設定項目は、公式のLarge Language Model configurationを参照してください。リクエスト識別子を利用する場合は、アプリ側のターンIDと対応を保存しておくと、どの設定・モデルへ送った処理かを追跡しやすくなります。

ただし、識別子へ認証情報や発話本文を埋め込むべきではありません。ランダムまたは非意味的なIDを使い、詳細はアクセス制御されたログ側で関連付けます。

AIコンパニオンやキャラクター対話をソーシャル体験へ組み込むユースケースは、Social Entertainment solutionでも扱われています。音声AIが交流の中心になるほど、AIであることの表示、停止手段、保存範囲、モデレーション方針をUI上でも明示する必要があります。

確認方法：文章品質ではなく契約違反をテストする

test_contract.py を作成します。

import copy
import unittest

from build_contract import load_manifest, validate


class ContractTest(unittest.TestCase):
    def setUp(self) -> None:
        self.manifest = load_manifest("conversation-manifest.json")

    def test_valid_manifest(self) -> None:
        validate(self.manifest)

    def test_rejects_missing_interruption_event(self) -> None:
        broken = copy.deepcopy(self.manifest)
        del broken["recoveryMatrix"]["USER_SPEECH_DURING_TTS"]

        with self.assertRaisesRegex(ValueError, "missing recovery events"):
            validate(broken)

    def test_rejects_unknown_local_phrase(self) -> None:
        broken = copy.deepcopy(self.manifest)
        broken["recoveryMatrix"]["LLM_DEADLINE"] = [
            "cancel_active_turn",
            "play_local:notDefined",
            "open_input",
        ]

        with self.assertRaisesRegex(ValueError, "unknown local phrase"):
            validate(broken)

    def test_interruption_must_cancel_old_turn(self) -> None:
        broken = copy.deepcopy(self.manifest)
        broken["recoveryMatrix"]["USER_SPEECH_DURING_TTS"] = [
            "stop_tts",
            "open_input",
        ]

        with self.assertRaisesRegex(ValueError, "cancel the active turn"):
            validate(broken)

    def test_provider_error_does_not_depend_on_llm(self) -> None:
        broken = copy.deepcopy(self.manifest)
        broken["recoveryMatrix"]["LLM_PROVIDER_ERROR"] = [
            "cancel_active_turn",
            "open_input",
        ]

        with self.assertRaisesRegex(ValueError, "LLM-independent"):
            validate(broken)


if __name__ == "__main__":
    unittest.main()

実行します。

python -m unittest -v

さらに実機では、少なくとも次を確認します。

  • AIの読み上げ中に話すと、再生が止まり、新しい入力を受け付ける
  • 割り込み前のLLM結果が遅れて到着しても読み上げない
  • LLM接続を意図的に失敗させても、復旧文が再度LLMを呼ばずに再生される
  • STTが確定できない場合、勝手に補完して回答しない
  • AIであることの表示と、マイク・会話を止める操作が見つけられる
  • ログに生音声、認証トークン、不要な個人情報が残らない

ネットワーク遅延は環境によって変わるため、llmDeadlineMs に万能な値はありません。少なくとも「ユーザー発話終了」「LLM要求開始」「最初の再生可能音声」「復旧開始」を計測し、端末種別やネットワーク条件ごとに分布を確認します。

判断表：どこをJSON化し、どこを人が決めるか

対象 マニフェスト化 LLMに任せる 人間の判断
名乗り方・口調 ○ 表現のみ 最終承認
回答に使える事実 ○ 事実から文章化 出典の承認
TTS停止 ○ × 条件を決定
遅延時の復旧文 ○ × 文面と待機時間を決定
有人対応への切り替え 条件を記述 提案まで 実行方針を決定
会話ログの保存範囲 ○ × 法務・運用判断

生成AIによって価値が下がるのは、プロンプトを毎回手で複製する作業です。一方で重要性が増すのは、「何をモデルへ任せないか」「失敗をどう観測するか」「誰が停止を決めるか」を仕様とテストへ落とす能力です。

注意点とトレードオフ

1. JSON化してもLLMの遵守は保証されない

system-prompt.txt はモデルへ意図を伝える手段であり、強制的なアクセス制御ではありません。禁止事項、課金、外部操作、個人情報処理などは、アプリ側の検証と権限制御を併用してください。

2. ローカル復旧文は自然さより確実性を優先する

障害時の文言を毎回LLMで生成すると、LLM障害から復旧するために同じLLMが必要になります。多少定型的でも、端末または信頼できるアプリ資産として保持する方が依存関係を減らせます。

3. 割り込みを許可すると誤検知も増える

周囲の音や別話者をユーザー発話として扱う可能性があります。発話開始の判定精度と、停止操作の即時性はトレードオフです。静かな環境だけでなく、スピーカー音量、イヤホン有無、複数人の会話を含めて確認します。

4. マニフェストへ秘密情報を書かない

生成物がCIログや成果物へ残る可能性があります。APIキー、認証トークン、個人の会話履歴はマニフェストに含めず、秘密管理基盤やアクセス制御されたストレージへ分離します。

5. 会話契約にもバージョン管理が必要

モデル名だけでなく、マニフェストのバージョンと生成物を同じリリース証跡へ残します。会話の変化がモデル更新によるものか、人格・復旧方針の変更によるものかを切り分けやすくなります。

まとめ

リアルタイム音声AIをデモから運用へ進めるとき、中心になるのは「自然に話すモデル」だけではありません。

  • 人格と根拠を機械可読な会話契約にする
  • 割り込み、遅延、障害復旧をLLMの外へ出す
  • 契約の欠落をCIで検出する
  • 実測した待ち時間から復旧条件を決める
  • 同意、停止、プライバシーの最終判断を人間に残す

まずは現在のシステムプロンプトから、名乗り方、事実、割り込み、障害時の文言を抜き出してみてください。そのうちTTS停止や復旧に関する記述があれば、プロンプトではなくアプリ側の recoveryMatrix へ移すのが実用的な第一歩です。

関係性の開示：筆者はTencent RTCの技術コンテンツ制作に関わる立場で本稿を執筆しています。実装上の製品情報は、本文に記載したTencent RTC公式ドキュメントを参照しました。

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?