16
3

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

Day 13

AI Agent on AWS Advent Calendar 2025

Day 18

Japan AWS Jr. Champions Advent Calendar 2025

@har1101

音声対話エージェントをStrands × AgentCore × Nova 2 Sonicで動かしてみる！

16
Posted at

こんにちは、ふくちです。

re:Invent 2025にて、以下の3つが新発表されました。

1.Strands Agents SDKがBidirectional Streaming(双方向ストリーミング)に対応

2.AgentCore RuntimeのWebSocket通信によるBidirectional Streamingに対応

3.Nova 2 Sonicが東京リージョンにも登場

ということで、音声をインターフェースとしたAIエージェントを構築する準備が整ってきました。
今回はこれらの要素をすべて取り入れたAIエージェントを構築してみようと思います。

GitHubは以下です。基本的にはcdkディレクトリ配下を御覧ください。

1.Strands Agentsで音声対話エージェントを作る

公式ドキュメントはExperimental(実験的機能)として紹介してくれています。

Python版のStrands Agents SDKを用いて、クラスメソッドのたかくにさんがわかりやすく手順をまとめてくださっているので、これと公式ドキュメントを参考にして進めていきます。

環境構築とコード記述

uvを用いてプロジェクトのディレクトリを作成し、必要なモジュールをインポートします。

ターミナル
$ uv init <プロジェクト名>
$ cd <プロジェクト名>

# 以下でOSごとに必要なAudioをインストール
$ brew install portaudio  # Mac OSの場合
$ sudo apt-get install portaudio19-dev python3-pyaudio  # Linuxの場合
# Windowsの場合は不要

# Nova 2 Sonicに必要なプロバイダー設定と、エージェントが使えるツールを追加
$ uv add "strands-agents[bidi-all, bidi]" "strands-agents-tools"

Audioのインストールができていないと、strands-agents[bidi-all, bidi]をインストールする際エラーになります。

続いて、以下のようなコードを書きます。ほぼQuick Startのコピペです。

main.py
import asyncio

from strands.experimental.bidi import BidiAgent
from strands.experimental.bidi.io import BidiAudioIO, BidiTextIO
from strands.experimental.bidi.models import BidiNovaSonicModel
from strands.experimental.bidi.tools import stop_conversation

from strands_tools import http_request

async def main() -> None:
    # モデルはNova 2 Sonicを使う
    model = BidiNovaSonicModel(
        model_id='amazon.nova-2-sonic-v1:0',
        provider_config={
            "audio": {
                "voice": "tiffany", # ボイス設定によって対応している言語が違うらしい
            }
        },
    )

    # stop_conversation toolは、ユーザーがエージェントの実行を口頭で停止できるようにします。
    agent = BidiAgent(
        model=model,
        tools=[http_request, stop_conversation],
        system_prompt="You are a helpful assistant that can use the http_request tool to search and get some information. Speak Japanese.",
    )

    audio_io = BidiAudioIO()
    text_io = BidiTextIO()
    await agent.run(inputs=[audio_io.input()], outputs=[audio_io.output(), text_io.output()])

if __name__ == "__main__":
    asyncio.run(main())

簡単に用語解説しておきます。

用語 説明
Strands Agents AWSが提供するオープンソースのAIエージェントSDK
Bedrockモデルと連携してエージェントを構築できる
BidiAgent Bidirectional(双方向)にリアルタイムの音声・テキストストリーミング会話ができるエージェント(ただしまだ実験的機能)
BidiNovaSonicModel Amazon Nova 2 Sonic(音声モデル)用のBidiAgent対応モデル
BidiInput / BidiOutput 双方向ストリーミングの入出力チャネル
音声やテキストのイベントを送受信する
BidiAudioIO ローカルのマイク・スピーカーを使用するI/O
(ローカルのみ、クラウド上では使用不可)

またNova 2 Sonicは、モデルカタログ上日本語に対応しているとありますが、コンソール上で選択することはまだできません。
image.png

Voice Typeによって変わるらしいのですが、東京リージョンで選べる2種類だとまだ日本語を選べませんでした(2025/12/7現在)。
image.png

