Amazon S3 Vectors にベクトルを投入して Bedrock で類似検索してみた

犬     → [0.21, 0.88, -0.12, ..., 0.07]
猫     → [0.23, 0.85, -0.10, ..., 0.05]   ← 犬と数値が近い(意味が近い)
自動車 → [-0.71, 0.02, 0.45, ..., 0.33]   ← 犬とは遠い(意味が遠い)

python3 -m venv .venv
source .venv/bin/activate
pip install -U boto3

import json
import boto3
from botocore.exceptions import ClientError

# 設定値
REGION = "ap-northeast-1"                    # S3 Vectors / Bedrock を利用するリージョン
BUCKET = "devio-s3-vectors-demo"             # 事前に作成しておくベクトルバケット名
INDEX = "devio-index"                        # 事前に作成しておくベクトルインデックス名
MODEL_ID = "amazon.titan-embed-text-v2:0"    # 使用する埋め込みモデル(1024 次元)

# AWS クライアント
bedrock = boto3.client("bedrock-runtime", region_name=REGION)   # 埋め込み生成用
s3v = boto3.client("s3vectors", region_name=REGION)             # ベクトル PUT 用

def preflight():
    # このリージョンにあるベクトルバケット一覧を取得
    try:
        buckets = s3v.list_vector_buckets().get("vectorBuckets", [])
    except ClientError as e:
        print(f"[preflight] list_vector_buckets failed: {e}")
        return False

    names = [b.get("vectorBucketName") for b in buckets]
    print(f"[preflight] buckets in {REGION}: {names}")
    if BUCKET not in names:
        # 指定バケットが見つからない(REGION 違い or 名前ミス)
        print(f"[preflight] bucket '{BUCKET}' が見つかりません。REGION か BUCKET を見直してください。")
        return False

    # バケット配下のインデックス一覧を取得
    try:
        indexes = s3v.list_indexes(vectorBucketName=BUCKET).get("indexes", [])
    except ClientError as e:
        print(f"[preflight] list_indexes failed: {e}")
        return False
    # インデックス名の一覧を抽出して表示
    index_names = [i.get("indexName") for i in indexes]
    print(f"[preflight] indexes in '{BUCKET}': {index_names}")
    if INDEX not in index_names:
        # 指定インデックスが未作成
        print(f"[preflight] index '{INDEX}' が見つかりません。INDEX 名を見直すか、まず作成してください。")
        return False

    print("[preflight] bucket / index の存在を確認")
    return True

# 投入対象のドキュメント。key はベクトルの一意識別子。
docs = [
    {"id": "doc-1", "text": "Amazon S3 Vectorsはベクトル検索をS3ネイティブに提供する新機能です。"},
    {"id": "doc-2", "text": "Amazon Bedrockは生成AIモデルをAPIで簡単に呼び出せるサービスです。"},
    {"id": "doc-3", "text": "Amazon OpenSearch Serviceは全文検索やベクトル検索が可能なマネージドサービスです。"},
    {"id": "doc-4", "text": "AWS Lambdaはサーバーレスでコードを実行できるコンピューティングサービスです。"},
]

# 埋め込み生成
def embed(text: str) -> list[float]:
    """
    Bedrock Titan Embeddings V2 でテキストを 1024 次元の float ベクトルに変換する。
    normalize=True にしておくとコサイン類似度の計算に有利。
    """
    body = json.dumps({"inputText": text, "dimensions": 1024, "normalize": True})
    res = bedrock.invoke_model(modelId=MODEL_ID, body=body)
    return json.loads(res["body"].read())["embedding"]

def main():
    # 事前チェック(失敗したら即終了)
    if not preflight():
        raise SystemExit(1)

    # 各ドキュメントを埋め込みベクトル化して PUT 用ペイロードに詰め替え
    vectors = []
    for d in docs:
        vectors.append({
            "key": d["id"],                                # ベクトルの一意 ID(あとで更新・削除で指定)
            "data": {"float32": embed(d["text"])},         # ベクトル本体(1024 次元の float 配列)
            "metadata": {                                  # 検索時のフィルタや、結果への付帯情報
                "source_text": d["text"],
                "category": "aws-service",
            },
        })

    # S3 Vectors にまとめて投入(同じ key で PUT すれば upsert 動作)
    s3v.put_vectors(
        vectorBucketName=BUCKET,
        indexName=INDEX,
        vectors=vectors,
    )
    print(f"PUT {len(vectors)} vectors")

