Amazon Bedrock WorkshopでBedrockの基本機能の確認

Amazon Bedrock Workshopで、Bedrockの主要機能のハンズオンを試してみた。

Jupyter Notebookベースで、実際にKnowledge BasesやAtrands AgentでAgentCore Runtime、model fine-tuningをPythonSDKベースで構築するハンズオン

学べる内容	触れる機能
テキスト生成	Converse API / InvokeModel / streaming / function calling
RAG	Knowledge Bases / RetrieveAndGenerate / Retrieve API
モデルカスタマイズ	fine-tuning / continued pre-training / distillation
Agents	DynamoDB・Knowledge Bases 連携、AgentCore Runtime へのデプロイ
Open-weight model 推論	OpenAI-compatible endpoint、tool use、structured output

CodeXに勉強順を聞く

入力プロンプト

このリポジトリをハンズオン教材として読む前提で、
学習順序と各章・各ファイルの目的を整理して。

01_Text_generation/ → Bedrock InvokeModel API / Converse API

InvokeModel APIでテキスト要約依頼のプロンプト (モデルを直接呼び出す一番基本的なAPI)

# Create request body for Claude Haiku 4.5
claude_body = json.dumps(
    {
        "anthropic_version": "bedrock-2023-05-31",
        "max_tokens": 1000,
        "temperature": 0.5,
        "messages": [{"role": "user", "content": [{"type": "text", "text": prompt}]}],
    }
)

# Send request to Claude Haiku 4.5
response = bedrock.invoke_model(
    modelId=MODELS["Claude Haiku 4.5"],
    body=claude_body,
    accept="application/json",
    contentType="application/json",
)
response_body = json.loads(response.get("body").read())

# Extract and display the response text
claude_summary = response_body["content"][0]["text"]
display_response(claude_summary, "Claude Haiku 4.5 (Invoke Model API)")

Converse API で要約文章をより簡潔にする依頼のプロンプト (チャット・会話用途向けに作られた統一API)

converse_request = {
    "messages": [
        {
            "role": "user",
            "content": [
                {
                    "text": f"Please provide a concise summary of the following text in 2-3 sentences. Text to summarize: {text_to_summarize}"
                }
            ],
        }
    ],
    "inferenceConfig": {
        "maxTokens": 500,
        "temperature": 0.4,
    },
}

# Call Claude Haiku 4.5 with Converse API
response = bedrock.converse(
    modelId=MODELS["Claude Haiku 4.5"],
    messages=converse_request["messages"],
    inferenceConfig=converse_request["inferenceConfig"],
)

# Extract the model's response
claude_converse_response = response["output"]["message"]["content"][0]["text"]
display_response(claude_converse_response, "Claude Haiku 4.5 (Converse API)")

Converse API でFunction Callingを使用する (登録された関数をLLMが判断して使用するTool)

weather_tool = {
    "tools": [
        {
            "toolSpec": {
                "name": "get_weather",
                "description": "Get current weather for a specific location",
                "inputSchema": {
                    "json": {
                        "type": "object",
                        "properties": {
                            "location": {
                                "type": "string",
                                "description": "The city name to get weather for",
                            }
                        },
                        "required": ["location"],
                    }
                },
            }
        }
    ],
    "toolChoice": {
        "auto": {}  # Let the model decide when to use the tool
    },
}

function_request = {
    "messages": [
        {
            "role": "user",
            "content": [
                {
                    "text": "What's the weather like in San Francisco right now? And what should I wear?"
                }
            ],
        }
    ],
    "inferenceConfig": {
        "temperature": 0.0,  # Use 0 temperature for deterministic function calling
        "maxTokens": 500,
    },
}

response = bedrock.converse(
    modelId=MODELS["Claude Haiku 4.5"],
    messages=function_request["messages"],
    inferenceConfig=function_request["inferenceConfig"],
    toolConfig=weather_tool,
)
print(json.dumps(response, indent=2))

02_Knowledge_Bases_and_RAG

OpenSearch Serverlessのベクトルインデックスの作成

# Define the configuration for the AOSS vector index
index_definition = {
    "settings": {
        "index.knn": "true",
        "number_of_shards": 1,
        "knn.algo_param.ef_search": 512,
        "number_of_replicas": 0,
    },
    "mappings": {
        "properties": {
            "vector": {
                "type": "knn_vector",
                "dimension": embedding_model_dim,
                "method": {"name": "hnsw", "engine": "faiss", "space_type": "l2"},
            },
            "text": {"type": "text"},
            "text-metadata": {"type": "text"},
        }
    },
}