image.png

詳しくはこちらをご参照ください。

ただ、英語を選んで日本語で喋ってと指示すると一応喋ってくれるらしいです。なのでこれでやってみます。

ローカルでテストしてみる

上記作成したコードを実行することで動作確認可能です。

ターミナル
$ uv run main.py

実際の動作はご自身の環境でご確認ください！

ここまでで、ローカル環境で動作するAIエージェントをStrands × Nova 2 Sonicで作ることができました。

(Option)会話セッションのライフサイクル管理

基本的に何もしなければ、セッションは無限に続きます。
なので不要になったらどこかで止める必要があります。方法としてはCtrl+cで止めるか、会話を終わらせましょう的なことを言うと、エージェント側がツールを使って自動で止まってくれます。

main.py
import asyncio

from strands.experimental.bidi import BidiAgent
from strands.experimental.bidi.io import BidiAudioIO, BidiTextIO
from strands.experimental.bidi.models import BidiNovaSonicModel
from strands.experimental.bidi.tools import stop_conversation

from strands_tools import http_request

async def main() -> None:
    # モデルはNova 2 Sonicを使う
    model = BidiNovaSonicModel(
        model_id='amazon.nova-2-sonic-v1:0',
        provider_config={
            "audio": {
                "voice": "tiffany",
            }
        },
    )

    # stop_conversation toolは、ユーザーがエージェントの実行を口頭で停止できるようにします。
    agent = BidiAgent(
        model=model,
        tools=[http_request, stop_conversation],
        system_prompt="You are a helpful assistant that can use the http_request tool to search and get some information. Speak Japanese.",
    )

    audio_io = BidiAudioIO()
    text_io = BidiTextIO()

+   try: 
+       # 明示的にやめるまで無限にセッションは続く
        await agent.run(
            inputs=[audio_io.input()],
            outputs=[audio_io.output(), text_io.output()] # 音声と文字両方でアウトプット
        )
+   except asyncio.CancelledError:
+       print("\nConversation cancelled by user")
+   finally:
+       # stop()メソッドは、必ずrun()メソッドのループを抜け出した後で実施する
+       await agent.stop()

if __name__ == "__main__":
    asyncio.run(main())

run()またはreceive()ループを抜けた後で、必ずagent.stop()を呼び出す必要があります。
ループ実行中にstop()を呼び出すと、エラーが発生する可能性があるそうです。

2.AgentCore Runtimeへデプロイする

続いて、先程作成したエージェントをAWS環境へデプロイします。
デプロイ先は、こちらもWebSocket通信での双方向ストリーミングが可能になった、AgentCore Runtimeです。

こちらもクラスメソッドのじんのさんがわかりやすくまとめてくださっているので、これも参考にさせていただきます。

ベースはAgentCore Runtimeデプロイ用CDKプロジェクトを用意しておき、そこにデプロイするためのAgent用モジュールを用意していきます。

bidi-agent/agent.py
import os
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from starlette.websockets import WebSocket, WebSocketDisconnect

from strands.experimental.bidi.agent import BidiAgent
from strands.experimental.bidi.models.nova_sonic import BidiNovaSonicModel
from strands.experimental.bidi.tools import stop_conversation

from strands_tools import http_request, calculator

# 環境変数からモデルID取得(デフォルトはNova 2 Sonic)
model_id = os.environ.get("MODEL_ID", "amazon.nova-2-sonic-v1:0")

# BedrockAgentCoreApp を使用
app = BedrockAgentCoreApp()

