音声AIに毎回同じ説明をするのは面倒です。一方で、雑談中の「最近は短い回答のほうが助かる」をAIが永久的な好みとして勝手に保存するのも困ります。

ここにある緊張は、パーソナライズしたいが、推測を事実に昇格させたくないというものです。別途アンケートやフィードバック用チャネルを作ると、今度は誰が確認し、いつ設定へ反映するのかが曖昧になります。

本稿では、リアルタイム音声コンパニオンの確定済み文字起こしからGeminiで「記憶候補」を抽出し、同じ会話内で本人が明示的に承認したものだけを保存するPython実装を作ります。

結論：LLMには「候補と根拠」だけを作らせる

会話メモリを安全に扱う最小構成は次のとおりです。

STTの途中結果ではなく、発話単位の確定テキストを受け取る
Geminiは許可された項目の候補、値、原文中の根拠をJSONで返す
アプリが根拠の存在、機微情報、重複を決定的に検査する
候補はpendingとして保存し、通常の会話は止めない
音声で具体的な確認を取り、承認された候補だけを設定へ昇格する
確認中に割り込まれても、承認や拒否として扱わない
LLM障害時は候補抽出だけを諦め、音声会話は継続する

LLMが得意なのは、表現の揺れた発話から意味的な候補を見つけることです。しかし、長期保存してよいか、センシティブか、本人が本当に意図したかはLLMの文章生成能力だけでは確定できません。そこは人間の明示操作とアプリケーションのルールで制御します。

まず失敗パターンを固定する

正常系より先に、保存してはいけないケースを決めます。

発話	LLMが作り得る候補	アプリの判断
「返事は短めだと助かる」	`response_style=short`	確認候補にする
「友達はホラーが好き」	`favorite_topic=horror`	本人の好みとは限らないため保存しない
「前は夜に話せたけど、今は分からない」	`preferred_call_time=night`	現在の希望が明確でないため保存しない
「はい」	直前候補への同意	汎用的すぎるため承認に使わない
メールアドレスを含む発話	連絡先	許可スキーマ外かつ機微情報として棄却する
AIの確認音声へユーザーが割り込む	拒否または同意	どちらとも解釈せず`pending`を維持する

特に重要なのは、JSON Schemaに適合した出力と、保存してよい情報は別物だという点です。構造化出力は形式崩れを減らしますが、意味上の誤推測までは防ぎません。

前提：音声経路とメモリ抽出経路を分離する

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

Tencent Conversational AI overview

本稿では、次の責任を分けます。

ユーザー音声
   │
   ▼
RTC / 音声伝送
   │
   ▼
STT ── partial transcript ──> 画面表示だけに利用
   │
   └── final transcript
            ├──> 会話LLM ──> TTS ──> RTC
            │
            └──> メモリ候補抽出（Gemini）
                       │
                       ▼
                決定的な検証
                       │
                       ▼
                  pending候補
                       │
                 本人への確認
                       │
              accepted / rejected

候補抽出を会話回答の必須経路に入れないのがポイントです。抽出が失敗しても、そのターンの返答や音声伝送まで失敗させません。

LLM接続について、Tencent RTCの公式ガイドではOpenAI互換モデルやエージェント基盤との接続、およびリクエスト識別子を使ったルーティング・観測の考え方が説明されています。

Large Language Model configuration

Geminiを使う場合も、会話状態から直接SDKを呼び散らかさず、後述するPreferenceExtractor境界へ閉じ込めます。Tencent Conversational AI側へ接続するときは、利用するモデルまたはOpenAI互換ゲートウェイの設定を公式ガイドに合わせてください。

完成するワークフロー

この実装では、長期メモリを二段階に分けます。

final transcript
  ↓
LLM抽出
  ↓
proposed_preferences     ← まだ会話へ反映しない
  ↓ 明示確認
accepted_preferences     ← 次のターン以降で利用可能

保存する候補には、少なくとも以下を持たせます。

CREATE TABLE preference_candidates (
  candidate_id   TEXT PRIMARY KEY,
  request_id     TEXT NOT NULL,
  source_turn_id TEXT NOT NULL,
  key             TEXT NOT NULL,
  proposed_value  TEXT NOT NULL,
  evidence        TEXT NOT NULL,
  status          TEXT NOT NULL,
  expires_at      TEXT NOT NULL
);

