Amazon SageMaker AI 上の Kokoro TTS を使って、AI エージェントからの低レイテンシ音声をストリーミングする

顧客が AI 音声エージェントに質問をして待っています。エージェントはすでに答えを持っており、複数段落にわたる長い応答を生成している途中ですが、顧客には 5 秒間何も聞こえません。答えが届く頃には、顧客は苛立ち、混乱し、もはや注意を払っていません。

テキスト読み上げの統合は、まさにこのギャップを生み出します。1 文の応答なら 600 ミリ秒で生成できるかもしれませんが、応答全体には 5 秒かかり、統合部分は合成を開始する前にそのすべてを待ってしまいます。

この記事では、そのギャップを埋める方法を紹介します。各文を、エージェントが生成した直後にホスト型の Kokoro-82M テキスト読み上げ (TTS) モデルへ Amazon SageMaker AI 上でストリーミングすることで、完成した応答ではなく最初の文から合成を開始できます。その音声をブラウザに返すにも独自のソリューションが必要なため、そのための 2 つのトランスポート方式と、それぞれが実際にもたらすコストについても取り上げます。

ソリューションの概要

このソリューションは、生成が完了するのを待つのではなく、生成と並行して合成をストリーミングすることでギャップを埋めます。エージェントが各文を生成すると、その文は直ちにホスト型の Kokoro-82M テキスト読み上げモデルへストリーミングされるため、エージェントが応答全体を話し終える前に音声再生を開始できます。Kokoro-82M は 8,200 万のパラメータのみを持ち、中間クラスの GPU インスタンス 1 台でも低コストで実行できるほど小さなモデルです。これにより、実際のトラフィックを処理する音声エージェントにとって、レイテンシと推論コストの両方を低く抑えられます。

Kokoro-82M は、SageMaker AI リアルタイムエンドポイント上のカスタムコンテナで、 ml.g5.xlarge GPU インスタンス上で実行されます。このコンテナは 2 つのルートを公開します: POST /invocationsは 1 つの完全な入力に対する応答をストリーミングし、 WS /invocations-bidirectional-streamは多数のメッセージに対して双方向で 1 つの接続を開いたままにします。

次の図は、クライアントからモデルへのリクエストパスを示しています:

図 1: 双方向ストリーム経由でクライアントから SageMaker AI リアルタイムエンドポイントへのリクエストパス

前提条件

開始する前に、以下が必要です:

• SageMaker AI エンドポイントを作成する権限を持つ AWS アカウント、Amazon Bedrock AgentCore の機能である AgentCore runtime、AWS Identity and Access Management (IAM) ロール、および Amazon Elastic Container Registry (Amazon ECR) リポジトリ。リポジトリには最低でも次の sagemaker:CreateEndpoint, sagemaker:CreateEndpointConfig, sagemaker:CreateModel, sagemaker:InvokeEndpointWithBidirectionalStream、 bedrock:InvokeModelWithResponseStream アクションが含まれている必要があります。
• サンプルコンテナは、既知のアドバイザリ (PYSEC-2026-2286、torch 2.10.0 で修正済み) のある古い PyTorch を固定しています。価値実証以外の使用の前に、この固定バージョンを見直してアップグレードしてください。
• AWS Cloud Development Kit (AWS CDK) v2。ターゲットのアカウントと AWS リージョンにインストールおよびブートストラップ済みであること。
• コンテナからデプロイされた SageMaker AI リアルタイムエンドポイント。コンテナは com.amazonaws.sagemaker.capabilities.bidirectional-streaming=true Docker ラベルを宣言していること。
• Python 3.12 以降、 uv、およびコンテナイメージをビルドするための Docker (基盤となる HTTP/2 クライアントは、Python 3.12 以降用のビルド済みホイールのみを提供しています)。
• 次をサポートする Python 用 AWS SDK (Boto3): invoke_endpoint_with_bidirectional_stream.
• WebSockets および Starlette または同等の ASGI フレームワークに関する知識。
• 本投稿で参照されているデプロイコードは、 GitHubリポジトリ から入手できます。

ウォークスルー

