リアルタイム音声AIが同じ発話に2回答するのを防ぐ：STT確定ログをPythonで実装する

結論：LLMへ渡すのは「最新テキスト」ではなく「確定した会話ターン」

Whisperなどで録音を後処理するパイプラインから、リアルタイム音声AIへ進むと、同じ設計をそのまま流用できない場面があります。

リアルタイムSTTでは、次のように認識結果が更新されるためです。

partial: 明日の
partial: 明日の天気を
final:   明日の予定を教えて

各イベントをそのままLLMへ送ると、AIは「天気」と「予定」の両方に回答しかねません。モデルを高性能にしても、この二重起動は直りません。問題は回答能力ではなく、どの文字列を一つのユーザー発話として確定するか、その所有者がいないことです。

本稿では、STTとLLMの間に次の役割を持つ確定ログを置きます。

• 同じ発話の古いrevisionを捨てる
• partialではLLMを起動しない
• final後に短い確定待ち時間を置く
• 一つの発話からLLMリクエストを一度だけ発行する
• 確定後に届いた訂正は、自動再回答せず人が確認できる状態にする
• LLM障害時も同じターンを無制限に再送しない

AIが自然な返答を作れることと、AIに発話確定権を渡すことは別です。リアルタイム音声で人が担うべき重要な判断は、プロンプト作成だけでなく、この確定条件と復旧条件の設計です。

前提：音声経路と判断経路を分ける

今回の対象は、AIコンパニオン、キャラクター会話、音声アシスタントなどのConversational AIシナリオです。

構成を一つの「音声AI API」として扱わず、以下に分離します。

ユーザー音声
  ↓
RTC／メディア転送
  ↓
音声認識（STT）
  ↓ partial / final / revision
確定ログ  ← 今回実装する範囲
  ↓ committed turn
OpenAI・GeminiなどのLLM
  ↓ response
音声合成（TTS）
  ↓
RTC／メディア転送

Tencent Conversational AIは、リアルタイム音声インタラクションと複数のLLMプロバイダーを接続するシナリオを扱います。全体像は公式ドキュメントで確認できます。

• Tencent Conversational AI overview
• Large Language Model configuration

LLM設定ではOpenAI互換モデルやエージェント基盤との接続、リクエスト識別子を使ったルーティング・観測が説明されています。本稿のrequest_idは、その接続境界まで一貫して引き回すためのアプリケーション側識別子です。具体的な製品設定項目は、利用時点の公式ドキュメントに従ってください。

動作環境

• Python 3.11以降
• 外部パッケージなし
• STT、LLM、TTSはテスト用に抽象化

まずディレクトリとファイルを作ります。

mkdir voice-turn-ledger
cd voice-turn-ledger
touch app.py

先に決めるデータモデル

重要なのは、テキストだけでなく発話の同一性と改訂番号を持つことです。

フィールド	役割
session_id	音声セッションを識別する
utterance_id	一続きのユーザー発話を識別する
revision	同じ発話内の認識更新順
text	そのrevision時点の文字列
is_final	STTが確定候補として通知したか
observed_ms	アプリがイベントを観測した時刻

finalを即時送信せず、設定可能なsettle_msだけ待ちます。これは特定製品の性能値ではなく、連続して届く訂正を吸収するためのアプリケーション設定です。実環境のSTTイベント間隔を計測して決めます。

状態は次のように遷移します。

collecting
   ├─ partial更新 ─→ collecting
   └─ final受信 ───→ settling
                         ├─ 新revision ─→ collecting / settling
                         └─ 待機完了 ───→ committed
                                             └─ 遅延訂正 ─→ correction_pending

committedからcollectingへは戻しません。戻すと、すでに読み上げた回答と内部履歴が食い違うためです。

コード：revision付き確定ログを作る

app.pyへ次を保存します。

from dataclasses import dataclass
from typing import Protocol


@dataclass(frozen=True)
class TranscriptEvent:
    session_id: str
    utterance_id: str
    revision: int
    text: str
    is_final: bool
    observed_ms: int


@dataclass
class TurnState:
    latest_revision: int = -1
    latest_text: str = ''
    eligible_at_ms: int | None = None
    committed_revision: int | None = None
    committed_text: str | None = None
    correction_pending: bool = False


