※本記事は、AWS Streaming Specialist Solutions ArchitectのSubham Rakshit氏による講演「Agentic AI with AWS Streaming」の内容を基に作成されています。本講演は、2026年3月12日にアイルランド・KilkennyのLyrath Convention Centreで開催された第5回 AWS AI and Data Conference 2026にて収録されたものです。動画の詳細情報やAWSのイベントについては https://go.aws/events でご覧いただけます。本記事では、講演の内容を要約しております。なお、本記事の内容は原登壇者の見解を正確に反映するよう努めていますが、要約や解釈による誤りがある可能性もありますので、正確な情報や文脈については、オリジナルの動画をご視聴いただくことをお勧めいたします。
1. セッション導入とAI時代におけるストリーミングの重要性
1.1 登壇者紹介とアジェンダ
Shubham： 皆さん、こんにちは。本日は、ストリーミングとメッセージングのサービスを使ってAIエージェントアプリケーションをどう強化するかについてのセッションにお越しいただき、ありがとうございます。私はShubham Rakshitと申しまして、AWSでデータストリーミングのソリューションアーキテクトを務めています。日頃はお客様と一緒に、スケーラブルで高性能なストリーミングアーキテクチャの構築をお手伝いしています。本日のセッションでは、ストリーミングシステムとAIエージェントアプリケーションをどう組み合わせれば、応答性が高く自律的なAIアーキテクチャを構築できるのか、というテーマでお話しします。
Shubham： まずアジェンダを簡単にご紹介します。最初に、AIの世界においてストリーミングがなぜ重要なのか、その文脈を設定します。続いて、ストリーミングやメッセージングのデータがどのようにアプリケーションの差別化要因になりうるのかをお話しし、リアルタイムのAIエージェントを構築する際に活用できる設計パターンをいくつか取り上げます。その後、従うべきグッドプラクティスについて触れ、最後にデモをお見せします。うまく動いてくれることを願っていますが、これはライブデモで、AIエージェントを使ったジオフェンシングのアラート処理をご覧いただく予定です。
1.2 AIが中心となる変化とデータ基盤強化の必要性
Shubham： さて、今は本当にデータにとってエキサイティングな時代です。AIが主役になりつつある今、ランドスケープ全体が変化していることに驚く方はいないでしょう。Harvard Business Reviewが実施した調査では、最高データ責任者（CDO）の89%が、何らかの形でAIに関する施策を検討していることが分かりました。しかしながら、組織の半数以上は、この変化に対応する準備がまだできていません。その理由は、データ基盤の不足にあります。AI施策で成功を収めるお客様は、基本に立ち返り、まずデータ基盤を強化することを徹底しているのです。
1.3 データ戦略の4つの課題
Shubham： ところが、その基盤強化に取り組む中で、お客様はAIアプリケーションを支えるデータ戦略の構築において、大きな課題に直面しています。一つ目は、チーム連携の分断です。組織にはアナリスト、エンジニア、データサイエンティストなど、さまざまな職種の人がいますが、それぞれが孤立して別々のツールを使っており、コンテキストの切り替えが頻繁に必要となって、連携の妨げになっています。二つ目は、データの断片化です。データはサイロ化され、異なるフォーマット、異なる場所に散らばっていて、AIや分析アプリケーションのための統一された基盤を築くにはまだ適していません。三つ目は、データとAIの双方にわたるエンドツーエンドのガバナンスです。パイプライン全体を通じてセキュリティ、プライバシー、責任ある利用を確保する必要があります。そして四つ目が、要求の厳しいAIワークロードの負荷を引き受けつつ効率も満たす、スケーラブルで高性能なアーキテクチャをどう構築するか、という点です。
1.4 ストリーミングの4つの利点
Shubham： こうした課題に対処するために、多くのお客様がストリーミング技術を採用しているのを目にします。ストリーミングはより大きな価値の実現にも役立ちます。その魅力的なメリットをいくつか挙げますと、まずデータをリアルタイムに取り込んで分析し、より速く洞察を得られること。次に、不要な情報をフィルタリングして除外できるため、後段のストレージコストを削減できること。三つ目に、ライブデータをAIモデルに供給することで、継続的な更新やトレーニングを可能にし、データのより良い分析を提供できること。そして最後に、イベント駆動アーキテクチャによって、すべてのシステムやアプリケーションを常に最新の状態に保てることです。これらを実現するために、AWSのストリーミングサービスを活用いただけます。これらは最も幅広く、最も奥行きのあるストリーミング機能のセットを提供しており、シンプルさ、スケーラビリティ、信頼性を確保しながら、データを取り込んで分析し、分析アプリケーションやAIアプリケーション向けに処理することができます。
2. AWSストリーミングサービスの全体像とAIエージェントの定義
2.1 3つの柱
Shubham： AWSのストリーミングサービスは、インフラストラクチャ、ムーブメント、プロセッシングという3つの異なる柱に整理されています。データの取り込みと保存、データの移動と変換、そしてデータの処理という流れを、それぞれに適したサービスでカバーできるようになっているのです。
2.2 主要サービス群
Shubham： まずインフラストラクチャの層では、データの取り込みと保存を行います。ここではクラウドネイティブなKinesis Data Streamを使うこともできますし、オープンソースのApache Kafkaを利用したい場合は、Amazon Managed Service for Apache Kafkaを活用することもできます。次にムーブメントの層、つまりデータの移動と変換については、Amazon Data Firehoseを使えます。これは以前Kinesis Data Firehoseと呼ばれていたもので、失礼しました、今はAmazon Data Firehoseという名称になっています。さらに、Amazon MSK Connectを使えば、リアルタイムのデータをレイクハウスやウェアハウス、AIモデルへと供給することができます。
Shubham： そして最後にプロセッシングの層では、Amazon Managed Service for Apache Flinkを利用できます。これはもう一つのオープンソース技術であるFlinkをベースにしたもので、ストリーミングジョインやデータのライブエンリッチメントといった複雑な処理を行い、データを継続的に更新して、より速く洞察を得ることができます。
2.3 会場への挙手アンケート
Shubham： ここで簡単に挙手をお願いしたいのですが、実際に今、データストリーミング技術を使っている方はどれくらいいらっしゃいますか。なるほど、いいですね。では、組織の中で何らかのAI施策に直接または間接的に関わっている、あるいはそうした取り組みを認識している方はどれくらいいらっしゃいますか。いいですね。それであれば、これからお話しするパターンのいくつかは、皆さんのワークロードに当てはまるかどうか、どう適用できるかを考えながら聞いていただけると役立つと思います。
2.4 AIエージェントの定義と特徴
Shubham： さて、AIエージェントの話を盛んにしていますが、ここでAIエージェントとは何かという文脈を整理しておきましょう。皆さんはすでにご存じでしょうし、今朝から何度もこのスライドを目にしてきたかもしれません。それでも文脈のために申し上げますと、AWSでは、AIエージェントを「AIを活用して、人間やシステムに代わって推論し、計画を立て、タスクを実行する自律的なソフトウェアシステム」と定義しています。
Shubham： このAIエージェントの優れている点は、反復的に思考できることです。結果を評価し、アプローチを更新しながら、ゴールに向かって作業を続けることができます。エージェントは単に質問に答えるだけではありません。探索と改良を通じて、特定の問題を解決するために働かせることもできるのです。リアルタイムデータの重要性と、それがどのようにAIを強化するのかについて議論してきましたので、次はストリーミングやメッセージングのサービスが、AIアプリケーションにとってどのように差別化要因となりうるのかを見ていきましょう。
3. ストリーミング／メッセージングが差別化要因となる理由
3.1 インテリジェンスと迅速な行動
Shubham： インテリジェンス（知性）はデータを活用することで得られる、というのは周知の事実です。ですから、継続的にストリーミングされるデータは、エージェントが活用できるリアルタイムのインテリジェンスを提供してくれます。しかし、インテリジェンスだけでは十分ではありません。組織にインパクトを与える意味のある結果を得るためには、そのインテリジェンスに対してより速く行動を起こす必要があるのです。
3.2 自律システムの要件とフィードバックループ
Shubham： AIエージェントを使えば、データを継続的に収集し、継続的に処理し、洞察を見つけ出し、さらに自律的に行動を起こす自律システムを考えることができます。そしてここで重要なのが、絶え間ないフィードバックループです。これによって、エージェントは過去の行動から学び、将来に向けてより良い意思決定ができるようになります。
3.3 ストリーミングストレージの役割
Shubham： では、データストリーミングサービスはこのアプローチにどう適合するのでしょうか。Amazon MSKやAmazon Kinesisを使えば、耐久性が高くスケーラブルなストリーミングストレージが手に入ります。これは高速度のイベントストリームを信頼性高く処理できます。一例を挙げますと、昨年のAmazon Prime Dayでは、Kinesis Data Streamsがピーク時に1秒あたり8億700万レコードを処理しました。このストリーミングストレージは、エージェントに継続的にデータを供給するためにも使えますので、ライブデータをエージェントに送り込んで、エージェントの認識を変化させることができます。
3.4 Flinkの役割
Shubham： ストリーミングストレージに保存したデータは、当然ながら何らかの変換を経る必要があります。この場合はストリーミング変換ですが、Flinkを使えば、特徴量の生成や埋め込みの準備が整うまで、すべての中間層にデータを保存しておけます。そしてFlinkを使うことで、エージェントの意思決定に向けた特徴量生成をリアルタイムに作り出せますし、マルチステップの推論プロセスに役立つデータのライブエンリッチメントも行えます。さらに、Flinkの非同期呼び出し機能を使えば、AIエージェントのアクションをトリガーすることもできます。
Shubham： Flinkはステートフルなストリーム処理システムです。つまり状態を持っている、言い換えれば独自のメモリのようなものを持っているということです。このメモリをバッファとして使ってデータを保存し、より多くのコンテキスト管理に活用できます。そして最後に、入ってくるデータを使ってリアルタイムの動的なコンテンツを構築できますので、エージェントが新鮮なデータを取得して、より速く、より良い形で意思決定を行う助けになります。
3.5 メッセージングサービスの役割
Shubham： ここではメッセージングサービスも大きな役割を果たします。SQSのようなサービスは、これまで何千ものお客様がマイクロサービスアーキテクチャを疎結合にするために役立ててきました。それと同じ役割を、エージェント型AIのプロセスでも果たすことができます。というのも、これらを使ってエージェント型AIアプリケーションを疎結合にできるからです。オーケストレーションのパターンを構築すれば、耐障害性のあるエージェント型アプリケーションを作ることができます。メッセージングサービスがサポートする非同期通信は、安全なリトライとメッセージの確認応答を確保することで、データの配信保証と順序保証を実現します。そして、これまで大量のデータをさばくためにスケールしてきたのと同じサービスが、今やAIエージェントの能力を支えることもできるのです。一例として、同じPrime Dayのイベントでは、SQSがピーク時に1秒あたり約1億6,600万イベントを処理しました。ですから、スケーラビリティについては、最も心配する必要のない要素だと言えます。
3.6 各業種でのユースケース例
Shubham： ここからは、お客様がストリーミングデータの上にエージェント型AIアプリケーションをどのように構築し始めているのか、いくつかのユースケースをご紹介します。私たちは皆、何らかのチャットボットとやり取りした経験があるはずですし、何らかのパーソナライズされたレコメンデーションを受け取ったこともあるでしょう。これらはすべてAIを使って構築されており、データもストリーミングの形で流れてきています。製造、オートメーション、金融など、あらゆる業種を横断して、AIの実装が見られるようになっています。具体的なユースケースとしては、異常検知、アラートのトリガー、予防保全といったものが挙げられます。
4. 設計パターン①：異常検知と反応
4.1 異常の定義と履歴継承
Shubham： ここからは、リアルタイムデータとAIエージェントを組み合わせた設計パターンをいくつか見ていきましょう。最初に取り上げるのは、ストリーミングとAIエージェントという二つの能力を融合させて、お客様が実際に構築しているとても一般的なパターンで、私たちは「異常検知と反応」と呼んでいます。
Shubham： では、異常とは何でしょうか。あるグラフを見ると、青い線が引かれていて、その線の外側にプロットされたいくつかの点があるのが分かります。人間の目であれば、それらが異常だと文字どおり見分けることができますよね。これは異常だ、と。つまり別の言い方をすれば、正常でないものが異常と呼ばれるわけです。しかし問題は、システムはどうやって何が異常かを判定するのか、ということです。まず私たちはデータの正常なパターンを学習します。そして、その正常なパターンをベースラインとして、イベントが流れてくるたびに継続的に取り込み、その正常なパターンに対してプロットしていきます。そして、どれだけ離れた位置にプロットされたか、ある閾値を超えているかどうかを見ます。もし閾値を超えていれば、それは異常と判定されるのです。
Shubham： ただし、履歴を引き継いでいく必要もあります。なぜなら、昨日は異常と判定されたものが、今日には新たな正常になっているかもしれないからです。
4.2 異常検知の手法とユースケース
Shubham： 異常検知について考えると、それを行う方法はさまざまにあります。一つは間違いなくルールベースの異常検知です。いくつかのルールを用意して、流れてくるデータに対してそのルールを実行し、異常を割り出します。しかし、この仕事を非常にうまくこなす機械学習モデルもあります。たとえばRandom Cut Forest、ARIMAモデル、XGBoostといったものは、いずれも異常検知の構築に使えます。
Shubham： そして異常検知は、まさに水平的な、どの業界にも当てはまるユースケースです。さまざまな業界セグメントで見られる異常検知のユースケースをいくつか挙げますと、まずゲーム業界では、エンゲージメントの急激な低下といったチャーン（解約）防止があります。サイバーセキュリティや金融では、不正検知や取引の異常があります。Eコマースでは価格の異常が、ヘルスケアでは患者のリスクアラートが見られます。これらはすべて何らかの異常です。ですから、ぜひ皆さんもご自身のドメインの視点から、何が異常とみなされるのか、どうやってデータストリーミングを行えるのか、そしてそれらの異常をどう継続的に監視し、どう反応するのかを考えてみていただきたいのです。
4.3 高レベルアーキテクチャ
Shubham： この異常検知の高レベルなアーキテクチャは、次のようになります。大きく二つの部分に分かれていて、一つはストリーム処理の部分、もう一つはエージェント型のワークフローの部分です。ストリーム処理の部分が行っているのは、まず生のイベントを取り込み、それらのイベントを保存して、データの処理を行い、異常を検知することです。
4.4 従来技術とAIの役割分担
Shubham： ここで注意深く見ていただきたいのですが、このストリーム処理の部分は、従来型のストリーム処理システムを使って実現できる部分です。ここにエージェント型AIを使う必要はありません。ルールベースのアラートシステムでもできますし、MLモデルを使ってもかまいません。実際、皆さんの多くはすでにこれを行っているでしょう。組織の中に、異常を予測するこうしたアーキテクチャをすでにお持ちのはずです。
Shubham： では、このアーキテクチャの次の部分は何かというと、それらの異常をAIを使ってどう分析し、どう反応するか、という部分です。何が起こりうるのか、根本原因は何か、なぜそれが起きているのか、そしてこの問題に対する潜在的な緩和策は何か、ということを扱います。異常が起きているなら、そこには必ずルールブックがあるはずだと考えてください。そのルールブックをAIエージェントに与えて、その異常に対するアクションを取らせるのです。するとエージェントは何をすべきかを判断できます。メールやさまざまなアラートで通知を送る必要があるかもしれませんし、アクセスをブロックするといった何らかのアクションを取る必要があるかもしれません。
4.5 実装構成
Shubham： これを実際の実装に落とし込むと、次のようになります。データソースからデータを受け取り、そのデータをKinesis Data StreamsまたはAmazon MSKに保存します。大量のデータが流れてくる中で、Apache Flinkを使ってそれらのデータを継続的に処理し、異常を計算します。Flinkの中にルールベースのアラートを書くこともできます。また、Flink自体からAIモデルやMLモデルを呼び出すこともできます。MLモデルはFlink上に直接ホストすることが可能で、たとえばRandom Cut Forestの実装はFlinkの中で行えます。あるいは、別のサードパーティのモデルやAPIに対して非同期に呼び出しを行い、そのイベントが異常かどうかを判定することもできます。そして次に、そのデータをAIエージェントに渡す必要があります。
5. AIエージェントの構築・デプロイ（Amazon Bedrock / AgentCore）
5.1 Bedrockのプロトコル・フレームワーク統合
Shubham： エージェントの話をしているわけですから、ここでAIエージェントをどのように構築しデプロイするのかを学んでいきましょう。そのために使えるのがAmazon Bedrockです。Amazon Bedrockは、MCPやAgent-to-Agentプロトコルといったオープンソースのプロトコルを統合しており、さらにStrands Agent、Crew AI、LangGraphなど、数多くのフレームワークにも対応しています。
5.2 基盤モデルの選択肢と最適化
Shubham： 基盤モデルの面では、DeepSeekのモデル、OpenAIのモデル、Anthropicのモデルなど、さまざまな基盤モデルを使う能力が提供されており、多様な選択肢が用意されています。加えて、コストとパフォーマンスの最適化や、利用できるガードレールの定義といった機能も備わっていて、AIエージェントを構築するための包括的でエンドツーエンドのプラットフォームを築くことができます。
5.3 AgentCoreの9コンポーネント
Shubham： このAgentCoreプラットフォームを掘り下げてみますと、これは9つの異なるコンポーネントを提供するプラットフォームです。具体的には、runtime、memory、identity、gateway、code interpreter、browser、observability、policy、evaluationの9つです。
Shubham： ここで重要なのは、これらのコンポーネントをミックス＆マッチできるという点です。ご自身のワークロードに対して、すべてを使う必要はありません。それは完全に用途次第です。たとえば、エージェントに短期記憶と長期記憶が必要であれば、AgentCore memoryを活用できます。MCPサーバーを介した簡素化されたディスカバリを行いたいのであれば、AgentCore gatewayを活用できます。AgentCore identityを使ってセキュリティを確保することもできますし、ポリシーを定義してエージェントの振る舞いを実行時に制御することも、AgentCore evaluationを使ってエージェントのパフォーマンスを測定し検証することもできます。そして、AIエージェントをデプロイする必要が出てきたら、AgentCore runtimeのプラットフォームを活用してデプロイすることができます。
5.4 オブザーバビリティの重要性
Shubham： 最後に、オブザーバビリティの部分が非常に重要です。舞台裏で何が起きているのかを観測できなければ、皆さんがどんなことをしても成功には結びつきません。あるエージェントから別のエージェントへのすべてのホップをトレースし、レイテンシを監視して、どこで多くの時間を費やしているのかを把握する必要があります。そうすることで、完全に本番運用に耐えるシステムを確保できるのです。
5.5 Flinkとの統合アーキテクチャ
Shubham： ここでアーキテクチャ図に話を戻しますと、私たちができるのはシンプルに、Flinkを使ってAgentCoreのAPIを呼び出し、Flinkで異常として検知されたデータを、AgentCore runtime上で動作しているエージェントに渡すということです。そしてエージェントは、そのデータに対して何をすべきかを判断し、いくつかのアクションを実行します。外部APIを呼び出すこともできますし、おそらく一連のステップを経て、それが最終的に偽陽性なのか、それとも本物の異常なのかを判定します。そのうえで、SlackやSMS、メールを使って通知するといった形でアラートを出すこともできますし、さらに一歩進んでアクセスをブロックするといったアクションを取ることもできるのです。
6. 設計パターン②：マルチエージェントワークフロー
6.1 単一エージェントからの出発と複雑化
Shubham： 二つ目のパターンは、マルチエージェントのワークフローです。エージェントを構築するとき、私たちは単一のエージェントを作っているわけではありません。多くの場合、エージェントのネットワークを構築することになります。では、それはどのようにして生まれてくるのでしょうか。最初に取りかかるときは、一つのエージェントから始めます。たとえば、異常応答エージェントを構築しているとしましょう。この異常応答エージェントは、すべての異常イベントを読み込みます。そして他のデータポイントと相関を取り、続いて履歴のパターンを確認し、それが不正かどうかを判断して、何らかの緩和策を決定します。最後に、そのための応答や通知も生成します。そしてこのエージェントには、MCPサービスを介してすべての外部システムやデータベースへのアクセス権が与えられています。すべてのナレッジベース、ポリシー、ルールなど、あらゆるものにアクセスできるわけです。
Shubham： ところが、構築を続けていくにつれて、プロンプトはどんどん複雑になっていきます。ほとんどのプロジェクトはこのように始まりますし、それはまったく問題ありません。しかし、本番運用に耐えるシステムに仕上げる段階になると、一つの大きなエージェントがボトルネックになりうるのです。なぜなら、すべてを一手にやろうとしているからです。ロジックが重くなるにつれて、当然ながらコストは膨らみ、エラーが起きやすくなり、動作も遅くなっていきます。
6.2 モノリシックエージェントの4つの課題
Shubham： つまり、スピードと制御の両方を失ってしまうのです。このモノリシックなエージェントが抱える課題を整理すると、4つの柱に分けられます。一つ目は認知的な課題です。あらゆるコンテキストを一手に抱えてジャグリングしている単一のエージェントは、結果として肥大化したプロンプトを生み、推論が弱くなり、分析的なタスクと生成的なタスクの間で論理的な衝突を起こします。そのため、意思決定の説明可能性も低下してしまいます。
Shubham： 二つ目は運用的な課題です。逐次的な推論はレイテンシの増加を招き、フォールトの分離も安全なリトライもできません。もしエージェントの一つが失敗すると、パイプライン全体がクラッシュしてしまいます。すべてを同時に一括で実行しなければならず、当然デバッグも難しくなります。三つ目はスケーラビリティの課題です。スケールさせることも、並列化することもできず、個々の推論を別々に独立して進化させることもできません。つまり、これらのステップのどれか一つが失敗すれば、コンテキスト全体をやり直さなければならず、高いトークンコストも発生してしまいます。そして四つ目はガバナンスの課題です。広範なアクセス権を持つ単一のエージェントは、監査も更新もセキュリティの確保も明らかに困難です。最小権限の原則にも反していますし、モジュール単位のテストもサポートできません。
6.3 専門エージェントへの分割
Shubham： では、どうすればよいのでしょうか。私たちにできるのは、その一つの大きなエージェントを、3つの異なるエージェントに分割することです。それぞれが特定のタスクを担う専門のエージェントになります。検知（detect）、評価（assess）、実行（act）という具合です。そして、これらのエージェントはそれぞれ専用のデータベースへのアクセス権を持ち、すべてのプロンプトを見渡すのではなく、自分がアクションを取るために必要なものだけにアクセスできるようになっています。そして最後に応答を返します。当然ながら、それぞれがより小さなコンテキストを扱うことになるため、処理は速くなります。また、独立してスケールさせたり進化させたりすることも容易になります。
6.4 マイクロサービス化と実装構成
Shubham： ここで重要なのは、エージェント間の連携をイベントキューやメッセージバスの技術を介して行えるという点です。これによって、最終的にはエージェントのマイクロサービスとも言えるものを構築することになります。これを実際の実装に落とし込むと、最初の異常検知アーキテクチャは次のような形になります。生のイベントを取り込み、それらの生イベントがKinesis Data StreamsまたはMSKのトピックに流れ込みます。そしてFlinkがそれを拾い上げて異常検知を行い、SQSキューに書き込みます。そこから3つのエージェントのセットが拾い上げますが、これは実際には逐次的にチェーンされています。ただし、ワークロードによっては並列にすることも可能です。そして、それぞれが仕事を終えた後でアクションを取ることができます。アクションは先ほどと同じで、外部APIの呼び出し、人間への通知、あるいはアクセスのブロックです。しかしここでは、一つの大きなエージェントに頼るのではなく、それを複数のエージェントに分割できているのです。
7. 設計パターン③：エージェント型イベント駆動アーキテクチャと考慮事項
7.1 既存パイプラインの拡張という考え方
Shubham： 次は、エージェント型のイベント駆動アーキテクチャです。私たちは皆、このアーキテクチャに馴染みがあると思います。マイクロサービスを構築し、それらをKinesis Data Streams、MSK、SQSキューといったストリーミング技術やメッセージキューを使って疎結合にする、というものです。これらのマイクロサービスは、ECS、EKS、あるいはLambda関数といったあらゆるコンピュートにデプロイされているかもしれません。そしてそれらは同じデータストリームから消費し、あらかじめ設定しておいたルールに従って処理を行います。これは決定論的なルールで動くマイクロサービスで、処理を終えると別のキューや別のトピックに書き戻し、そこから次のマイクロサービスが拾い上げて、プロセスが続いていきます。
Shubham： ここで私たちが目にしているのは、お客様がこのアーキテクチャをエージェント型のイベント駆動アーキテクチャへと進化させられるようになっている、ということです。それが意味するのは、既存のパイプライン全体を置き換える必要はない、ということです。そのパイプラインを、エージェント型のサービスで拡張（augment）すればよいのです。そのエージェント型サービスは、同じキューから消費し、LLMエージェントを使ってデータを処理し、その結果をエコシステムに戻します。
7.2 ハイブリッドパターンによる段階的進化
Shubham： そして、ワークロードに応じてどのマイクロサービスを呼び出すべきかに基づいて、従来型のルールベースのマイクロサービスを呼ぶのか、それともLLMを使ったマイクロサービスを呼ぶのかを決定します。このハイブリッドなパターンによって、決定論的なシステムから自律的なシステムへと、段階的に進化させていくことができるのです。
7.3 レイテンシとインテリジェンスのトレードオフ
Shubham： さて、ここからは考慮事項の部分に入ります。エージェントをどうデプロイすべきかを考えるときに、皆さんができることについてです。一つは、レイテンシとインテリジェンスのトレードオフという問題が出てきます。エージェントが賢くなればなるほど、動作は遅くなります。大きなプロンプトなどを使って深い推論能力を持たせると、その分だけ遅くもなるのです。ですから、レイテンシとインテリジェンスのトレードオフのバランスをうまく取ることに注意を払う必要があります。
7.4 LLM呼び出しの最適化
Shubham： 二つ目に、ストリームを流れてくるすべてのイベントに対してLLMを呼び出すのではなく、という点です。ストリームには何百万ものイベントが流れているかもしれません。もし流れてくるすべてのイベントに対してLLMを呼び出していたら、膨大なレイテンシとコストが上乗せされてしまい、結果的にストリーミングパイプラインを持つ意味そのものが失われてしまいます。そこでの推奨は、何が異常なのかを見極める何らかの方法を用意することです。そのためには、MLモデルやルールベースのアラートといった既存のシステムを活用し、その上でAIエージェントを使ってさらに先の意思決定を行わせればよいのです。データをある程度バッチにまとめてAIエージェントを呼び出すこともできます。そうすればバッチ単位で呼び出せるようになり、呼び出し回数も削減できます。
Shubham： また、クロスリージョンの推論を検討することもできます。グローバルなアーキテクチャを構築している場合、トラフィックがスパイクする時間帯には、あるリージョンでのモデル推論が遅くなるかもしれません。その場合、自動的に別のリージョンにルーティングして、より速く応答を得ることができます。これによって、トラフィックがスパイクしている間でも、モデルから速く応答を受け取れるようにできるのです。
7.5 キャッシュ・疎結合・オブザーバビリティ
Shubham： さらに、頻出する応答はできる限り構造化された形でキャッシュするようにしてください。モデルからJSON形式などの特定のフォーマットで構造化された応答を得て、それをキャッシュしておけば、繰り返しのタスクのたびにモデルを呼び出す代わりに、そのキャッシュを使えます。これらはシンプルなものですが、皆さんがマイクロサービスや従来型のアプリケーションで活用してきた原則そのものであり、エージェント型アプリケーションでも同じように活用すべきものです。エージェントをキューで疎結合にしてください。もう一つ重要なのがオブザーバビリティです。呼び出しをトレースし、どこでホップが起きているのかを理解するために、オブザーバビリティを活用してください。
7.6 ビジネス成功指標と実行時制御
Shubham： そして最後の一つが、最も重要です。それは、ビジネスの成功指標から始める、ということです。コードを一行でも書き始める前に、何をもって成功とするのかを定義してください。そして、AgentCoreのpolicyとevaluationを使って、エージェントを実行時に制御し、そのエージェントが本当に正しい成功指標を提供しているかどうかを見極めてください。問題を解決するためにエージェントを使うことで、実際に価値を得られているのか、ということです。それでは、次のデモの部分に移りたいと思います。
8. ライブデモ：AIエージェントによるジオフェンスアラート処理
8.1 ユースケース背景と違反リスク
Shubham： これは、AIエージェントを使ったジオフェンスのアラート処理に関するデモです。何が起きるかと言いますと、フリート管理の会社、特に自動車セクターの企業を考えてみてください。たとえばあるフリート管理会社は、何百台、何千台もの車両を運用していて、どの時点においても自社の車両がどこにあるのかを把握する必要があります。そこで彼らが行うのが、車両をジオタグ付けすることです。たとえばKilkennyで、あるフリート管理会社がトラックを運用しているとしましょう。彼らはそのトラックがKilkennyを決して離れないようにしたいと考えます。Kilkennyのエリア内だけで稼働し、Dublinのエリアには行かないようにしたい、というわけです。
Shubham： ではなぜそれが必要なのでしょうか。もし車両が立ち入りを制限された区域に無許可でアクセスするようなことがあれば、セキュリティ上のリスクになりえます。あるいは、荷物を時間どおりに配送できないといった運用上の非効率を招くこともあります。さらにコンプライアンスの問題にもなりえます。規制された区域に入ってしまった場合、フリート管理会社はその責任を負うことになります。ですから、これらは総じて、運用コストの増加、顧客満足度の低下、そして潜在的な収益の損失という形で、会社にとってのコストにつながるのです。
8.2 課題とAIによるアラート選別
Shubham： では、彼らは実際にどうするのでしょうか。彼らは基本的に境界線、つまりペリメーターを作ります。緯度と経度を使ってKilkennyの地図を定義します。これが、車両が走行すべきエリアです。そして、IoTデバイスを使って車両からリアルタイムの追跡データを取得し、車両がそのペリメーターを越えたかどうかを常時チェックします。もし越えていれば、それは違反であり、対応に取りかかります。
Shubham： ところが、ここで起きる課題があります。もし何百台、何千台もの車両を運用していれば、大量のアラートが生成されることになります。そして、その多くが偽陽性（フォールスポジティブ）でありえます。それらのアラートの多くは、例外として発生したものなのです。たとえば、車両が何らかの不具合を抱えていて、ガレージに行く必要があり、そのガレージがたまたまジオフェンシングのペリメーターの外側に当たるエリアの近くにある、といった具合に、Kilkennyの境界を越えざるをえない場合があるわけです。ですから、何が実行可能な洞察（アクショナブルなインサイト）なのか、彼らにとって何が重要なのかを理解するには、それらのイベントをすべてまとめて分析することが必要になります。
Shubham： ここでAIを使えるのです。それらのアラートをAIで分析して、なぜ違反が起きたのかを理解し、価値の高いアラートだけに対してアクションを取ればよいのです。本質的には、アナリストが流れてくるすべてのアラートの分析に時間を費やすのではなく、AIにその土台となる地ならしの作業をやらせて、アナリストはそうした違反が起きたときにどう対応するかという、価値の高い意思決定に時間を使えるようになります。
8.3 アーキテクチャ全体
Shubham： アーキテクチャは次のようになっています。車両から、IoT Coreのデバイス経由で車両のテレメトリデータを取得し、それがKinesis Data Streamに入ります。これを生イベント、つまり車両テレメトリデータと呼びます。次にFlinkがイベントを処理しますが、その前に、DynamoDBのデータベースからジオフェンスのルールを取得します。そしてそのルールを使って、車両がルールに従って範囲内にいるかどうかを判定します。もし従っていなければ、例外、つまり違反を発生させ、それがジオフェンスアラートのストリームの中で「ジオフェンスアラート」と呼ばれます。
Shubham： そして、AgentCore runtimeにデプロイされたエージェントを呼び出し、このエージェントがなぜこれが起きたのかを分析します。そのためにエージェントはLLMモデルを呼び出すことができます。また、AgentCore memoryへのアクセス権も持っていて、短期記憶と長期記憶を使い、過去のパターンを理解し保存します。たとえば、その車両が直近1週間で何回違反したか、といったことです。これは極めて重要なデータポイントになります。そして、このエージェントの応答を、SQSキューを介して別のエージェント、つまりアラートパブリッシャーのエージェントに渡します。そのエージェントは流れてくるデータを見て、それをメールの形式に整形し、SNSトピックを通じてアナリストに送り返すのです。
8.4 デモ実演とデータ構造
Shubham： では、コンソールに戻ってデモがどう見えるかをお見せします。すでにシミュレータを動かしてあります。シミュレータが何をしているかというと、データを生成しています。実際のシステムであれば、IoTデバイスから直接データを取得することになりますが、ここではシミュレータを使ってこの種のデータを模擬しています。データストリームは二つあります。車両テレメトリのデータストリームには、データビューアのタブがあって、データが流れてきているかどうかを確認できます。Kinesis Data Streamの中のシャードを一つ選んで、レコードが流れてきているかを見ますと、車両テレメトリのデータストリームにいくつかレコードが流れてきているのが分かります。
Shubham： また、Flinkアプリケーションも動いていて、このデータをすべて常時処理しています。もし正しく処理されていれば、ジオフェンスアラートのデータストリームに何らかのアラートが生成されるはずです。シャードを覗いてみますと、データが流れてきているのが確認できます。これは直近のデータで、15時49分のものです。データの構造をお見せしますと、車両テレメトリのイベントは、vehicle ID（車両ID）があり、そのイベントが紐づけられているgeofence ID（ジオフェンスID）があり、車両デバイスから送られてくる緯度と経度が定義されています。このデータがFlinkで処理されると、vehicle ID、緯度と経度に加えて、違反の情報が含まれるようになります。なぜ違反が起きたのか、その名称や説明などの情報です。Flinkは、DynamoDBのテーブルからジオフェンスのルールを取得しますが、そのテーブルにはgeofence IDがあり、そのペリメーターをどう定義するかは緯度と経度のポリゴン（多角形）で表現されています。流れてくるすべてのイベントがそのポリゴンの内側にあるかどうかをチェックし、外側にあれば違反、内側にあれば正常、と判定しているのです。
8.5 Kiroによるスペック駆動開発
Shubham： 最後にエージェントを通っていきますので、エージェントについて少しご説明します。私はこれをKiroを使って開発しました。Kiroをご存じない方のために申し上げますと、Kiroはスペック駆動開発（spec-driven development）ができるエージェント型のIDEです。ここにプロンプトがあり、エージェントが何をすべきかというタスクを基本的に定義しています。また、すべての違反をどう扱うかというプレイブックも用意してあります。たとえば、事前承認された移動や認可済みの立ち入りといった、いくつかの明確な例外があるかもしれません。それを見つけた場合は、たとえ違反であっても、その例外のおかげでアラートなしと判断されます。こうして、ここで偽陽性を削減しているのです。
8.6 JSON出力とアラート振り分け
Shubham： 私はモデルに対して、応答をJSON形式で返すよう強制しています。JSON形式は構造化されたフォーマットなので扱いやすく、このフォーマットを使えばデータを構造化された形で保存して操作できるからです。これらのフィールドのいくつかを見てみますと、decision（決定）というフィールドがあって、これはalert（アラート）かno alert（アラートなし）のいずれかになります。そしてseverity（重大度）のフィールドがあり、これはlow（低）、medium（中）、high（高）、critical（重大）のいずれかになります。highとcriticalのものだけを、SQSキューを介して次のエージェントに送っています。
8.7 メール整形と効果
Shubham： SQSキューの先には、SNSパブリッシャーのエージェントがもう一つあって、そのプロンプトには、これをどうメールとして整形するかが基本的に書かれています。現在の車両のステータス、なぜアラートが発生したのか、そのアラートの影響は何か、そして推奨されるアクションは何か、といったことについて記述します。では、今まさに生成されているアラートのいくつかを見てみましょう。このアラートを見てみますと、これは今まさに流れてきたアラートで、15時51分のものです。IDを持つ車両について述べていて、その位置を示し、車両に関する情報を示しています。どのようにアラートが発生したのか、なぜアラートが発生したのか、そのアラートの影響は何か、そしてアナリストが確認できる推奨アクションを示しています。これらはすべてプレイブックの中で定義してエージェントに与えておくことができるので、エージェントはそこから正しい推奨を提供できるのです。
Shubham： さて、もしこの作業をすべてアナリストがやらなければならないとしたら、すべての異常を分析するのは確かに大きな負担になっていたでしょう。たとえば何百もの異常が発生しているとして、その数を絞り込んで、1時間あたりそのうちの5%だけにフィルタリングできれば、アナリストは分析作業をする代わりに、なぜ違反が起きたのかを突き止めることに時間を使えるようになるのです。