LiveKitとTavusで爆速実装するリアルタイムAIアバター

はじめに

リアルタイム音声・映像アプリケーションを開発できる LiveKit で作った対話型アプリに、Tavus の実写品質AIアバターを導入する方法を紹介します。

「WebRTC周りの実装を重くしたくない」「アバター付きの対話体験を最短で作りたい」という方向けです。

LiveKitとは

LiveKitは、リアルタイムな音声・映像・データ通信をアプリに実装できるオープンソースのWebRTCプラットフォームです。
クロスプラットフォーム対応のSDKやスケーラビリティが特徴で、近年は Agents（AIエージェント）向け機能も強化されています。複雑な通信サーバーを自作せずとも、対話型アプリを構築しやすいのが魅力です。

Tavusとは

Tavusは、実写品質のAIアバターを生成・操作できる動画プラットフォームです。API主導で設計されており、テキストに応じたリアルタイムなリップシンクや表情生成が可能です。
LiveKitと組み合わせることで、低遅延な“アバター対話”をアプリに組み込めます。

導入方法

前提（用意するもの）

• Tavusアカウント（API Key取得のため）
• Tavusの Persona ID / Replica ID
• LiveKit Agentsの実行環境（ローカル or サーバ）
• 動作確認用のLiveKitフロントエンド（または接続できるクライアント）

1. LiveKit AI Agents セットアップ

まずは公式ドキュメントに沿ってプロジェクトをセットアップします。
https://docs.livekit.io/agents/start/voice-ai/

2. インストール（uvの例）

uv add "livekit-agents[tavus]~=1.2"

3. Tavus API Keyの取得

Tavusに登録後、開発者コンソールからAPI Keyを取得します。
https://platform.tavus.io/dev/api-keys

取得したキーを .env ファイルへ設定します。

TAVUS_API_KEY=tv_sk_...

4. アバターの選択（Persona / Replica）

Tavusのダッシュボードで使いたいアバターを選択し、以下を控えます。

• Persona ID
• Replica ID

5. agent.pyの実装

LiveKitのエージェントコードにTavusを組み込みます。

from livekit import agents
from livekit.agents import AgentServer, AgentSession, RoomOutputOptions
from livekit.plugins import tavus

server = AgentServer()

@server.rtc_session()
async def my_agent(ctx: agents.JobContext):
    # エージェントセッションの定義（STT, LLM, TTS等を設定）
    session = AgentSession(
        # ... stt, llm, tts, etc.
    )

    # Tavusアバターのセッションを作成
    avatar = tavus.AvatarSession(
        replica_id="r79e...",  # 取得した Replica ID
        persona_id="p9a8...",  # 取得した Persona ID
    )

    # アバターを開始し、ルームへの参加を待機
    await avatar.start(session, room=ctx.room)

    # ユーザーとのエージェントセッションを開始
    await session.start(
        # ... room, agent, room_input_options, etc....
    )

6. 実行

uv run agent.py dev

実行結果

LiveKitのフロントエンドと接続すると、以下のようにアバターが表示されます。

結論

LiveKitで実装した対話型アプリに、Tavusの設定を追加するだけで、比較的簡単にAIアバターを導入できました。

執筆時点の挙動では、日本語のリップシンク（口の動き）が完全に一致しない場面もあったため、今後のアップデートに期待しています。

おまけ ： LangGraphとの連携

実用的な例として、自前の LangGraph ワークフローと接続した「AIカスタマーサポートエージェント」を作ってみました。 STT/LLM/TTSにはAzure OpenAIを使用しています。

agent.py

# 必要なインポートは省略しています (livekit, openai, silero等)

class AvatarAgent(Agent):
    def __init__(self) -> None:
        super().__init__(
            instructions="""
                あなたは親切なカスタマーサポートです。
            """,
            stt=openai.STT.with_azure(
                model="gpt-4o-transcribe",
                azure_deployment=os.environ["AZURE_OPENAI_STT_DEPLOYMENT_NAME"],
                api_key=os.environ.get("AZURE_OPENAI_STT_API_KEY"),
                azure_endpoint=os.environ.get("AZURE_OPENAI_STT_ENDPOINT"),
                api_version=os.environ.get("OPENAI_STT_API_VERSION"),
            ),
            llm=openai.LLM.with_azure(
                azure_deployment=os.environ["AZURE_OPENAI_DEPLOYMENT_NAME"],
                api_key=os.environ.get("AZURE_OPENAI_API_KEY"),
                azure_endpoint=os.environ.get("AZURE_OPENAI_ENDPOINT"),
                api_version=os.environ.get("OPENAI_API_VERSION"),
            ),
            tts=openai.TTS.with_azure(
                azure_deployment=os.environ["AZURE_OPENAI_TTS_DEPLOYMENT_NAME"],
                api_key=os.environ.get("AZURE_OPENAI_TTS_API_KEY"),
                azure_endpoint=os.environ.get("AZURE_OPENAI_TTS_ENDPOINT"),
                api_version=os.environ.get("OPENAI_TTS_API_VERSION"),
                voice="onyx",
            ),
            vad=silero.VAD.load(),
        )

    @function_tool
    async def answer_with_langgraph(self, context: RunContext, message: str) -> str:
        """
        Use this tool to produce the final answer for the given user message.
        Always call this tool. Return plain text only.
        """
        print("入力クエリ: ", message)
        try:
            # LangGraphのワークフローを実行
            state = await run_workflow(message)
            ans = (state.get("final_answer") or state.get("draft_answer") or "").strip()
            
            if not ans:
                return "うまく答えを準備できませんでした。別の聞き方で試してみてください。"
            return ans
        except Exception as e:
            logger.exception("LangGraph failed: %s", e)
            return "内部処理でエラーが発生しました。少し時間をおいてもう一度お試しください。"

async def entrypoint(ctx: JobContext):
    agent = AvatarAgent()
    await ctx.connect()
    
    # ユーザーデータを含むAgentSessionを作成
    userdata = UserData(ctx=ctx)
    session = AgentSession[UserData](
        userdata=userdata, 
        turn_detection=MultilingualModel() # 会話のターン検知モデル
    )

    # Tavusアバターセッションの作成
    avatar = tavus.AvatarSession(
        replica_id="...",
        persona_id="..."
    )

    # アバターを開始
    await avatar.start(session, room=ctx.room)

    # エージェントセッションを開始（アバターが話すため音声出力を有効化）
    await session.start(
        room=ctx.room,
        room_output_options=RoomOutputOptions(
            audio_enabled=True, 
        ),
        agent=agent
    )

if __name__ == "__main__":
    cli.run_app(
        WorkerOptions(
            entrypoint_fnc=entrypoint,
            initialize_process_timeout=300
        )
    )

参考資料

• LiveKit Docs : Tavus virtual avatar integration guide
• Youtube : LiveKit & Tavus: Realtime AI Avatars
• Tavus Docs : Introduction