@dataclass(frozen=True)
class CommittedTurn:
    session_id: str
    utterance_id: str
    revision: int
    text: str
    request_id: str


class TurnLedger:
    def __init__(self, settle_ms: int = 350):
        if settle_ms < 0:
            raise ValueError('settle_ms must be non-negative')
        self.settle_ms = settle_ms
        self.states: dict[tuple[str, str], TurnState] = {}

    def ingest(self, event: TranscriptEvent) -> str:
        key = (event.session_id, event.utterance_id)
        state = self.states.setdefault(key, TurnState())

        # 再送または順序が逆転した古いイベントは状態を戻さない
        if event.revision <= state.latest_revision:
            return 'stale'

        state.latest_revision = event.revision
        state.latest_text = event.text.strip()

        # すでに発行したターンは自動再発行しない
        if state.committed_revision is not None:
            if state.latest_text != state.committed_text:
                state.correction_pending = True
                return 'late_correction'
            return 'committed_duplicate'

        if not state.latest_text:
            state.eligible_at_ms = None
            return 'empty'

        if event.is_final:
            state.eligible_at_ms = event.observed_ms + self.settle_ms
            return 'settling'

        # より新しいpartialが来たら、それ以前のfinal候補を無効化する
        state.eligible_at_ms = None
        return 'collecting'

    def commit_due(self, now_ms: int) -> list[CommittedTurn]:
        committed: list[CommittedTurn] = []

        for (session_id, utterance_id), state in self.states.items():
            if state.committed_revision is not None:
                continue
            if state.eligible_at_ms is None or state.eligible_at_ms > now_ms:
                continue

            state.committed_revision = state.latest_revision
            state.committed_text = state.latest_text

            request_id = (
                f'{session_id}:{utterance_id}:r{state.latest_revision}'
            )
            committed.append(
                CommittedTurn(
                    session_id=session_id,
                    utterance_id=utterance_id,
                    revision=state.latest_revision,
                    text=state.latest_text,
                    request_id=request_id,
                )
            )

        return committed

    def needs_review(self, session_id: str, utterance_id: str) -> bool:
        state = self.states[(session_id, utterance_id)]
        return state.correction_pending


class LanguageModel(Protocol):
    def answer(self, text: str, request_id: str) -> str:
        ...


class FakeModel:
    def __init__(self, fail_request_ids: set[str] | None = None):
        self.calls: list[str] = []
        self.fail_request_ids = fail_request_ids or set()

    def answer(self, text: str, request_id: str) -> str:
        self.calls.append(request_id)
        if request_id in self.fail_request_ids:
            raise TimeoutError('model timeout')
        return f'受け付けた発話: {text}'


@dataclass(frozen=True)
class ResponseRecord:
    request_id: str
    status: str
    spoken_text: str


class ResponseCoordinator:
    def __init__(self, model: LanguageModel):
        self.model = model
        self.records: dict[str, ResponseRecord] = {}

    def handle(self, turn: CommittedTurn) -> ResponseRecord:
        # commit_dueが誤って複数回呼ばれてもLLMを再起動しない
        existing = self.records.get(turn.request_id)
        if existing is not None:
            return existing

        try:
            text = self.model.answer(turn.text, turn.request_id)
            record = ResponseRecord(turn.request_id, 'completed', text)
        except TimeoutError:
            # モデルに作らせない固定の復旧文
            record = ResponseRecord(
                turn.request_id,
                'recoverable_error',
                '応答を生成できませんでした。もう一度話してください。',
            )

        self.records[turn.request_id] = record
        return record

ここではLLMの出力をそのままTTSへ送る代わりに、ResponseRecordへ保存しています。本番では、このレコードをTTSキューへ渡します。

ポイントは、タイムアウト時の復旧文をLLMへ再度考えさせていないことです。障害中のモデルへ「障害を説明して」と再依頼すると、待ち時間と失敗回数を増やしてしまいます。

手順：認識結果の揺れを再現する

同じapp.pyの末尾へ検証コードを追加します。

