自然で応答性の高い本番品質の音声エージェントを構築することは、音声から音声へのモデルのオーケストレーション、低遅延の音声ストリーミングの管理、接続ライフサイクルの処理など、複雑な工学的課題です。さらに、Web、モバイル、デスクトップアプリケーション全体で一貫したエクスペリエンスを提供する必要もあります。この記事では、StreamのVision AgentsオープンソースフレームワークをAmazon BedrockおよびAmazon Nova 2 Sonicと組み合わせて、数分で本番環境に対応したリアルタイム音声エージェントを構築する方法を紹介します。アーキテクチャの内部動作、コード例、関数呼び出し、自動再接続、多言語音声サポートなどの高度な機能についても説明します。
課題は、複雑なシステムの統合にあります。リアルタイム音声ストリーミングインフラ、音声認識、言語モデル、テキスト読み上げサービスは、それぞれ独自のレイテンシ特性と障害モードを持っています。一般的な音声対話は、ユーザーのマイクからの音声キャプチャ、音声からテキストへのストリーミング、言語モデルによる処理、応答生成、テキストから音声への変換、ユーザーへの配信というプロセスを経ます。これらすべてを数百ミリ秒以内に完了する必要があり、遅延が生じると会話の流れが途切れ、ユーザーをイライラさせます。さらに、本番環境では、ネットワーク接続の不安定さ、ブラウザの互換性問題、セッションタイムアウト、サービスのグレースフルデグラデーションなどに対処する必要があります。Vision Agentsはインフラストラクチャの複雑さを抽象化し、AIエクスペリエンスをカスタマイズする柔軟性を提供します。
このソリューションは、3つの主要コンポーネントで構成されています。Amazon Nova 2 Sonicは、Amazon Bedrockを通じて利用可能な音声から音声への基盤モデルで、リアルタイム双方向音声ストリーミング、ネイティブのターン検出、関数呼び出し機能を備えています。StreamのVision Agentsは、リアルタイム音声・ビデオAIエージェント向けのオープンソースPythonフレームワークで、25以上の統合を備えたプラグインベースのアーキテクチャ、本番デプロイツール、React、iOS、Android、Flutter、React Native用のクライアントSDKを提供します。Streamのエッジネットワークは、グローバルに分散されたエッジネットワークで、通常500ms未満の参加時間と30ms未満の音声レイテンシを実現し、クライアントとエージェントバックエンド間のリアルタイムトランスポートレイヤーを提供します。これらのコンポーネントが連携して完全なスタックを構成します。Streamがリアルタイムメディアトランスポートとクライアント側のエクスペリエンスを処理し、Amazon Nova 2 SonicがAIインテリジェンスを提供し、Vision Agentsがそれらを結び付けるグルーコードを提供します。
アーキテクチャは、Streamのエッジネットワークがメディアブローカーとして機能し、ユーザーデバイスとVision Agentワーカープロセス間の音声転送を処理します。ユーザーが話すと、音声はキャプチャされ、暗号化されてRTP over UDPで最も近いStream SFUに送信されます。SFUはWebRTC接続を終了し、NATトラバーサルと帯域幅推定を処理し、音声トラックをVision Agentワーカーに転送します。ワーカーは音声を生のPCMにデコードし、BedrockリアルタイムAPIを介してAmazon Nova Sonicにストリーミングし、応答音声フレームを再エンコードしてRTPとしてSFU経由でクライアントに配信します。エンドツーエンドのレイテンシは通常500ミリ秒未満です。音声アクティビティ検出（VAD）はワーカーで実行され、ブラウザのエコーキャンセレーションがエージェント自身の出力によるVADループの再トリガーを防ぎます。
コード実装は非常にシンプルで、プロジェクトディレクトリを作成し、Vision AgentsをインストールしてAWSとStreamの認証情報を設定し、エージェント作成と通話参加のロジックを含むmain.pyファイルを記述するだけです。以下の例は基本的なエージェントを示しています。
async def create_agent(**kwargs) -> Agent:
agent = Agent(
edge=getstream.Edge(),
agent_user=User(name="Helpful Assistant", id="agent"),
instructions="You are a helpful voice assistant. Be concise and friendly.",
llm=aws.Realtime(
model="amazon.nova-2-sonic-v1:0",
region_name="us-east-1",
voice_id="matthew",
),
)
return agent
async def join_call(agent: Agent, call_type: str, call_id: str, **kwargs) -> None:
call = await agent.create_call(call_type, call_id)
async with agent.join(call):
await asyncio.sleep(2)
await agent.llm.simple_response(
text="Greet the user warmly and ask how you can help."
)
await agent.finish()30行未満のコードで、Amazon Nova Sonicを搭載した完全に機能するリアルタイム音声エージェントを作成でき、StreamクライアントSDKからアクセス可能です。この統合は、関数呼び出し、自動再接続、多言語音声サポートなどの高度な機能もサポートしており、開発者はインフラストラクチャではなくビジネスロジックに集中できます。