KDDIアジャイル開発センター(KAG)の2025年度エンジニア新人研修コンテンツの一部です。
このハンズオン編の前に、ホワイトボーディング中心の座学編があります。
- AWSアカウントに、自分のIAMユーザーでサインイン
- リージョンをオレゴンに切り替え(以降、すべてオレゴンで作業します)
- Bedrockコンソールへ移動し、すべてのモデルを有効化
- CloudShellを起動
できる人は、自分のMacでVS Codeを使って作業実施してもOK。
事前に aws configure
でIAM認証情報を設定しておくこと。
1-1. LLMのAPIを叩いてみよう!
Amazon BedrockのConverse APIを使います。
1-1.py
# 必要なPyhtonライブラリをインポート
import boto3
# AWS SDK for Pythonで、Bedrock用のAPIクライアントを作成
client = boto3.client("bedrock-runtime")
# Converse APIで推論を行う
response = client.converse(
modelId="us.anthropic.claude-sonnet-4-20250514-v1:0",
messages=[{
"role": "user",
"content": [{"text": "KAGのゆるキャラの名前は?"}]
}]
)
# APIレスポンスから、生成テキストのみを取り出してプリント
print(response["output"]["message"]["content"][0]["text"])
1-2. ストリーミング出力させてみよう!
ConverseSetream APIを使います。
1-2.py
# 必要なPyhtonライブラリをインポート
import boto3
# AWS SDK for Pythonで、Bedrock用のAPIクライアントを作成
client = boto3.client("bedrock-runtime")
# Converse Stream APIでストリーミング推論を行う
response = client.converse_stream(
modelId="us.anthropic.claude-sonnet-4-20250514-v1:0",
messages=[{
"role": "user",
"content": [{"text": "KAGのゆるキャラの名前は?"}]
}]
)
# ストリーミングレスポンスを自動待機し、チャンクごとにプリント
for event in response['stream']:
if 'contentBlockDelta' in event:
if 'delta' in event['contentBlockDelta']:
if 'text' in event['contentBlockDelta']['delta']:
print(event['contentBlockDelta']['delta']['text'], end='', flush=True)
print() # 最後に改行
1-3. 簡単なフロントエンドを付けよう
Pythonライブラリ「Streamlit」を使います。
1-3.py
# 必要なPythonライブラリをインポート
import streamlit as st
import boto3
# フロントエンド
st.title("おしえて! Bedrock")
prompt = st.text_input("質問を入力", "KAGのゆるキャラの名前は?")
# AWS SDK for Pythonで、Bedrock用のAPIクライアントを作成
client = boto3.client("bedrock-runtime")
# ボタンをクリックしたら実行
if st.button("送信"):
# Converse APIで推論を行う
response = client.converse(
modelId="us.anthropic.claude-sonnet-4-20250514-v1:0",
messages=[{
"role": "user",
"content": [{"text": prompt}]
}]
)
# 回答を表示
answer = response["output"]["message"]["content"][0]["text"]
st.write(answer)
CloudShell
# 1タブ目
pip install streamlit
streamlit run 1-3.py
# 2タブ目
ssh -p 443 -R0:localhost:8501 a.pinggy.io
2-1. RAGアーキテクチャを作ってみよう!
Meta社のベクトルインデックス作成OSS「Faiss」を使います。
2-1.py
# 必要なPythonライブラリをインポート
import boto3
import json
import numpy as np
import faiss
# AWS SDK for Pythonで、Bedrock用のAPIクライアントを作成
client = boto3.client("bedrock-runtime")
# 生成AIモデルを設定
llm = "us.anthropic.claude-sonnet-4-20250514-v1:0"
embedding_model = "cohere.embed-multilingual-v3"
# ========================================
# 社内文書を定義
# ========================================
documents = [
{
"id": 0,
"title": "かぐたん",
"content": "かぐたんはKAG社のSlackチャットボットです。"
},
{
"id": 1,
"title": "カグカグ",
"content": "カグカグはKAG社のゆるキャラです。"
}
]
# ========================================
# 社内文書をベクトルに変換
# ========================================
print("【ベクトルデータを準備】")
# ドキュメントをベクトル化
embeddings = []
for doc in documents:
print(f"ベクトルに変換中: {doc['title']}")
# ドキュメントの埋め込みを生成
response = client.invoke_model(
modelId=embedding_model,
body=json.dumps({
"texts": [doc['content']],
"input_type": "search_document"
})
)
response_body = json.loads(response["body"].read())
embedding = np.array(response_body["embeddings"][0])
embeddings.append(embedding)
print()
# NumPy配列に変換し、ベクトルを正規化
embeddings_array = np.array(embeddings).astype('float32')
faiss.normalize_L2(embeddings_array)
# FAISSインデックスを作成
dimension = embeddings_array.shape[1] # ベクトルの次元数(Cohere: 1024次元)
index = faiss.IndexFlatIP(dimension) # 内積(コサイン類似度)でインデックス作成
index.add(embeddings_array) # ベクトルを追加
# ========================================
# 検索を実行
# ========================================
query = "KAG社のゆるキャラの名前は?"
# クエリをベクトル化
response = client.invoke_model(
modelId=embedding_model,
body=json.dumps({
"texts": [query],
"input_type": "search_query"
})
)
response_body = json.loads(response["body"].read())
# NumPy配列に変換し、ベクトルを正規化
query_embedding = np.array(response_body["embeddings"][0]).reshape(1, -1).astype('float32')
faiss.normalize_L2(query_embedding)
# コサイン類似度で検索(上位1件を取得)
similarities, indices = index.search(query_embedding, 1)
retrieved_docs = []
for i, (sim, idx) in enumerate(zip(similarities[0], indices[0])):
doc = documents[idx]
retrieved_docs.append(doc)
context = retrieved_docs[0]['content']
print("【検索結果】")
print("クエリ: ", query)
print("結果: ", retrieved_docs)
print()
# ========================================
# 推論を実行
# ========================================
prompt = f"質問: {query} / コンテキスト: {context}"
print("【プロンプト】")
print(prompt)
print()
# LLMに推論を実行
response = client.converse(
modelId=llm,
messages=[{
"role": "user",
"content": [{"text": prompt}]
}]
)
# レスポンスを表示
print("【LLMの回答】")
print(response["output"]["message"]["content"][0]["text"])
CloudShell
pip install numpy faiss-cpu
python 2-1.py
2-2. LangChainでRAGを簡単に書いてみよう!
LLMアプリでよく使う処理を簡単に書ける「LangChain」を使います。
2-2.py
# 必要なPythonライブラリをインポート
from langchain_core.documents import Document
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_aws import ChatBedrockConverse, BedrockEmbeddings
from langchain_community.vectorstores import FAISS
# 生成AIモデルを設定
llm = ChatBedrockConverse(
model_id="us.anthropic.claude-sonnet-4-20250514-v1:0",
temperature=0
)
embeddings = BedrockEmbeddings(
model_id="cohere.embed-multilingual-v3"
)
# ========================================
# 社内文書を定義
# ========================================
documents = [
Document(
page_content="かぐたんはKAG社のSlackチャットボットです。",
metadata={"title": "かぐたん", "id": 0}
),
Document(
page_content="カグカグはKAG社のゆるキャラです。",
metadata={"title": "カグカグ", "id": 1}
)
]
# ========================================
# 社内文書をベクトルに変換
# ========================================
print("【ベクトルデータを準備】")
vectorstore = FAISS.from_documents(
documents=documents,
embedding=embeddings
)
print("ベクトルストアの構築が完了しました。")
print()
# ========================================
# 検索を実行
# ========================================
query = "KAG社のゆるキャラの名前は?"
# Retrieverを実行(上位1件を取得)
retriever = vectorstore.as_retriever(search_kwargs={"k": 1})
retrieved_docs = retriever.invoke(query)
context = retrieved_docs[0].page_content
print("【検索結果】")
print("クエリ: ", query)
print("結果:", retrieved_docs)
print()
# ========================================
# LLMの推論
# ========================================
# プロンプトテンプレートを定義
prompt = ChatPromptTemplate.from_template(
"質問: {question} / コンテキスト: {context}"
)
print("【プロンプト】")
print(f"質問: {query} / コンテキスト: {context}")
print()
# LCELを使ってRAGチェーンを構築
rag_chain = (
{"context": retriever, "question": RunnablePassthrough()}
| prompt
| llm
| StrOutputParser()
)
# RAGチェーンを実行
print("【LLMの回答】")
result = rag_chain.invoke(query)
print(result)
CloudShell
pip install langchain langchain-community langchain-aws
python 2-2.py
2-3. クラウド上にRAGアプリを構築してみよう!
AWSのマネージドサービス「Bedrockナレッジベース」を使います。
3-1. Function calling(Tool Use)を使ってみよう!
Converse APIのTool Use機能を使います。
3-1.py
# 必要なPythonライブラリをインポート
import boto3
import json
import urllib.request
# AWS SDK for Pythonで、Bedrock用のAPIクライアントを作成
client = boto3.client("bedrock-runtime")
input = "2025年6月の祝日はいつ?"
llm = "us.anthropic.claude-sonnet-4-20250514-v1:0"
# 祝日を取得する関数
def get_japanese_holidays(year):
"""指定された年の日本の祝日を取得する"""
url = f"https://holidays-jp.github.io/api/v1/{year}/date.json"
with urllib.request.urlopen(url) as response:
data = response.read()
holidays = json.loads(data)
return holidays
# 関数をLLMのツールとして定義
tools = [{
"toolSpec": {
"name": "get_japanese_holidays",
"description": "指定された年の日本の祝日一覧を取得します",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"year": {
"type": "integer",
"description": "祝日を取得したい年(例: 2024)"
}
},
"required": ["year"]
}
}
}
}]
# ========================================
# 1回目の推論
# ========================================
print("【推論1回目】")
print("ユーザーの入力: ", input)
response = client.converse(
modelId=llm,
messages=[{
"role": "user",
"content": [{"text": input}]
}],
toolConfig={"tools": tools}
)
# 初回のレスポンスを処理
message = response["output"]["message"]
print("LLMの回答: ", message["content"][0]["text"])
# Tool Useの要否をチェック
tool_use = None
for content_item in message["content"]:
if "toolUse" in content_item:
tool_use = content_item["toolUse"]
print("ツール要求: ", tool_use)
print()
break
# ========================================
# 2回目の推論
# ========================================
if tool_use:
# 実際のツール実行:APIを呼び出して祝日を取得
year = tool_use['input']['year']
holidays = get_japanese_holidays(year)
tool_result = {
"year": year,
"holidays": holidays,
"count": len(holidays)
}
print("【アプリから直接、ツール実行して結果を取得】")
print(tool_result)
print()
# 2回目の推論用に、1回目の会話履歴+ツール実行結果を入力
messages = [
{
"role": "user",
"content": [{"text": input}] # 1回目の質問
},
{
"role": "assistant",
"content": message["content"] # 1回目の回答
},
{
"role": "user",
"content": [{
"toolResult": {
"toolUseId": tool_use["toolUseId"],
"content": [{
"json": tool_result # ツール実行結果
}]
}
}]
}
]
# 2回目の回答
final_response = client.converse(
modelId=llm,
messages=messages,
toolConfig={"tools": tools}
)
print("【推論2回目】")
print("ユーザーの入力: (ツール実行結果)")
print("LLMの回答: ", final_response["output"]["message"]["content"][0]["text"])
3-2. AIエージェントを作ってみよう!
AIエージェントを簡単に書けるAWS製のOSS「Strands Agents SDK」を使います。
3-2.py
# 必要なPythonライブラリをインポート
import json
import urllib.request
from strands import Agent, tool
# Strandsのデコレーターでツールを定義
@tool
def get_holidays(year):
url = f"https://holidays-jp.github.io/api/v1/{year}/date.json"
with urllib.request.urlopen(url) as response:
data = response.read()
holidays = json.loads(data)
return holidays
# エージェントを作成
agent = Agent(
model="us.anthropic.claude-sonnet-4-20250514-v1:0",
system_prompt="あなたは日本の祝日を調べるプロフェッショナルです。",
tools=[get_holidays],
)
# エージェントを実行
agent("2025年6月の祝日はいつ?")
CloudShellのPythonが3.9と古くてStrandsが動かないので、アップグレードします。
CloudShell
# Python 3.12をインストール
sudo yum install python3.12
# シンボリックリンクを修正
sudo ln -sf /usr/bin/python3.12 /usr/bin/python3
# pipもアップグレード
sudo python -m ensurepip --upgrade
CloudShell
pip install strands-agents
python 3-2.py
3-3. Streamlitで簡単なフロントエンドを付けよう
Pythonライブラリ「Streamlit」を使います。
3-3.py
# 必要なPythonライブラリをインポート
import json
import asyncio
import urllib.request
import streamlit as st
from strands import Agent, tool
# フロントエンド
st.title("おしえて! Strandsエージェント")
question = st.text_input("質問を入力", "2025年6月の祝日はいつ?")
# Strandsのデコレーターでツールを定義
@tool
def get_holidays(year):
url = f"https://holidays-jp.github.io/api/v1/{year}/date.json"
with urllib.request.urlopen(url) as response:
data = response.read()
holidays = json.loads(data)
return holidays
# エージェントを作成
agent = Agent(
model="us.anthropic.claude-sonnet-4-20250514-v1:0",
system_prompt="あなたは日本の祝日を調べるプロフェッショナルです。",
tools=[get_holidays],
)
# 非同期ストリーミング処理
async def process_stream(question, container):
text_holder = container.empty()
response = ""
shown_tools = set()
# エージェントからのストリーミングレスポンスを処理
async for chunk in agent.stream_async(question):
if isinstance(chunk, dict):
event = chunk.get('event', {})
# ツール実行を検出して表示
if 'contentBlockStart' in event:
tool_use = event['contentBlockStart'].get('start', {}).get('toolUse', {})
tool_id = tool_use.get('toolUseId')
# バッファをクリア
if response:
text_holder.markdown(response)
response = ""
# ツール実行のメッセージを表示
container.info("ツールを実行中…")
text_holder = container.empty()
# テキストを抽出してリアルタイム表示
if text := chunk.get('data'):
response += text
text_holder.markdown(response)
# ボタンを押したら生成開始
if st.button("質問する"):
with st.spinner("回答を生成中…"):
container = st.container()
asyncio.run(process_stream(question, container))
CloudShell
# 1タブ目
streamlit run 3-3.py
# 2タブ目
ssh -p 443 -R0:localhost:8501 a.pinggy.io
3-4. クラウド上にAIエージェントを構築してみよう!
AWSのマネージドサービス「Bedrockエージェント」を使います。
Views: 0