LoginSignup
2
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@tetutetu214(てつてつ)

S3 Vectors を AWS CLIで作成ハンズオン

Posted at

1.はじめに

1.1.背景

2025/7/15、AWSから S3 Vectorsと呼ばれる機能がプレビューされました。
Amazon S3 Vectors

早速ですが AWS CLIを利用して、とりあえず動くものを作りながら学びを深めたいと思います。

1.2.構成

システム構成図

2.ハンズオン

2.1.前提

2.1.1.実行環境

環境 設定 バージョン
環境 AWS CloudShell 2.27.55

2.2.ベクトル S3作成

1.CloudShellアップデート - `2.27.51`よ`S3 Vectors`に対応しているとのことで、最新の CLIを取得しておく 
# aws cliバージョンアップ
curl "https://awscli.amazonaws.com/awscli-exe-linux-x86_64.zip" -o "awscliv2.zip"
unzip awscliv2.zip
sudo ./aws/install --update

# バージョン確認
aws --version

# S3 Vectorsテスト
aws s3vectors help
2.変数設定 
# 環境変数設定
export AWS_REGION="us-east-1"
export VECTOR_BUCKET_NAME="sangokushi-rag-bucket-$(date +%s)"
export VECTOR_INDEX_NAME="sangokushi-index"
3.ベクトル S3作成 ベクトル（数値の配列）を効率的に保存・検索するためのバケット。 ベクトル検索に最適化された専用ストレージで、中にベクトルインデックスを複数作成することが可能。 
# ベクトルS3作成
aws s3vectors create-vector-bucket --vector-bucket-name "${VECTOR_BUCKET_NAME}" --region "${AWS_REGION}"

# 構築確認
aws s3vectors list-vector-buckets --region us-east-1
4.変数設定 
# 環境変数設定
export VECTOR_BUCKET_NAME="sangokushi-rag-bucket-1752990275"
export VECTOR_INDEX_NAME="sangokushi-index"
5.ベクトルインデックス作成 実際のベクトルデータを格納し、類似検索を実行する。 次元数（1024）や距離計算方法（コサイン）を定義し、ここにベクトル化したデータを投入する。 
# ベクトルインデックス作成
aws s3vectors create-index \
    --vector-bucket-name "${VECTOR_BUCKET_NAME}" \
    --index-name "${VECTOR_INDEX_NAME}" \
    --data-type "float32" \
    --dimension 1024 \
    --distance-metric "cosine" \
    --metadata-configuration '{"nonFilterableMetadataKeys":["text"]}' \
    --region us-east-1

2.2.データ配置スクリプト 作成

今回のサンプルデータとして Wikipediaから要約された三国志関連での情報を取得する

1.keywordsテキストの作成 
cat > keywords.txt << 'EOF'
# 三国志関連人物（#はコメント行）
劉備
曹操
孫権
諸葛亮
関羽
張飛
趙雲
司馬懿
周瑜
呂布
EOF
2.メインスクリプトの作成

Wikipedia REST APIを利用してWikipediaの要約ページ内容を取得
取得した内容 右記項目(。！？、・\s（）()「」『』【】])で分割。

# 読点分割版スクリプト作成
cat > wikipedia_data_loader_detailed.py << 'EOF'
#!/usr/bin/env python3
"""
Wikipedia データ収集・詳細分割・ベクトル化・S3 Vectors投入ツール
使用方法: python3 wikipedia_data_loader_detailed.py
"""
import boto3
import json
import os
import uuid
import re
import requests
import time
from typing import List, Dict

# 環境変数確認
VECTOR_BUCKET_NAME = os.environ.get('VECTOR_BUCKET_NAME')
VECTOR_INDEX_NAME = os.environ.get('VECTOR_INDEX_NAME')
AWS_REGION = os.environ.get('AWS_REGION', 'us-east-1')

KEYWORDS_FILE = 'keywords.txt'

def load_keywords() -> List[str]:
    """keywords.txtからキーワード読み込み"""
    keywords = []
    with open(KEYWORDS_FILE, 'r', encoding='utf-8') as f:
        for line in f:
            line = line.strip()
            if line and not line.startswith('#'):
                keywords.append(line)
    
    print(f"[INFO] {len(keywords)}個のキーワードを読み込み")
    return keywords

