ウェブサイト検索

GPU ドロップレット上の OpenAI、LiveKit、Deepgram を使用して、ビジョンと音声機能を備えたリアルタイム AI チャットボットを構築する


導入

このチュートリアルでは、DigitalOcean GPU Droplet にデプロイされた OpenAI、LiveKit、Deepgram を使用して、ビジョンおよび音声機能を備えたリアルタイム AI チャットボットを構築する方法を学びます。このチャットボットは、ユーザーとリアルタイムで会話し、カメラからキャプチャした画像を分析し、正確かつタイムリーな応答を提供できます。

先進テクノロジーによるチャットボット機能の強化

このチュートリアルでは、3 つの強力なテクノロジーを活用してリアルタイム AI チャットボットを構築します。それぞれが、DigitalOcean の GPU ドロップレットによって提供される堅牢なインフラストラクチャを利用しながら、チャットボットの機能を強化する特定の目的を果たします。

  1. OpenAI API: OpenAI API は、ユーザー入力に基づいて人間のようなテキスト応答を生成します。 GPT-4o のような高度なモデルを採用することで、チャットボットはコンテキストを理解し、有意義な会話を行い、ユーザーのクエリに正確な回答を提供できるようになります。これは、ユーザーが理解され、評価されていると感じるインタラクティブなエクスペリエンスを作成するために重要です。

  2. LiveKit: LiveKit は、ユーザーとチャットボット間のリアルタイムの音声およびビデオ通信を容易にします。これにより、シームレスな対話エクスペリエンスを作成でき、ユーザーがチャットボットに話しかけて音声応答を受け取ることができるようになります。これは、ユーザーを自然に関与させ、対話をより個人的かつ直感的にできる音声対応チャットボットを構築するために不可欠です。

  3. ディープグラム: ディープグラムは音声認識に使用され、話し言葉をテキストに変換します。これにより、チャットボットはユーザーの音声入力を効率的に処理できるようになります。 Deepgram の機能を統合することで、チャットボットがユーザーのコマンドとクエリを正確に理解できるようになり、全体的な対話の品質が向上します。これは、ユーザーエンゲージメントを維持するために迅速かつ正確な応答が必要なリアルタイム設定では特に重要です。

GPU ドロップレットを使用する理由: DigitalOcean の GPU ドロップレットを利用することは、これらの AI モデルとリアルタイム通信に必要な集中的な処理を強化し、処理するために必要な計算および GPU インフラストラクチャを提供するため、このセットアップでは特に有益です。 GPU は AI/ML ワークロードの実行用に最適化されており、モデル推論とビデオ処理タスクが大幅に高速化されます。これにより、負荷が重い場合でもチャットボットが迅速かつ効率的に応答を提供できるようになり、ユーザー エクスペリエンスとエンゲージメントが向上します。

前提条件

始める前に、次のものが揃っていることを確認してください。

  • DigitalOcean クラウド アカウント。
  • GPU ドロップレットがデプロイされ、実行されています。
  • Python プログラミングの基本的な知識。
  • GPT-4o モデルを使用するために設定された OpenAI API キー。
  • LiveKit サーバーが GPU ドロップレット上で稼働しています。
  • ディープグラム API キー。

ステップ 1 - GPU ドロップレットをセットアップする

1.新しいプロジェクトの作成 - クラウド コントロール パネルから新しいプロジェクトを作成し、それを GPU ドロップレットに関連付ける必要があります。

2.GPU ドロップレットの作成 - DigitalOcean アカウントにログインし、新しい GPU ドロップレットを作成し、OS としてAI/ML Ready を選択します。この OS イメージは、必要なすべての NVIDIA GPU ドライバーをインストールします。 GPU ドロップレットの作成方法については、公式ドキュメントを参照してください。

3.認証用の SSH キーを追加 - GPU ドロップレットでの認証には SSH キーが必要です。SSH キーを追加することで、端末から GPU ドロップレットにログインできます。