def run_demo() -> None:
    ledger = TurnLedger(settle_ms=350)
    model = FakeModel()
    coordinator = ResponseCoordinator(model)

    events = [
        TranscriptEvent('s1', 'u1', 1, '明日の', False, 1000),
        TranscriptEvent('s1', 'u1', 2, '明日の天気を', False, 1100),
        TranscriptEvent('s1', 'u1', 3, '明日の予定を教えて', True, 1200),
    ]

    assert [ledger.ingest(event) for event in events] == [
        'collecting',
        'collecting',
        'settling',
    ]

    # final直後にはまだ発行しない
    assert ledger.commit_due(1500) == []

    # 1200 + 350を過ぎたので一度だけ発行する
    turns = ledger.commit_due(1550)
    assert len(turns) == 1
    assert turns[0].text == '明日の予定を教えて'

    first = coordinator.handle(turns[0])
    second = coordinator.handle(turns[0])

    assert first == second
    assert len(model.calls) == 1
    print(first)

    # 確定後の遅延訂正。二つ目の回答は自動生成しない
    result = ledger.ingest(
        TranscriptEvent(
            's1', 'u1', 4, '明日の予定を変更したい', True, 1800
        )
    )
    assert result == 'late_correction'
    assert ledger.commit_due(3000) == []
    assert ledger.needs_review('s1', 'u1') is True

    # LLMタイムアウト時も固定文でターンを閉じる
    failed_id = 's2:u2:r1'
    failing_model = FakeModel({failed_id})
    failing_coordinator = ResponseCoordinator(failing_model)
    failed_turn = CommittedTurn(
        session_id='s2',
        utterance_id='u2',
        revision=1,
        text='おすすめを教えて',
        request_id=failed_id,
    )
    failed = failing_coordinator.handle(failed_turn)
    assert failed.status == 'recoverable_error'
    assert len(failing_model.calls) == 1
    print(failed)


if __name__ == '__main__':
    run_demo()

実行します。

python app.py

想定される出力は次のとおりです。

ResponseRecord(request_id='s1:u1:r3', status='completed', spoken_text='受け付けた発話: 明日の予定を教えて')
ResponseRecord(request_id='s2:u2:r1', status='recoverable_error', spoken_text='応答を生成できませんでした。もう一度話してください。')

assertion errorが出ず、model.callsが1件なら、同一ターンからLLMが二重起動されていません。

Tencent Conversational AIへ組み込む位置

実サービスでは、各コンポーネントを次の境界で接続します。

1. RTC経由の音声をSTTへ渡す
2. STT固有のイベントをTranscriptEventへ正規化する
3. TurnLedger.commit_due()が返したターンだけをLLMへ送る
4. request_idをLLM接続側の観測・ルーティング情報と関連付ける
5. 成功した回答だけをTTSへ送る
6. TTS音声をRTC経由でユーザーへ返す

Tencent Conversational AIのLLM設定でOpenAI互換モデルへ接続する場合も、STTイベントを直接モデルへ流すのではなく、この確定境界をアプリケーション側に残します。

Geminiを利用する場合は、LanguageModelを満たすアダプターへ置き換えます。OpenAIとGeminiを比較するときも、同じCommittedTurnを入力にすれば、STTの揺れとモデル品質を混同せずに評価できます。

class GeminiAdapter:
    def answer(self, text: str, request_id: str) -> str:
        # 利用する公式SDKでtextを送信する。
        # request_idはアプリログとモデル呼び出しの対応付けに使う。
        raise NotImplementedError


class OpenAICompatibleAdapter:
    def answer(self, text: str, request_id: str) -> str:
        # 接続先の公式仕様に従って実装する。
        raise NotImplementedError

ここで具体的なエンドポイント名や設定フィールドを固定していないのは、接続先と利用方式によって異なるためです。Tencent RTC側の設定は、Large Language Model configurationを参照してください。

モデル選定で見るべきもの

リアルタイム音声では、「どちらのLLMが賢いか」だけでは選べません。確定ログを固定したうえで、次を分けて判断します。

判断項目	機械的に確認するもの	人が決めるもの
重複防止	1ターンあたりの呼び出し回数	訂正時に再質問を促すか
復旧	タイムアウト、空応答、形式不正	固定文の表現と有人導線
応答内容	禁止語、長さ、必須形式	キャラクターらしさ、違和感
モデル切替	同じ入力ログでの成功・失敗	品質と運用負荷の許容範囲
記録	request_idによる追跡	保存期間、同意、閲覧権限

LLMが実際に得意なのは、確定した文脈に対する応答生成や、表現の調整です。一方、発話の確定、再送の可否、保存範囲、安全上の停止条件まで自動的に正しく決めるわけではありません。