def get_wikipedia_summary(keyword: str) -> Dict:
    """Summary APIでWikipedia記事取得"""
    url = f"https://ja.wikipedia.org/api/rest_v1/page/summary/{keyword}"
    
    try:
        response = requests.get(url, timeout=10)
        
        if response.status_code == 404:
            print(f"[WARN] {keyword}: 記事が見つかりません")
            return None
            
        if response.status_code != 200:
            print(f"[ERROR] {keyword}: HTTPエラー {response.status_code}")
            return None
            
        data = response.json()
        extract = data.get('extract', '')
        
        if len(extract) < 20:
            print(f"[WARN] {keyword}: 記事が短すぎるためスキップ ({len(extract)}文字)")
            return None
            
        print(f"[SUCCESS] {keyword}: {len(extract)}文字の要約を取得")
        return {
            'title': data.get('title', keyword),
            'text': extract,
            'url': data.get('content_urls', {}).get('desktop', {}).get('page', ''),
            'description': data.get('description', '')
        }
        
    except Exception as e:
        print(f"[ERROR] {keyword}: 取得エラー {e}")
        return None

def split_into_detailed_chunks(article: Dict) -> List[Dict]:
    """詳細分割でチャンク作成"""
    text = article['text']
    title = article['title']
    
    # より細かい分割（読点、中黒、括弧なども含む）
    parts = re.split(r'[。！？、・\s（）\(\)「」『』【】]', text)
    
    chunks = []
    current_chunk = ""
    chunk_id = 1
    
    for part in parts:
        part = part.strip()
        if not part:
            continue
            
        # 短すぎる部分は結合
        if len(part) < 3:
            current_chunk += part
        else:
            # 現在のチャンクを保存
            if current_chunk and len(current_chunk) >= 5:
                chunks.append({
                    'title': title,
                    'text': current_chunk.strip(),
                    'url': article['url'],
                    'description': article['description'],
                    'chunk_id': f"{title}-detailed-{chunk_id}"
                })
                chunk_id += 1
            
            # 新しいチャンク開始
            current_chunk = part
    
    # 最後のチャンク
    if current_chunk and len(current_chunk) >= 5:
        chunks.append({
            'title': title,
            'text': current_chunk.strip(),
            'url': article['url'],
            'description': article['description'],
            'chunk_id': f"{title}-detailed-{chunk_id}"
        })
    
    return chunks

def generate_embeddings(chunks: List[Dict]) -> List[List[float]]:
    """チャンクをベクトル化"""
    bedrock = boto3.client('bedrock-runtime', region_name=AWS_REGION)
    embeddings = []
    
    print(f"[INFO] {len(chunks)}個のチャンクをベクトル化中...")
    
    for i, chunk in enumerate(chunks, 1):
        try:
            body = json.dumps({"inputText": chunk['text']})
            response = bedrock.invoke_model(
                modelId='amazon.titan-embed-text-v2:0',
                body=body
            )
            
            result = json.loads(response['body'].read())
            embedding = result['embedding']
            embeddings.append(embedding)
            
            if i % 5 == 0:
                print(f"[INFO] 進捗: {i}/{len(chunks)}")
                
        except Exception as e:
            print(f"[ERROR] ベクトル化エラー ({i}/{len(chunks)}): {e}")
            embeddings.append(None)
            continue
    
    valid_embeddings = [emb for emb in embeddings if emb is not None]
    print(f"[SUCCESS] ベクトル化完了: {len(valid_embeddings)}個")
    return valid_embeddings

def upload_to_s3vectors(chunks: List[Dict], embeddings: List[List[float]]) -> bool:
    """S3 Vectorsにアップロード"""
    s3vectors = boto3.client('s3vectors', region_name=AWS_REGION)
    
    vectors = []
    for chunk, embedding in zip(chunks, embeddings):
        if embedding is None:
            continue
            
        vector = {
            'key': str(uuid.uuid4()),
            'data': {'float32': embedding},
            'metadata': {
                'text': chunk['text'],
                'title': chunk['title'],
                'source': 'wikipedia_detailed_chunked',
                'url': chunk.get('url', ''),
                'description': chunk.get('description', ''),
                'chunk_id': chunk.get('chunk_id', '')
            }
        }
        vectors.append(vector)
    
    print(f"[INFO] {len(vectors)}個のベクトルをS3 Vectorsにアップロード中...")
    
    try:
        response = s3vectors.put_vectors(
            vectorBucketName=VECTOR_BUCKET_NAME,
            indexName=VECTOR_INDEX_NAME,
            vectors=vectors
        )
        print("[SUCCESS] アップロード完了!")
        return True
        
    except Exception as e:
        print(f"[ERROR] アップロードエラー: {e}")
        return False