4.GPU ドロップレットを完成させて作成する - 上記の手順がすべて完了したら、新しい GPU ドロップレットを完成させて作成します。

ステップ 2 - LiveKit アカウントをセットアップし、GPU ドロップレットに CLI をインストールする

まず、アカウントを作成するか、LiveKit Cloud アカウントにサインインして LiveKit プロジェクトを作成する必要があります。 プロジェクト設定 ページの LIVEKIT_URLLIVEKIT_API_KEY、および LIVEKIT_API_SECRET 環境変数が必要になるため、メモしてください。チュートリアルの後半で。

LiveKit CLIをインストールする

以下のコマンドは、GPU ドロップレットに LiveKit CLI をインストールします。

curl -sSL https://get.livekit.io/cli | bash

LiveKit Cloud ユーザーの場合、Cloud プロジェクトで CLI を認証して API キーとシークレットを作成できます。これにより、資格情報を毎回手動で提供しなくても CLI を使用できるようになります。

lk cloud auth

次に、指示に従ってブラウザからログインします。

デバイスを追加し、この手順の前半で作成した LiveKit プロジェクトへのアクセスを承認するように求められます。

ステップ 3 - 既存の LiveKit テンプレートからエージェントをブートストラップする

このテンプレートは、構築するための実用的な音声アシスタントを提供します。テンプレートには次のものが含まれます。

  • 基本的な音声インタラクション
  • 音声のみのトラックのサブスクリプション
  • 音声アクティビティ検出 (VAD)
  • 音声テキスト変換 (STT)
  • 言語モデル (LLM)
  • テキスト読み上げ (TTS)

注: デフォルトでは、サンプル エージェントは STT に Deepgram を使用し、TTS と LLM に OpenAI を使用します。ただし、これらのプロバイダーを使用する必要はありません。

単純な Python 音声エージェントのスターター テンプレートのクローンを作成します。

lk app create

これにより、アプリのデプロイに使用できる複数の既存の LiveKit テンプレートが提供されます。

voice-assistant-frontend                                                                                                                                        
transcription-frontend                                                                                                                                        
token-server                                                                                                                                               
multimodal-agent-python                                                                                                                                                 
multimodal-agent-node                                                                                                  
voice-pipeline-agent-python                                                                                                                       
voice-pipeline-agent-node                                                                                                                                                     
android-voice-assistant                                                                                                                                                         
voice-assistant-swift                                                                                                                                                                
outbound-caller-python

voice-pipeline-agent-python テンプレートを使用します。

lk app create --template voice-pipeline-agent-python

次に、 プロンプトが表示されたらアプリケーション名OpenAI API キー、およびディープグラム API キーを入力します。 Deepgram と OpenAI を使用していない場合は、他のサポートされているプラグインをチェックアウトできます。

Cloning template...
Instantiating environment...
Cleaning up...
To setup and run the agent:

       cd /root/do-voice-vision-bot
       python3 -m venv venv
       source venv/bin/activate
       pip install -r requirements.txt
       python3 agent.py dev

ステップ 4 - 依存関係をインストールし、仮想環境を作成する

まず、前の手順で作成したアプリケーションのディレクトリに切り替えます。

cd <app_name>

テンプレートから作成されたファイルを一覧表示できます。

ls
LICENSE  README.md  agent.py  requirements.txt

ここで、agent.py は、AI チャットボットのロジックとソース コードを含むメイン アプリケーション ファイルです。

次に、以下のコマンドを使用して Python 仮想環境を作成し、アクティブ化します。

apt install python3.10-venv
python3 -m venv venv

次の API キーを環境に追加します。

export LIVEKIT_URL=<>
export LIVEKIT_API_KEY=<>
export LIVEKIT_API_SECRET=<>
export DEEPGRAM_API_KEY=<>
export OPENAI_API_KEY=<>

LIVEKIT_URLLIVEKIT_API_KEY、および LIVEKIT_API_SECRET は、LiveKit プロジェクトの設定ページにあります。