この境界を明示すると、「AIに自分の判断を奪われる」という漠然とした不安も整理できます。人の役割は全文を手作業で書くことではなく、何を確定事実としてモデルへ渡し、失敗時にどこで止めるかを設計することへ移ります。

確認方法：回答文より発行回数を検証する

1. partialだけでは回答しない

partial → partial → 無音

期待結果：commit_due()は空配列。LLM呼び出しは0件。

2. finalが訂正されたら新しいrevisionだけを使う

final r3 → settle中にfinal r4 → 待機完了

期待結果：r4のみを一度発行。

3. 古いイベントが遅れて届いても状態を戻さない

final r4 → partial r2

期待結果：r2はstale。

4. 確定後の訂正で二重回答しない

commit r4 → final r5

期待結果：correction_pending=Trueとなり、自動回答は増えない。UIには「聞き取り結果が修正されました」などの確認操作を用意します。

5. LLM障害時に無限再試行しない

期待結果：同じrequest_idに対するモデル呼び出しは一度。固定の復旧文を返し、次のユーザー発話を新しいutterance_idとして受け付けます。

6. 実機で追加確認する

• 短い相槌が独立したutterance_idになるか
• 無音検出とSTT finalの順序が端末やネットワークで変わらないか
• TTS再生中のユーザー発話を別ターンとして取得できるか
• 割り込み時に古いTTSを停止できるか
• 再接続後に以前のfinalが再送されないか
• session_idをまたいで会話履歴が混ざらないか

割り込み停止は確定ログとは別の制御です。ユーザーが話し始めた時点でTTSを止めることと、その発話をLLMへ確定送信することを同じ条件にしないでください。

注意点とトレードオフ

settle_msを長くすると安定するが、返答開始は遅くなる

固定値を勘で決めず、次を記録します。

• 最初のfinalから最後のrevisionまでの時間
• 一つの発話で届いたrevision数
• commit後に訂正が届いた割合
• commitから最初のTTS再生までの時間

その分布を見て設定を変更します。全ユーザー、全言語、全ネットワークに共通する最適値は前提にしません。

finalは意味的な正しさを保証しない

is_final=Trueは、アプリがその文字列を事実として保存してよいという意味ではありません。決済、予約、医療、安全に関わる操作では、認識結果を画面や音声で復唱し、ユーザーの明示確認を別ターンで取得します。

確定後の訂正を無視し続けない

二重回答を防ぐために自動再送を止めても、訂正そのものを隠してはいけません。次のどれかをプロダクト要件として選びます。

• 訂正表示だけ行い、次の発話を待つ
• 「今の発話を修正しますか」と確認する
• 高リスク操作なら処理を保留して再確認する

生の文字起こしを無制限に保存しない

観測性のために全文保存が必要とは限りません。request_id、状態遷移、時間、エラー種別だけで追跡できるケースもあります。音声や文字起こしを保存する場合は、利用目的、保存期間、削除手段、閲覧権限をユーザーに示してください。

ソーシャル用途では安全導線も会話設計に含める

AIコンパニオンやキャラクター会話は、Tencent RTCのSocial Entertainmentシナリオにも含まれます。

• Social Entertainment solution

ただし、親密な会話表現を生成できることと、ユーザーの同意、安全性、モデレーション、有人対応が不要になることは同義ではありません。停止ボタン、ミュート、履歴削除、通報・有人導線はLLMの判断から独立させます。

まとめ

バッチ文字起こしでは、最後に完成した文章だけを見れば済みます。リアルタイム音声AIでは、その完成までに届く途中結果もシステムの入力です。

そのため、STTとLLMの間に次の契約を置く必要があります。

• 発話はutterance_idで識別する
• 更新順はrevisionで比較する
• partialは表示用、finalは確定候補として扱う
• 確定したターンは一度だけ発行する
• 遅延訂正を二つ目の自動回答に変えない
• LLM障害時の復旧文はアプリ側で管理する

OpenAI、Gemini、別のLLMへ切り替えても、この確定ログは残せます。モデル変更より先に会話ターンの発行条件を固定すると、回答品質とリアルタイム制御を別々に改善できます。

---

関係開示：筆者はTencent RTCのコンテンツ制作に関わる立場です。本稿の実装上の参照には、Tencent RTCの公式ドキュメントを使用しました。