AWS Strands Agents で構築する AI エージェント LINE Bot

クラウド×AI 勉強会 〜プロバイダー別最新動向・LINE Bot・アーキテクチャ実装事例集〜 AWS Strands Agents で構築 で構築 する AI エージェント する エージェント LINE Bot

自己紹介 井 上 奏 人 @ so hjin_uv 🚀 専門領域 • SNSマーケティングツール開発 • フロントエンド: Vue.js • バックエンド: Rust ユニークビジョン株式会社 エンジニア 👏 特技 & 今日の話 趣味: 料理、野球観戦など 特技: 片手拍手 👏

📱 作ったもの 🎯 🤖 AI コンシェルジュ 明日渋谷でランチしたいです イタリアンで予算2000円以内 現在地から30分以内で行けると ころ 12:34 🤖 📋 おすすめプランをご提案しま す！ 🍝 ⭐ 4.3 (Google評価) 💰 ランチ: 1,200円〜1,800円 📍 渋谷区道玄坂2-23-12 🚃 パスタ・デ・ココ 渋谷店 アクセス 現在地 → JR山手線 → 渋谷駅 所要時間: 約25分 運賃: 200円 マ ル チ エ ー ジ ェ ン ト LI NE Bo t • 実在店舗のみ提案（Places API） • リアルタイム交通情報（Maps API） • 営業時間・混雑状況も考慮

システム構成 AI エージェント層 ユーザー接点 📱 LINE Bot 🧠 実行環境 外部API連携 📍 Google Places External API Master Planner → ⚡ Lambda/ECS/EKS Messaging API Python + Flask Frontend Infrastructure Nova Premier → Main Agent 🍝 🚃 → 🗺️ Maps API External API 📅 Gourmet Agent Transport Agent Calendar Specialist Specialist MCP

🔧 Strands Agents の基本機能 の基本機能 SDK の特徴・モデル駆動・AWS 統合

⚙️🎯 事前準備 Strands Agents を実際に試すための準備 必要な環境とツール: 実際にハンズオンで試したい方向け 📋 シ ス テ ム 要件 ☁️ A W S 設定 Python環境 認証設定 Python 3.10+ 必須 uv パッケージマネージャー推奨 インストール # uvでの推奨インストール方法 uv add strands-agents uv add strands-agents-tools # 従来のpipでも可能 pip install strands-agents strands-agents-tools # AWS CLI設定（必須） aws configure ⚠️ 重要: Bedrockモデルアクセス有効化 AWS Console → Bedrock → Model access Claude 3.7 Sonnet の有効化が必要 デフォルトリージョン: us-west-2 IAM 権限 bedrock:InvokeModel bedrock:InvokeModelWithResponseStream

Strands Agentsとは

🚀
GitHub
Python AI

でオープンソース公開の

エージェントSDK

@tool デコレータでツール定義 | モデルが自動でツール選択・実行
20+ 標準ツール内蔵 | MCP統合で外部サービス連携

📦 最小構成（3行で 作成）
from strands import Agent
# エージェント作成（デフォルト設定）
agent = Agent()

# 実行
agent("Strands Agentsについて教えてください。")

🔧 実用的な エ ージ ェ ン ト
from strands import Agent, tool
from strands.models import BedrockModel
@tool
def get_weather(city: str):
"""指定した都市の天気を取得"""
# 天気API呼び出し
return f"{city}の天気は晴れです"
# カスタム設定のエージェント
agent = Agent(
model=BedrockModel(
model_id="us.amazon.nova-lite-v1:0",
region_name="us-east-1"
),
tools=[get_weather],
system_prompt="あなたは親切な天気案内アシスタントです。"

デフォルト設定:

)

モデル: Claude 3.7 Sonnet (Bedrock)
system_prompt: 基本的なアシスタント
AWS CLI認証が必要

agent("東京の天気を教えて")
主要パラメータ:

model : 使用するLLMモデル
tools : 利用可能なツール一覧
system_prompt : エージェントの役割・性格

🤖🎯 モデル駆動アプローチ モデルの推論能力を活用した開発手法 ツール提供: 機能を @tool で定義、モデルが使い方を自動学習 自動判断: モデルがコンテキストを理解してツール選択・組み合わせ 🔧 シ ン プ ル な 実装 @tool def search_restaurants(area: str, genre: str = None): """指定エリアでレストラン検索""" return places_api.search(area, genre) @tool def get_route(origin: str, dest: str): """出発地から目的地への経路検索""" return maps_api.directions(origin, dest) # モデルが自動でツール選択・実行 agent = Agent(tools=[search_restaurants, get_route]) agent("渋谷でイタリアンランチして、新宿駅への行き方も教えて") ✨ モ デ ル 駆動の 特徴 自動ツール選択 ユーザーの意図を分析 適切なツールを自動選択 複数ツールの組み合わせ実行 柔軟な対応 新しいツール追加時の再学習不要 自然言語での複雑な指示に対応 状況に応じた最適な処理フロー 開発効率 if-elif分岐の大幅削減 ツール定義に集中できる モデルが処理ロジックを担当

標準ツール群 🚀 標準で 20+ 種類のツールを提供 🌐 Web & A P I http_request : HTTP リクエスト 📁 フ ァ イ ル & シ ス テ ム 🎨 マ ル チ モ ーダ ル file_read/write: ファイル操作 + 許可要求 image_reader: 画像読込 retrieve: Bedrock RAG 検索 shell: シェル実行 generate_image: 画像生成 slack: Slack 統合 python_repl: Python 実行環境 speak: 音声合成 use_aws: AWS サービス連携 current_time: 現在時刻取得 🔧 使用例 from strands import Agent from strands_tools import current_time, http_request, shell # 標準ツールだけで構成されたエージェント agent = Agent(tools=[current_time, http_request, shell]) # モデルが自動でツールを選択・実行 agent("今の時刻を教えて、天気を調べて、それをファイルに保存して") # 実際の実行順序（モデルが自動判断）: # 1. current_time() → "2025年1月30日 14:30" # 2. http_request(url="天気API") → "東京は晴れ、気温12℃" # 3. shell(command="echo '...' > weather.txt") → ファイル保存完了

💬🎯 会話管理とストリーミング 実用的なチャットボット開発に必須の機能 長期間の会話対応: コンテキスト自動管理でユーザー体験向上 リアルタイム応答: ストリーミング対応で自然な対話 📚 会話の 履歴管理機能 自動履歴管理 ⚡ ス ト リ ーミ ン グ 機能 リアルタイム応答 会話が長くなってもコンテキストを保持 回答が生成されながら表示 window_size で履歴サイズ調整 ChatGPTのような体験を実現 トークン使用量を効率的に制御 ユーザーの待ち時間を削減 from strands import Agent from strands.conversation import SlidingWindowConversationM # 会話管理機能付きエージェント agent = Agent( conversation_manager=SlidingWindowConversationManager( window_size=10 # 直近10回の会話を保持 ) ) # 長い会話でも前後の文脈を保持 agent("前回の話の続きで...") 対話型アプリケーション LINE Bot、Slack Bot で活用 Streamlit との組み合わせ WebSocket 対応 # ストリーミング対応 for chunk in agent.stream("長い回答が必要な質問"): print(chunk, end="", flush=True)

@toolで自作ツールを作成
シンプルなツール定義
@tool
def search_restaurants(area: str, genre: str = None)
"""Google Places API でレストラン検索"""
# Google Places API 呼び出し
places_result = places_client.text_search(
query=f"{area} {genre} レストラン",
language="ja"

モデルが理解すること
自動的に推測される情報

関数名から機能を理解
引数の型からパラメータ仕様
docstringで詳細な使い方
内蔵ツールとの組み合わせ

)
# 結果をフィルタリング・整形
restaurants = []
for place in places_result.get('results', []):
restaurants.append({
'name': place['name'],
'rating': place.get('rating'),
'address': place.get('formatted_address')
})
return restaurants

# モデルが自動で組み合わせ実行
agent("渋谷でイタリアン、予算2000円")
# 実際の実行順序:
# 1. http_request: エリア情報収集
# 2. search_restaurants: 検索実行
# 3. retrieve: 過去データ参照

🔧🔧 AWS 連携ツール (1/2) A W S リ ソ ース へ の 自然な ア ク セ ス 標準ツールで AWS 操作: use_aws 、 retrieve で直接 AWS サービス利用 認証自動継承: 既存の IAM ロール・プロファイルをそのまま活用 🔍 use_aw s ツ ール from strands import Agent from strands_tools import use_aws agent = Agent(tools=[use_aws]) # S3バケット一覧取得 agent("S3のバケット一覧を教えて") # → 内部でuse_awsが自動実行 # 明示的な呼び出しも可能 result = agent.tool.use_aws( service_name="s3", operation_name="list_buckets", region="us-west-2" ) 特徴: boto3 の全サービス・操作に対応、破壊的操作は確認プロンプト

🔧📚 AWS 連携ツール (2/2) retri eve ツ ール from strands_tools import retrieve agent = Agent(tools=[retrieve]) # Bedrock Knowledge Base 検索 agent("社内ドキュメントから過去の事例を検索して") # → 自動でKnowledge Base検索 # 明示的な検索 result = agent.tool.retrieve( query="AWS Bedrock 料金体系", knowledge_base_id="KB123456" ) 特徴: Bedrock Knowledge Base との自然な連携 🎯 開発体験の 向上 認証の透明性: IAM ロール、プロファイル、VPC 設定をそのまま継承 コード簡素化: AWS 操作も自然言語で指示可能 本番実績: Amazon Q Developer、AWS Glue 等で実用化

🔗🔗 MCP 統合：外部ツール連携 Model Context Protocol (MCP) ツールの標準化: 外部サービスとの統合を標準化するプロトコル 7,000+ アプリ: Zapier連携で膨大なサービス群へアクセス 🌐 M CP サ ーバ ー対応 ローカルMCPサーバー：開発環境 リモートMCPサーバー：本番環境 Streamable HTTP：リアルタイム通信 🔌 対応サ ービ ス 例 生産性ツール Gmail、Slack、Teams Google Drive、OneDrive from mcp import stdio_client, StdioServerParameters from strands.tools.mcp import MCPClient # MCPサーバー接続 mcp_client = MCPClient( lambda: stdio_client(StdioServerParameters( command="uvx", args=["awslabs.aws-documentation-mcp-server@latest" )) ) Notion、Asana 開発・分析 GitHub、GitLab Jupyter、Tableau AWS Services 実装が簡単: Strands Agentsがイベントループを自動処理 # ツール取得とエージェント作成 with mcp_client: tools = mcp_client.list_tools_sync() agent = Agent(tools=tools)

対応モデル一覧 🤖 Strands Agents で使用可能なモデル モデル非依存: 複数プロバイダーに対応、同じコードで切り替え可能 🌟 AWS Bedrock 豊富な選択肢: AWS Bedrock経由 + 直接API両方に対応 対応モデル 🚀 A maz o n No va（A W S 製） Nova Premier: マルチモーダル、高度推論 🦙 M eta Ll ama Llama 3.3 70B: 最新、高効率（3.1 405B相当性能） Nova Lite: 軽量タスク、コスト効率 Llama 3.2 90B Vision: マルチモーダル、企業向け Nova Micro: 超軽量、超高速 Llama 3.2 11B Vision: 軽量ビジョンモデル 🧠 A nthro pi c Cl aude Claude 4 Sonnet : 最新、高性能、ハイブリッド推論 Claude 4 Opus: 最高品質、長時間タスク対応 Llama 3.2 3B/1B: 軽量、エッジ・オンデバイス対応 🏢 そ の 他の 主要モ デ ル Amazon Titan: AWS製、汎用・マルチモーダル Claude 3.7 Sonnet : 標準、ハイブリッド推論対応 Cohere Command : 企業向け、RAG最適化 Claude 3.5 Haiku: 高速、低コスト AI21 Jamba: 効率的推論、長文対応

モデル利用方法

🔧 AWS Bedrock
基本的な 使い 方

from strands.models import BedrockModel
# Amazon Nova Premier
model = BedrockModel(
model_id="us.amazon.nova-premier-v1:0",
region_name="us-east-1"
)
# Claude 3.7 Sonnet via Bedrock
model = BedrockModel(
model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0",
region_name="us-east-1"
)
特徴: IAM認証、Guardrails統合、企業向け運用機能

🔗

直接API（APIキー必要）

OpenA I

A nthro pi c

from strands.models.openai import OpenAIModel
import os

from strands.models.anthropic import AnthropicModel
import os

model = OpenAIModel(
client_args={"api_key": os.getenv("OPENAI_API_KEY")

model = AnthropicModel(
client_args={"api_key": os.getenv("ANTHROPIC_API_KE

model_id="gpt-4o",
params={"temperature": 0.7}
)
インストール:

model_id="claude-3-7-sonnet-20250219",
params={"temperature": 0.7}
)

pip install 'strands-agents[openai]'

インストール:

pip install 'strands-agents[anthropic]'

🤝 マルチエージェント機能 Workflow・Graph・Swarm・A2A

マルチエージェント機能の使い分け 🔄 Wo rk f l o w 🕸️ G raph 🐝 S w arm 手順を決めて順番に実行 専門チームで役割分担して解決 複数のAIで相談して決める 使用場面: 使用場面: 使用場面: 「毎回同じ作業の繰り返し」 「複雑な問題を専門家チームで分析した 「一人だと思いつかない…」 「ステップ1→2→3の順番が大事」 い」 「どの案がベスト？」 「途中で失敗したらやり直し」 「階層的な役割分担で品質向上したい」 「色んな角度から検討したい」 具体例: ファイル取得→データ加工→結果保存 在庫確認→注文処理→決済→配送手配 レポート生成→承認→メール送信 「情報の流れを制御して一貫性確保した い」 具体例: 具体例: 旅行プラン：予算・交通・グルメ担当で相 談 市場調査：Executive → Economic Dept → 商品企画：マーケ・技術・デザイン視点で 専門分析チーム 議論 システム設計：Coordinator → Frontend + レストラン選び：3人のAIが競って最適解提 Backend + DB Team 案 コンテンツ制作：Manager → Research + Writing + Review 🎯 ど れ を 使え ば い い ？ 普通の作業: 通常のツール → 決まった手順: Workflow → 専門チーム分担: Graph → みんなで相談: Swarm

Workflow: 多段階処理の自動化 📋 ツールの連続実行を自動管理 workflow ツール: 複数のツールを順次実行 依存関係処理: 前のステップ結果を次に自動引継ぎ 🔧 実装例 from strands import Agent from strands_tools import http_request # 1. 専門エージェントを個別に作成 researcher = Agent( system_prompt="Webから情報を収集する研究員", tools=[http_request] ) analyst = Agent( system_prompt="情報を検証・分析する分析官" ) writer = Agent( system_prompt="明確なレポートを作成する執筆者" ) # 2. ワークフロー関数で連携 def research_workflow(query): # 研究員が情報収集 research_data = researcher(f"「{query}」について調査") # 分析官が検証 analysis = analyst(f"この情報を分析: {research_data}") # 執筆者がレポート作成 report = writer(f"分析結果をレポート化: {analysis}") return report 🎯 活用シ ーン データ処理パイプライン file_read → データ取得 python_repl → データ加工 http_request → API送信 file_write → 結果保存 メリット ツール実行の自動化 エラー発生時の再試行 処理状況の追跡可能 特徴: エージェントがタスクを分析してworkflowを自動構築

Agent Graph: エージェント関係の可視化

🕸️

複数エージェントの関係性を定義・管理

agent_graph ツール: エージェント間の接続を定義
可視化機能: 複雑なマルチエージェントシステムの構造把握

🔧 実装例（概念）

from strands import Agent
from strands_tools import agent_graph
# 1. 専門エージェントを作成
coordinator = Agent(system_prompt="中央管理者")
gourmet = Agent(system_prompt="グルメ専門家")
transport = Agent(system_prompt="交通案内係")
weather = Agent(system_prompt="天気予報士")
# 2. agent_graphツールで関係定義（概念）
# Star topology: coordinatorが中心
# - coordinator → gourmet（レストラン検索依頼）
# - coordinator → transport（経路検索依頼）
# - coordinator → weather（天気確認依頼）
# 3. 実行例
def plan_outing(request):
# コーディネーターが各専門家に問い合わせ
food = coordinator("グルメエージェントに問い合わせ: " + requ
route = coordinator("交通エージェントに問い合わせ: " + food
weather_info = coordinator("天気エージェントに問い合わせ: "
return f"プラン: {food}, {route}, {weather_info}"

🎯 実際に こ ん な 時に 使う
複雑な問題を分担して解決したい時

「市場分析を経済・技術・社会の専門家チームで分析したい」
「大きなタスクを専門エージェントに分けて効率化したい」
「エージェント間の役割分担と情報の流れを制御したい」
具体的な使い方
市場調査: Executive → Economic Dept → Market Research +

Financial Analysis
システム設計: Coordinator → Frontend Team + Backend Team + DB

Team
コンテンツ制作: Manager → Research + Writing + Review の順次処

理
実用的なメリット

複雑なタスクを専門チームで効率的に処理
階層化された役割分担で品質向上
構造化された情報フローで一貫性確保

Swarm: 協調的な問題解決 🐝 複数エージェントの集合知を活用 swarm ツール: 複数エージェントで並列問題解決 協調パターン: collaborative、competitive、hybridモード 🔧 実装例（概念） from strands import Agent from strands_tools import swarm # 1. Swarmエージェント群を作成 research_agent = Agent(system_prompt="研究者：データ収集") creative_agent = Agent(system_prompt="創造者：アイデア生成") critical_agent = Agent(system_prompt="批評家：課題指摘") summary_agent = Agent(system_prompt="要約者：統合・整理") # 2. 協調型Swarm実行（概念） def collaborative_swarm(task): # 各エージェントが並行して考察 ideas = [ research_agent(task), creative_agent(task), critical_agent(task) ] # 要約エージェントが統合 🎯 3つ の パ タ ーン を 使い 分け Collaborative（みんなで相談） 例：旅行プラン「予算・交通・グルメ担当で話し合い」 お互いの意見を聞きながら案を練る 一人では思いつかないアイデアが生まれる Competitive（コンペ形式） 例：レストラン選び「3人のAIが別々に提案→ベスト選出」 それぞれ独立して案を出す 色んな選択肢から最適解を選べる Hybrid （状況に応じて） 例：商品企画「最初は競争→良いアイデア出たら協調」 臨機応変にパターンを切り替え return summary_agent(f"次の意見を統合: {ideas}") # 3. 競争型Swarm（各自が独立して最適解を提案） def competitive_swarm(task): solutions = [agent(task) for agent in [research_agent, # 最良の解を選択するロジック return select_best(solutions) 要するに: 一人で考えるより、複数で考えた方が良いアイデアが出る

A2A実装（実験的） 🧠 🍝 マ ス タ ープ ラ ン ナ ー（A 2A ク ラ イ ア ン ト） # planner.py (実験的機能の概念例) from strands import Agent, tool from strands.models import BedrockModel # A2Aは実験的機能のため、実際のimportパスは変更の可能性あり from a2a.client import A2AClient class PlannerAgent: def __init__(self): # A2A通信用のツール定義 @tool async def delegate_to_gourmet(query: str, area: str """グルメエージェントに委譲""" client = A2AClient("http://localhost:8001") グ ル メ エ ージ ェ ン ト （A 2A サ ーバ ー） # gourmet_server.py from strands import Agent from strands.multiagent.a2a import A2AAgent from strands.models import BedrockModel from src.tools.gourmet_tools import search_restaurants def main(): # グルメ専門エージェント作成 gourmet_agent = Agent( model=BedrockModel( model_id="us.amazon.nova-lite-v1:0", region_name="us-east-1" ), name="Gourmet Agent", description="レストラン検索専門エージェント", tools=[search_restaurants], system_prompt="""グルメ専門エージェントです。 ユーザーの好みに合うレストランを提案します。""" return await client.send_message(query, {"area" # マスタープランナー作成 self.agent = Agent( model=BedrockModel( model_id="us.amazon.nova-premier-v1:0", region_name="us-east-1" ), tools=[delegate_to_gourmet], system_prompt="""プランナーエージェントです。 専門エージェントと連携してプランを作成します。""" ) async def create_plan(self, user_request: str): return await self.agent.arun(user_request) # 使用例 planner = PlannerAgent() result = await planner.create_plan("渋谷でイタリアンランチしたい ) # A2Aサーバーとして起動 a2a_agent = A2AAgent(agent=gourmet_agent, port=8001) a2a_agent.serve() # 独立サービスとして実行 if __name__ == "__main__": main()

☁️ AWS 運用機能 運用機能 Guardrails・運用監視・メトリクス

AWS Bedrock Guardrails 🎯 コンテンツフィルタリング ⚙️ カ ス タ マ イ ズ 可能な 設定 自動検出・ブロック機能 フィルタリングレベル 有害・不適切コンテンツ 禁止用語の定義 個人情報漏洩防止 トピック別制限設定 企業ポリシー違反 感度レベル調整 基本フロー 実用的な運用 AWS Console でガードレール作成 段階的な警告・ブロック フィルタリングルール設定（有害コンテンツ、個人情報等） 監査ログでの追跡 Strands Agents で活用 違反パターンの分析 # AWS Console で作成したガードレールを指定 model = BedrockModel( "us.amazon.nova-premier-v1:0", guardrail_id="gr-1234567890abcdef", guardrail_version="1" ) # 自動的にガードレールが適用 response = agent("機密情報を含む質問") # → 適切にブロック・代替回答を提供 # Console で作成

運用監視：標準メトリクス機能

📊

デフォルトで監視機能内蔵

設定不要: エージェント実行で自動的にメトリクス収集
商用利用: 非機能要件・コスト管理に必須の情報を簡単取得

🔍 実際の 取得例

from strands import Agent

📈 OpenTel emetry統合
本格的な観測可能性

agent = Agent()

分散トレーシング対応

result = agent("今後、AIはどのように進化しますか？")

OTEL標準で任意のプラットフォーム選択

# 実行結果のメトリクス取得
print(f"使用トークン数: {result.metrics.accumulated_usage['to
print(f"実行時間: {sum(result.metrics.cycle_durations):.2f} 秒
print(f"サイクル数: {result.metrics.cycle_count} 回")
print(f"レイテンシ: {result.metrics.accumulated_metrics['late

実行結果

使用トークン数: 373
実行時間: 8.61 秒
サイクル数: 1 回

エンタープライズレベルの監視
# OpenTelemetry設定（環境変数）
import os
os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = "your-endpoint"
os.environ["OTEL_EXPORTER_OTLP_HEADERS"] = "Authorization=B
# strands-agents[otel]をインストール
# pip install strands-agents[otel]
agent = Agent(
trace_attributes={
"session.id": "session-123",
"user.id": "[email protected]"
}

レイテンシ: 8160.00 秒

)
# 自動的にOTELトレース・メトリクス送信

📊 まとめ 実用化ポイント・開発効率・メリット

実用化での重要ポイント 🎯 Strands Agents + AWS ✅ 適用推奨シ ーン 🗣️ 自然対話アプリケーション → モデルがコンテキストを理解しツール選択 🚀 プロトタイプ～本格開発 → @tool で段階的な機能追加が可能 ☁️ AWS エコシステム活用 → 既存インフラ・サービスとの統合 🔗 マルチ API 統合システム → MCP で標準化されたツール連携 が適用できるシーン ⚠️ 検討が 必要な シ ーン 🎯 決定論的処理 → LLM の確率的性質が不適合 🔒 厳密フロー制御 → モデルが予期外のツール選択の可能性

開発効率とメリット 🚀 開発速度の 劇的改善 💰 運用コ ス ト 削減 従来のエージェント開発 AWS での本番実績 # 複雑な分岐処理が必要 if "レストラン" in user_input: if "検索" in user_input: result = search_restaurants() elif "予約" in user_input: result = make_reservation() elif "交通" in user_input: # さらに分岐... Amazon Q Developer: Strands で構築済み AWS Glue: 本番環境で稼働中 VPC Reachability Analyzer: 実用化済み 開発期間の短縮効果 従来: プロトタイプ→本番まで数ヶ月 Strands: 数日から数週間で新エージェント提供 Strands Agents @tool def search_restaurants(area: str): ... @tool def make_reservation(name: str): ... @tool def get_directions(origin: str): ... # モデルが自動判断・実行 agent = Agent(tools=[search_restaurants, make_reservation, 開発効率: 複雑な分岐処理が不要、@toolで機能追加が簡単 効果: 市場投入時間の大幅短縮 コスト例（公式価格） Nova Premier: 入力$2.50/出力$12.50 per 1M tokens Nova Lite: 入力$0.06/出力$0.24 per 1M tokens Nova Micro: 入力$0.035/出力$0.14 per 1M tokens

ありがとうございました