Audio Customization

参照元: LiveKit Agents Documentation ロードマップ: 学習ロードマップ

LiveKit Agentsはエージェントの音声をカスタマイズする3つの機能を提供する。TTSキャッシュ、発音カスタマイズ、音量調整だ。このページは /agents/multimodality/audio/customization/ に独立しており、以前は audio/#cached-tts-in-tools のアンカーセクションだったものが昇格した構成になっている。

graph TD
    A["Audio Customization"] --> B["Cached TTS"]
    A --> C["Pronunciation"]
    A --> D["Volume"]
    B --> B1["Pre-synthesized"]
    B --> B2["Auto caching by key"]
    B --> B3["Tool hold message"]
    C --> C1["Text replacement via tts_node"]
    C --> C2["SSML tags"]
    D --> D1["audioop.mul in tts_node"]
    D --> D2["audioop.mul in realtime_audio_output_node"]

Cached TTS — TTSキャッシュ

固定フレーズ（挨拶、保留メッセージ、エラープロンプト等）を事前に音声合成しておき、session.say(text, audio=...) に渡すことでTTS API呼び出しをスキップし、レイテンシとコストを削減する。text はトランスクリプト・チャットコンテキスト用、audio は実際の再生用という役割分離の設計になっている。

Pre-synthesized（事前合成・録音済み）

audio_frames_from_file() で音声ファイルを読み込み、session.say() の audio 引数に渡す。フレーズが固定・安定しているケースに適する。

from livekit.agents.utils.audio import audio_frames_from_file
 
await session.say(
    "Your phrase",
    audio=audio_frames_from_file(path, sample_rate=24000, num_channels=1),
)

管理方法としては、WAVファイルをコードと同階層に置く方法、起動時にTTS APIで合成する方法、外部ストレージ（S3等）から取得する方法がある。実戦的には「起動時TTS合成」が推奨される。テキストとコードが1箇所に集まり、TTSプロバイダー変更時も音声が追従し、サンプリングレート不一致事故を防げるためだ。

Automatic caching（テキストキーによる自動キャッシュ）

初回合成時に dict[str, list[AudioFrame]] にテキストをキーとしてキャッシュし、2回目以降はキャッシュヒットしたフレームを session.say() に渡す。同じフレーズがセッション内・セッション跨ぎで繰り返し使われるケースに適する。

tts_cache: dict[str, list[rtc.AudioFrame]] = {}
 
async def say_cached(session, tts, text):
    if text not in tts_cache:
        stream = tts.synthesize(text)
        frames = []
        async for event in stream:
            frames.append(event.frame)
        tts_cache[text] = frames
 
    async def audio_gen():
        for frame in tts_cache[text]:
            yield frame
 
    await session.say(text, audio=audio_gen())

フレームワーク標準機能ではない理由は、メモリ管理の責任をユーザーに委ねるためだ。LLM生成テキストをキャッシュするとヒット率が低くメモリが肥大化する。またパイプライン出力（LLM生成音声）のキャッシュも、カスタムTTSノード内で実装可能だが、キャッシュルックアップに全文が必要になりtime-to-first-byteが増加するトレードオフがある。

Cached TTS in a tool call（ツール内保留メッセージ）

Function toolの実行開始時にキャッシュ済みの保留メッセージを流し、外部APIの結果が先に返ったら即座に中断するパターン。音声エージェントにおける最も実戦的なキャッシュTTSユースケースだ。

HOLD_FRAMES: list[rtc.AudioFrame] = []
 
async def preload_hold_message(tts):
    global HOLD_FRAMES
    async for event in tts.synthesize("Let me check that for you."):
        HOLD_FRAMES.append(event.frame)
 
class MyAgent(Agent):
    @function_tool()
    async def check_order_status(self, context, order_id: str) -> str:
        async def cached_audio():
            for frame in HOLD_FRAMES:
                yield frame
 
        # awaitしない — 並行実行のため
        hold_handle = context.session.say(
            "Let me check that for you.",
            audio=cached_audio(),
            add_to_chat_ctx=False,
        )
 
        result = await fetch_order_status(order_id)
 
        if not hold_handle.interrupted and not hold_handle.done():
            hold_handle.interrupt()
 
        return result

このパターンの要点は4つある。say() を await しないことで保留メッセージとAPI呼び出しを並行実行すること。add_to_chat_ctx=False で保留メッセージがチャット履歴を汚染しないこと。hold_handle.interrupt() でAPI先返却時に即座に切り替えること。そして起動時に1回だけTTS合成するためツール呼び出しごとのレイテンシがゼロになることだ。

Pronunciation — 発音カスタマイズ

テキスト置換（custom tts_node）

tts_node 内でテキストを正規表現で置換し、TTSエンジンに渡す前に読み方を調整する。ストリーミング対応で、AsyncIterable[str] のチャンク単位で処理する。\b で単語境界マッチ、re.IGNORECASE で大文字小文字を無視する。自分でTTS合成するのではなく、テキストだけ操作して標準パイプラインに委譲する設計だ。

SSML（Speech Synthesis Markup Language）

W3C定義のXMLベースマークアップ言語。TTSエンジンに発音・ピッチ・速度・ポーズ・強調をタグで指示する。主なタグとして phoneme（IPA表記で発音指定）、say-as（テキストの解釈方法指定）、break（ポーズ挿入）、emphasis（強調）、prosody（ピッチ・速度・音量）がある。

ElevenLabsはSSMLのサブセットをサポートする。phoneme、break、say-as は動作するが、prosody はプロバイダー依存で挙動が異なる。LiveKit Agentで利用する場合、custom tts_node 内でテキストを <speak> タグでラップしてTTSに渡す。ただしストリーミング環境ではチャンク境界でSSMLタグが途切れる可能性があるため注意が必要だ。

実戦的な判断としては、略語の読み方（API → A P I）はテキスト置換、固有名詞の精密な発音制御にはSSML phoneme、意図的なポーズには break を使い、基本はテキスト置換で対応するのが推奨される。

Volume — 音量調整

tts_node または realtime_audio_output_node 内で、Python標準ライブラリ audioop.mul() を使ってAudioFrameの振幅を調整する。

import audioop
 
async def adjust_volume(input_frames):
    async for event in input_frames:
        frame = event.frame
        adjusted_data = audioop.mul(
            frame.data,
            frame.sample_width,
            1.5,  # 倍率
        )
        yield voice.SynthesizedAudio(
            text=event.text,
            data=rtc.AudioFrame(
                data=adjusted_data,
                sample_rate=frame.sample_rate,
                num_channels=frame.num_channels,
                samples_per_channel=frame.samples_per_channel,
            ),
            request_id=event.request_id,
        )

既存フレームを直接書き換えず、新しいAudioFrameを生成する。倍率の上限は2.0程度が目安で、上げすぎるとクリッピングで音が割れる。audioop はPython 3.13で非推奨となっており、将来的にはnumpy等への移行が必要だ。