if __name__ == "__main__":
    main()

python put_vectors.py

(.venv) developers-io %python put_vectors.py
[preflight] buckets in ap-northeast-1: ['devio-s3-vectors-demo']
[preflight] indexes in 'devio-s3-vectors-demo': ['devio-index']
[preflight] bucket / index の存在を確認
PUT 4 vectors

import boto3
s3v = boto3.client("s3vectors", region_name="ap-northeast-1")
res = s3v.list_vectors(
    vectorBucketName="devio-s3-vectors-demo",
    indexName="devio-index",
    returnMetadata=True,
)
for v in res["vectors"]:
    print(f"- key={v['key']} text={v['metadata']['source_text']}")

python list.py

- key=doc-4  text=AWS Lambdaはサーバーレスでコードを実行できるコンピューティングサービスです。
- key=doc-3  text=Amazon OpenSearch Serviceは全文検索やベクトル検索が可能なマネージドサービスです。
- key=doc-2  text=Amazon Bedrockは生成AIモデルをAPIで簡単に呼び出せるサービスです。
- key=doc-1  text=Amazon S3 Vectorsはベクトル検索をS3ネイティブに提供する新機能です。

import json
import boto3

REGION = "ap-northeast-1"
BUCKET = "devio-s3-vectors-demo"
INDEX = "devio-index"
MODEL_ID = "amazon.titan-embed-text-v2:0"

bedrock = boto3.client("bedrock-runtime", region_name=REGION)
s3v = boto3.client("s3vectors", region_name=REGION)

def embed(text: str) -> list[float]:
    body = json.dumps({"inputText": text, "dimensions": 1024, "normalize": True})
    res = bedrock.invoke_model(modelId=MODEL_ID, body=body)
    return json.loads(res["body"].read())["embedding"]

# ① 質問文を用意 → ② ベクトル化
query_text = "S3にベクトルを直接保存できるサービスは？"
query_vec = embed(query_text)

# ③ S3 Vectors に「近いものを top 3 ちょうだい」
res = s3v.query_vectors(
    vectorBucketName=BUCKET,
    indexName=INDEX,
    queryVector={"float32": query_vec},
    topK=3,
    returnMetadata=True,
    returnDistance=True,
)

# ④ 結果を表示
for r in res["vectors"]:
    print(f"- key={r['key']} distance={r['distance']:.4f} text={r['metadata']['source_text']}")

(.venv) developers-io %python query_vectors.py
- key=doc-1 distance=0.5082 text=Amazon S3 Vectorsはベクトル検索をS3ネイティブに提供する新機能です。
- key=doc-2 distance=0.7358 text=Amazon Bedrockは生成AIモデルをAPIで簡単に呼び出せるサービスです。
- key=doc-3 distance=0.7530 text=Amazon OpenSearch Serviceは全文検索やベクトル検索が可能なマネージドサービスです。

import json
import boto3

# 設定値(put_vectors.py と揃える)
REGION = "ap-northeast-1"
BUCKET = "devio-s3-vectors-demo"
INDEX = "devio-index"
MODEL_ID = "amazon.titan-embed-text-v2:0"

# AWS クライアント
bedrock = boto3.client("bedrock-runtime", region_name=REGION)   # 埋め込み生成用
s3v = boto3.client("s3vectors", region_name=REGION)             # ベクトル PUT 用

# 埋め込み生成関数
def embed(text: str) -> list[float]:
    """
    Bedrock Titan Embeddings V2 でテキストを 1024 次元の float ベクトルに変換する。
    put_vectors.py の embed() と同じ設定(次元数・normalize)にすること。
    """
    body = json.dumps({"inputText": text, "dimensions": 1024, "normalize": True})
    res = bedrock.invoke_model(modelId=MODEL_ID, body=body)
    return json.loads(res["body"].read())["embedding"]

# 追加投入するドキュメント(category = "database")
# メタデータで category を切ることで、後の Query 時に $eq: "database" などで絞り込める。
extra = [
    {"id": "doc-5", "text": "Amazon RDS for PostgreSQLはリレーショナルデータベースのマネージドサービスです。", "category": "database"},
    {"id": "doc-6", "text": "Amazon DynamoDBはNoSQLのマネージドキーバリューデータベースです。",           "category": "database"},
]