仮想環境をアクティブ化します。

source venv/bin/activate

注: Debian/Ubuntu システムでは、次のコマンドを使用して python3-venv パッケージをインストールする必要があります。

apt install python3.10-venv

次に、アプリが動作するために必要な依存関係をインストールしましょう。

python3 -m pip install -r requirements.txt

ステップ 5 - AI エージェントにビジョン機能を追加する

ビジョン機能をエージェントに追加するには、以下のインポートと関数を使用して agent.py ファイルを変更する必要があります。

まず、これらのインポートを既存のものと一緒に追加することから始めましょう。 vinano などのテキスト エディタを使用して agent.py ファイルを開きます。

vi agent.py

以下のインポートを既存のものと一緒にコピーします。

from livekit import rtc
from livekit.agents.llm import ChatMessage, ChatImage

これらの新しいインポートには次のものが含まれます。

  • rtc: LiveKit のビデオ機能へのアクセス
  • ChatMessage および ChatImage: LLM に画像を送信するために使用するクラス

ビデオのサブスクリプションを有効にする

entrypoint 関数で ctx.connect() 行を見つけます。 AutoSubscribe.AUDIO_ONLYAutoSubscribe.SUBSCRIBE_ALL に変更します。

await ctx.connect(auto_subscribe=AutoSubscribe.SUBSCRIBE_ALL)

注: vi または nano テキストを使用して agent.py ファイルを編集および変更することが難しい場合GPU ドロップレット上のエディター。 agent.py ファイルの内容をローカル システムにコピーし、VSCode などのコード エディターで必要な編集を行ってから、更新されたコードをコピーして貼り付けることができます。

これにより、アシスタントはオーディオだけでなくビデオ トラックも受信できるようになります。

ビデオフレーム処理を追加

これら 2 つのヘルパー関数をインポートの後かつ prewarm 関数の前に追加します。

async def get_video_track(room: rtc.Room):
   """Find and return the first available remote video track in the room."""
   for participant_id, participant in room.remote_participants.items():
       for track_id, track_publication in participant.track_publications.items():
           if track_publication.track and isinstance(
               track_publication.track, rtc.RemoteVideoTrack
           ):
               logger.info(
                   f"Found video track {track_publication.track.sid} "
                   f"from participant {participant_id}"
               )
               return track_publication.track
   raise ValueError("No remote video track found in the room")

この機能は、すべての参加者を検索して、利用可能なビデオ トラックを見つけます。処理するビデオ フィードを見つけるために使用されます。

次に、フレームキャプチャ機能を追加します。

async def get_latest_image(room: rtc.Room):
   """Capture and return a single frame from the video track."""
   video_stream = None
   try:
       video_track = await get_video_track(room)
       video_stream = rtc.VideoStream(video_track)
       async for event in video_stream:
           logger.debug("Captured latest video frame")
           return event.frame
   except Exception as e:
       logger.error(f"Failed to get latest image: {e}")
       return None
   finally:
       if video_stream:
           await video_stream.aclose()

この関数の目的は、ビデオ トラックから 1 つのフレームをキャプチャし、リソースを適切にクリーンアップすることです。 aclose() を使用すると、メモリ バッファやビデオ デコーダ インスタンスなどのシステム リソースが解放され、メモリ リークの防止に役立ちます。

LLM コールバックを追加する

ここで、entrypoint 関数内に、LLM が応答を生成する直前に最新のビデオ フレームを挿入する以下のコールバック関数を追加します。 agent.py ファイル内で entrypoint 関数を検索します。