# Create an OpenSearch index
response = os_client.indices.create(index=aoss_index_name, body=index_definition)

Knowledge Baseの作成

# Vector Storage Configuration
storage_config = {
    "type": "OPENSEARCH_SERVERLESS",
    "opensearchServerlessConfiguration": {
        "collectionArn": aoss_collection["createCollectionDetail"]["arn"],
        "vectorIndexName": aoss_index_name,
        "fieldMapping": {
            "vectorField": "vector",
            "textField": "text",
            "metadataField": "text-metadata",
        },
    },
}

# Knowledge Base Configuration
knowledge_base_config = {
    "type": "VECTOR",
    "vectorKnowledgeBaseConfiguration": {"embeddingModelArn": embedding_model_arn},
}

response = bedrock_agent_client.create_knowledge_base(
    name=bedrock_kb_name,
    description="Amazon shareholder letter knowledge base.",
    roleArn=bedrock_kb_execution_role_arn,
    knowledgeBaseConfiguration=knowledge_base_config,
    storageConfiguration=storage_config,
)

bedrock_kb_id = response["knowledgeBase"]["knowledgeBaseId"]

S3データソースとナレッジベースとの接続

# Data Source Configuration
data_source_config = {
    "type": "S3",
    "s3Configuration": {
        "bucketArn": f"arn:aws:s3:::{s3_bucket_name}",
        # "inclusionPrefixes":["*.*"]   # you can use this if you want to create a KB using data within s3 prefixes.
    },
}

# Vector Ingestion Configuration
vector_ingestion_config = {
    "chunkingConfiguration": {
        "chunkingStrategy": "FIXED_SIZE",
        "fixedSizeChunkingConfiguration": {"maxTokens": 512, "overlapPercentage": 20},
    }
}

response = bedrock_agent_client.create_data_source(
    name=bedrock_kb_name,
    description="Amazon shareholder letter knowledge base.",
    knowledgeBaseId=bedrock_kb_id,
    dataSourceConfiguration=data_source_config,
    vectorIngestionConfiguration=vector_ingestion_config,
)

bedrock_ds_id = response["dataSource"]["dataSourceId"]

S3のデータソースからOpenSearch Serverlessへ埋め込みベクトル変換・保存をする

# Start an ingestion job
response = bedrock_agent_client.start_ingestion_job(
    knowledgeBaseId=bedrock_kb_id, dataSourceId=bedrock_ds_id
)

bedrock_job_id = response["ingestionJob"]["ingestionJobId"]

RetrieveAndGenerate API (ナレッジベースから情報を取得し、回答を生成する)

user_query = "How does Amazon use technology to better serve its customers?"

response = bedrock_agent_client.retrieve_and_generate(
    input={"text": user_query},
    retrieveAndGenerateConfiguration={
        "type": "KNOWLEDGE_BASE",
        "knowledgeBaseConfiguration": {
            "knowledgeBaseId": bedrock_kb_id,
            "modelArn": model_arn,
        },
    },
)

Retrieve API (ナレッジベースから情報を取得のみ)

# Implement the `retrieve` function
def retrieve(user_query, kb_id, num_of_results=5):
    return bedrock_agent_client.retrieve(
        retrievalQuery={"text": user_query},
        knowledgeBaseId=kb_id,
        retrievalConfiguration={
            "vectorSearchConfiguration": {
                "numberOfResults": num_of_results,
                "overrideSearchType": "HYBRID",  # optional
            }
        },
    )

04_Agents

#strands_nova.py
# Python 組み込みライブラリ:
import os
import uuid

# 外部依存関係:
from bedrock_agentcore.runtime import BedrockAgentCoreApp  # サーバーのひな形
import boto3  # Python 用 AWS SDK
from strands import Agent, tool  # エージェントフレームワーク
from strands.models import BedrockModel
from strands_tools import current_time, retrieve

# 渡された環境変数を読み込みます
kb_id = os.environ["KNOWLEDGE_BASE_ID"]
kb_name = os.environ["KNOWLEDGE_BASE_NAME"]
db_table_name = os.environ["DB_TABLE_NAME"]
model_id = os.environ["BEDROCK_MODEL_ID"]
system_prompt = os.environ["BEDROCK_MODEL_SYSTEM_PROMPT"]