CREATE UNIQUE INDEX candidate_dedup
ON preference_candidates(request_id, key, evidence);

request_idはLLM呼び出し、source_turn_idは会話ターンを追跡するための識別子です。同じIDをログへ残しておけば、「どの発話から、どの抽出処理を経て候補が生まれたか」を調査できます。

手順1：検証環境を作る

Python 3.11以降を想定します。

mkdir confirmed-voice-memory
cd confirmed-voice-memory

python -m venv .venv
source .venv/bin/activate

pip install pydantic google-genai pytest

Geminiの認証情報と、利用するモデル名は環境変数へ渡します。

export GEMINI_API_KEY='...'
export GEMINI_MODEL='利用するモデル名'

モデル名をコードへ固定しないことで、候補抽出ロジックとモデル選定を分けられます。

手順2：LLMの出力範囲を小さくする

memory.pyを作成します。

from __future__ import annotations

import json
import os
import re
import uuid
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from typing import Literal, Protocol

from google import genai
from google.genai import types
from pydantic import BaseModel, Field


PreferenceKey = Literal[
    "favorite_topic",
    "preferred_call_time",
    "response_style",
]


class ExtractedPreference(BaseModel):
    key: PreferenceKey
    value: str = Field(min_length=1, max_length=80)
    evidence: str = Field(min_length=1, max_length=200)
    sensitive: bool


class PreferenceBatch(BaseModel):
    items: list[ExtractedPreference] = Field(default_factory=list, max_length=3)


class PreferenceExtractor(Protocol):
    def extract(self, transcript: str) -> PreferenceBatch:
        ...


class GeminiExtractor:
    def __init__(self) -> None:
        self.client = genai.Client(api_key=os.environ["GEMINI_API_KEY"])
        self.model = os.environ["GEMINI_MODEL"]

    def extract(self, transcript: str) -> PreferenceBatch:
        prompt = f"""
あなたは音声コンパニオンの設定候補抽出器です。
次の制約を必ず守ってください。

- ユーザー本人が現在の希望として明示した内容だけを候補にする
- 他人の好み、過去だけの習慣、推測、皮肉、否定された内容は除外する
- evidenceには入力中に実在する連続した原文をそのまま入れる
- 住所、連絡先、健康、政治、宗教、金融、認証情報などは候補にしない
- 候補がなければitemsを空配列にする
- 候補は最大3件

入力:
{transcript}
""".strip()

        response = self.client.models.generate_content(
            model=self.model,
            contents=prompt,
            config=types.GenerateContentConfig(
                response_mime_type="application/json",
                response_schema=PreferenceBatch,
            ),
        )

        if not response.text:
            raise ValueError("Gemini returned an empty response")

        return PreferenceBatch.model_validate_json(response.text)

許可するキーをLiteralで限定しているため、モデルが住所や勤務先など別の項目を追加してもPydanticの検証を通りません。

ただし、モデルがsensitive=falseと返しただけで安全と判断してはいけません。次の手順でアプリ側の検査を追加します。

手順3：原文根拠のない候補を捨てる

同じmemory.pyへ追加します。

SENSITIVE_PATTERNS = [
    re.compile(r"[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}"),
    re.compile(r"\b(?:\d[ -]?){10,13}\b"),
]


@dataclass
class ProposedPreference:
    candidate_id: str
    request_id: str
    source_turn_id: str
    key: PreferenceKey
    value: str
    evidence: str
    status: Literal["pending", "accepted", "rejected", "expired"]
    expires_at: datetime


def contains_sensitive_text(text: str) -> bool:
    return any(pattern.search(text) for pattern in SENSITIVE_PATTERNS)


def validate_candidate(
    transcript: str,
    item: ExtractedPreference,
) -> bool:
    # LLMが存在しない引用を生成した場合は棄却する。
    if item.evidence not in transcript:
        return False

    # モデル自身の機微判定は信用せず、アプリ側でも止める。
    if item.sensitive:
        return False

    if contains_sensitive_text(item.evidence):
        return False

    return True