async def before_llm_cb(assistant: VoicePipelineAgent, chat_ctx: llm.ChatContext):
       """
       Callback that runs right before the LLM generates a response.
       Captures the current video frame and adds it to the conversation context.
       """
       try:
           if not hasattr(assistant, '_room'):
               logger.warning("Room not available in assistant")
               return
           latest_image = await get_latest_image(assistant._room)
           if latest_image:
               image_content = [ChatImage(image=latest_image)]
               chat_ctx.messages.append(ChatMessage(role="user", content=image_content))
               logger.debug("Added latest frame to conversation context")
           else:
               logger.warning("No image captured from video stream")
       except Exception as e:
           logger.error(f"Error in before_llm_cb: {e}")

このコールバックは効率的なコンテキスト管理の鍵です。アシスタントが応答しようとしているときにのみ視覚的な情報を追加します。すべてのメッセージに視覚情報が追加されると、LLM のコンテキスト ウィンドウがすぐにいっぱいになってしまい、非常に非効率でコストが高くなります。

システムプロンプトを更新する

entrypoint 関数内で initial_ctx 作成を見つけて、ビジョン機能が含まれるように更新します。

initial_ctx = llm.ChatContext().append(
   role="system",
   text=(
       "You are a voice assistant created by LiveKit that can both see and hear. "
       "You should use short and concise responses, avoiding unpronounceable punctuation. "
       "When you see an image in our conversation, naturally incorporate what you see "
       "into your response. Keep visual descriptions brief but informative."
   ),
)

アシスタント構成を更新する

entrypoint 関数内で VoicePipelineAgent 作成を見つけて、コールバックを追加します。

assistant = VoicePipelineAgent(
   vad=ctx.proc.userdata["vad"],
   stt=openai.STT(),
   llm=openai.LLM(),
   tts=openai.TTS(),
   chat_ctx=initial_ctx,
   before_llm_cb=before_llm_cb
)

ここでの主な更新は before_llm_cb パラメーターです。このパラメーターは、以前に作成したコールバックを使用して最新のビデオ フレームを会話コンテキストに挿入します。

音声とビジョン機能を備えた最終的な agent.py ファイル

必要な関数とインポートをすべて追加した後の agent.py ファイルは次のようになります。

from asyncio.log import logger
from livekit import rtc
from livekit.agents.llm import ChatMessage, ChatImage
import logging
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 openai, deepgram, silero

async def get_video_track(room: rtc.Room):
   """Find and return the first available remote video track in the room."""
   for participant_id, participant in room.remote_participants.items():
       for track_id, track_publication in participant.track_publications.items():
           if track_publication.track and isinstance(
               track_publication.track, rtc.RemoteVideoTrack
           ):
               logger.info(
                   f"Found video track {track_publication.track.sid} "
                   f"from participant {participant_id}"
               )
               return track_publication.track
   raise ValueError("No remote video track found in the room")

async def get_latest_image(room: rtc.Room):
   """Capture and return a single frame from the video track."""
   video_stream = None
   try:
       video_track = await get_video_track(room)
       video_stream = rtc.VideoStream(video_track)
       async for event in video_stream:
           logger.debug("Captured latest video frame")
           return event.frame
   except Exception as e:
       logger.error(f"Failed to get latest image: {e}")
       return None
   finally:
       if video_stream:
           await video_stream.aclose()

def prewarm(proc: JobProcess):
   proc.userdata["vad"] = silero.VAD.load()