# 各ドキュメントを埋め込みベクトル化して PUT 用ペイロードに詰め替え
vectors = [
    {
        "key": d["id"],                                    # ベクトルの一意 ID(同じ key で PUT すると upsert)
        "data": {"float32": embed(d["text"])},             # ベクトル本体(1024 次元)
        "metadata": {                                      # 検索時のフィルタ・付帯情報
            "source_text": d["text"],
            "category": d["category"],
        },
    }
    for d in extra
]

# S3 Vectors にまとめて投入
s3v.put_vectors(vectorBucketName=BUCKET, indexName=INDEX, vectors=vectors)
print(f"PUT {len(vectors)} extra docs")

(.venv) developers-io %python put_extra_vectors.py
PUT 2 extra docs

import json
import boto3

# 設定値
REGION = "ap-northeast-1"
BUCKET = "devio-s3-vectors-demo"
INDEX = "devio-index"
MODEL_ID = "amazon.titan-embed-text-v2:0"

# AWS クライアント
bedrock = boto3.client("bedrock-runtime", region_name=REGION)   # クエリ文の埋め込み生成用
s3v = boto3.client("s3vectors", region_name=REGION)             # ベクトル検索用

# 埋め込み生成関数
def embed(text: str) -> list[float]:
    """
    Bedrock Titan Embeddings V2 でテキストを 1024 次元の float ベクトルに変換。
    PUT 時と同じモデル・同じ次元数・同じ normalize 設定でベクトル化しないと比較が成り立たない。
    """
    body = json.dumps({"inputText": text, "dimensions": 1024, "normalize": True})
    res = bedrock.invoke_model(modelId=MODEL_ID, body=body)
    return json.loads(res["body"].read())["embedding"]

# クエリ文をベクトル化(2 回のクエリで使い回す)
qv = embed("データを保存するサービスは？")

# ===== ① フィルタなし =====
# 意味の近い順にそのまま上位 5 件を取得する。
# category に関係なく「意味の近さ」だけで並ぶことを確認する。
print("=== フィルタなし ===")
res = s3v.query_vectors(
    vectorBucketName=BUCKET,
    indexName=INDEX,
    queryVector={"float32": qv},   # 「これに近いものを探して」の元ベクトル
    topK=5,                        # 上位何件返すか
    returnMetadata=True,           # PUT 時に付けた metadata も返してもらう
    returnDistance=True,           # スコア(距離)も返してもらう
)
for r in res["vectors"]:
    # distance は小さいほど意味が近い(コサインの場合)
    print(f"  {r['key']:6s} d={r['distance']:.4f} cat={r['metadata']['category']:12s} {r['metadata']['source_text']}")

# ===== ② filter で category = 'database' に絞り込み =====
# メタデータで属性検索を効かせる。ベクトル検索(意味の近さ)と
# 属性フィルタ(完全一致)を 1 回のクエリで同時に効かせられる。
print("\n=== filter: category = 'database' ===")
res = s3v.query_vectors(
    vectorBucketName=BUCKET,
    indexName=INDEX,
    queryVector={"float32": qv},
    topK=5,
    filter={"category": {"$eq": "database"}},   # ← ここでカテゴリ絞り込み($eq は MongoDB 風の演算子)
    returnMetadata=True,
    returnDistance=True,
)
for r in res["vectors"]:
    print(f"  {r['key']:6s} d={r['distance']:.4f} cat={r['metadata']['category']:12s} {r['metadata']['source_text']}")

python query_with_filter.py

(.venv) developers-io %python query_with_filter.py
=== フィルタなし ===
  doc-5  d=0.7092 cat=database     Amazon RDS for PostgreSQLはリレーショナルデータベースのマネージドサービスです。
  doc-6  d=0.7481 cat=database     Amazon DynamoDBはNoSQLのマネージドキーバリューデータベースです。
  doc-3  d=0.7551 cat=aws-service  Amazon OpenSearch Serviceは全文検索やベクトル検索が可能なマネージドサービスです。
  doc-4  d=0.7899 cat=aws-service  AWS Lambdaはサーバーレスでコードを実行できるコンピューティングサービスです。
  doc-1  d=0.8058 cat=aws-service  Amazon S3 Vectorsはベクトル検索をS3ネイティブに提供する新機能です。

=== filter: category = 'database' ===
  doc-5  d=0.7092 cat=database     Amazon RDS for PostgreSQLはリレーショナルデータベースのマネージドサービスです。
  doc-6  d=0.7481 cat=database     Amazon DynamoDBはNoSQLのマネージドキーバリューデータベースです。