このウォークスルーでは、まずエンドポイントとエージェントをデプロイし、その後、インクリメンタルストリーミングをエンドツーエンドで機能させるためのクライアント側の変更について順を追って説明します。

エンドポイントとエージェントをデプロイする

Kokoro-82Mエンドポイントとエージェントランタイムは、それぞれが独自のAWS CDKスタックをラップする2つのCLIツールを通じてデプロイされます:

uv run speech-infra deploy kokoro-82m
uv run agent-infra deploy --bedrock-model-id <your-bedrock-model-id>

speech-infra deploy は、Kokoro-82Mエンドポイントをプロビジョニングします。これには、基盤となるIAMロールを作成し、モデルの重みをAmazon Simple Storage Service (Amazon S3) に同期する、基盤スタックとモデルキャッシュスタックが含まれます。次にエンドポイントスタックがコンテナイメージをビルドし、SageMaker AIのモデル、エンドポイント設定、およびエンドポイントを作成します。 agent-infra deploy は、StrandsエージェントをホストするAmazon Bedrock AgentCoreランタイムをプロビジョニングします。このエージェントは、直前にデプロイしたエンドポイントを呼び出すため、先にエンドポイントをデプロイしてください。

ステップ 1: SageMaker AIが実際に送信するWebSocketフレームタイプを処理する

SageMaker AIが使用する双方向トランスポートは、クライアントのメッセージのすべての部分をバイナリWebSocketフレームとしてコンテナに配信します。StarletteのWebSocketハンドラーを記述する一般的な方法は、 receive_text()を呼び出すことです。これは message["text"] を読み取り、そのキーが存在しない場合は KeyError を送出します。このクライアントが送信するすべてのメッセージはバイナリとして届くため、そのハンドラーはすべてのリクエストで失敗します。

この障害はよく誤診されます。転送する汎用の例外ハンドラーは str(e) クライアントへのターン KeyError('text') 次の内容を読み取るエラーフレームに 'text'「これは項目の欠落であり、フレーミングの問題ではありません。エンドポイント自体が報告しています InService 接続を受け付け、すべてのメッセージに対して毎回エラーを返して応答するため、終始その状態が続きます。

どちらのフレームタイプも明示的に受け入れるようにし、片方が現れると決めつけないようにします:

from starlette.websockets import WebSocket, WebSocketDisconnect

async def _receive_message(websocket: WebSocket) -> str:
    message = await websocket.receive()
    if message["type"] == "websocket.disconnect":
        raise WebSocketDisconnect(message.get("code", 1000))
    payload = message.get("text")
    if payload is None:
        payload = message.get("bytes", b"").decode("utf-8")
    return payload

テキストフィールドが存在しない場合、このコードはフレームタイプを一つに決め打ちするのではなく、バイトをUTF-8としてデコードします。バイナリフレームを送信してテストしてください。この障害モードでは、明白なクラッシュではなく、整形式のエラーフレームと正常なエンドポイントステータスが返されるためです。

ステップ 2: メッセージ間でセッションを開いたままにする

コンテナが1つのメッセージの音声の合成を終えた後、HTTP のサイクルが終わるのと同じようにセッションを閉じたくなるでしょう。しかしここではそれは間違いです。あるメッセージの音声と synthesis_complete 制御フレームに戻ると、コンテナループは待機状態に戻るべきです。 receive() 同じまだ開いているセッションでの次のメッセージに対して使用します。接続を閉じるのは、クライアントが明示的に送信した場合のみにしてください。 {"type": "close"} メッセージ。

次の図は、1つのセッションが2つの文を順に運び、最後にのみ閉じる様子を示しています:

図2：2つの文を連続して伝送する一つの双方向セッションで、終了時にのみクローズする

このようにして、1つの持続的なセッション上で複数の文を次々と送信でき、文の間に再接続のコストがかかりません。1つのレスポンスを受けるたびにすぐ接続を閉じると、文ごとに新しい接続が必要になり、レイテンシ削減のメリットの大半が失われてしまいます。

ステップ3:エージェントが生成しながら文をストリーミングする