async def entrypoint(ctx: JobContext):

   async def before_llm_cb(assistant: VoicePipelineAgent, chat_ctx: llm.ChatContext):
       """
       Callback that runs right before the LLM generates a response.
       Captures the current video frame and adds it to the conversation context.
       """
       try:
           if not hasattr(assistant, '_room'):
               logger.warning("Room not available in assistant")
               return
           latest_image = await get_latest_image(assistant._room)
           if latest_image:
               image_content = [ChatImage(image=latest_image)]
               chat_ctx.messages.append(ChatMessage(role="user", content=image_content))
               logger.debug("Added latest frame to conversation context")
           else:
               logger.warning("No image captured from video stream")
       except Exception as e:
           logger.error(f"Error in before_llm_cb: {e}")

   initial_ctx = llm.ChatContext().append(
       role="system",
       text=(
           "You are a voice assistant created by LiveKit that can both see and hear. "
           "You should use short and concise responses, avoiding unpronounceable punctuation. "
           "When you see an image in our conversation, naturally incorporate what you see "
           "into your response. Keep visual descriptions brief but informative."
       ),
   )

   logger.info(f"connecting to room {ctx.room.name}")
   await ctx.connect(auto_subscribe=AutoSubscribe.SUBSCRIBE_ALL)

   # Wait for the first participant to connect
   participant = await ctx.wait_for_participant()
   logger.info(f"starting voice assistant for participant {participant.identity}")

   # Configure the voice pipeline agent
   agent = VoicePipelineAgent(
       vad=ctx.proc.userdata["vad"],
       stt=deepgram.STT(),
       llm=openai.LLM(),
       tts=openai.TTS(),
       chat_ctx=initial_ctx,
       before_llm_cb=before_llm_cb  # Add the callback here
   )

   agent.start(ctx.room, participant)

   # Greet the user when the agent starts
   await agent.say("Hey, how can I help you today?", allow_interruptions=True)

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

エージェントをテストする

アシスタントを起動し、以下をテストします。

python3 agent.py dev
  • 音声インタラクションのテスト: マイクに向かって話して、チャットボットの応答を確認します。

  • ビジョン機能のテスト: ビデオ カム ストリームを通じてチャットボットにオブジェクトを識別するよう依頼します。

コンソールに次のログが表示されます。

2024-12-30 08:32:56,167 - DEBUG asyncio - Using selector: EpollSelector
2024-12-30 08:32:56,168 - DEV  livekit.agents - Watching /root/do-voice-vision-bot
2024-12-30 08:32:56,774 - DEBUG asyncio - Using selector: EpollSelector
2024-12-30 08:32:56,778 - INFO livekit.agents - starting worker {"version": "0.12.5", "rtc-version": "0.18.3"}
2024-12-30 08:32:56,819 - INFO livekit.agents - registered worker {"id": "AW_cjS8QXCEnFxy", "region": "US East", "protocol": 15, "node_id": "NC_OASHBURN1A_BvkfVkdYVEWo"}

次に、オーディオとビデオの両方を公開するクライアントを使用して、アプリを LiveKit ルームに接続する必要があります。これを行う最も簡単な方法は、ホストされたエージェント プレイグラウンドを使用することです。

そのため、このエージェントは通信するためにフロントエンド アプリケーションを必要とします。 livekit-examples でサンプル フロントエンドの 1 つを使用したり、クライアント クイックスタートの 1 つに従って独自のフロントエンドを作成したり、ホストされているサンドボックス フロントエンドの 1 つに対して即座にテストしたりできます。

この例では、既存のホスト型エージェント プレイグラウンドを使用します。システムのブラウザでこの https://agents-playground.livekit.io/ を開き、LiveKit プロジェクトに接続するだけです。プロジェクトが自動的に入力されるはずです。

仕組み

エージェントに対する上記の変更により、アシスタントは次のことができるようになります。

  1. オーディオ ストリームとビデオ ストリームの両方に接続します。

  2. 以前と同様にユーザーの音声を聞きます。

  3. 各応答を生成する直前に、次のようにします。

  • 現在のビデオ フレームをキャプチャします。
  • 会話コンテキストに追加します。
  • 応答を通知するために使用します。

4.必要な場合にのみフレームを追加して、コンテキストをクリーンに保ちます。

結論

おめでとう! DigitalOcean GPU Droplets 上の OpenAI、LiveKit、Deepgram を使用して、ビジョンと音声機能を備えたリアルタイム AI チャットボットを構築することに成功しました。この強力な組み合わせにより、アプリケーションの効率的でスケーラブルなリアルタイムの対話が可能になります。

AI エージェントの構築の詳細については、LiveKit の公式ドキュメントと API リファレンスを参照してください。