この検査は完全な個人情報検出器ではありません。目的は「LLMに安全判定を丸投げしない」ことです。本番では対象地域の法令、保存方針、モデレーション要件に合わせた検出処理へ置き換えてください。

手順4：候補抽出を会話の成否から切り離す

続いて、実行時の状態を追加します。

class MemoryRuntime:
    def __init__(self, extractor: PreferenceExtractor) -> None:
        self.extractor = extractor
        self.candidates: dict[str, ProposedPreference] = {}
        self.preferences: dict[PreferenceKey, str] = {}
        self.events: list[dict] = []

    def ingest_final_transcript(
        self,
        source_turn_id: str,
        transcript: str,
    ) -> list[ProposedPreference]:
        request_id = str(uuid.uuid4())

        try:
            batch = self.extractor.extract(transcript)
        except Exception as exc:
            # 候補抽出の失敗で、通常の音声会話を失敗させない。
            self._record(
                "extraction_failed",
                request_id=request_id,
                source_turn_id=source_turn_id,
                error=type(exc).__name__,
            )
            return []

        proposed: list[ProposedPreference] = []

        for item in batch.items:
            if not validate_candidate(transcript, item):
                self._record(
                    "candidate_discarded",
                    request_id=request_id,
                    source_turn_id=source_turn_id,
                    key=item.key,
                )
                continue

            candidate = ProposedPreference(
                candidate_id=str(uuid.uuid4()),
                request_id=request_id,
                source_turn_id=source_turn_id,
                key=item.key,
                value=item.value,
                evidence=item.evidence,
                status="pending",
                expires_at=datetime.now(timezone.utc) + timedelta(minutes=10),
            )
            self.candidates[candidate.candidate_id] = candidate
            proposed.append(candidate)

            self._record(
                "candidate_proposed",
                candidate_id=candidate.candidate_id,
                request_id=request_id,
                source_turn_id=source_turn_id,
                key=candidate.key,
            )

        return proposed

    def _record(self, event: str, **fields: object) -> None:
        self.events.append({
            "event": event,
            "at": datetime.now(timezone.utc).isoformat(),
            **fields,
        })

    def dump_jsonl(self) -> str:
        return "\n".join(
            json.dumps(event, ensure_ascii=False)
            for event in self.events
        )

partial transcriptからこのメソッドを呼ばないでください。途中結果には訂正前の単語や未完了の否定表現が含まれます。

たとえば「長い返事が……いや、短いほうがいい」という発話を途中で解析すると、相反する候補が作られます。発話確定後も意味が曖昧なら、候補を作らないほうへ倒します。

手順5：「はい」では承認しない

確認用の音声は、候補を具体的に読み上げます。

「返答は短めがよい」という設定を記憶してもよいですか。保存する場合は「この設定で保存して」、保存しない場合は「記憶しないで」と言ってください。

実装を追加します。

ACCEPT_PHRASES = {
    "この設定で保存して",
    "この好みを覚えておいて",
}

REJECT_PHRASES = {
    "記憶しないで",
    "この候補は削除して",
}


def normalize_confirmation(text: str) -> str:
    return re.sub(r"[\s、。！？!?]", "", text)


NORMALIZED_ACCEPT = {
    normalize_confirmation(value) for value in ACCEPT_PHRASES
}
NORMALIZED_REJECT = {
    normalize_confirmation(value) for value in REJECT_PHRASES
}


def apply_confirmation(
    runtime: MemoryRuntime,
    candidate_id: str,
    transcript: str,
) -> Literal["accepted", "rejected", "ambiguous", "expired"]:
    candidate = runtime.candidates[candidate_id]
    now = datetime.now(timezone.utc)

    if now >= candidate.expires_at:
        candidate.status = "expired"
        runtime._record("candidate_expired", candidate_id=candidate_id)
        return "expired"

    normalized = normalize_confirmation(transcript)

    if normalized in NORMALIZED_ACCEPT:
        candidate.status = "accepted"
        runtime.preferences[candidate.key] = candidate.value
        runtime._record(
            "candidate_accepted",
            candidate_id=candidate_id,
            key=candidate.key,
        )
        return "accepted"

    if normalized in NORMALIZED_REJECT:
        candidate.status = "rejected"
        runtime._record("candidate_rejected", candidate_id=candidate_id)
        return "rejected"

    runtime._record(
        "confirmation_ambiguous",
        candidate_id=candidate_id,
    )
    return "ambiguous"