def main():
    print("=== Wikipedia 詳細分割データ収集・投入ツール ===")
    print(f"バケット: {VECTOR_BUCKET_NAME}")
    print(f"インデックス: {VECTOR_INDEX_NAME}")
    print()
    
    # 1. キーワード読み込み
    keywords = load_keywords()
    
    # 2. Wikipedia要約取得
    print(f"\n[INFO] Wikipedia要約取得開始...")
    articles = []
    
    for keyword in keywords:
        article = get_wikipedia_summary(keyword)
        if article:
            articles.append(article)
    
    if not articles:
        print("[ERROR] 取得できた記事がありません")
        exit(1)
        
    print(f"[SUCCESS] {len(articles)}個の要約を取得完了")
    
    # 3. 詳細分割
    print(f"\n[INFO] 詳細分割開始...")
    all_chunks = []
    
    for article in articles:
        chunks = split_into_detailed_chunks(article)
        all_chunks.extend(chunks)
        print(f"[INFO] {article['title']}: {len(chunks)}個のチャンクに分割")
    
    print(f"[SUCCESS] 総チャンク数: {len(all_chunks)}個")
    
    # 4. ベクトル化
    print(f"\n[INFO] ベクトル化開始")
    embeddings = generate_embeddings(all_chunks)
    
    # 5. S3 Vectorsアップロード
    print(f"\n[INFO] S3 Vectorsアップロード開始")
    
    valid_chunks = []
    valid_embeddings = []
    for chunk, embedding in zip(all_chunks, embeddings):
        if embedding is not None:
            valid_chunks.append(chunk)
            valid_embeddings.append(embedding)
    
    if valid_chunks and valid_embeddings:
        success = upload_to_s3vectors(valid_chunks, valid_embeddings)
        if success:
            print(f"\n[SUCCESS] 完了! {len(valid_embeddings)}個の詳細チャンクベクトルを投入しました")
        else:
            print(f"\n[ERROR] アップロード失敗")
    else:
        print(f"\n[ERROR] アップロードできるデータがありません")

if __name__ == "__main__":
    main()
EOF
3.上記スクリプトの実行画面 
# スクリプト実行
python wikipedia_data_loader_detailed.py

# レスポンス
=== Wikipedia 詳細分割データ収集・投入ツール ===
バケット: sangokushi-rag-bucket-1752990275
インデックス: sangokushi-index

[INFO] 10個のキーワードを読み込み

[INFO] Wikipedia要約取得開始...
[SUCCESS] 劉備: 34文字の要約を取得
[SUCCESS] 曹操: 79文字の要約を取得
[SUCCESS] 孫権: 32文字の要約を取得
[SUCCESS] 諸葛亮: 44文字の要約を取得
[SUCCESS] 関羽: 85文字の要約を取得
[SUCCESS] 張飛: 123文字の要約を取得
[SUCCESS] 趙雲: 67文字の要約を取得
[SUCCESS] 司馬懿: 150文字の要約を取得
[SUCCESS] 周瑜: 111文字の要約を取得
[SUCCESS] 呂布: 87文字の要約を取得
[SUCCESS] 10個の要約を取得完了

[INFO] 詳細分割開始...
[INFO] 劉備: 2個のチャンクに分割
[INFO] 曹操: 7個のチャンクに分割
[INFO] 孫権: 3個のチャンクに分割
[INFO] 諸葛亮: 2個のチャンクに分割
[INFO] 関羽: 7個のチャンクに分割
[INFO] 張飛: 9個のチャンクに分割
[INFO] 趙雲: 4個のチャンクに分割
[INFO] 司馬懿: 9個のチャンクに分割
[INFO] 周瑜: 10個のチャンクに分割
[INFO] 呂布: 6個のチャンクに分割
[SUCCESS] 総チャンク数: 59個

[INFO] ベクトル化開始
[INFO] 59個のチャンクをベクトル化中...
[INFO] 進捗: 5/59
[INFO] 進捗: 10/59
[INFO] 進捗: 15/59
[INFO] 進捗: 20/59
[INFO] 進捗: 25/59
[INFO] 進捗: 30/59
[INFO] 進捗: 35/59
[INFO] 進捗: 40/59
[INFO] 進捗: 45/59
[INFO] 進捗: 50/59
[INFO] 進捗: 55/59
[SUCCESS] ベクトル化完了: 59個

[INFO] S3 Vectorsアップロード開始
[INFO] 59個のベクトルをS3 Vectorsにアップロード中...
[SUCCESS] アップロード完了!

[SUCCESS] 完了! 59個の詳細チャンクベクトルを投入しました
4.ベクトルデータ確認 
# 投入されたベクトル確認
aws s3vectors list-vectors --vector-bucket-name "${VECTOR_BUCKET_NAME}" --index-name "${VECTOR_INDEX_NAME}" --region us-east-1 --max-results 10