セッションの挙動が整ったところで、レイテンシ問題に対するクライアント側の修正は、2つの形式で届くテキスト——最初から分かっている完全な文字列、あるいはライブのトークンストリーム——を正規化するクラスです。どちらの形式でも同じ出力を生成します。完全な文であり、それぞれが完成した時点ですぐに送信できる状態になります。

import re

_SENTENCE_RE = re.compile(r"(?<=[.!?])\s+")


class IncrementalSentenceChunker:
    def __init__(self) -> None:
        self._buffer = ""

    def feed(self, fragment: str) -> list[str]:
        self._buffer += fragment
        parts = _SENTENCE_RE.split(self._buffer)
        if len(parts) == 1:
            return []
        self._buffer = parts[-1]
        return [s.strip() for s in parts[:-1] if s.strip()]

    def flush(self) -> str | None:
        remainder = self._buffer.strip()
        self._buffer = ""
        return remainder or None

電話 feed() テキストの断片が到着するたびに1回実行されます。新しく完成した文を返し、末尾の未完成テキストは保持します。Call flush() 一度、入力の最後で、末尾の句読点がなくても残っている部分的なテキストをすべて解放するために実行されます。これにより、ピリオドが到着する前に終了したストリームでも、最後の文が失われなくなります。

これは並行合成ではありません。各双方向ハンドラは厳密に逐次実行のループです。メッセージを受信し、その音声を合成し、その音声を送信し、送信 synthesis_completeその後、次のメッセージを受信します。文2の音声は、文1がまだ送信されている間には生成されません。逐次送信によって得られるものは、より限定的です。文1が存在した瞬間に合成が開始され、応答全体を待つ必要がなくなるという点です。最初の文と最後の文の間に実際の生成時間がかかる複数文の応答の場合、これがレイテンシ削減のすべてです。

ライブエンドポイントに対して、1セッションで3文からなる入力を送信して検証しました。3つのチャンクすべてが、各チャンクに送信したテキストと一致する音声付きで到着しました:

ステップ 4: 音声をクライアントに返すためのトランスポートを選択する

テキストを少しずつモデルへストリーミングすることで、問題の半分は解決します。残る半分は、生成された音声を、それをレンダリングする主体、つまり多くの場合はブラウザへ送り出すことです。エージェントやチャットのフロントエンドの多くは、すでにServer-Sent Events(SSE)経由でテキストをストリーミングしています。SSEは一方向のプロトコルで、ほとんどのブラウザがネイティブに対応しています。SSEはテキストプロトコルであり、生のバイナリペイロードを運ぶようには作られていません。既存のSSEパイプラインに音声を追加しようとするチームは、自然と既に持っているパターンに頼ることになります。つまり、音声をbase64エンコードし、JSONイベントに包むのです。

私たちは24本のクリップでその手法のコストを測定し、Amazon Transcribeでエンコーディング間に品質の違いがないことを確認した上で、MP3とAACを比較しました:

Base64はすべてのペイロードをおよそ期待どおりの33%膨らませます。より分かりにくいコストはフレームアラインメントです。SSEイベントは空行で区切られますが、内部のオーディオはMP3フレームに区切られています。この2つの方式は、トランスポートがデータをチャンクに分割する方法と必ずしも一致しません。実際に測定したあるレスポンスでは、4つの論理的なMP3フレームが6つの配信チャンクとして到着し、そのうち2つはフレームの途中で終わっていました。各配信チャンクを1つの完全な単位として扱うコンシューマは、普通の健全なトラフィックでも失敗します。解決策は、受信バイトをバッファリングし、区切り文字が見つかった時点でのみ処理を行うことです:

buffer = b""
for event in resp["Body"]:
    if "PayloadPart" not in event:
        continue
    buffer += event["PayloadPart"]["Bytes"]
    while b"\n\n" in buffer:
        frame, buffer = buffer.split(b"\n\n", 1)
        handle(frame)

今日、モデル向けの経路で使われている代替手段は、base64とSSEフレーミングを完全に回避します。これは、増分テキスト入力をすでに運んでいる同じ双方向セッション上で、生のバイナリオーディオとJSON制御フレームを多重化し、1つのルールで両者を区別します。

import json


def _decode_control_frame(chunk: bytes) -> dict | None:
    try:
        decoded = json.loads(chunk.decode("utf-8"))
    except (UnicodeDecodeError, json.JSONDecodeError):
        return None
    return decoded if isinstance(decoded, dict) else None