「はい」は通常会話にも頻出します。確認音声の途中でユーザーが「はい、ところで明日の……」と割り込んだ場合まで承認にすると、意図しない保存が起きます。そのため、ここでは具体的なフレーズだけを受理します。

音声UXとして不自然なら、アプリ画面に「保存」「今回は保存しない」ボタンを併設する方法もあります。音声だけにこだわるより、ユーザーが結果を確認できることを優先します。

割り込みと復旧をどう扱うか

確認音声をTTSで再生中にユーザーの発話が始まった場合は、次の順番にします。

1. ユーザー発話開始を検知
2. 確認TTSの再生をキャンセル
3. 候補はpendingのまま維持
4. ユーザーの新しい発話を通常ターンとして処理
5. 明示的な確認文なら対象候補へ適用
6. それ以外なら会話を続け、候補は期限切れまで保留

割り込み自体を拒否とみなす必要はありません。また、TTSの再生完了を保存条件にしてもいけません。保存条件は、あくまで本人の明示確認です。

切断後に再接続した場合、古い確認を突然読み上げると文脈が失われます。再接続時は次のどちらかを製品ポリシーとして選びます。

すべてのpending候補を期限切れにする
候補一覧を画面表示し、ユーザー操作で確認を再開する

会話相手らしさを優先するコンパニオンでも、同意を自然さのために省略しないことが重要です。AIバーチャルコンパニオンを含むソーシャル用途の位置付けは、Tencent RTCのソリューションページでも確認できます。

Social Entertainment solution

確認方法：文章の自然さではなく状態遷移をテストする

test_memory.pyを作ります。

from memory import (
    ExtractedPreference,
    MemoryRuntime,
    PreferenceBatch,
    apply_confirmation,
)


class FakeExtractor:
    def __init__(self, items):
        self.items = items

    def extract(self, transcript: str) -> PreferenceBatch:
        return PreferenceBatch(items=self.items)


def test_explicit_evidence_becomes_pending():
    runtime = MemoryRuntime(FakeExtractor([
        ExtractedPreference(
            key="response_style",
            value="short",
            evidence="返事は短めだと助かる",
            sensitive=False,
        )
    ]))

    candidates = runtime.ingest_final_transcript(
        "turn-1",
        "返事は短めだと助かる",
    )

    assert len(candidates) == 1
    assert candidates[0].status == "pending"
    assert runtime.preferences == {}


def test_hallucinated_evidence_is_discarded():
    runtime = MemoryRuntime(FakeExtractor([
        ExtractedPreference(
            key="favorite_topic",
            value="science fiction",
            evidence="SFが大好き",
            sensitive=False,
        )
    ]))

    candidates = runtime.ingest_final_transcript(
        "turn-2",
        "最近は何を見るか決めていない",
    )

    assert candidates == []


def test_generic_yes_does_not_accept():
    runtime = MemoryRuntime(FakeExtractor([
        ExtractedPreference(
            key="response_style",
            value="short",
            evidence="短めに答えて",
            sensitive=False,
        )
    ]))

    candidate = runtime.ingest_final_transcript(
        "turn-3",
        "短めに答えて",
    )[0]

    result = apply_confirmation(runtime, candidate.candidate_id, "はい")

    assert result == "ambiguous"
    assert candidate.status == "pending"
    assert runtime.preferences == {}


def test_explicit_confirmation_accepts():
    runtime = MemoryRuntime(FakeExtractor([
        ExtractedPreference(
            key="response_style",
            value="short",
            evidence="短めに答えて",
            sensitive=False,
        )
    ]))

    candidate = runtime.ingest_final_transcript(
        "turn-4",
        "短めに答えて",
    )[0]

    result = apply_confirmation(
        runtime,
        candidate.candidate_id,
        "この設定で保存して。",
    )

    assert result == "accepted"
    assert runtime.preferences["response_style"] == "short"