# レスポンス
{
    "nextToken": "2GDD6KniBdXLB-J6DuvcP32-nZOuVCcwPJ8ZcmqgcDdySyTGBEXaPdsYOAOTSaTd__tfK4aQheBO-LzL4MwllGp0yGtq3FdSkRsLPv8x5dbROh9gQDH3_qzrYX1R_g",
    "vectors": [
        {
            "key": "a969e08f-1eda-4afc-802f-6cf166ef8efe"
        },
    ...以下省略(同様の Keyが10個表示される)

2.3.データ検索スクリプト 作成

RAGに質問(質問内容をベクトル化)する スクリプトを作成

1.検索スクリプト作成 
# 検索スクリプト作成
cat > query_sangokushi.py << 'EOF'
#!/usr/bin/env python3
"""
三国志RAGシステム クエリツール
使用方法: python3 query_sangokushi.py "劉備について教えて"
"""
import boto3
import json
import os
import sys

# 環境変数
VECTOR_BUCKET_NAME = os.environ.get('VECTOR_BUCKET_NAME')
VECTOR_INDEX_NAME = os.environ.get('VECTOR_INDEX_NAME')
AWS_REGION = os.environ.get('AWS_REGION', 'us-east-1')

def embed_question(question: str):
    """質問をベクトル化"""
    bedrock = boto3.client('bedrock-runtime', region_name=AWS_REGION)
    
    body = json.dumps({"inputText": question})
    response = bedrock.invoke_model(
        modelId='amazon.titan-embed-text-v2:0',
        body=body
    )
    
    result = json.loads(response['body'].read())
    return result['embedding']

def search_vectors(query_vector, top_k=3):
    """ベクトル検索"""
    s3vectors = boto3.client('s3vectors', region_name=AWS_REGION)
    
    response = s3vectors.query_vectors(
        vectorBucketName=VECTOR_BUCKET_NAME,
        indexName=VECTOR_INDEX_NAME,
        queryVector={'float32': query_vector},
        topK=top_k,
        returnMetadata=True,
        returnDistance=True
    )
    
    return response['vectors']

def main():
    if len(sys.argv) < 2:
        print("使用方法: python3 query_sangokushi.py '質問文'")
        print("例: python3 query_sangokushi.py '劉備について教えて'")
        sys.exit(1)
    
    question = sys.argv[1]
    
    print(f"[INFO] 質問: {question}")
    print(f"[INFO] ベクトル化中...")
    query_vector = embed_question(question)
    
    print(f"[INFO] 検索中...")
    results = search_vectors(query_vector)
    
    print(f"\n=== 検索結果 ===")
    for i, result in enumerate(results, 1):
        metadata = result.get('metadata', {})
        distance = result.get('distance', 0)
        
        print(f"\n{i}. タイトル: {metadata.get('title', 'N/A')}")
        print(f"   類似度: {1-distance:.3f}")
        print(f"   内容: {metadata.get('text', 'N/A')}")
        print(f"   URL: {metadata.get('url', 'N/A')}")

if __name__ == "__main__":
    main()
EOF

3.検索実行

3.1.検索実行

1.検索実行 
# 検索する質問
python query_sangokushi.py "蜀の武将は？"

=== 検索結果 ===

1. タイトル: 張飛
   類似度: 0.663
   内容: 中国後漢末期から三国時代の蜀の将軍
   URL: https://ja.wikipedia.org/wiki/%E5%BC%B5%E9%A3%9B

2. タイトル: 趙雲
   類似度: 0.603
   内容: 中国後漢末期から三国時代にかけての蜀漢の武将
   URL: https://ja.wikipedia.org/wiki/%E8%B6%99%E9%9B%B2

3. タイトル: 諸葛亮
   類似度: 0.524
   内容: 中国後漢末期から三国時代の蜀漢の政治家武将軍師
   URL: https://ja.wikipedia.org/wiki/%E8%AB%B8%E8%91%9B%E4%BA%AE

3.2.検索改善

分割数を分ければ分けるほど、検索類似度も上昇が見込める。
※ MeCabなどの品詞によるベクトル分割までは未実施。単純な分割でのベクトル化を実施。

項番 チャンク分け方 分割数(10偉人を挿入しての分割された数) 検索類似度(上位3つの平均)
1 無分割 10分割 ≒0.39
2 [。! ？、] 21分割 ≒0.43
3 [。！？、・\s（）()「」『』【】] 59分割 ≒0.60

4.リソース削除

リソース削除 
# ベクトルインデックス削除
aws s3vectors delete-index \
    --vector-bucket-name "${VECTOR_BUCKET_NAME}" \
    --index-name "${VECTOR_INDEX_NAME}" \
    --region us-east-1

# ベクトルバケット削除
aws s3vectors delete-vector-bucket \
    --vector-bucket-name "${VECTOR_BUCKET_NAME}" \
    --region us-east-1

5.おわりに

5.1.得られた知見

  • S3 VectorsDB作成方法

5.2.今後の課題

  • 品詞による詳細なベクトル分割の実装
  • ベクトルDBより取得できた内容を集約して返信するLLMの拡張
2
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
2
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?