API を使った呼び出し
Python（boto3）を使って、プログラムから Amazon Bedrock を呼び出してみましょう。
前提条件
- Python 3.9 以上がインストールされていること
- AWS CLI が設定されていること（
aws configure 済み）
- pip が使用可能であること
ステップ 1: 環境のセットアップ
必要なパッケージのインストール
AWS 認証情報の確認
aws sts get-caller-identity
正常に実行されれば、AWS 認証情報が設定されています。
ステップ 2: 基本的な API 呼び出し
シンプルな例
以下のコードを bedrock_basic.py として保存します：
| bedrock_basic.py |
|---|
| import boto3
import json
# Bedrock Runtime クライアントを作成
client = boto3.client(
service_name='bedrock-runtime',
region_name='us-east-1'
)
# リクエストボディを作成
request_body = {
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 1024,
"messages": [
{
"role": "user",
"content": "AWSのLambdaとは何ですか？50文字以内で説明してください。"
}
]
}
# API を呼び出し
response = client.invoke_model(
modelId='anthropic.claude-3-haiku-20240307-v1:0',
body=json.dumps(request_body),
contentType='application/json',
accept='application/json'
)
# レスポンスを解析
response_body = json.loads(response['body'].read())
print(response_body['content'][0]['text'])
|
実行
期待される出力例
AWS Lambdaは、サーバー管理不要でコードを実行できるサーバーレスコンピューティングサービスです。
ステップ 3: システムプロンプトの使用
システムプロンプトを追加して、AIの振る舞いをカスタマイズします：
| bedrock_system_prompt.py |
|---|
| import boto3
import json
client = boto3.client(
service_name='bedrock-runtime',
region_name='us-east-1'
)
# システムプロンプトを追加
request_body = {
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 2048,
"system": """あなたは優秀なAWSアーキテクトです。
以下のルールに従って回答してください：
- 日本語で回答する
- 具体的なユースケースを含める
- メリットとデメリットを説明する
- 必要に応じてサンプルコードを提供する""",
"messages": [
{
"role": "user",
"content": "ECSとEKSの違いを教えてください。"
}
]
}
response = client.invoke_model(
modelId='anthropic.claude-3-5-sonnet-20240620-v1:0',
body=json.dumps(request_body),
contentType='application/json',
accept='application/json'
)
response_body = json.loads(response['body'].read())
print(response_body['content'][0]['text'])
|
ステップ 4: ストリーミングレスポンス
長い回答をリアルタイムで受け取るには、ストリーミングを使用します：
| bedrock_streaming.py |
|---|
| import boto3
import json
client = boto3.client(
service_name='bedrock-runtime',
region_name='us-east-1'
)
request_body = {
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 2048,
"messages": [
{
"role": "user",
"content": "Pythonでウェブスクレイピングを行う方法を、手順を追って説明してください。"
}
]
}
# ストリーミングで呼び出し
response = client.invoke_model_with_response_stream(
modelId='anthropic.claude-3-5-sonnet-20240620-v1:0',
body=json.dumps(request_body),
contentType='application/json',
accept='application/json'
)
# ストリーミングレスポンスを処理
print("=== ストリーミング開始 ===")
for event in response['body']:
chunk = json.loads(event['chunk']['bytes'])
if chunk['type'] == 'content_block_delta':
print(chunk['delta'].get('text', ''), end='', flush=True)
print("\n=== ストリーミング終了 ===")
|
ストリーミングのメリット
- ユーザーが回答を早く確認できる
- 長い回答でも待ち時間が短く感じられる
- チャットアプリケーションに最適
ステップ 5: 会話の継続（マルチターン）
複数のメッセージをやり取りする会話を実装します：
| bedrock_conversation.py |
|---|
| import boto3
import json
client = boto3.client(
service_name='bedrock-runtime',
region_name='us-east-1'
)
def chat(messages, system_prompt=None):
"""Bedrockとチャットする関数"""
request_body = {
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 2048,
"messages": messages
}
if system_prompt:
request_body["system"] = system_prompt
response = client.invoke_model(
modelId='anthropic.claude-3-haiku-20240307-v1:0',
body=json.dumps(request_body),
contentType='application/json',
accept='application/json'
)
response_body = json.loads(response['body'].read())
return response_body['content'][0]['text']
# 会話履歴を保持
conversation_history = []
# 会話を開始
print("=== 会話開始（'quit'で終了）===\n")
while True:
user_input = input("You: ")
if user_input.lower() == 'quit':
print("会話を終了します。")
break
# ユーザーメッセージを追加
conversation_history.append({
"role": "user",
"content": user_input
})
# AIの応答を取得
assistant_response = chat(
conversation_history,
system_prompt="あなたは親切なAWSサポートエンジニアです。"
)
# AIの応答を履歴に追加
conversation_history.append({
"role": "assistant",
"content": assistant_response
})
print(f"\nAssistant: {assistant_response}\n")
|
ステップ 6: エラーハンドリング
本番環境では、適切なエラーハンドリングが重要です：
| bedrock_error_handling.py |
|---|
| import boto3
import json
from botocore.exceptions import ClientError
client = boto3.client(
service_name='bedrock-runtime',
region_name='us-east-1'
)
def invoke_bedrock(prompt, model_id='anthropic.claude-3-haiku-20240307-v1:0'):
"""エラーハンドリング付きでBedrockを呼び出す"""
request_body = {
"anthropic_version": "bedrock-2023-05-31",
"max_tokens": 1024,
"messages": [{"role": "user", "content": prompt}]
}
try:
response = client.invoke_model(
modelId=model_id,
body=json.dumps(request_body),
contentType='application/json',
accept='application/json'
)
response_body = json.loads(response['body'].read())
return response_body['content'][0]['text']
except ClientError as e:
error_code = e.response['Error']['Code']
error_message = e.response['Error']['Message']
if error_code == 'AccessDeniedException':
print(f"アクセス拒否: モデルへのアクセス権限がありません")
elif error_code == 'ThrottlingException':
print(f"スロットリング: リクエスト制限に達しました。少し待ってから再試行してください")
elif error_code == 'ModelNotReadyException':
print(f"モデル準備中: モデルがまだ利用可能ではありません")
elif error_code == 'ValidationException':
print(f"バリデーションエラー: リクエストの形式が正しくありません")
else:
print(f"エラー ({error_code}): {error_message}")
return None
except Exception as e:
print(f"予期しないエラー: {e}")
return None
# 使用例
result = invoke_bedrock("AWSとは？")
if result:
print(result)
|
モデル ID 一覧
| モデル |
モデル ID |
| Claude 3.5 Sonnet |
anthropic.claude-3-5-sonnet-20240620-v1:0 |
| Claude 3 Haiku |
anthropic.claude-3-haiku-20240307-v1:0 |
| Claude 3 Sonnet |
anthropic.claude-3-sonnet-20240229-v1:0 |
| Titan Text Express |
amazon.titan-text-express-v1 |
ハンズオン課題
課題: FAQ ボットを作成
以下の要件を満たすシンプルな FAQ ボットを作成してみましょう：
- システムプロンプトで「あなたはAWSの FAQ ボットです」と設定
- ユーザーの質問に対して回答を生成
- 会話履歴を保持してコンテキストを維持
- エラーハンドリングを実装
次のステップ
API の基本的な使い方を学んだら、次はプロンプトエンジニアリングのテクニックを学びましょう。
プロンプトエンジニアリングへ進む