【新人研修】LLMアプリ開発入門（ハンズオン編）

KDDIアジャイル開発センター（KAG）の2025年度新人研修コンテンツの一部です。

このハンズオン編の前に、ホワイトボーディング中心の座学編があります。

環境準備

• AWSアカウントに、自分のIAMユーザーでサインイン
• リージョンをオレゴンに切り替え（以降、すべてオレゴンで作業します）
• Bedrockコンソールへ移動し、すべてのモデルを有効化
• CloudShellを起動

できる人は、自分のMacでVS Codeを使って作業実施してもOK。
事前に aws configure でIAM認証情報を設定しておくこと。

1. LLMのAPI入門

1-1. LLMのAPIを叩いてみよう！

Amazon BedrockのConverse APIを使います。

CloudShell

nano 1-1.py

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"])

CloudShell

python 1-1.py

1-2. ストリーミング出力させてみよう！

ConverseSetream APIを使います。

CloudShell

nano 1-2.py

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()  # 最後に改行

CloudShell

python 1-2.py

1-3. 簡単なフロントエンドを付けよう

Pythonライブラリ「Streamlit」を使います。

CloudShell

nano 1-3.py

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. RAG（Retrieval-Augmented Generation）入門

2-1. RAGアーキテクチャを作ってみよう！

Meta社のベクトルインデックス作成OSS「Faiss」を使います。

CloudShell

nano 2-1.py

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」を使います。

CloudShell

nano 2-2.py

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. AIエージェント入門

3-1. Function calling（Tool Use）を使ってみよう！

Converse APIのTool Use機能を使います。

CloudShell

nano 3-1.py

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"])

CloudShell

python 3-1.py

3-2. AIエージェントを作ってみよう！

AIエージェントを簡単に書けるAWS製のOSS「Strands Agents SDK」を使います。

CloudShell

nano 3-2.py

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

この辺でCloudShellのストレージ溢れエラーになる方も多いと思います。
そのときは環境を削除して、再度シェルを開いてリトライください。

3-3. Streamlitで簡単なフロントエンドを付けよう

Pythonライブラリ「Streamlit」を使います。

CloudShell

nano 3-3.py

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エージェント」を使います。

使用したコード