はじめに
AWSは2023年に、生成AIを簡単に利用できるマネージドサービスとして「Amazon Bedrock」を正式リリースしました。このサービスは、Claude（Anthropic）、Llama 2（Meta）、Titan（Amazon）、Mistral AIなど、さまざまな大規模言語モデル（LLM）にアクセスできる統一インターフェースを提供します。本記事では、Dockerコンテナを使用してAmazon Bedrockを活用する方法を解説し、実践的なユースケースやサンプルコードを紹介します。
Amazon Bedrockとは
Amazon Bedrockは、AWSが提供する生成AIのためのフルマネージドサービスです。主な特徴は以下の通りです：
- 複数のファウンデーションモデルへのアクセス: Claude、Llama 2、Titan、Mistral AIなど、トップクラスの大規模言語モデルを単一のAPIで利用可能
- プライバシーとセキュリティ: データはAWSのインフラ内で処理され、モデルの訓練には使用されない
- カスタマイズ可能: 特定のユースケースに合わせてモデルをファインチューニング可能
- AWSエコシステムとの統合: 他のAWSサービスと簡単に連携
DockerとBedrockを組み合わせる利点
DockerコンテナとAmazon Bedrockを組み合わせることで、以下のような利点が得られます：
- 環境の一貫性: 開発、テスト、本番環境で同じ設定を使用可能
- スケーラビリティ: コンテナオーケストレーションツールと組み合わせて、需要に応じて自動的にスケール
- 移植性: クラウド環境やオンプレミス環境で一貫して実行可能
- 分離: アプリケーションと依存関係を分離し、競合を防止
- CI/CD統合: 継続的インテグレーション/デリバリーパイプラインとの簡単な統合
準備：Bedrockへのアクセス設定
Amazon Bedrockを使用するには、まずAWSアカウントでサービスへのアクセスを設定する必要があります。
1. Bedrockへのアクセス権限を持つIAMロールの作成
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"bedrock:InvokeModel",
"bedrock:InvokeModelWithResponseStream",
"bedrock:GetModelCustomizationJob",
"bedrock:ListModelCustomizationJobs",
"bedrock:ListFoundationModels"
],
"Resource": "*"
}
]
}
2. AWS CLIの設定
aws configure
# Access Key ID、Secret Access Key、リージョン、出力形式を入力
Docker環境の構築
Bedrockを利用するPythonアプリケーションのDockerfile
FROM python:3.9-slim
WORKDIR /app
# 必要なライブラリをインストール
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
# アプリケーションコードをコピー
COPY . .
# アプリケーションの起動
CMD ["python", "app.py"]
requirements.txt
boto3>=1.28.0
streamlit>=1.24.0
pandas>=1.5.3
matplotlib>=3.7.2
Bedrockを活用したDockerアプリケーション例
例1：テキスト生成AIチャットボット
以下は、Streamlitを使用したシンプルなチャットボットアプリケーションの例です。
app.py
import streamlit as st
import boto3
import json
import base64
from datetime import datetime
# Bedrockクライアントの初期化
bedrock_runtime = boto3.client(
service_name='bedrock-runtime',
region_name='us-east-1' # Bedrockが利用可能なリージョンを指定
)
# アプリケーションのタイトル
st.title("Bedrock AIチャットボット")
# セッション状態の初期化
if "messages" not in st.session_state:
st.session_state.messages = []
# モデル選択
model_id = st.selectbox(
"AIモデルを選択してください:",
[
"anthropic.claude-v2",
"anthropic.claude-instant-v1",
"meta.llama2-13b-chat-v1",
"amazon.titan-text-express-v1"
]
)
# メッセージ履歴の表示
for message in st.session_state.messages:
with st.chat_message(message["role"]):
st.markdown(message["content"])
# ユーザー入力
if prompt := st.chat_input("メッセージを入力してください"):
# ユーザーメッセージをセッションに追加
st.session_state.messages.append({"role": "user", "content": prompt})
# ユーザーメッセージの表示
with st.chat_message("user"):
st.markdown(prompt)
# AIの応答を生成
with st.chat_message("assistant"):
message_placeholder = st.empty()
full_response = ""
# モデルに応じたリクエスト形式の調整
if "anthropic" in model_id:
request_body = {
"prompt": f"\n\nHuman: {prompt}\n\nAssistant:",
"max_tokens_to_sample": 1000,
"temperature": 0.7,
"top_p": 0.9,
}
elif "meta.llama" in model_id:
request_body = {
"prompt": f"<s>[INST] {prompt} [/INST]",
"max_gen_len": 1000,
"temperature": 0.7,
"top_p": 0.9,
}
else: # amazon.titan
request_body = {
"inputText": prompt,
"textGenerationConfig": {
"maxTokenCount": 1000,
"temperature": 0.7,
"topP": 0.9,
}
}
try:
# Bedrockモデル呼び出し
response = bedrock_runtime.invoke_model(
modelId=model_id,
body=json.dumps(request_body)
)
# レスポンスの解析
response_body = json.loads(response['body'].read())
if "anthropic" in model_id:
full_response = response_body.get('completion', '')
elif "meta.llama" in model_id:
full_response = response_body.get('generation', '')
else: # amazon.titan
full_response = response_body.get('results', [{}])[0].get('outputText', '')
# 応答の表示
message_placeholder.markdown(full_response)
# 応答をセッションに追加
st.session_state.messages.append({"role": "assistant", "content": full_response})
except Exception as e:
st.error(f"エラーが発生しました: {str(e)}")
# 履歴のクリアボタン
if st.button("会話履歴をクリア"):
st.session_state.messages = []
st.experimental_rerun()
例2：画像生成アプリケーション
Bedrock Stable Diffusionモデルを使用して画像を生成するアプリケーションです。
image_app.py
import streamlit as st
import boto3
import json
import base64
from io import BytesIO
from PIL import Image
# Bedrockクライアントの初期化
bedrock_runtime = boto3.client(
service_name='bedrock-runtime',
region_name='us-east-1'
)
# アプリケーションのタイトル
st.title("Bedrock 画像生成アプリ")
# プロンプト入力
prompt = st.text_area("画像の説明を入力してください:",
"青い海と白い砂浜のトロピカルビーチ")
# パラメータ設定
col1, col2 = st.columns(2)
with col1:
negative_prompt = st.text_input("ネガティブプロンプト:",
"低品質, ぼやけた, 歪んだ")
num_inference_steps = st.slider("推論ステップ数:", 20, 100, 50)
with col2:
guidance_scale = st.slider("ガイダンススケール:", 1.0, 10.0, 7.5)
seed = st.number_input("シード値 (ランダムの場合は0):", 0, 10000, 0)
# 画像生成ボタン
if st.button("画像を生成"):
with st.spinner("画像を生成中..."):
try:
# リクエストボディの作成
request_body = {
"text_prompts": [
{
"text": prompt
}
],
"seed": seed if seed > 0 else None,
"cfg_scale": guidance_scale,
"steps": num_inference_steps
}
if negative_prompt:
request_body["text_prompts"].append({
"text": negative_prompt,
"weight": -1.0
})
# Stability.aiモデルの呼び出し
response = bedrock_runtime.invoke_model(
modelId="stability.stable-diffusion-xl-v1",
body=json.dumps(request_body)
)
# レスポンスの解析
response_body = json.loads(response['body'].read())
if "artifacts" in response_body and len(response_body["artifacts"]) > 0:
image_data = base64.b64decode(response_body["artifacts"][0]["base64"])
image = Image.open(BytesIO(image_data))
st.image(image, caption="生成画像", use_column_width=True)
# 画像のダウンロードボタン
buffered = BytesIO()
image.save(buffered, format="PNG")
img_bytes = buffered.getvalue()
st.download_button(
label="画像をダウンロード",
data=img_bytes,
file_name=f"bedrock_image_{seed}.png",
mime="image/png"
)
else:
st.error("画像の生成に失敗しました。")
except Exception as e:
st.error(f"エラーが発生しました: {str(e)}")
例3：データ分析と要約アプリケーション
CSVデータを分析し、そのデータについてAIが要約するアプリケーションです。
data_analyzer.py
import streamlit as st
import boto3
import json
import pandas as pd
import matplotlib.pyplot as plt
import io
# Bedrockクライアントの初期化
bedrock_runtime = boto3.client(
service_name='bedrock-runtime',
region_name='us-east-1'
)
# アプリケーションのタイトル
st.title("Bedrock データ分析・要約アプリ")
# ファイルアップロード
uploaded_file = st.file_uploader("CSVファイルをアップロードしてください", type=["csv"])
if uploaded_file is not None:
# データの読み込みと表示
try:
df = pd.read_csv(uploaded_file)
st.subheader("データプレビュー")
st.dataframe(df.head())
# 基本統計情報
st.subheader("基本統計情報")
st.write(df.describe())
# データの可視化
st.subheader("データ可視化")
fig_col1, fig_col2 = st.columns(2)
# 数値列の抽出
numeric_columns = df.select_dtypes(include=['float64', 'int64']).columns.tolist()
if len(numeric_columns) > 0:
with fig_col1:
selected_column = st.selectbox("ヒストグラム用の列を選択:", numeric_columns)
fig, ax = plt.subplots()
ax.hist(df[selected_column].dropna(), bins=20)
ax.set_title(f"{selected_column}のヒストグラム")
st.pyplot(fig)
if len(numeric_columns) > 1:
with fig_col2:
x_column = st.selectbox("X軸:", numeric_columns, index=0)
y_column = st.selectbox("Y軸:", numeric_columns, index=min(1, len(numeric_columns)-1))
fig2, ax2 = plt.subplots()
ax2.scatter(df[x_column], df[y_column])
ax2.set_xlabel(x_column)
ax2.set_ylabel(y_column)
ax2.set_title(f"{x_column} vs {y_column}")
st.pyplot(fig2)
# データの要約をAIに依頼
if st.button("AIにデータ分析を依頼"):
with st.spinner("AIがデータを分析中..."):
# データの概要を文字列に変換
buffer = io.StringIO()
df.info(buf=buffer)
data_info = buffer.getvalue()
# 統計情報を文字列に変換
data_stats = df.describe().to_string()
# AIへのプロンプト
prompt = f"""
以下のデータセットの分析と洞察を提供してください:
データ情報:
{data_info}
基本統計:
{data_stats}
このデータについて以下の観点から分析してください:
1. データセットの概要と特徴
2. 主要な傾向やパターン
3. 異常値や特筆すべき点
4. ビジネスや意思決定に活用できるインサイト
5. さらなる分析のための提案
専門的かつ簡潔に説明してください。
"""
# Bedrockのモデルリクエスト
request_body = {
"prompt": f"\n\nHuman: {prompt}\n\nAssistant:",
"max_tokens_to_sample": 2000,
"temperature": 0,
"top_p": 0.9,
}
try:
response = bedrock_runtime.invoke_model(
modelId="anthropic.claude-v2",
body=json.dumps(request_body)
)
response_body = json.loads(response['body'].read())
analysis = response_body.get('completion', '')
st.subheader("AIによるデータ分析")
st.markdown(analysis)
except Exception as e:
st.error(f"AI分析中にエラーが発生しました: {str(e)}")
except Exception as e:
st.error(f"ファイル処理中にエラーが発生しました: {str(e)}")
Dockerコンテナのビルドと実行
作成したアプリケーションをDockerコンテナとしてビルドし、実行する手順は以下の通りです。
ビルド
# テキスト生成アプリケーションのビルド
docker build -t bedrock-chat-app .
# 画像生成アプリケーションのビルド
docker build -t bedrock-image-app -f Dockerfile.image .
# データ分析アプリケーションのビルド
docker build -t bedrock-data-app -f Dockerfile.data .
実行
# AWS認証情報をコンテナに渡して実行
docker run -p 8501:8501 \
-e AWS_ACCESS_KEY_ID=your_access_key \
-e AWS_SECRET_ACCESS_KEY=your_secret_key \
-e AWS_REGION=us-east-1 \
bedrock-chat-app
セキュリティを強化するために、AWS IAMロールを使用する方法もあります：
# EC2インスタンスのIAMロールを使用
docker run -p 8501:8501 \
-v ~/.aws:/root/.aws:ro \
bedrock-chat-app
AWS ECSでのデプロイ
Amazon Elastic Container Service (ECS)を使用して、Bedrockアプリケーションをデプロイする方法を紹介します。
タスク定義の作成
{
"family": "bedrock-app",
"networkMode": "awsvpc",
"executionRoleArn": "arn:aws:iam::your-account-id:role/ecsTaskExecutionRole",
"taskRoleArn": "arn:aws:iam::your-account-id:role/BedrockAppRole",
"containerDefinitions": [
{
"name": "bedrock-app",
"image": "your-account-id.dkr.ecr.us-east-1.amazonaws.com/bedrock-app:latest",
"essential": true,
"portMappings": [
{
"containerPort": 8501,
"hostPort": 8501,
"protocol": "tcp"
}
],
"logConfiguration": {
"logDriver": "awslogs",
"options": {
"awslogs-group": "/ecs/bedrock-app",
"awslogs-region": "us-east-1",
"awslogs-stream-prefix": "ecs"
}
}
}
],
"requiresCompatibilities": [
"FARGATE"
],
"cpu": "1024",
"memory": "2048"
}
サービスの作成
ECSクラスターにサービスを作成し、タスク定義をデプロイします：
aws ecs create-service \
--cluster your-cluster \
--service-name bedrock-app-service \
--task-definition bedrock-app:1 \
--desired-count 1 \
--launch-type FARGATE \
--network-configuration "awsvpcConfiguration={subnets=[subnet-12345678],securityGroups=[sg-12345678],assignPublicIp=ENABLED}" \
--load-balancers "targetGroupArn=arn:aws:elasticloadbalancing:us-east-1:your-account-id:targetgroup/bedrock-app-tg/1234567890abcdef,containerName=bedrock-app,containerPort=8501"
実践的なユースケース
1. カスタマーサポート自動化
Bedrockモデルを使用して、顧客からの問い合わせに自動的に回答するシステムを構築できます。このシステムは、以下のような機能を提供します：
- FAQ情報を基に、一般的な質問に回答
- 複雑な問い合わせは人間のエージェントに転送
- 顧客の感情分析を行い、対応の優先度を決定
2. コンテンツ生成パイプライン
マーケティングや出版業界向けに、コンテンツを自動生成するパイプラインを構築できます：
- ブログ記事のドラフト作成
- ソーシャルメディア投稿の自動生成
- 製品説明の多言語翻訳と最適化
3. データ分析と意思決定支援
ビジネスデータを分析し、意思決定を支援するシステムを構築できます：
- 売上データの分析と予測
- 市場トレンドのサマリー作成
- データに基づく推奨事項の提示
パフォーマンスとコスト最適化
コンテキスト窓の最適化
Bedrockモデルでは、入力トークン数が料金に影響します。コンテキスト窓を効率的に使用するためのテクニックを紹介します：
def optimize_context(prompt, max_tokens=8000):
"""長いプロンプトを最大トークン数内に収める"""
# 簡易的なトークン数推定（正確なトークン化はモデルによって異なる）
estimated_tokens = len(prompt.split()) * 1.3
if estimated_tokens <= max_tokens:
return prompt
# トークン数が超過する場合は、重要な部分だけを保持
important_parts = extract_important_parts(prompt)
return important_parts
def extract_important_parts(prompt):
# ここに重要な部分を抽出するロジックを実装
# 例：最新の情報を優先、質問に直接関連する部分を保持など
pass
キャッシング戦略
同じ質問や似たような質問に対する応答をキャッシュすることで、コストとレイテンシーを削減できます：
import hashlib
import redis
import json
# Redisキャッシュの初期化
redis_client = redis.Redis(host='localhost', port=6379, db=0)
def get_cached_response(prompt, model_id):
"""キャッシュから応答を取得"""
# プロンプトとモデルIDからキャッシュキーを生成
cache_key = hashlib.md5(f"{prompt}:{model_id}".encode()).hexdigest()
# キャッシュをチェック
cached_response = redis_client.get(cache_key)
if cached_response:
return json.loads(cached_response)
return None
def cache_response(prompt, model_id, response, ttl=3600):
"""応答をキャッシュに保存"""
cache_key = hashlib.md5(f"{prompt}:{model_id}".encode()).hexdigest()
redis_client.setex(cache_key, ttl, json.dumps(response))
バッチ処理
大量のリクエストを処理する場合は、バッチ処理を使用してスループットを向上させます：
import asyncio
import boto3
from concurrent.futures import ThreadPoolExecutor
async def process_batch(prompts, model_id="anthropic.claude-v2", max_workers=5):
"""複数のプロンプトを並行処理"""
bedrock_runtime = boto3.client('bedrock-runtime')
async def process_single(prompt):
request_body = {
"prompt": f"\n\nHuman: {prompt}\n\nAssistant:",
"max_tokens_to_sample": 1000,
"temperature": 0.7,
}
response = await loop.run_in_executor(
executor,
lambda: bedrock_runtime.invoke_model(
modelId=model_id,
body=json.dumps(request_body)
)
)
response_body = json.loads(response['body'].read())
return response_body.get('completion', '')
loop = asyncio.get_event_loop()
executor = ThreadPoolExecutor(max_workers=max_workers)
tasks = [process_single(prompt) for prompt in prompts]
return await asyncio.gather(*tasks)
セキュリティベストプラクティス
Bedrock APIキーやAWS認証情報を安全に管理するためのベストプラクティスを紹介します。
1. シークレット管理
Docker Secretsや環境変数ではなく、AWS Secrets ManagerやParameterStoreを使用してシークレットを管理します：
import boto3
from botocore.exceptions import ClientError
def get_secret():
"""AWS Secrets Managerからシークレットを取得"""
secret_name = "bedrock/api-keys"
region_name = "us-east-1"
session = boto3.session.Session()
client = session.client(
service_name='secretsmanager',
region_name=region_name
)
try:
response = client.get_secret_value(SecretId=secret_name)
if 'SecretString' in response:
return json.loads(response['SecretString'])
except ClientError as e:
raise e
return None
2. IAMロールとポリシー
最小権限の原則に従って、必要最小限のアクセス権限を持つIAMロールを使用します：
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"bedrock:InvokeModel"
],
"Resource": [
"arn:aws:bedrock:us-east-1::foundation-model/anthropic.claude-v2",
"arn:aws:bedrock:us-east-1::foundation-model/stability.stable-diffusion-xl-v1"
]
},
{
"Effect": "Allow",
"Action": [
"secretsmanager:GetSecretValue"
],
"Resource": [
"arn:aws:secretsmanager:us-east-1:your-account-id:secret:bedrock/api-keys-*"
]
}
]
}
3. ネットワークセキュリティ
VPCエンドポイントを使用して、プライベートネットワーク内からBedrockにアクセスします：
# BedrockのVPCエンドポイントを作成
aws ec2 create-vpc-endpoint \
--vpc-id vpc-12345678 \
--service-name com.amazonaws.us-east-1.bedrock-runtime \
--subnet-ids subnet-12345678 subnet-87654321 \
--security-group-ids sg-12345678 \
--vpc-endpoint-type Interface
まとめ
本記事では、AWSの最先端サービス「Amazon Bedrock」をDockerで活用する方法について解説しました。主なポイントは以下の通りです：
- Bedrockの主要機能と特徴
- Docker環境でBedrockを利用するための基本セットアップ
- テキスト生成、画像生成、データ分析など、様々なユースケース向けのサンプルアプリケーション
- コンテナのビルド・実行方法とAWS ECSでのデプロイ
- パフォーマンスとコスト最適化のテクニック
- セキュリティベストプラクティス
DockerとAWS Bedrockを組み合わせることで、スケーラブルで一貫性のある生成AI環境を構築できます。これにより、プロダクション環境での生成AIアプリケーションの開発と運用が大幅に効率化されます。ぜひ、この記事で紹介した手法を活用して、独自のAIサービスを構築してみてください。
コメント