はじめに
大言語モデル（LLM）の実力については、今更言うまでもないでしょう。しかし、「チャットができること」と「通話ができること」は全くの別物です。
今年より前、AI をリアルタイムの音声対話シナリオに導入するには、通常、STT（音声認識）→ LLM（意味理解）→ TTS（音声合成）という 3 つのパイプラインを自分で繋ぎ合わせる必要がありました。その上、VAD（音声活動検出）、文の区切り、割り込み、並行処理といった大量のエッジケースに対処しなければなりませんでした。少しでも処理を誤ると、ユーザーが「うん」と相槌を打っただけで AI が言葉を遮ってしまったり、認識の遅延が 2〜3 秒に膨れ上がり、音声対話というよりはトランシーバーでのやり取りのようになってしまったりしました。
しかし、LiveKit Agents というフレームワークに出会ったことで、その状況は一変しました。
この記事では、チュートリアルを書くつもりはありません（それらはネット上にすでにたくさんあります）。私が共有したいのは、実際の IM（インスタントメッセージング）や通話ビジネスにおいて、LiveKit Agents をバックエンドアーキテクチャに組み込む際に直面した典型的な課題とその解決策です。
一、 選定：なぜ Twilio や Agora ではなく LiveKit なのか
まず簡単に背景を説明します。すでに IM と WebRTC 通話機能を備えている既存のシステムに、AI 音声アシスタントの機能を追加する必要がありました。
市場にある選択肢は、大きく分けて以下の 3 つのカテゴリに分類されます。
| 選択肢 | 代表例 | 評価 |
|---|---|---|
| 完全マネージド型 | Twilio Voice + OpenAI | 手間はかからないが、高コスト。データが自社管理下になく、カスタマイズの余地が少ない。 |
| 自作パイプライン | 自前で STT → LLM → TTS を結合 | 柔軟性は高いが、開発量が膨大。VAD、並行処理、文の区切りなどを全て自前で実装する必要がある。 |
| Agent フレームワーク | LiveKit Agents, Vocode, Pipecat | 折衷案。低レイヤーの詳細はフレームワークが処理し、開発者はビジネスロジックの記述に集中できる。 |
最終的に LiveKit Agents を選択した理由は非常にシンプルです。**「Agent API の抽象化が最も優れているから」**です。
LiveKit Agents は、「リアルタイムの音声・ビデオ会議室に AI アシスタントを参加させる」という処理を、標準的な Job/Worker モデルとして抽象化しています。
会議室の作成 ──> LiveKit Server が Job を発行 ──> Worker が Job を消費 ──> Agent が会議室に参加 ──> 対話開始
このモデルは、既存の IM 通話フローに非常にうまく適合しました。私のシステムでは、通話用の会議室を作成するロジックはすでに実装されていたため、会議室作成後に AI を参加者の一人として追加するだけで済みました。
二、 アーキテクチャの分解
実際のデプロイアーキテクチャは以下の通りです。
クライアント（ブラウザ / アプリ）
↕ (WebRTC 音声ストリーム)
LiveKit Server（セルフホスト）
↕ (音声ストリーム)
Python Agent Worker（LiveKit Agents）
↕ (HTTP: コンテキストの取得 + 履歴の送信)
Go メインバックエンド（ビジネスサービス）
設計における最も重要な原則はただ一つです。**「Agent はビジネスロジックを持たず、『音声 ↔ テキスト』の変換のみを行う」**ということです。
すべてのビジネス知識、ユーザー情報、チャット履歴は、HTTP インタフェースを介してメインバックエンドから取得し、System Prompt の形で LLM に注入されます。また、Agent が認識したテキストや AI の返答も、HTTP 経由でメインバックエンドに書き戻され、永続化されます。
Agent の入口となるコードは非常にシンプルで、コアプロセスはわずか数ステップです。
async def entrypoint(ctx: JobContext):
room_name = ctx.room.name
# 1. 業務バックエンドから現在の会議室のコンテキストを取得する
room_context = await fetch_room_context(room_name)
system_prompt = build_system_prompt(room_context)
# 2. 会議室に接続し、音声のみ自動サブスクライブしてユーザーの参加を待つ
await ctx.connect(auto_subscribe=AutoSubscribe.AUDIO_ONLY)
participant = await ctx.wait_for_participant()
# 3. Agentを構築し、LLM + VAD + TTSを注入する
agent = Agent(
instructions=system_prompt,
llm=google.realtime.RealtimeModel(
api_key=google_api_key, voice="Aoede",
),
tts=google.beta.GeminiTTS(api_key=google_api_key),
vad=ctx.proc.userdata["vad"],
allow_interruptions=True,
)
# 4. セッションを開始する
session = AgentSession()
await session.start(agent, room=ctx.room)
await session.say("AIアシスタントが会議室に参加しました")
フレームワークが WebRTC 接続、音声コーデック、エンドツーエンドの遅延制御など、音声ストリームの下位レイヤーの処理をすべて隠蔽してくれるため、開発者は LLM の instructions とイベントコールバックだけに集中できます。
これにより、以下の 3 つのメリットが得られます。
- Agent の軽量化と代替可能性の維持：将来的に STT/TTS プロバイダを変更したい場合や、AI 機能を LLM から別のモデルに切り替える場合でも、Agent 側のコード修正量はごくわずかで済みます。
- ビジネスロジックの統一：ユーザー管理、権限管理、履歴管理などはすべてメインの Go バックエンドで行われるため、Agent は業務ロジックを理解する必要がありません。
- デプロイのデカップリング：Agent Worker は独立してスケーリング可能です。音声処理が集中するシナリオでは、メインバックエンドに影響を与えることなく、Worker の数を増やすことができます。
三、 フルデュプレックス（全二重）vs ハーフデュプレックス（半二重）
初期の多くの AI 音声ソリューションは、ハーフデュプレックス（半二重）方式でした。ユーザーが話し終わると AI が返答し、話し終わった側が「マイクを解放する」という流れです。これはインタラクションとして非常に不自然であり、実際の人間同士の会話では、相手が話し終えるのを一言も発さずに待つことはありません。
LiveKit Agents は 1.x バージョンで allow_interruptions（割り込み許可）メカニズムを導入し、VAD（音声活動検出）モジュールと組み合わせることで、真のフルデュプレックス（全二重）対話を実現しました。
全二重を実現する 3 つのコアメカニズム
- VAD（Silero）による継続的な検出：ユーザーが話しているかどうか、そしていつ話し終えたかをリアルタイムで判定します。
- ストリーミング TTS のサポート：AI の返答は、文全体が生成されてから再生されるのではなく、生成されながらリアルタイムで再生されます。
- 割り込みロジック：ユーザーが話し始めると、AI の再生は即座に停止し、ユーザーが話し終えるのを待ってから再度応答します。
これら 3 つのメカニズムの連携が、ユーザーが感じる「対話の滑らかさ」を決定します。開発段階で最も頻繁に遭遇した問題は、VAD の感度が高すぎて AI の発話が頻繁に遮られたり、逆に感度が低すぎてユーザーが話し終えた後も AI が反応せずに無言の時間が流れてしまうことでした。
VAD モデルは、通話のたびにダウンロードされるのを防ぐため、Worker の起動時にメモリにプリロードしておく必要があります。
def prewarm(proc: JobProcess):
"""Workerプロセス起動時に一度だけ実行され、Silero VADモデルをプリロードする"""
proc.userdata["vad"] = silero.VAD.load()
そして、Worker のエントリーポイントで以下のように登録します。
if __name__ == "__main__":
cli.run_app(
WorkerOptions(
entrypoint_fnc=entrypoint,
prewarm_fnc=prewarm,
)
)
这个 prewarm メカニズムは、LiveKit Agents の非常に優れた設計の一つです。音声処理モデルのロードには通常、数百ミリ秒から数秒かかります。通話開始時に毎回ロードすると、ユーザーが最初に話した際に明らかなラグが生じます。プロセスレベルでプリロードしておくことで、ユーザーの最初の一文字目から正確に検出することができます。
[!TIP]
チューニングの経験則：VAD のしきい値と沈黙タイムアウト時間は、実際のユースケースの会話速度や周囲の騒音レベルに合わせて調整する必要があります。デフォルトのパラメータはとりあえず動作するレベルのものであり、本番運用には個別のチューニングが不可欠です。
四、 重要な決定：マルチモーダルか、それとも STT+LLM+TTS の 3 点セットか
Agent のアーキテクチャ設計において、議論の余地がある重要なポイントが一つあります。それは「LLM に音声の入出力を直接担当させるべきか否か」です。
-
プラン A（従来の分離型アプローチ）：
$$\text{ユーザー音声} \longrightarrow \text{STT (テキスト変換)} \longrightarrow \text{LLM (テキスト生成)} \longrightarrow \text{TTS (音声変換)} \longrightarrow \text{再生}$$ -
プラン B（マルチモーダルアプローチ）：
$$\text{ユーザー音声} \longrightarrow \text{マルチモーダル LLM (音声を直接処理)} \longrightarrow \text{再生}$$
この決定は、遅延、コスト、およびコードの複雑さに直接影響します。
私の現在の実装では、プラン B のマルチモーダルな RealtimeModel アプローチを採用しています。このアプローチには非常に明確なメリットがあります。
- コード量の劇的な削減：個別の STT および TTS コンポーネントを登録する必要がなく、LLM が一括して入出力を処理します。
- 低レイ延：テキストのシリアライズおよびデシリアライズのプロセスが省略されます。
- 感情や口調の自然なサポート：マルチモーダルモデルはユーザーのトーン（語調）を感知できるため、AI の返答もより「人間味」を帯びたものになります。
ただし、マルチモーダルアプローチには代償もあります。選択できるモデルの範囲が非常に狭く、特定のクラウドプロバイダへの依存度が高くなります。将来的にローカル環境でホストするモデルに切り替えたい場合、現時点ではマルチモーダルアプローチを採用するのはあまり現実的ではありません。
五、 会話コンテキストの同期問題
これは実際の開発プロセスで直面した最も厄介なエンジニアリング上の問題でした。
AI が通話中に話した内容を IM のメッセージ履歴に同期し、ユーザーが後から「会話履歴」として見返せるようにする必要があります。しかし、非同期の音声通話を行っている最中、IM のテキストチャット画面にはメッセージがどのように表示されるべきでしょうか？
私は最終的に以下のような設計を採用しました。
- Agent がテキスト書き起こし（トランスクリプト）をリアルタイムでバックエンドに送信：ユーザーが話した各発言が認識されるたび、また AI の返答が生成されるたびに、HTTP を介して非同期でバックエンドに書き込まれます。
- バックエンドで「メッセージ」として永続化：IM の通常のテキストメッセージと同じメッセージテーブルを共有し、「音声書き起こし」であることを示すフラグを付与します。
- 通話終了時に要約処理を実行：長時間の通話の場合は会話の要約を生成し、特別なメッセージ形式で参加者にプッシュ送信します。
実装としては、Agent のイベントコールバック内で非同期の HTTP プッシュを実行します。
@session.on("conversation_item_added")
def on_item_added(ev):
item = ev.item
if not isinstance(item, llm.ChatMessage):
return
content = item.text_content
if not content:
return
if item.role == "user":
# ユーザーの発言 → 業務バックエンドにプッシュして永続化
asyncio.create_task(
push_utterance(room_name, participant.identity, content)
)
elif item.role == "assistant":
# AIの返答 → 同様に業務バックエンドにプッシュして永続化
asyncio.create_task(
push_utterance(room_name, "ai_assistant", content)
)
そして、push_utterance 自体は状態を持たないシンプルな HTTP POST リクエストです。
async def push_utterance(room_name: str, identity: str, content: str):
url = f"{backend_url}/api/agent/utterance"
payload = {
"room_name": room_name,
"speaker_identity": identity,
"content": content,
"language": "zh-CN",
}
async with session.post(url, json=payload, timeout=aiohttp.ClientTimeout(total=5)) as resp:
if resp.status != 200:
logger.warning(f"Push failed: {resp.status}")
避けるべき落とし穴
- 非同期処理におけるファンアウト（Fan-out）問題：Agent は同時に複数の部屋を処理する可能性があるため、コールバック内の非同期タスクで例外処理を適切に行わないと、特定の発言が欠落する原因になります。
- 順序の保証：HTTP リクエストは非同期で送信されるため、サーバーへの到着順序は保証されません。そのため、バックエンド側で送信時刻のタイムスタンプに基づいて再ソートを行う必要があります。
- リトライと冪等性（Idempotency）：ネットワークの揺らぎによって同じ発言が複数回プッシュされる可能性があるため、バックエンド側で重複排除を行う必要があります。
六、 デプロイと運用に関するいくつかのノウハウ
Agent Worker のデプロイには Docker を使用していますが、network_mode: host を採用したことは極めて重要な決定でした。
音声ストリームは、処理のステップ（ホップ）が増えるたびに遅延が蓄積されます。Docker のデフォルトの bridge ネットワークモードを使用した場合、音声データはコンテナ内部の NAT 転送を経由することになり、帯域幅とリアルタイム性が極めて重視される通話シナリオでは無視できないオーバーヘッドになります。host モードではホストマシンのネットワークスタックを直接共有するため、往復の遅延を約 5〜10ms 削減することができます。
また、自己署名証明書の SSL 検証スキップについても記述しておきます。LiveKit Server をイントラネットにデプロイする場合、ほぼ確実に自己署名証明書が使用されますが、Python の aiohttp や urllib はデフォルトで SSL 検証を行います。すべてのリクエストに対して毎回 ssl=False を指定するのは非常に煩雑であるため、モジュールレベルで一度だけモンキーパッチ（monkey-patch）を適用するのが最も手軽な解決策です。
import ssl
import aiohttp
# グローバルでSSL検証をスキップ（自社ネットワーク内の自己署名証明書を使用する環境に限定）
ssl._create_default_https_context = ssl._create_unverified_context
_original_init = aiohttp.TCPConnector.__init__
def _patched_init(self, *args, **kwargs):
kwargs["ssl"] = False
_original_init(self, *args, **kwargs)
aiohttp.TCPConnector.__init__ = _patched_init
Docker デプロイ時の docker-compose.yml の設定例は以下の通りです。
services:
axiom-agent:
build: .
env_file:
- .env
network_mode: "host" # ホストマシンのネットワークスタックを直接利用
deploy:
resources:
limits:
memory: 1G
logging:
driver: "json-file"
options:
max-size: "10m"
max-file: "3"
七、 コスト分析
マルチモーダルモデルを使用する場合のコスト構造は、従来の分離型（STT+LLM+TTS）アプローチとは異なります。従来の分離型アプローチにおける、1 時間あたりのアクティブ通話のおおよそのコストは以下の通りです。
- STT（音声認識）：約 $0.26
- LLM（意味理解）：约 $0.50
- TTS（音声合成）：約 $0.30
- 合計：約 $1.06 / 時間
一方、マルチモーダルアプローチでは、STT と TTS のコストが LLM の利用料金に「内包」される形になります。単価自体は一見すると高く見えますが、複数のサービスをシリアルに接続するためのオーバーヘッドやネットワークの往復遅延が削減されるため、実際の総合的なコストパフォーマンスはマルチモーダルの方が優れているケースが多いです。
[!NOTE]
目安として AI 通話のコストを 1 時間あたり 1〜2 ドル以内に抑えられるのであれば、多くの B 端（エンタープライズ向け）ユースケースにおいて十分に許容可能な範囲と言えます。
八、 現在も模索しているいくつかの開発方向
現在のソリューションで基本的な対話の閉環は実現できていますが、さらにユーザー体験を向上させるために模索しているテーマがいくつかあります。
- より自然な割り込み（Interruption）：現在の VAD ベースの割り込みは十分に機能していますが、複数人が参加する会話シナリオにおいては、AI は「いつ自分が発言すべきか（空気を読むこと）」を判断できず、ユーザーが話し終えるのをただ受動的に待つことしかできません。より細粒度な対話状態管理を導入できれば、AI の会話への参加感がさらに向上するはずです。
- TTS の感情制御：現在の TTS はパラメータ調整によってトーンを変更できますが、LLM の返答内容とリアルタイムに連動させることはまだできていません。ユーザーを慰める際には声を柔らかくし、重要事項を確認する際には引き締まった口調にするなど、状況に応じた感情制御ができれば、体験は全く異なるものになるでしょう。
- 長時間の会話における Token 管理：通話時間が 1 時間を超えると会話コンテキストが非常に長くなり、Token 消費量とモデルの応答遅延が急増します。単なるスライディングウィンドウ方式では初期の会話情報が失われてしまうため、より優れた要約と記憶（Memory）メカニズムが必要です。
- 環境ノイズとバックグラウンド消音：実際の使用環境におけるバックグラウンドノイズ（キーボード打鍵音、周囲の他人の話し声など）は、VAD の精度に悪影響を与え、Agent に「自分が遮られた」と誤解させる原因になります。ノイズキャンセリングのプリプロセスを 1 ステップ追加することで、体験の大幅な向上が見込めます。
九、 おわりに
LiveKit Agents は非常に優れたフレームワークであり、AI 音声対話を「研究室のおもちゃ」から「本番環境で使えるツール」の段階へと押し上げました。
しかし、フレームワークが解決してくれるのはあくまで「どうやって接続するか」という問題であり、「どうやって体験を良くするか」という問題までは解決してくれないことも認める必要があります。
ユーザー体験を決定づけるのは、VAD パラメータの地道なチューニングであり、既存のビジネスロジックと会話コンテキストの有機的な結合であり、1 ミリ秒単位の遅延の削減です。これらはすべて、実際のビジネスシナリオの中で泥臭く磨き上げていく必要があります。