AWS re:Invent 2025の4つのアップデートで作る 音声×分析スケジューリングエージェント
358 Views
December 26, 25
スライド概要
2025年12月24日に開催した社内イベント「AWSランチセッション第13回」の発表資料です。
AWS re:Invent 2025 にて発表された以下4つのアップデートを使って、分析処理をスケジュール実行できる音声会話AIエージェントを作ってみました。
・Nova 2 Sonic
・Bedrock AgentCore Runtime 双方向ストリーミング
・Lambda Durable Functions
・S3 Access Points for FSx for NetApp ONTAP
Nova 2 SonicとAgentCore Runtimeの双方向ストリーミングによる自然な会話から、DurableFunctionsをスケジューラーとして使用し、S3 Access Points for FSx for NetApp ONTAPでオンプレデータを分析します。
音声会話エージェントとの会話動画は以下からご覧ください。
「双方向ストリーミング対応エージェントとの音声会話」
https://www.docswell.com/s/kkmtyyz/vM52L9J-2025-12-26-5696
SFとコンピュータが好き
関連スライド
各ページのテキスト
2025-12-24 AWSランチセッション 第13回 AWS re:Invent 2025の4つのアップデートで作る 音声×分析スケジューリングエージェント 山崎 拓也
山崎 拓也 所属: SIer 仕事: • AWS案件のアプリやインフラのリード • 社内AWSサポート 好き: 低レイヤ、SF、AWS AWSアワード: • 2024～2025 Japan AWS Top Engineer • 2022～2025 Japan All AWS Certifications Engineer
会話で分析処理をスケジュールできるAIエージェントを作ってみました 組み合わせた４つのアップデート • Nova 2 Sonic • Bedrock AgentCore Runtime 双方向ストリーミング • Lambda Durable Functions • S3 Access Points for FSx for NetApp ONTAP
構成図 ❷ ❶ ❸ ❺ ❹ オンプレに見立てたONTAP
処理の流れ ❸ 指定された日時まで待つ ❺ 指定された日時にクエリで分析 ❷ 分析開始ツール呼び出し ❷ ❶ ❶ ユーザーから音声会話で 「明日20時に分析開始して」 ❸ ❺ ❹ オンプレデータを自動レプリ （今回はAWS上から） ❹
クライアント画面（AWSのサンプルそのまま）
クライアント画面（AWSのサンプルそのまま） ユーザーとエージェントとの 音声会話の文字起こし ユーザー 「明日の午後八時に分析を開始してください」 エージェント 「分析は明日の午後8時（日本時間）に開始する よう予約されました」
デモ動画：https://www.docswell.com/s/kkmtyyz/vM52L9J-2025-12-26-5696 双方向ストリーミングによる自然な会話
分析処理はONTAP内データをAthenaでクエリし、ログ出力するのみ
❷ ❶ ❸ ❺ ❹
Nova 2 Sonicは自然でリアルタイムな音声会話ができるAIモデル • 業界トップクラスの会話品質、価格設定、音声理解 • 日本語未対応 • イベントライフサイクルは複雑 いろんな言語を話せます 日本語は？ 未対応だけどちょとなら イベントライフサイクル：https://docs.aws.amazon.com/nova/latest/nova2-userguide/sonic-event-lifecycle.html Amazon Nova 2 Sonic の紹介: 会話型 AI 向けの新しい音声変換モデル：https://aws.amazon.com/jp/blogs/news/introducing-amazon-nova-2-sonic-next-generation-speech-to-speech-model-for-conversational-ai/
❷ ❶ ❸ ❺ ❹
Agent Core Runtimeの双方向ストリーミングでAIとより自然に会話 • AIが聞き取りと応答を同時に行える • 発言をかぶせたり、会話の方向転換ができる • WebSocketで双方向通信 • 今回はAWSのサンプル実装に、分析予約ツールを加えて使用 https://github.com/awslabs/amazon-bedrock-agentcore-samples/tree/main/01-tutorials/01-AgentCore-runtime/06-bi-directional-streaming 明日の午後に… 午後ですね。何時ですか？ あ、やっぱり明後日の― 了解。明後日の何時ですか？ 19時で。 Amazon Bedrock AgentCore Runtime が双方向ストリーミングのサポートを開始：https://aws.amazon.com/jp/about-aws/whats-new/2025/12/bedrock-agentcore-runtime-bi-directional-streaming/
❷ ❶ ❸ ❺ ❹
ひとつのLambda関数だけで、1年間うごくワークフローを構築できる Lambda Durable Functions • コード内でステップを定義してワークフローを実現 • ステップ完了時、状態がチェックポイントとして保存される • 1回の実行は15分、チェックポイントからリプレイして最大1年実行可能 [NEW LAUNCH] Deep Dive on AWS Lambda durable functions (CNS380)：https://www.youtube.com/watch?v=XJ80NBOwsow
Durable Functionsをスケジューラーとして使う • Waitを使えばスケジューラーができる • 停止中はコンピューティング料金無料（データ保持料などはかかる）
実行時の冪等性を担保すること • 実行ごとに出力が変わる処理は必ずSTEPで囲み、チェックポイントに保存する 4時間後、リプレイ ここでWAITに0秒を渡すため、例外で落ちる
❷ ❶ ❸ ❺ ❹
ONTAPはNetApp社の高機能ストレージOS • NFS、SMB、iSCSIなど多様なプロトコルに対応 • 高度なデータ管理機能 • AWSではFSx for NetApp ONTAPとして利用できる • SnapMirror： • ONTAPクラスタ間の差分レプリケーション機能 • ファイル単位ではなくブロック単位で高速・効率的にレプリする • FlexCache： • ONTAPクラスタ間のリモートキャッシュ機能 • READされたデータをブロック単位で効率的にキャッシュする 「NetApp ONTAP」って何？その概要を5分で解説！（基本機能、データ保護機能、効率化機能）：https://www.netapp.com/ja/blog/guide-for-netapp-ontap/ Amazon FSx for NetApp ONTAP とは何ですか?：https://docs.aws.amazon.com/ja_jp/fsx/latest/ONTAPGuide/what-is-fsx-ontap.html SnapMirrorデータ レプリケーション： https://www.netapp.com/ja/snapmirror-data-replication/ FlexCacheとは何ですか。： https://www.netapp.com/ja/data-storage/what-is-flex-cache/
ONTAP内のデータに、S3としてAWSサービスから直接アクセス • SnapMirrorでオンプレデータをレプリケートし、そのままAthenaなどの AWSサービスから使用できる • Amazon FSx for NetApp ONTAP Advent Calendar 2025がおすすめ https://qiita.com/advent-calendar/2025/fsxforontap スケジュールでレプリケート Amazon FSx for NetApp ONTAP が Amazon S3 と統合され、シームレスなデータアクセスが可能になりました： https://aws.amazon.com/jp/blogs/news/amazon-fsx-for-netapp-ontap-now-integrates-with-amazon-s3-for-seamless-data-access/
FlexCacheのボリュームは今のところ未対応でした • FlexCacheに対応すると、オンプレのデータをより効率よくAWSでも扱える
まとめ • AIとの双方向ストリーミングによる会話はとても快適 • Lambda Durable Functionsは使い方次第で可能性が広がる • ONTAPのデータをAWSで簡単活用 • re:Invent 2025のアップデートは他にも沢山！
参考 • S3アクセスポイント経由でAmazon FSx for NetApp ONTAPのデータにAthenaからクエリ実行してみた https://qiita.com/kanako_kodera_16/items/b49512653301ec68637b • Amazon FSx for NetApp ONTAPのSnapMirror設定方法 https://blog.serverworks.co.jp/fsx-ontap-snapmirror • [Amazon FSx for NetApp ONTAP] FlexCacheを試してみた https://dev.classmethod.jp/articles/amazon-fsx-for-netapp-ontap-flexcache/ • aws-samples / amazon-fsx-workshop / netapp-ontap/JP/09-flexcache https://github.com/aws-samples/amazon-fsx-workshop/tree/master/netapp-ontap/JP/09-flexcache
（補足）06-bi-directional-streamingのツール定義更新部分
• client側実装も同様に更新する
{
"toolSpec": {
"name": "startAnalytics",
"description": "指定された時刻・タイムゾーンで分析を実行するためのDurable Functionsを起動する",
"inputSchema": {
"json": '''{
"type": "object",
"properties": {
"date": {
"type": "string",
"description": "yyyy-mm-ddの形となるが、人間にそれを求めてはいけない。例えば「明日の19時」という入力に対して、現在日時を求める他のツールを使
用して上記yyyy-mm-ddの形で時刻を求めること"
},
"hour": {
"type": "number",
"description": "分析を実行する時刻。24時間表記とする"
},
"minute": {
"type": "number",
"description": "分析を実行する分"
},
"timezone": {
"type": "string",
"description": "分析を実行するタイムゾーン。pythonで解釈できる文字列を設定する必要があるが、人間にそれを求めてはいけない。"
}
},
"required": ["date", "hour", "minute", "timezone"]
}'''
}
}
}
（補足）06-bi-directional-streamingのツール呼び出し更新部分
• Lambda Durable Functionsを非同期で呼び出す
if toolName == "getdatetool":
from datetime import datetime, timezone
result = datetime.now(timezone.utc).strftime('%A, %Y-%m-%d %H:%M:%S') + " in UTC"
elif toolName == "startanalytics":
content = json.loads(content)
payload = {
"date": content["date"],
"hour": content["hour"],
"minute": content["minute"],
"timezone": content["timezone"],
}
resp = self.lambda_client.invoke(
FunctionName="start-analytics:latest",
InvocationType="Event",
Payload=json.dumps(payload).encode("utf-8"),
)
result = "指定時刻での分析を予約しました"
（補足）Lambda関数 start-analyticsの実装
from datetime import datetime
from zoneinfo import ZoneInfo
from aws_durable_execution_sdk_python.config import Duration
from aws_durable_execution_sdk_python.context import StepContext, durable_step
from aws_durable_execution_sdk_python.execution import durable_execution
import awswrangler as wr
@durable_step
def get_wait_seconds(step_context: StepContext, event: dict) -> int:
year, month, day = map(int, event['date'].split("-"))
start_analytics_datetime = datetime(
year=year,
month=month,
day=day,
hour=int(event['hour']),
minute=int(event['minute']),
tzinfo=ZoneInfo(event['timezone']),
)
now = datetime.now(ZoneInfo(event['timezone']))
delta = start_analytics_datetime - now
return int(delta.total_seconds())
@durable_execution
def lambda_handler(event, context):
# 渡されるevent
# {'date': '2025-12-22', 'hour': 19.0, 'minute': 0.0, 'timezone': 'Asia/Tokyo'}
# eventと現在日時の差から、分析実行日時までのwait秒数を求めるステップ
wait_seconds = context.step(get_wait_seconds(event))
# 分析実行日時までwaitする
context.wait(Duration.from_seconds(wait_seconds))
# 分析実（今回はAthenaでselectするだけ）
df = wr.athena.read_sql_query(
sql="SELECT * FROM default.ontap_table LIMIT 10",
database="default",
workgroup="primary",
)
print(df)
ご清聴ありがとうございました。