JSONオブジェクトとしてデコードできないチャンクはオーディオであり、受け取ったままの形で、経路のどこにもエンコード処理を介さずに転送されます。注意すべきエッジケースが1つあります。生のPCMオーディオは、正当な理由でバイト 0x7b（ASCII文字 {）から始まることがあり、これはJSONオブジェクトを開くのと同じ文字です。このルールは、チャンク全体が有効なJSONとしてパースされることを要件とし、最初のバイトだけでは判断しないため、そのオーディオを誤分類することを回避します。PCMフレームがJSONとしてパースされることは実質的にまずありません。

注意: テキスト用にすでに動作するSSEパイプラインをフロントエンドが持っているなら、SSEイベント内にbase64エンコードしたオーディオを載せるのが実用的な選択です。ペイロードがおよそ33%増え、バッファと再構築のステップが加わりますが、すでに運用しているインフラを再利用できます。音声経路を新規に構築する場合、あるいはレイテンシを優先する場合は、統合されたバイナリとJSONのアプローチが両方のコストを回避し、テキストをすでに運んでいる接続を再利用します。

ステップ5: 実際のエージェントにまとめる

より完全なエンドツーエンドのケースは、1トークンずつ応答を構築するライブの生成エージェントであり、文1が存在するまで文4を知る術はありません。この例はStrandsエージェントとして実行され、Amazon Bedrockをバックエンドとし、Amazon Bedrock AgentCoreランタイムにデプロイされ、ステップ4のパターンでKokoroエンドポイントを呼び出します。この設計では3つの通信経路が接続します。Server-Sent Eventsを呼び出し元に返すブラウザ向け経路、テキスト生成のためのエージェントからAmazon Bedrockへのストリーミング呼び出し、そして合成のためのエージェントとKokoroエンドポイント間の双方向セッションです。

次の図は、これら3つの通信経路と、生成と合成のオーバーラップが発生する場所を示しています:

図3: 呼び出し元、Amazon Bedrock、Kokoroエンドポイントを結ぶ3つの通信経路

リクエストペイロードには mode フィールドが含まれ、値は "bidi" または "batch"のいずれかで、同じエントリポイント関数内の2つのコードパスに振り分けられます。両者間のベンチマークで変化する変数は1つだけです。それは合成が開始されるタイミングです。 run_bidiでは、2つのタスクが並行して実行されます。一方はBedrockのテキストストリームを読み取り、完成した文をチャンカーに供給します。もう一方はKokoroセッションを読み取り、到着した順にオーディオチャンクを出力します。両者は1つの共有キューにプッシュするため、コンシューマは真の到着順にイベントを取り出せます。早い文のオーディオは、Bedrockが後の文をまだ生成している間に呼び出し元に到達できます。 run_batch は意図的にその正反対で、1回の合成呼び出しの前に応答全体を待ち、公正なベースラインとして維持されます。

逆方向、つまりエージェントコンテナから呼び出し元への経路はbase64を使いますが、その理由はステップ4で述べたものとは異なります。 BedrockAgentCoreAppのエントリポイントは、SDKが自動的に data: {json}\n\n フレームへ変換する、プレーンなJSONシリアライズ可能な辞書を返す必要があります。生のバイナリフレームを渡す方法はありません。これはJSONネイティブのマネージドランタイムによる帰結であり、ステップ4で測定し、脱却した選択によるものではありません。 run_bidi と run_batch の両方がこの経路で同一のコストを払うため、両者の測定された差は純粋にタイミング、すなわち最初のオーディオチャンクが出力される時点の違いです。

テストと検証

我々は、TTSトランスポート自体が最初のオーディオまでの時間に与える寄与を、同じ267文字・4文のテキストブロックを2つの方法、すなわち1つのメッセージとして送る場合（バッチ）と最初の文だけを送る場合（インクリメンタル）で送信することによって切り分けました。ライブエンドポイントに対して各5回の試行を実施しました。

先に測定したとおり、最初の文だけを送ることで、最初のオーディオバイトまでの平均時間は329.3 msから156.7 msに短縮され、相対改善率は52.4%で、その改善はTTSトランスポート自体に起因します。

次に、実際の言語モデルがテキストを生成する、より完全なエンドツーエンドのケースを測定しました。同一のプロンプトを run_bidi と run_batch の両方で、同じライブのBedrockモデルとエンドポイントに対して実行し、変数は合成の開始タイミングだけにしました。

同じ手法を用いた結果、bidiモードの最初のオーディオまでの時間は、ローカルコンテナに対してはバッチモードより2.4倍速く、デプロイ済みのAgentCoreランタイムに対しては3.5倍速くなりました。

次のチャートは、両環境におけるこの差を示しています:

図4: ローカルおよびデプロイ済みランタイムにおける、バッチモードとbidiモードの最初のオーディオまでの相対時間

実際のネットワークホップを1つ追加すると、デプロイしたケースでは差は縮まるどころか、さらに広がります。 run_bidi は、そのホップを、生成と合成の間で既に利用しているオーバーラップの中に吸収します。 run_batch には、そのコストを隠せるようなオーバーラップがなく、一度に1つのことしか行わないため、追加されたネットワークオーバーヘッドのミリ秒はすべてそのシリアル待ちに直接上乗せされます。

自分のデプロイメントに対してこの比較を再現するには、エージェントコンテナが公開している同じbidi-vs-batchテストを実行します:

uv run agent-infra demo --endpoint-url <your-endpoint-url> --prompt "<your prompt>"

これにより、プロンプトが両方のモードを通して送られ、最初の audio_chunk イベントの agent_start に対するタイムスタンプがそれぞれ報告されます。代わりにエラーが表示される場合は、ステップ1の KeyError: 'text' ステップ 1 の症状。報告するエンドポイント InService と報告するエンドポイントは、デプロイの問題ではなくこのバグを示しています。

クリーンアップ

継続的な課金を避けるために、このウォークスルーが作成したリソースをデプロイの逆順に、まずエージェント、次にエンドポイントの順で削除してください:

uv run agent-infra destroy --bedrock-model-id <your-bedrock-model-id>
uv run speech-infra destroy kokoro-82m

agent-infra destroy はAmazon Bedrock AgentCoreランタイムリソースを削除します。 speech-infra destroy はSageMaker AIのモデル、エンドポイント設定、およびエンドポイントを削除します。どちらも、モデルウェイトをキャッシュするAmazon S3バケットの空にする操作や、コンテナイメージを保持するAmazon ECRリポジトリの削除は行いません。これらは複数のモデル間で共有できるためです。もはや不要な場合は、Amazon S3バケットを空にして削除し、Amazon ECRリポジトリを手動で削除してください。保存されたモデルチェックポイントとコンテナイメージは、削除されるまで少額のストレージ料金が発生し続けます。

まとめ

私たちは、音声エージェントから回避可能なレイテンシーの原因の一つ、つまりエージェントの完全な応答を待ってから音声合成を開始することを取り除く方法と、もう一つの目立たないコストを定量化する方法を示しました。SSE内のbase64エンコードされたオーディオはペイロードを約3分の1膨張させ、バッファリングしない場合、MP3フレーム境界とずれる可能性があります。エージェントが文を生成した時点で直ちにSageMaker AIの双方向エンドポイントへストリーミングすることで、単体では最初の音声までの時間を半分以上削減し、デプロイされたエージェントでは2.4倍から3.5倍削減できました。オーディオをテキストとしてエンコードする代わりに、そのセッション上でバイナリオーディオとJSON制御フレームを多重化することで、素朴なSSEリレーで必要となるサイズの膨張とフレームの再調整の処理を取り除けます。

すでにSSEでテキストをストリーミングしているエージェントに音声を追加する場合は、まずステップ4のbase64アプローチから始め、レイテンシーが優先事項になった時点で統合トランスポートを見直してください。詳細については、 Amazon SageMaker AIリアルタイム推論ドキュメント および Amazon Bedrock AgentCoreドキュメントを参照してください。このパターンをご自身のアーキテクチャに適用する際にサポートが必要な場合は、AWSアカウントチームにお問い合わせください。

著者について