@app.websocket
async def websocket_handler(websocket: WebSocket, context):
    """
    AgentCore Runtime から /ws に来た WebSocket 接続を受け、
    Strands の BidiAgent(Nova 2 Sonic)にブリッジする。

    クライアントは BidiAudioInputEvent / BidiTextInputEvent 形式の
    JSONイベントを送信し、BidiAudioStreamEvent / BidiTranscriptStreamEvent
    等のイベントをJSONで受信する。

    Args:
        websocket: Starlette WebSocketオブジェクト
        context: RequestContext (session_id, request_headers等を含む)
    """
    await websocket.accept()
    print("[Server] WebSocket connected")
    print(f"[Server] Context: {context}")

    # Nova Sonic モデルの設定
    print("[Server] Creating model...")
    model = BidiNovaSonicModel(
        model_id=model_id,
        provider_config={
            "audio": {
                "voice": "tiffany",
            }
        },
    )
    print("[Server] Model created")

    # BidiAgent の設定
    # stop_conversation toolはユーザーが口頭でエージェントを停止できるようにする
    print("[Server] Creating agent...")
    agent = BidiAgent(
        model=model,
        tools=[calculator, http_request, stop_conversation],
        system_prompt="You are a helpful assistant. Speak Japanese.",
    )
    print("[Server] Agent created")

    try:
        print("[Server] Starting agent.run()...")
        # WebSocketのreceive_json/send_jsonを直接I/Oとして使用(ドキュメントに記載あり)
        await agent.run(
            inputs=[websocket.receive_json],
            outputs=[websocket.send_json],
        )
        print("[Server] agent.run() completed")

    except WebSocketDisconnect:
        print("[Server] Client disconnected")
    except Exception as e:
        print(f"[Server] Error: {e}")
        import traceback
        traceback.print_exc()
    finally:
        print("[Server] Cleanup...")
        try:
            await agent.stop()
        except Exception as e:
            print(f"[Server] Stop error: {e}")
        try:
            await websocket.close()
        except Exception:
            pass
        print("[Server] Done")


if __name__ == "__main__":
    print("Starting WebSocket server with BedrockAgentCoreApp on port 8080...")
    app.run()
bidi-agent/Dockerfile
FROM public.ecr.aws/docker/library/python:3.13-slim
WORKDIR /app