# DynamoDB テーブルリソースを取得します
dynamodb = boto3.resource('dynamodb')
db_table = dynamodb.Table(db_table_name)

# Bedrock AgentCore アプリを初期化します
app = BedrockAgentCoreApp()

@tool
def get_booking_details(booking_id: str, restaurant_name: str) -> dict:
    """Get the relevant details for booking_id in restaurant_name
    Args:
        booking_id: the id of the reservation
        restaurant_name: name of the restaurant handling the reservation

    Returns:
        booking_details: the details of the booking in JSON format
    """

    try:
        response = db_table.get_item(
            Key={"booking_id": booking_id, "restaurant_name": restaurant_name}
        )
        if "Item" in response:
            return response["Item"]
        else:
            return f"No booking found with ID {booking_id}"
    except Exception as e:
        return str(e)
    
@tool
def delete_booking(booking_id: str, restaurant_name:str) -> str:
    """delete an existing booking_id at restaurant_name
    Args:
        booking_id: the id of the reservation
        restaurant_name: name of the restaurant handling the reservation

    Returns:
        confirmation_message: confirmation message
    """
    try:
        response = db_table.delete_item(Key={'booking_id': booking_id, 'restaurant_name': restaurant_name})
        if response['ResponseMetadata']['HTTPStatusCode'] == 200:
            return f'Booking with ID {booking_id} deleted successfully'
        else:
            return f'Failed to delete booking with ID {booking_id}'
    except Exception as e:
        return str(e)

@tool
def create_booking(date: str, hour: str, restaurant_name: str, guest_name: str, num_guests: str) -> str:
    """Create a new booking at restaurant_name
    Args:
        date: The date of the booking in the format YYYY-MM-DD. 
        hour:the hour of the booking in the format HH:MM"
        restaurant_name:The name of the restaurant handling the reservation"
        guest_name: The name of the customer to have in the reservation"
        num_guests: The number of guests for the booking"

    Returns:
        confirmation_message: confirmation message
    """
    
    print(f"Creating reservation for {num_guests} people at {restaurant_name}, {date} at {hour} in the name of {guest_name}")
    try:
        booking_id = str(uuid.uuid4())[:8]
        db_table.put_item(
            Item={
                'booking_id': booking_id,
                'restaurant_name': restaurant_name,
                'date': date,
                'name': guest_name,
                'hour': hour,
                'num_guests': num_guests
            }
        )
        return f"Reservation created with booking id: {booking_id}"
    except Exception as e:
        print(e)
        return "Failed to create booking."



model = BedrockModel(
    model_id=model_id
)
agent = Agent(
    model=model,
    tools=[current_time, retrieve, get_booking_details, create_booking, delete_booking],
    system_prompt=system_prompt
)

@app.entrypoint # <-- 呼び出し関数をデコレートします
def strands_agent_bedrock(payload):
    """
    Invoke the agent with a payload
    """
    user_input = payload.get("prompt")
    print("User input:", user_input)
    response = agent(user_input)
    return response.message['content'][0]['text']

if __name__ == "__main__":
    app.run() # <-- `app.run()` でエージェントの実行を AgentCoreRuntime に制御させます

Dockerfileの作成

from bedrock_agentcore_starter_toolkit import Runtime
from boto3.session import Session

agentcore_runtime = Runtime()
response = agentcore_runtime.configure(
    entrypoint="strands_nova.py",
    auto_create_execution_role=True,
    auto_create_ecr=True,
    memory_mode="NO_MEMORY",
    requirements_file="requirements.txt",
    region=Session().region_name,
    agent_name="restaurant_helper",
)
print("Created Docker file:", response.dockerfile_path)

AgentCore RunTimeを起動

launch_result = agentcore_runtime.launch(
    env_vars={
        "KNOWLEDGE_BASE_ID": kb_id,
        "KNOWLEDGE_BASE_NAME": kb_name,
        "DB_TABLE_NAME": db_table_name,
        "BEDROCK_MODEL_ID": "us.amazon.nova-2-lite-v1:0",
        "BEDROCK_MODEL_SYSTEM_PROMPT": system_prompt,
    }
)

参考文献