Posted inAI

LiveKit AgentsとPythonで構築するリアルタイム音声エージェント：超低遅延会話型AI実践ガイド

AI tutorial - IT technology blog
AI tutorial - IT technology blog

課題提起：なぜ従来の音声ボットは不自然に感じられるのか

過去10年間に構築されてきた音声ボットの大半は、非効率なリクエスト・レスポンス方式に依存しています。クライアントが無音を検知するまで音声を録音し、WAVファイルにパッケージ化してHTTP POSTリクエストを送信します。バックエンドはそこから音声認識（STT）による文字起こしを待ち、LLMに問い合わせ、音声合成（TTS）でオーディオを生成します。その結果、発信者は最初の1音を聞くまでに、通常3〜6秒もの無音状態（デッドエア）に耐えなければなりません。

人間同士の会話において、これほどの遅延は許容されません。自然な対話における発話交代の間の時間は200〜300ミリ秒程度です。無音が600ミリ秒を超えると、会話は途端にぎこちなく、機械的な印象を与えてしまいます。

さらに悪いことに、標準的なREST構成ではユーザーの割り込み（バージイン）に対応できません。ボットが話している最中にユーザーが口を挟んでも、クライアントはバッファ内のストリームを再生し終えるまで古い音声を流し続けてしまいます。本番環境で通用する音声エージェントを構築するには、音声バッファの動的な中断（トランケーション）と、低遅延トランスポートを介した生音声のストリーミングという2つの根本課題を解決する必要があります。

コア概念：WebRTC、VAD、そしてLiveKitフレームワーク

1秒未満の応答速度で会話を実現するには、継続的な音声ストリーミングが不可欠です。これを可能にするのが、以下の3つのアーキテクチャ上のアプローチです。

1. リアルタイムトランスポートとしてのWebRTC

RESTやWebSocketはTCP上で動作します。ネットワークの一時的な不安定が発生すると、TCPはパケット再送のために処理を停止（ストール）させ、顕著な音声ジッターを引き起こします。WebRTCはTCPの再送処理を排除し、UDPベースのRTPおよびSRTPチャネルを採用しています。ネイティブのジッターバッファと適応型ビットレート制御（ABR）により、安定した回線であれば往復（RTT）の音声遅延を120ミリ秒未満に抑えることができます。

2. ストリーミング音声パイプライン

従来の順次バッチ処理から脱却しましょう。高性能な音声エンジンは、20ミリ秒単位の音声チャンクを以下の4つのパイプラインステージを通じてストリーミング処理します。

  • 音声活動検知（VAD）： 16kHzの未加工PCM音声フレームを検査します。Silero VADはCPU上で直接動作し、30ミリ秒以内に人間の発話を識別します。
  • ストリーミング音声認識（STT）： 発話中に部分的な音声チャンクを順次文字起こしします。Deepgram Nova-2などのプロバイダーは、約150〜200ミリ秒で逐次テキストを出力します。
  • ストリーミング大規模言語モデル（LLM）： 確定した発話セグメントを受け取り、即座にトークンを生成します。GPT-4o-miniのような高速モデルは、250〜400ミリ秒で最初のトークンを出力します。
  • ストリーミング音声合成（TTS）： 段落全体の完了を待つことなく、フレーズ単位で音声を合成します。クライアントへの最初の音声バッファは約180ミリ秒でストリーミング配信されます。

3. LiveKit Agentsのアーキテクチャ

LiveKitは、オープンソースのWebRTCメディアサーバーとPythonフレームワーク（livekit-agents）を提供します。ICEネゴシエーション、シグナリング、トラックのミキシングといった複雑なコードを独自に実装する必要はありません。Pythonサービスは自律的なWebRTCルーム参加者として機能します。ユーザーが接続すると、LiveKitサーバーがワーカープロセスにジョブをディスパッチします。ワーカーは音声トラックをアタッチしてパイプラインを起動し、音声入力から応答出力までのターンアラウンドタイム800ミリ秒未満を目指します。

ハンズオン実践：Pythonによるリアルタイム音声エージェントの構築

ステップ1：環境構築

Python 3.10以降を使用して仮想環境をセットアップします。LiveKitエージェントランタイムと公式のプロバイダープラグインをインストールしましょう。

mkdir voice-agent && cd voice-agent
python3 -m venv .venv
source .venv/bin/activate

pip install livekit-agents livekit-plugins-openai livekit-plugins-silero livekit-plugins-deepgram python-dotenv

ステップ2：環境変数の設定

LiveKitインスタンス（Cloudまたはセルフホスト）および各AIモデルの認証情報を設定します。これらを.envファイルに保存します。

LIVEKIT_URL=wss://your-project.livekit.cloud
LIVEKIT_API_KEY=your_api_key
LIVEKIT_API_SECRET=your_api_secret

OPENAI_API_KEY=your_openai_key
DEEPGRAM_API_KEY=your_deepgram_key

ステップ3：エージェントワーカーの実装