# pyaudio build に必要なもの（assert.h を含む）
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    portaudio19-dev \
    && rm -rf /var/lib/apt/lists/*

RUN pip install --no-cache-dir uv
ENV UV_SYSTEM_PYTHON=1 UV_COMPILE_BYTECODE=1

COPY requirements.txt requirements.txt
RUN uv pip install --system -r requirements.txt
# 追加：pyaudio が入ってないならここでビルドを落とす
RUN python -c "import pyaudio; print('pyaudio ok:', pyaudio.__version__)"
RUN uv pip install --system "aws-opentelemetry-distro>=0.10.1"

ENV AWS_REGION=ap-northeast-1
ENV AWS_DEFAULT_REGION=ap-northeast-1
ENV DOCKER_CONTAINER=1

RUN useradd -m -u 1000 bedrock_agentcore
USER bedrock_agentcore

COPY . .
EXPOSE 8080

CMD ["opentelemetry-instrument", "python", "-m", "agent"]
bidi-agent/requirements.txt
bedrock-agentcore
strands-agents-tools
strands-agents[bidi]
pyaudio>=0.2.14
starlette

これで無事に動くはずです。
簡単に用語解説もしておきます。

用語 説明
AgentCore Runtime 任意のOSSエージェント(Strands, LangChain等)をサーバーレスでホスティングするコンテナ
BedrockAgentCoreApp AgentCore Runtime用のPythonフレームワーク
@app.websocket WebSocketエンドポイント（/ws）を定義するデコレータ
port 8080固定
RequestContext WebSocketハンドラに渡されるコンテキスト（session_id, request_headers等）

実装解説

以下、実装の解説です。StrandsとAgentCoreとそれ以外の要素が絡み合ってるので、できるだけわかりやすいように整理していこうと思います。

全体アーキテクチャ

イメージとしては以下です。実際のアプリケーションをイメージすると、オーディオ機能付きWebアプリ(クライアント)からRuntimeのWebSocketエンドポイントに接続し、Strands Agentsと対話するようなイメージです。
image.png

この時、クライアントとRuntimeの間にAPI Gateway+Lambdaを噛ませるべきなのかがわかっていません。
一応API GatewayにもWebSocketはありますが…実装が複雑になりそうだなぁと思っています。

それぞれの役割

BidiAgent

  • BidiAgentは音声・テキストをリアルタイムにやりとりするためのエージェント
  • Nova SonicなどのSpeech to Speechモデルと組み合わせて音声アシスタントを作れる
  • 入出力はBidiInput/BidiOutputというI/Oチャネル経由で行う
  • ローカル用にはBidiAudioIO(マイク・スピーカー）、BidiTextIO(ターミナルなど)がある
  • WebSocket経由の場合はwebsocket.receive_json/websocket.send_jsonを直接使用するのが良さそう

AgentCore Runtime

  • 任意のOSSエージェントを安全にホスティングするためのサーバレスランタイム
  • WebSocketモードでは、コンテナ側にport 8080の/wsエンドポイントがある前提
  • BedrockAgentCoreAppの@app.websocketデコレータを使うと、この条件に合わせて/wsを自動で生成

なぜWebSocket I/Oが必要か

冒頭で記載したローカル用のサンプル(Strands Agents単体)は端末のマイク・スピーカーを前提としています。

agent = BidiAgent(model=BidiNovaSonicModel())
audio_io = BidiAudioIO()
await agent.run(inputs=[audio_io.input()], outputs=[audio_io.output()])

しかし、AgentCoreのコンテナにはマイクがないためそのままデプロイしても動かないのです。そこで、

  • BidiAgent自体はサーバ側(AgentCore Runtime)に置く
  • I/OはWebSocketに差し替え
  • クライアントがマイク入力を取り、BidiAudioInputEvent等のJSONをWebSocketで送信する
  • BidiAgentはクライアントからの音声入力を受け取り、Nova Sonicが生成した音声を返す

とするようにしています。

実装上の修正点についても触れておきましょう。
ローカル実装からの大きな変更点としては、直接Audio I/Oを用いるのではなく、WebSocket用のI/Oを使用するように修正した点です。

ローカル版とデプロイ版の差分
- audio_io = BidiAudioIO()
- text_io = BidiTextIO()
  await agent.run(
-   inputs=[audio_io.input()],
-   outputs=[audio_io.output(), text_io.output()]
+   inputs=[websocket.receive_json],
+   outputs=[websocket.send_json],
  )

詳しくは以下のドキュメントに記載されています。

WebSocketとStarlette

クライアント-サーバー間におけるWebSocket通信を行うために、今回はStarletteというものを使っています。

Starletteは軽量のASGI(Asynchronous Server Gateway Interface)フレームワーク/ツールキットで、Pythonで非同期Webサービスを構築するのに最適とのことです。

公式サイトに特徴が記載されているのですが、上2つが以下になっていたので、採用してみました。

  • 軽量で複雑さの少ない HTTP Web フレームワーク
  • WebSocket サポート

また、StarletteはFastAPIの基盤として動いているようです。
先程のIOのドキュメントではWebSocket通信のためにFastAPIが採用されていましたが、以下サイトをたまたま見つけたので使ってみることにしました。

曰く、

Starlette はネイティブで WebSocket をサポートしています。
FastAPI も WebSocket をサポートしていますが、実装の詳細は Starlette と似ています。Starlette は WebSocket 処理のインターフェイスを直接公開しており、開発者がより深くカスタマイズするのに便利です。

と。今回は特にFastAPIでないといけない理由が特に無く、むしろより軽量で高速なStarletteを使った方が動作が軽くなるのでは、と思って採用しています。

とにもかくにも、これでクライアントからの通信をWebSocketで受け取れるようになります。

イベント形式（Strands Bidi Events）

続いてはI/Oの形式です。、開発者だけが知っていれば良いようなものです。
簡単に言うと、BidiTextInputEvent・BidiAudioInputEvent等のイベントをJSON形式で送受信する形式になっています。

BidiAudioInputEvent(音声入力)

クライアントからサーバーへの入力形式は以下のとおりです。

{
    "audio": "<Base64でエンコードされたオーディオ文字列>",
    "format": "pcm",
    "sample_rate": 16000,
    "channels": 1
}

見慣れない用語が出てきました。以下にまとめておきます。

フィールド 型 説明
audio string Base64でエンコードされた、
モデルに送信するオーディオ文字列
format string 音声フォーマット
"pcm", "wav", "opus", "mp3"のいずれか
sample_rate number サンプリングレート(Hz)
16000, 24000, 48000のいずれか
channels number チャンネル数
1=モノラル、2=ステレオ

BidiAudioStreamEvent(音声出力)

サーバーからクライアントへの出力形式は以下のとおりです。

{
    "type": "bidi_audio_stream",
    "audio": "<base64 encoded PCM>",
    "format": "pcm",
    "sample_rate": 16000,
    "channels": 1
}

見ていただくと分かる通り、ほとんど入力と一緒です。

フィールド 型 説明
type string イベント種別。"bidi_audio_stream"が固定で入る
audio string Base64でエンコードされた、
モデルが出力したオーディオ文字列
format string 音声フォーマット
"pcm", "wav", "opus", "mp3"のいずれか
sample_rate number サンプリングレート(Hz)
16000, 24000, 48000のいずれか
channels number チャンネル数
1=モノラル、2=ステレオ

これらのパラメータの詳細については別途ブログで取り上げます。私もきちんと理解できているわけではないので…
ひとまずここでは、上記のような入力パラメータがあるということをご理解いただければと思います。

詰まったポイント・トラブルシューティング

今回、CDKの「deploy-time-build」Constructを使用してエージェントをビルド・デプロイしています。
その際にいくつかエラーが出たので、その解消法についても記載しておきます。

エラー1.pyaudioビルド失敗

CodeBuildでrequirements.txtのインストール実行時に、「assert.hが無い」ということでエラーになりました。

#11 2.384 self._finalize_license_expression()
#11 2.384 In file included from src/pyaudio/device_api.h:7,
#11 2.384 from src/pyaudio/device_api.c:1:
#11 2.384 /usr/local/include/python3.13/Python.h:20:10: fatal error: assert.h: No
#11 2.384 such file or directory
#11 2.384 20 | #include // assert()
#11 2.384 | ~~~~~~~~~^~~~~~~~~~
#11 2.384 compilation terminated.
#11 2.384 error: command '/usr/bin/gcc' failed with exit code 1
#11 2.384
#11 2.384 hint: This error likely indicates that you need to install a library
#11 2.384 that provides "assert.h" for pyaudio@0.2.14
#11 2.384 help: pyaudio (v0.2.14) was included because strands-agents[bidi]
#11 2.384 (v1.19.0) depends on pyaudio
#11 ERROR: process "/bin/sh -c uv pip install -r requirements.txt" did not complete successfully: exit code: 1

Dockerfile:18
16 | COPY requirements.txt requirements.txt
17 | # Install from requirements file
18 | >>> RUN uv pip install -r requirements.txt
19 | RUN uv pip install aws-opentelemetry-distro>=0.10.1
20 |
ERROR: failed to build: failed to solve: process "/bin/sh -c uv pip install -r requirements.txt" did not complete successfully: exit code: 1

原因

以下、チャッピーの助けを借りた部分を多分に含みます。文章自体は人間が打ってます。適宜ソース確認はしましたが、間違っている場合はご指摘ください。

これはpyaudioをインストール中に、C言語の拡張モジュール(pyaudio._portaudio)をコンパイルしようとし、その途中でPython.hがassert.hを読み込もうとして失敗した、ということだそうです。

※Python.hは、C言語やC++からPythonの機能を利用したり、Pythonの拡張モジュールを作成する際に必要となるC言語のヘッダファイルです。

※assert.hはC言語の標準ライブラリのヘッダファイルです。中にはassert()というマクロが定義されており、想定が崩れていないかをチェックするようなものになっています。

ではなぜこのassert.hが無かったのか、と言う話ですが、今回使っていたDockerベースが「python:3.13-slim」だったのが原因のようです。

-slim版は、Pythonを実行するのに最低限のパッケージのみが含まれたイメージであり、開発用ツールが入っていないそうです。assert.hはDebianだと主に[「libc6-dev」のようなC標準ライブラリの開発用パッケージに含まれます。
※実際にlibc6-devのファイル一覧にはassert.hが入っています。

そしてこの-slim版Dockerfileの中身を確認すると、一時的にlibc6-devがインストールされた後、最終的にはapt-get purge --auto-removeコマンドで削除されているようでした。
なので、-slim版にはこのassert.hが存在せず、ビルドでも失敗するという状況だったようです。

なぜpyaudioを入れるとコンパイルが必要になるのか？

そもそもpyaudioとは、Pythonで音声を扱う際に使うことができるライブラリです。
マイクからの入力を録音したり、音声データを再生したり、リアルタイム処理が可能なようです。

そしてこのpyaudioは、Pythonからマイク/スピーカーを扱うためにPortAudioというCライブラリに橋渡しする仕組みだそうです。PyPIのページでは「PortAudio v19へのPythonバインディング」と説明されていました。

このタイプのライブラリは、インストール時に

  • 既に用意された完成品があればそれを置くだけ
  • 完成品がない・もしくは環境に合わない場合はソースからCコンパイルして作る

という挙動になるようです。
今回は後者を実行しようとしたけれど、assert.hが無いため失敗したという感じでした。

対応策

Dockerfileに以下を追加することで、解決しました。

Dockerfile
# pyaudio build に必要なもの（assert.h を含む）
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    portaudio19-dev \
    && rm -rf /var/lib/apt/lists/*

ここで「build-essential」と「portaudio19-dev」をインストールしています。

build-essentialは、Debianパッケージをビルドするのに必須とされるもの一式をまとめて入れるためのメタパッケージのようです。もちろんlibc6-devおよびassert.hも入っています。
これでビルドが進むようになりました。

そしてportaudio19-devも追加していますが、これはpyaudioが依存するためです。ビルド時にportaudioのヘッダやライブラリが必要になるため、追加しています。

これでCodeBuildでのビルド実行時エラーは解消することができました。

エラー2.ビルドはできたのに、サーバーが動かない

エラー1を解消してビルド・デプロイはできたのですが、サーバーが動きませんでした。
CloudWatch Logsを確認すると、以下のようなログが出ていました。

Traceback (most recent call last):
File "<frozen runpy>", line 198, in _run_module_as_main
File "<frozen runpy>", line 88, in _run_code
File "/app/agent.py", line 14, in <module>
from strands.experimental.bidi.agent import BidiAgent
File "/usr/local/lib/python3.13/site-packages/strands/experimental/bidi/__init__.py", line 18, in
from .io.audio import BidiAudioIO
File "/usr/local/lib/python3.13/site-packages/strands/experimental/bidi/io/__init__.py", line 3, in
from .audio import BidiAudioIO
File "/usr/local/lib/python3.13/site-packages/strands/experimental/bidi/io/audio.py", line 15, in <module>
import pyaudio
ModuleNotFoundError: No module named 'pyaudio'

pyaudioが存在していないようです。
私の感覚としては、Runtimeは音声デバイス不要なのでpyaudioも不要だと思っていたのですが、実際にはStrands Agentsの実装に引っ張られる形で必要になることがわかりました。

上記のログのとおりなのですが、"/usr/local/lib/python3.13/site-packages/strands/experimental/bidi/__init__.py"の18行目で、BidiAudioIOをインポートしています。

更にその後、File "/usr/local/lib/python3.13/site-packages/strands/experimental/bidi/io/audio.py"の15行目でpyaudioをインポートしています。

よって、Runtimeが音声デバイスを用いるからpyaudioが必要なわけではなく、Strands Agentsのbidi-agentを使う際の依存関係によってpyaudioが必要になるという形でした。

ということで、requirements.txtにpyaudioを追加してあげています。現状(2025/12/14)の最新は0.2.14です。

bidi-agent/requirements.txt
bedrock-agentcore
strands-agents-tools
strands-agents[bidi]
pyaudio>=0.2.14
starlette

まとめ

自分たちオリジナルの音声エージェントを開発して、ハンズフリーでできることを増やしていきましょう！

16
3
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
16
3

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?