def test_sensitive_candidate_is_discarded():
    runtime = MemoryRuntime(FakeExtractor([
        ExtractedPreference(
            key="preferred_call_time",
            value="evening",
            evidence="連絡先はuser@example.com",
            sensitive=False,
        )
    ]))

    candidates = runtime.ingest_final_transcript(
        "turn-5",
        "連絡先はuser@example.com",
    )

    assert candidates == []

実行します。

pytest -q

Geminiへの実接続を確認する前に、FakeExtractorで状態遷移を固定してください。実モデルだけでテストすると、同じ入力でも候補数や表現が変わり、アプリ側の安全条件が検証しにくくなります。

RTC統合時の検証チェックリスト

実際の音声セッションへ接続したら、以下を手作業でも確認します。

STTの途中結果から候補が作られない
候補抽出中でも通常回答とTTSが継続する
Geminiのタイムアウトや不正JSONで会話が終了しない
原文に存在しないevidenceが棄却される
候補生成だけではユーザー設定が変わらない
「はい」「うん」だけでは承認されない
確認TTSへの割り込みで候補が自動承認されない
割り込み後に古い確認音声が再開されない
期限切れ候補を承認できない
同じrequest_idとsource_turn_idで処理経路を追跡できる
保存済みの好みをユーザーが閲覧・削除できる
生の文字起こしを保存する期間が明示されている

Tencent Conversational AIとの統合では、実際のSDKや管理画面にある正式なイベント・設定を利用してください。本稿のingest_final_transcriptやapply_confirmationはアプリケーション側の境界であり、Tencent RTCのAPI名ではありません。

判断フレームワーク：何を自動保存してよいか

候補の種類ごとに保存方法を変えると、過剰な同意画面を避けられます。

情報	例	推奨処理
現在ターンだけの指示	「もう少し短く」	セッション内で即時反映し、長期保存しない
低リスクな継続設定	「今後も短めに答えて」	候補化し、明示確認後に保存
文脈依存の推測	「最近夜に話すことが多い」	保存せず、必要なら質問する
センシティブ情報	健康、認証、住所、金融	通常の会話メモリから除外する
外部操作につながる設定	予約、購入、連絡	別の確認・権限・監査経路を使う

候補を多く抽出するほど便利になるとは限りません。確認回数が増えると会話が途切れます。最初はresponse_styleなど、ユーザーが効果を理解しやすく、削除もしやすい項目だけに限定するのが現実的です。

注意点とトレードオフ

1. 根拠文字列の完全一致は保守的

句読点やSTTの表記揺れで、正しい候補まで棄却することがあります。ただし、最初から曖昧一致へ広げると、モデルが作った似た文章を根拠として通す可能性があります。

まず完全一致で運用し、棄却ログを確認してから、正規化規則を一つずつ追加してください。

2. 構造化出力でも意味は保証されない

PydanticとJSON Schemaが保証するのは主に形式です。「友達の好み」を「本人の好み」と誤認するような意味上の失敗は、固定テスト、根拠照合、確認操作で補います。

3. 確認を急ぐとターンテイキングが悪化する

好みが見つかるたびに会話へ割り込むと、AIがユーザーの話を奪います。確認は話題の区切り、会話終了前、設定画面などへ遅延できます。ただし、遅延中の候補を既成事実として回答生成へ使ってはいけません。

4. ログにもプライバシー境界が必要

観測可能性のために識別子は必要ですが、生の音声や全文文字起こしを無期限に残す必要はありません。候補、根拠、保持期間、削除手段を分け、運用担当を決めてください。

まとめ

LLMによる意味抽出は、音声コンパニオンが会話からユーザーの希望を見つける工程には役立ちます。しかし、抽出結果をそのまま長期メモリへ入れる設計では、人間がどこで判断したのか分からなくなります。

実装上の要点は次の5つです。

確定済み文字起こしだけを解析する
LLM出力を小さなスキーマへ制限する
原文根拠と機微情報をアプリ側で検査する
候補と保存済み設定を別の状態にする
割り込みや障害を承認として解釈しない

「AIが覚えてくれる」ではなく、AIが候補を見つけ、人間が何を覚えさせるか決められる状態を目標にすると、便利さと制御を両立しやすくなります。

関係開示: 本稿はTencent RTCのコミュニティ向けコンテンツとして作成しており、実装上の製品情報はTencent RTC公式ドキュメントを参照しています。