以下のコードをagent.pyとして保存します。ここでは、Silero VAD、Deepgram STT、OpenAIモデルを連携させ、割り込み可能なループを構築するVoicePipelineAgentを活用します。

import os
from dotenv import load_dotenv
from livekit.agents import (
    AutoSubscribe,
    JobContext,
    JobProcess,
    WorkerOptions,
    cli,
    llm,
)
from livekit.agents.pipeline import VoicePipelineAgent
from livekit.plugins import deepgram, openai, silero

load_dotenv()

def prewarm_process(proc: JobProcess):
    # 発信者参加時のコールドスタートを回避するため、
    # VADのネットワーク重みをメモリに事前ロード
    proc.userdata["vad"] = silero.VAD.load()

async def entrypoint(ctx: JobContext):
    # ルームに参加して発信者の音声トラックを受信
    await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)

    initial_chat_ctx = llm.ChatContext().append(
        role="system",
        text=(
            "あなたはインフラチーム向けの自動テクニカルサポート音声アシスタントです。"
            "回答は簡潔かつ明瞭で、端的な表現を心がけてください。出力内容はそのまま音声として読み上げられるため、"
            "マークダウン、箇条書き、絵文字は使用しないでください。"
        ),
    )

    agent = VoicePipelineAgent(
        vad=ctx.proc.userdata["vad"],
        stt=deepgram.STT(model="nova-2"),
        llm=openai.LLM(model="gpt-4o-mini"),
        tts=openai.TTS(voice="alloy"),
        chat_ctx=initial_chat_ctx,
        allow_interruptions=True,
        interrupt_speech_duration=0.5,
    )

    agent.start(ctx.room)
    await agent.say("こんにちは！接続が完了し、準備が整いました。本日はどのシステムのトラブルシューティングを行いますか？")

if __name__ == "__main__":
    cli.run_app(WorkerOptions(entrypoint_fnc=entrypoint, prewarm_fnc=prewarm_process))

ステップ4：エージェントの実行とテスト

開発モードでエージェントをローカル実行します。

python agent.py dev

テストのために独自のフロントエンドを用意する必要はありません。agents-playground.livekit.ioにアクセスし、URLとAPI認証情報を入力して接続するだけです。

ヘッドセットに向かって話し、パイプラインの動作を確認してみましょう。

  1. Deepgramが200ミリ秒以内に文字起こしテキストをターミナルへ出力します。
  2. Silero VADが500ミリ秒の無音を検知し、発話の区切りを認識します。
  3. OpenAIがトークンを生成し、約650〜800ミリ秒で合成音声が再生されます。
  4. ボットが回答している最中に話しかけてみてください。エージェントが音声を検知し、100ミリ秒未満で発話を中断して聞き取り状態へと切り替わります。

ステップ5：本番運用のためのFunction Callingの追加

音声ボットの真価は、会話の途中で実際のシステムに問い合わせを行う際に発揮されます。@llm.ai_callableデコレータを使用してPythonツールを定義します。

class ServerOps(llm.FunctionContext):
    @llm.ai_callable(description="本番ホストのCPUおよびメモリ使用率を確認する")
    def get_server_metrics(self, hostname: str) -> str:
        if "db" in hostname.lower():
            return f"{hostname}のCPU使用率は88%、メモリ使用率は92%です。"
        return f"{hostname}のCPU使用率は14%、メモリ使用率は35%です。"

# fnc_ctxをVoicePipelineAgentに渡す:
# agent = VoicePipelineAgent(..., fnc_ctx=ServerOps())

「db-node-01のヘルス状態を確認してくれますか？」と話しかけてみましょう。エージェントは音声合成を保留し、get_server_metricsを実行して結果を解析し、メトリクスを自然な言葉で読み上げます。

まとめ：プロトタイプから堅牢な音声インフラへの移行

WebRTCによるメディアストリーミングは、会話型AIのあり方を根本から変えます。メディアトランスポートやパケット損失のリカバリをLiveKitに任せることで、対話の往復レイテンシを人間が躊躇を感じる閾値よりもはるかに低く抑えられます。

本番環境へデプロイする前に、以下の3つの運用要素を必ず確認してください。

  • ワーカーの事前ウォーミング（Prewarming）： prewarm_fnc内でVADのモデル重みやトークナイザーを事前ロードしておきます。これを怠ると、初回の接続時に1.2〜2秒の遅延が発生します。
  • クラウドの近接配置（Colocation）： リージョン間通信による不要なホップをなくすため、モデルのエンドポイントからPing値で20ミリ秒以内の距離（例：OpenAIやDeepgram向けにはus-east-1など）にワーカーノードを配置します。
  • サイジング設計： 音声フレームの処理はCPUを継続的に消費します。一般的な2 vCPU / 4GB RAM構成のワーカーノード1台で、安定して処理できる同時接続音声ルーム数は15〜25程度です。これを超える負荷には水平オートスケーリングで対応します。
Share:
Tags: