初めに
こんにちは。
Cocone Engineeringで主に決済に関するサーバサイドを担当しています、Kと申します。
私の業務内容としては、自社が運営するゲームの決済周りのシステム開発・保守・運用を行っており、特にゲーム内課金や決済システムに関する業務を中心に担当しています。
今回は、自宅でAgents for Amazon Bedrockを用いて作成したAIエージェントの簡単なサンプルアプリをご紹介させていただきます。
AIエージェントとは
AIエージェントとは、特定のタスクや目的に基づいて自律的に意思決定を行い、行動を実行するソフトウェアプログラムのことです。これにより、人間の代わりに問題を解決したり、業務を自動化したり、ユーザーとの対話が可能になります。
AIエージェントは、カスタマーサポート、データ分析、ゲームAIなど、さまざまな分野で利用されています。特に、複雑なタスクを自動化し、効率的に処理する能力が評価されています。
作成するサンプルアプリの概要
このサンプルアプリでは、以下の目的に沿ったAIエージェントを作成します。
このAIエージェントアプリは、コインに関する情報を簡単に取得し、コインの売買をサポートする便利なツールです。ユーザーは対話を通じて、コインの残高確認や相場価格の問い合わせを行うことができ、コインの種類や枚数を指定して即座に購入することも可能です。
さらに、マーケット全体のコイン購入履歴を調べることで、市場の動向を把握し、より良い売買判断に役立てることができます。このアプリは、シンプルな操作でコインの管理や売買がスムーズに行えるよう設計されており、ユーザーの意思決定を効率的にサポートします。
Amazon Bedrockで作成を開始
まず、AWSコンソールでAmazon Bedrockのページを開きます。
最初に「モデルアクセス」セクションから、今回使用するBedrock基盤モデル（Foundation Model: FM）へのアクセスを有効にします。今回使用するのは、Anthropic社のClaude 3 Sonnetで、リージョンはバージニア北部です。
モデルの有効化が完了したら、「オーケストレーション」のエージェントセクションに移り、エージェントの作成を開始します。以下の設定でエージェントを作成しました。
- エージェント名: agent-coin-market-assistant
- エージェントの説明: コインの購入を手伝います。
- エージェント向けの指示: コインを購入するには、コインの名前と枚数を知っておく必要があります。お客様には丁寧に説明を行ってください。
エージェントの詳細を入力した後、次にアクショングループを作成していきます。アクショングループとは、AIエージェントが特定のタスクを達成するために、一連のアクション（操作や活動）をグループ化したものです。今回は、アクショングループタイプとして「関数の詳細で定義」を選択しました。
- アクショングループ名: action-group-coin-market-assistant
- 説明: ユーザーは、現在所持している各コインの枚数や相場価格を確認し、コインの種類と枚数を指定して購入することができます。
「新しいLambda関数をすばやく作成する」を選択した後、以下の3つのアクショングループを作成します。
- get-current-coin-balances: 現在所持しているそれぞれのコインの枚数を表示します。
- get-current-coin-prices: 現在のそれぞれのコインの相場価格（円）を表示します。
- purchase-coin: コイン名と枚数を渡すことで、特定のコインを任意の枚数購入できます。
このうち、「purchase-coin」関数には、引数を設定します。設定内容は以下の通りです。
- coin_name: 購入するコインの名称（タイプ: string, 必須: True）
- quantity: 購入するコインの枚数（タイプ: integer, 必須: True）
次に、作成したアクショングループに基づいて、新規に生成されたLambda関数を編集します。AWSコンソールで該当のLambda関数を開くと、サンプルのコードが表示されているので、それを以下のコードに書き換えます。
- AWSコンソールのLambdaセクションに移動し、作成されたLambda関数を選択します。
- Lambda関数の編集画面に移動すると、サンプルコードが表示されています。このサンプルコードを削除し、以下のコードに書き換えます。
import json
import boto3
s3 = boto3.client('s3')
bucket_name = 'agent-coin-market-assistant-bucket'
def convert_params_to_dict(params_list):
params_dict = {}
for param in params_list:
name = param.get("name")
value = param.get("value")
if name is not None:
params_dict[name] = value
return params_dict
def get_balances():
file_name = 'balances.json'
response = s3.get_object(Bucket=bucket_name, Key=file_name)
return json.loads(response['Body'].read().decode('utf-8'))
def get_prices():
file_name = 'prices.json'
response = s3.get_object(Bucket=bucket_name, Key=file_name)
return json.loads(response['Body'].read().decode('utf-8'))
# コインを購入後、支払った金額の合計を返却
def purchase_coin(coin_name, quantity):
# 現在残高
balances = get_balances()
# 指定されたコインの残高を探して更新
coin_found = False
for coin in balances:
if coin['coin_name'] == coin_name:
coin['quantity'] += quantity
coin_found = True
break
if not coin_found:
raise ValueError(f"Coin {coin_name} not found in balances.")
file_name = 'balances.json'
# 更新されたbalancesを保存
# balancesデータをJSON形式にシリアライズ
balances_data = json.dumps(balances)
# S3にbalances.jsonを上書き保存
s3.put_object(Bucket=bucket_name, Key=file_name, Body=balances_data)
# 購入による支払額の算出
coin_price = None
coin_prices = get_prices()
for coin in coin_prices:
if coin['coin_name'] == coin_name:
coin_price = coin['price']
break
return coin_price * quantity
def lambda_handler(event, context):
print(event)
actionGroup = event['actionGroup']
function = event['function']
param = convert_params_to_dict(event.get('parameters', []))
if actionGroup == 'action-group-coin-market-assistant':
if function == 'get-current-coin-balances':
balances = get_balances()
body = f"Your current balances:\n{json.dumps(balances)}."
elif function == 'get-current-coin-prices':
prices = get_prices()
body = f"The prices of the coins available for sale:\n{json.dumps(prices)}."
elif function == 'purchase-coin':
coin_name = param['coin_name']
quantity = int(param['quantity'])
total_price = purchase_coin(coin_name, quantity)
body = (f"{coin_name} coin(s) were purchased in the amount of {quantity}."
f" The total price paid is {total_price} yen.")
else:
body = f"Unknown function {function} for action group {actionGroup}."
else:
body = f"Unknown action group {actionGroup}."
# Format the output as expected by the agent
responseBody = {
"TEXT": {
"body": body
}
}
action_response = {
'actionGroup': actionGroup,
'function': function,
'functionResponse': {
'responseBody': responseBody
}
}
function_response = {'response': action_response, 'messageVersion': event['messageVersion']}
print(f"Response: {function_response}")
return function_response
Lambda関数のコードを編集し終えたら、変更を保存し「Deploy」ボタンを押してLambda関数を更新します。通常、このLambda関数を使用して外部サービスや自社サービスなど、既存のシステムと統合しますが、今回は簡易的に、S3に保存された残高情報やコインの価格情報のJSONファイルをDBの代わりとして使用します。これにより、残高情報の更新やコインの価格情報の取得を行います。
S3に作成したバケット名
agent-coin-market-assistant-bucket
バケットに保存したファイル
- balances.json:
[
{
"coin_name": "gold",
"quantity": 0
},
{
"coin_name": "silver",
"quantity": 0
},
{
"coin_name": "copper",
"quantity": 0
}
]
- prices.json
[
{
"coin_name": "gold",
"price": 10000
},
{
"coin_name": "silver",
"price": 5000
},
{
"coin_name": "copper",
"price": 1000
}
]
作成したLambda関数内で、S3のデータを読み書きするために、IAMコンソールで「action-group-coin-market-assistant」のロールに「AmazonS3FullAccess」の権限を付与します。これにより、Lambda関数がS3バケットにアクセスし、残高情報やコインの価格情報を扱えるようになります。
保存が完了したら、AWSコンソールのAmazon Bedrockのエージェント画面に戻り、テスト画面にある「準備」ボタンを押します。これで、作成したエージェントをテストする準備が整いました。
残高が正しく取得できるかどうか確認しましょう。
残高を正しく取得でき、これで動作確認が完了しました。
次に、「purchase_history_2024-10-19.json」というファイルを用意し、同様にS3バケットに保存します。これは、2024年10月19日の全ユーザーの購入履歴を含んでいます。
- purchase_history_2024-10-19.json
[
{
"coin_name": "silver",
"purchase_time": "2024-10-19T00:15:40",
"price": 5000
},
{
"coin_name": "copper",
"purchase_time": "2024-10-19T00:23:54",
"price": 1000
},
{
"coin_name": "copper",
"purchase_time": "2024-10-19T00:52:48",
"price": 1000
},
...
]
先ほどエージェントを作成した際に、Code Interpreterを有効にしておくことで、エージェントはデータの処理や情報の抽出が可能になります。これにより、エージェントを呼び出す際にドキュメントを含めることもでき、テスト画面でも複数のドキュメントを添付してテストすることができます。
今回は、AWS SDK for Python (Boto3) と Streamlit を使用してエージェントを呼び出します。そのため、まずエージェントエイリアスとバージョンを作成する必要があります。エージェントIDとエイリアスIDは、エージェントを呼び出す際に必要となるため、これらをメモしておきます。Boto3のバージョンは1.35.46を使用しています。
また、「purchase_history_2024-10-19.json」ファイルはエージェントが呼び出すため、AmazonBedrockExecutionRoleForAgentsのIAMロールに「AmazonS3FullAccess」の権限を付与しました。
以下に、Streamlitでエージェントを呼び出すためのソースコードを示します。
import streamlit as st
import boto3
import uuid
import os
# StreamlitのUIを作成
st.title("CoinMarketBot")
# 定数の定義
AGENT_ID = os.getenv('AGENT_ID')
AGENT_ALIAS_ID = os.getenv('AGENT_ALIAS_ID')
# A unique identifier for each user
MEMORY_ID = os.getenv('MEMORY_ID')
# Bedrock Agentのクライアント
client = boto3.client("bedrock-agent-runtime")
# セッションIDを生成
if "session_id" not in st.session_state:
st.session_state.session_id = str(uuid.uuid4())
# 会話履歴を初期化
if "messages" not in st.session_state:
st.session_state.messages = []
# 過去の会話履歴を表示
for message in st.session_state.messages:
with st.chat_message(message["role"]):
for content in message["content"]:
# テキストの処理
if content["type"] == "text":
st.write(content["content"])
# ファイルの処理
elif content["type"] == "file":
st.image(content["bytes"], caption=content["name"])
# チャット入力欄を定義
if prompt := st.chat_input("メッセージを送信する"):
# ユーザーの入力を画面に表示
with st.chat_message("user"):
st.markdown(prompt)
# ユーザーのメッセージをセッションに追加
st.session_state.messages.append({"role": "user", "content": [{"type": "text", "content": prompt}]})
# AIの応答を取得し表示
with st.chat_message("assistant"):
# ユーザーからの入力を取得し、エージェントに送信
response = client.invoke_agent(
agentId=AGENT_ID,
agentAliasId=AGENT_ALIAS_ID,
sessionId=st.session_state.session_id,
inputText=prompt,
memoryId=MEMORY_ID,
enableTrace=False,
endSession=False,
sessionState={
"files": [
{
"name": "purchase_history_2024-10-19.json",
"source": {
"sourceType": "S3",
"s3Location": {
"uri": "s3://agent-coin-market-assistant-bucket/purchase_history_2024-10-19.json"
},
},
"useCase": "CODE_INTERPRETER",
}
]
},
)
if stream := response.get("completion"):
result = []
for event in stream:
# ファイルがある場合
if "files" in event:
files = event["files"]["files"]
for file in files:
# 画像を表示
st.image(file["bytes"], caption=file["name"])
# ファイル情報をリストに保存
result.append(
{
"type": "file",
"name": file["name"],
"bytes": file["bytes"],
}
)
# チャンク（テキスト）がある場合
if "chunk" in event:
chunk = event["chunk"]
answer = chunk["bytes"].decode()
st.markdown(answer)
# テキストをリストに保存
result.append({"type": "text", "content": answer})
# メッセージの保存
st.session_state.messages.append(
{"role": "assistant", "content": result}
)
では、実際に動作確認を行い、成果物を確認してみましょう。
- 現在の残高を確認します
エージェントに現在のコイン残高を問い合わせます。
⇩- マーケットのコイン価格を確認します
現在のマーケットで取引されているコインの価格情報をエージェントから取得します。
⇩- コインを購入します
購入するコインの種類と枚数を指定して、エージェントを通じてコインを購入します。
最後に、コイン残高が先程の購入を反映して増えていることを確認し、今後の購入の参考にするため、他ユーザーの過去の購入履歴も調べてみましょう。
- 現在のコイン残高を確認します
エージェントに現在のコイン残高を問い合わせ、先程の購入が反映されていることを確認します。
⇩- 2024年10月19日の他ユーザーの購入履歴を確認します
他ユーザーが2024年10月19日に購入したコインの履歴を、エージェントを通じて確認し、今後の購入の参考にします。
以上で、全ての操作が問題なく動作していることを確認できました。今回は簡単なサンプルアプリでしたが、これを基に様々な機能の拡張が可能です。たとえば、複雑なデータ分析や自動化されたトランザクション管理など、応用範囲は広がります。
ぜひ、お手元で動作を検証してみてください！
参考資料
参考にさせていただいた書籍
- 「Amazon Bedrock 生成AIアプリ開発入門 [AWS深掘りガイド] 」
参考にさせていただいたサイト
- Amazon Web Services ブログ
「Amazon Bedrock のエージェントがメモリ保持とコード解釈をサポートするようになりました (プレビュー中)」
https://aws.amazon.com/jp/blogs/news/agents-for-amazon-bedrock-now-support-memory-retention-and-code-interpretation-preview/- Boto3 1.35.47 documentation
「invoke_agent」
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/bedrock-agent-runtime/client/invoke_agent.html
まとめ
今回のブログでは、Amazon Bedrockを使ってAIエージェントを構築し、コインの残高確認や購入履歴の取得を通じて、エージェントの動作を確認しました。
このサンプルアプリは、私自身の勉強の一環として、基本的な機能を組み合わせて作成したものであり、特に業務内でこのようなAIツールを利用しているわけではありません。そのため、実際に業務で使用するには、さらなる検証が必要になるでしょう。
ただ、このようなAIを用いたシステムは、今後さまざまな職種で使われる可能性が高いため、最新の技術情報をキャッチアップしておくことは、エンジニアにとって非常に重要だと感じています。
私自身も、普段の業務において、小さな部分からでもAIを活用できるよう、日々の勉強を怠らないよう努めていきたいと思います。
最後までお読みいただき、ありがとうございました。