この記事について

Amazon S3 Vectorsは、ベクトルデータを保管・検索するためのS3のバケットを作成することで、ベクトル検索ができるサービスです。

この記事では、S3 VectorsをAPIから利用する手順を中心に解説します
また、ベクトルデータベースに比べて何が嬉しいのかを解説します
記事の最後に、Natureの論文に対してRAGを実施、S3 Vectorsを使うことでRAGの回答が大幅に良くなることを確認します

前提: ベクトル検索って何です？

ベクトル検索は意味が似ているものを検索します。

元々、AWSでベクトル検索をするためのサービスには、Kendra、OpenSearch（ElasticSearch）などがあります。

S3 Vectorsの良いところその1: 値段が安い

比較的安いと言われるKendra Gen AIのAPIの400エラーが見たかったので、Kendraを15分だけ立ち上げました。データ登録もせず、たった2回だけAPIリクエストを投げました。
何もせずに、15分動かすだけで請求書が届きます。

Kendraをこのまま1カ月動かすと、月230ドルかかります。
OpenSearch Serverlessはもう少し高く、最低料金で運用しても月350ドルかかります。

一方、S3 Vectorsも検証のために1週間ほど使いました。

こちらは料金が安すぎて請求書に金額が反映されませんでした。

もちろん完全な無料ではなく、公式の料金モデルケースでは「月11ドルかかるよ」と書いてあるのですが、月11ドルのモデルケースをよく読むと1000万件のデータがある環境で、月100万回の検索をかけたときの料金だそうです。

それだけ使ってもKendraのおよそ1/20の料金です。
個人や小規模で使う分には無料だと考えていいくらいの値段です。

S3 Vectorsの良いところその2: 良くない縛りがない

これまでのAWSのベクトルデータベースは、以下のような構造になっていました。

OpenSearch Serverlessを図にしていますが、Kendraも同じ構成です。

データの変換処理、データを保管する場所、どちらもベクトルデータベースの中に入っています。
利用するプログラムやユーザーは、検索対象のデータを渡して、そのデータに似た検索結果（チャンク）を受け取ります。

3つほど良くない縛りがありました。

良くない縛りその1. 検索結果がチャンクになる

チャンクは数百～数千文字程度に短く区切ったデータのことです。

ベクトル検索は、チャンクの意味がどれくらい似ているのかを調べるための検索処理です。埋め込みLLMとベクトル検索の制約で、文章はチャンクに短く区切る必要があります。細切れにしたチャンクを検索結果として返すのはベクトル検索の都合です。

昔のLLM、それこそGPT-3.5のコンテキストサイズは4Kでしたから、RAGで返ってくるデータはチャンク程度の小さなデータである必要があり、ベクトル検索の都合とLLMの都合が一致していました。

一方で、Claude Sonnet 4のコンテキストサイズはその頃の50倍あります。今のLLMにとっては、ベクトルデータベースはファイル全体や章全体をそのまま返してくれれば処理できます。低い検索精度のせいで無関係なデータが混じっていたとしても、LLMが精査できます。

検索結果として、わざわざ細切れにしたチャンクを返す必要がありません。

良くない縛りその2. 使わなくても塩漬けにできない

時間課金のデータベースにデータを持っているので、使わないデータベースをとりあえず料金をかけずに持っておく、たまに使う、ということができません。

良くない縛りその3. ライブラリの恩恵を受けられない

今のベクトル化はとても簡単で、下のコードで実現できます。
litellmですから、どのベンダーの埋め込みLLMでも同じソースです。

import litellm

# Titan Embeddingの実行関数を定義する
# ベクトル化する。引数にモデル名と次元数を書く
result = litellm.embedding(
    model="bedrock/amazon.titan-embed-text-v2:0",
    dimensions=1024, # 次元数。小さいほど料金と精度が下がる
    input=["ベクトル化したい文字列"],
)

# 戻り値はpydanticのオブジェクト
# 少し変換するだけで、S3 Vectorsで検索や登録ができる
print(result.data)

さらにRAGがしたいのなら、Mirascopeで以下のように書きます。

from mirascope import llm, prompt_template

@llm.call( # モデルを指定する
    provider="bedrock",
    model="amazon.nova-micro-v1:0",
)
@prompt_template( # RAGのプロンプトを書く
    """
    SYSTEM: excerptsの情報を元に、ユーザーからの質問に回答してください
    <excerpts>{excerpts}</excerpts>

    USER: {query}
    """
)
def execute_rag(query: str):
    # ベクトル検索を実行する
    search_result = {ベクトル検索の検索結果}
    # プロンプトに検索結果を連携する
    return {
        "computed_fields": {
            "excerpts": search_result["vectors"]
        }
    }

このexecute_rag関数を呼ぶだけでRAGができます。

ライブラリが発達していない頃は、LangChainを使って長々とした処理を書く必要がありました。LLMも今ほどの精度がなく、無関係な情報が混じるとそれに振り回されました。

ベクトルデータベースで、ベクトルを意識せずにベクトル検索ができること、簡単にベクトル検索の精度が出せることは大きなメリットでした。

今はライブラリの恩恵で簡単に実装ができますし、多少関係のない情報が混じってもLLMが上手いことやってくれます。

ベクトルデータベースのために独特なクエリ構文を覚えて、ライブラリの恩恵を受けずに実装をするメリットは昔ほど大きくありません。

S3 Vectorsでどうなったのか

S3 Vectorsでは、上であげた3つの縛りを解消しました。

検索結果としてチャンクを返さない
S3 Vectorsの中にデータやチャンクを保管しない
時間課金ではなく、クエリへの課金にする
ライブラリの恩恵を受けられるよう、全て呼び出し側で処理する

S3 Vectorsは、チャンクや元データを保管しません。
S3 Vectorsは、チャンクを変換したベクトルデータと、そのベクトルに紐づけたJSONデータを検索結果として返すため、実装側でJSONと元データの紐づけをします。

実装の流れ

まず、実装の流れを簡単に説明します。

この記事で使うライブラリと環境

この記事では以下のライブラリと環境を利用します
AWSのサンプルはboto3だけでやっているのですが、LiteLLMを使うとコード量を大きく減らせます。

AWS CLI
Python 3.13
boto3 1.39.9
python-dotenv 1.1.1
LiteLLM
Mirascope

実装の方法

LiteLLMを使うと、簡単にS3Vectorsの登録と検索を実装できます。

テキストをベクトル化するには以下のように書きます。

import litellm

# Titan Embeddingの実行関数を定義する
# ベクトル化する。引数にモデル名と次元数を書く
result = litellm.embedding(
    model="bedrock/amazon.titan-embed-text-v2:0",
    dimensions=1024, # 次元数。小さいほど料金と精度が下がる
    input=["ベクトル化したい文字列"],
)

# ベクトルデータがfloat32の配列に入っている
# なお、ベクトル化するテキストを配列で渡すので、結果も配列で返ってくる
print(result.data)

litellm.embeddingを呼ぶだけで、Bedrockを使ってテキストをベクトルに変換することができます。

あとは変換結果のfloat32の配列をput_vectorsに渡すだけです。

# S3 Vectorsに登録する
s3_vector.put_vectors(
    vectorBucketName=vector_bucket_name, # バケット名
    indexName=index_name, # インデックス名
    vectors=[
        {
            # ベクトルデータを登録する
            "data": {
                "float32": result.data[0].embedding,
            },
            # ユニークなキーを設定する
            "key": "unique-name",
            # 検索時に一緒に取得したい任意の値を登録する
            "metadata": {
                "text": "ベクトル化したい文字列"
            }
        }
    ],
)

これでデータの登録が完了しました。
クエリで参照するときも同様に、float32の配列をそのまま渡せば検索できます。

# ベクトル化したデータを使って、似たデータを検索する
search_result = s3_vector.query_vectors(
    vectorBucketName=vector_bucket_name, # バケット名
    indexName=index_name, # インデックス名
    returnMetadata=True, # metadataを取得する
    queryVector={
        "float32": result.data[0].embedding,
    },
)

# 検索結果を見る
print(search_result)

実行結果は以下の通りです。

取得できる値

{
    // ...一緒にとれるリクエストデータは中略
    "vectors": [
        {
            "key": "unique-name", 
            "metadata": {
                "text": "ベクトル化したい文字列"
            }
        }
    ]
}

ベクトル検索の結果を使ったRAGも、Mirascopeを使って簡単に書くことができます。

from mirascope import llm, prompt_template

@llm.call( # モデルを指定する
    provider="bedrock",
    model="amazon.nova-micro-v1:0",
)
@prompt_template( # RAGのプロンプトを書く
    """
    SYSTEM: excerptsの情報を元に、ユーザーからの質問に回答してください
    <excerpts>{excerpts}</excerpts>

    USER: {query}
    """
)
def execute_rag(query: str):
    # ベクトル検索を実行する
    search_result = {queryから検索した、s3_vector.query_vectorsの実行結果}
    # プロンプトに検索結果を連携する
    return {
        "computed_fields": {
            "excerpts": search_result["vectors"]
        }
    }

以下のようにして実行します。

print(execute_rag("質問したいことがら"))

RAGを使って質問をして、ベクトル検索の結果をもとに回答を受け取ることができます。

作ってみよう

実際に沿った形で実装します。

環境変数を定義しておく

まず環境変数を定義しておきます。
今回は.envファイルを作成して、そこに環境変数を書き込みます。

.env

BUCKET_NAME="作成するバケットの名前"
INDEX_NAME="defaults"
AWS_REGION_NAME="us-east-1"
AWS_DEFAULT_REGION="us-east-1"
DIMENSIONS=1024

バケットを作る

バケットを作るには、以下のようにcreate_vector_bucketを呼び出します。

def create_bucket(
    *,
    s3_vector, # boto3のクライアント
    vector_bucket_name: str, # バケット名
    **kwargs,
):
    """
    S3 Vectorsのバケットを作成する関数
    """

    # バケットを作成する
    try:
        s3_vector.create_vector_bucket(
            vectorBucketName=vector_bucket_name,
        )
    except Exception:
        print("failed to create bucket")

インデックスを作る

インデックスを作るには、以下のようにcreate_indexを呼び出します。
delete_indexでインデックスを削除してから呼び出すことで、洗い替えることができます。
また、データがあるかどうかはget_indexのNotFoundException例外で判定できます。

def create_index(
    *,
    s3_vector, # boto3のクライアント
    vector_bucket_name: str, # バケット名
    index_name: str, # インデックス名
    dimensions: int, # 次元数
    **kwargs,
):
    """
    S3 Vectorsのインデックスを作成する関数
    もしすでにインデックスがあるのなら、削除してから洗い替える
    """

    try:
        parameter = {
            "vectorBucketName": vector_bucket_name,
            "indexName": index_name,
        }
        # インデックスを取得、存在すれば削除する
        s3_vector.get_index(**parameter)
        s3_vector.delete_index(**parameter)
    except s3_vector.exceptions.NotFoundException:
        print("Index not found")
        pass

    # 新しいインデックスを作成する
    s3_vector.create_index(
        vectorBucketName=vector_bucket_name,
        indexName=index_name,
        dimension=dimensions,  # Embeddingsの引数に渡す次元数
        dataType="float32",  # データタイプはfloat32で固定
        distanceMetric="cosine",  # 距離はコサイン類似度を使用
    )

データを登録する

登録はput_vectors関数を実行します。

登録するデータは文字の配列形式で渡します。
また、登録するときにmetadataを指定することで、検索結果としてmetadataに指定したJSONを受け取れるようになります。

from typing import List, TypedDict

class VectorData(TypedDict):
    """
    ベクトルデータを表す辞書データ
    """

    text: str
    description: str


def put_text_vectors(
    *,
    s3_vector, # boto3のクライアント
    titan_embedding, # litellmのembedding関数
    vector_bucket_name: str, # バケット名
    index_name: str, # インデックス名
    target_data: List[VectorData], # 登録するデータ
    **kwargs,
):
    """
    S3 Vectorsにテキストを登録する関数

    引数
    target_data: 登録するテキストのリスト
    """
    # ベクトル化したデータを登録する
    s3_vector.put_vectors(
        vectorBucketName=vector_bucket_name,
        indexName=index_name,
        vectors=[
            {
                # ベクトルデータを登録する
                "data": {
                    "float32": e.embedding,
                },
                # ユニークなキーを設定する
                "key": f"vector-{index}",
                # 検索にヒットしたときに一緒に取得できるJSONを定義する
                # ※これでフィルタをすることもできる
                "metadata": target_data[index],
            }
            # テキストをベクトル化する
            for index, e in enumerate(
                titan_embedding(
                    # 対象のデータをTitan Embeddingに渡す
                    input=[t["text"] for t in target_data],
                ).data  # 結果はdataに入っているのでdataを連携
            )
        ],
    )

クエリをする

クエリはquery_vectors関数を実行します。
検索するデータは文字列で渡します。

def query_text_vectors(
    *,
    s3_vector, # boto3のクライアント
    titan_embedding, # litellmのembedding関数
    vector_bucket_name: str, # バケット名
    index_name: str, # インデックス名
    target_data: str, # クエリテキスト
    **kwargs,
):
    """
    S3 Vectorsからテキストを検索する関数

    引数
    target_data: 検索するテキスト
    """

    # ベクトル化したデータを登録する
    results = s3_vector.query_vectors(
        vectorBucketName=vector_bucket_name,
        indexName=index_name,
        # MetaDataを検索結果として返す
        returnMetadata=True,
        # クエリをベクトル化する
        queryVector={
            "float32": titan_embedding(
                input=[target_data],
            )
            .data[0]
            .embedding,  # 最初の要素のベクトルを取得
        },
        topK=1,  # 最も似ている結果を抽出する
    )

    return results

実行する

上の4つの関数をs3_vector_functions.pyに定義して、実行するための関数を別ファイルに書きだします。

app.py

import litellm
from os import environ
from functools import partial
import boto3
import dotenv
from s3_vector_functions import (
    create_index,
    put_text_vectors,
    query_text_vectors,
    create_bucket,
)

# 環境変数を読み込む
dotenv.load_dotenv()

# S3 Vectorsのクライアントを作成する
s3_vector = boto3.client("s3vectors")

# Titan Embeddingの実行関数を定義する
# partialは、何度も使う関数の引数を省略するための関数
titan_embedding = partial(
    litellm.embedding,
    model="bedrock/amazon.titan-embed-text-v2:0",
    dimensions=int(environ["DIMENSIONS"]),
)

# それぞれの関数に渡す共通引数を定義する
parameters = {
    "s3_vector": s3_vector,
    "titan_embedding": titan_embedding,
    "vector_bucket_name": environ["BUCKET_NAME"],
    "index_name": environ["INDEX_NAME"],
    "dimensions": int(environ["DIMENSIONS"]),
}

# バケットを作成する
create_bucket(**parameters)

# インデックスを初期化する、もしデータが既に入っていればリセットする
create_index(**parameters)

# テキストを登録する
put_text_vectors(
    **parameters,
    # 登録するデータ
    target_data=[
        {
            "text": "ボウルに卵を割り入れ、箸で卵白を切るように溶く。Aを加えて混ぜ合わせる。",
            "description": "卵を混ぜ合わせる",
        },
        {
            "text": "小さな容器にサラダ油を入れ、小さく折りたたんだキッチンペーパーを浸けておく。",
            "description": "火にかける準備をする",
        },
        {
            "text": "卵焼き器を中火にかけ、油を含んだキッチンペーパーで油をなじませる。かすかに煙が出るくらいまで中火で熱する。",
            "description": "卵を火にかける",
        },
        {
            "text": "気泡ができたら菜箸でつぶし、半熟状になったら奥から手前に卵を巻く。",
            "description": "卵を巻く",
        },
    ],
)

# テキストを検索する
print(query_text_vectors(
    **parameters,
    # 検索するデータ
    target_data="形を整える",
))

実行すると

この関数を実行すると、「形を整える」に最も近いテキストが検索されます。
以下の結果が返ってきます

気泡ができたら菜箸でつぶし、半熟状になったら奥から手前に卵を巻く。

{'ResponseMetadata': {'RequestId': 'f53730bb-f71e-4882-8821-348b9f716a53', 'HostId': '', 'HTTPStatusCode': 200, 'HTTPHeaders': {'date': 'Mon, 21 Jul 2025 07:19:23 GMT', 'content-type': 'application/json', 'content-length': '183', 'connection': 'keep-alive', 'x-amz-request-id': 'f53730bb-f71e-4882-8821-348b9f716a53', 'access-control-allow-origin': '*', 'vary': 'origin, access-control-request-method, access-control-request-headers', 'access-control-expose-headers': '*'}, 'RetryAttempts': 0}, 'vectors': [{'key': 'vector-3', 'metadata': {'text': '気泡ができたら菜箸でつぶし、半熟状になったら奥から手前に卵を巻く。', 'description': '卵を巻く'}}]}

検索クエリを「卵を焼く」にすると、以下の結果が返ってきます

卵焼き器を中火にかけ、油を含んだキッチンペーパーで油をなじませる。かすかに煙が出るくらいまで中火で熱する。

{'ResponseMetadata': {'RequestId': '9fbaa2fd-b185-439e-acf9-e0ef836f4ea0', 'HostId': '', 'HTTPStatusCode': 200, 'HTTPHeaders': {'date': 'Mon, 21 Jul 2025 02:31:16 GMT', 'content-type': 'application/json', 'content-length': '244', 'connection': 'keep-alive', 'x-amz-request-id': '9fbaa2fd-b185-439e-acf9-e0ef836f4ea0', 'access-control-allow-origin': '*', 'vary': 'origin, access-control-request-method, access-control-request-headers', 'access-control-expose-headers': '*'}, 'RetryAttempts': 0}, 'vectors': [{'key': 'vector-2', 'metadata': {'text': '卵焼き器を中火にかけ、油を含んだキッチンペーパーで油をなじませる。かすかに煙が出るくらいまで中火で熱する。'}]}

このように、「卵を焼く」のテキストで検索をして、metadataに指定した本文を検索結果として受け取ることができます。

RAGにする

この検索結果をRAGにするには、以下のようにします。

import litellm
from os import environ
from functools import partial
import boto3
import dotenv
from s3_vector_functions import (
    query_text_vectors,
)
from mirascope import llm, prompt_template

# 環境変数を読み込む
dotenv.load_dotenv()

# S3 Vectorsのクライアントを作成する
s3_vector = boto3.client("s3vectors")

# Titan Embeddingの実行関数を定義する
titan_embedding = partial(
    litellm.embedding,
    model="bedrock/amazon.titan-embed-text-v2:0",
    dimensions=int(environ["DIMENSIONS"]),
)

# それぞれの関数に渡す引数を定義する
parameters = {
    "s3_vector": s3_vector,
    "titan_embedding": titan_embedding,
    "vector_bucket_name": environ["BUCKET_NAME"],
    "index_name": environ["INDEX_NAME"],
    "dimensions": int(environ["DIMENSIONS"]),
}

@llm.call(
    provider="bedrock",
    model="amazon.nova-micro-v1:0",
)
@prompt_template(
    """
    SYSTEM:
    excerptsの情報を元に、ユーザーからの質問に回答してください

    <excerpts>{excerpts}</excerpts>

    USER: {query}
    """
)
def execute_rag(query: str):
    """ RAGを実行する """
    # 検索処理を実行する
    rag_data = query_text_vectors(
        **parameters,
        target_data=query,
    )
    # プロンプトに検索結果を連携する
    return {"computed_fields": {"excerpts": rag_data["vectors"]}}


print(
    execute_rag("形を整えるにはどうすればよいですか")
)

実行すると、ベクトル検索の検索結果をもとにしたLLMの回答を得ることができます。
実行結果は以下の通りです。

卵を形を整えるためには、以下の手順に従ってください：

気泡ができる: 卵を茹でる際に、鍋に水を入れ、沸騰させます。沸騰したら、卵をゆっくりと水に落とします。この時、卵の表面に気泡が形成されるはずです。

気泡をつぶす: 気泡が形成されたら、菜箸を使って卵の表面の気泡をつぶします。この工程は、卵が均一に茹でられるようにするために重要です。

半熟状態になる: 鍋で茹で続け、卵が半熟になるまで調整します。半熟状態とは、卵黄が固まっているが、まだ少し液体感がある状態のことです。

卵を巻く: 半熟になったら、鍋の蓋を少し開けて、卵を奥から手前にゆっくりと巻きます。この動作は、卵が均一に形を整え、美しい形になるようにします。
これらの手順に従うことで、卵がきれいな形で茹で上がります。

卵焼きの質問だとはプロンプトに書いていないのですが、ベクトル検索の結果が卵焼きの調理方法であるため、卵焼きの調理方法を回答してくれています。

S3 Vectorsで精度が上がるということ

ソースコードは省略しますが、Natureの論文に対してRAGを実施します。

クエリは以下の通りです。

鳥インフルエンザはどのように伝播しますか

検索結果(チャンク長: 400)は以下の通りでした。

have long been recognized as reservoirs for avian influenza viruses (AIVs), with infection from \nlow pathogenic avian influenza (LPAI) often yielding asymptomatic \xadinfection1. Although some strains of H5 and \nH7 LPAI possess the potential to rapidly become highly pathogenic (HPAI) during replication in gallinaceous \npoultry where there is high density, such occurrences typically present limited threat
（鳥インフルエンザウイルス（AIV）の貯蔵庫として長年認識されてきました。低病原性鳥インフルエンザ（LPAI）による感染は、通常は無症状の感染を引き起こします。H5およびH7型のLPAIの一部株は、高密度飼育下にある鶏類家禽において複製中に急速に高病原性（HPAI）に変化する可能性を有していますが、このような事例は通常、限定的な脅威を呈します。）

まず、従来の方法に従って、この検索結果のチャンクをコンテキストにしてNova Proに回答させます。

鳥インフルエンザ（Avian Influenza Virus, AIV）は、主に以下の方法で伝播します：

1. **直接接触**：
   - 感染した鳥類との直接的な接触（例：糞、唾液、鼻汁）。
   - 感染した鳥類の死体との接触。

2. **間接接触**：
   - 感染した鳥類が接触した物体（例：餌やり器具、衣類、車両）との接触。
   - 感染した鳥類の排泄物が付着した土壌や水との接触。

3. **空気中伝播**：
   - 感染した鳥類が排出するエアロゾル粒子を介した伝播。特に高密度の養鶏場では、空気中のウイルス粒子が広範囲に拡散する可能性があります。     

4. **野鳥の移動**：
   - 渡り鳥や野生の水鳥が感染源となり、長距離を移動してウイルスを拡散する可能性があります。

5. **人為的な移動**：
   - 感染した鳥類やその製品（例：卵、肉）の移動。
   - 養鶏場間の人員や機器の移動。

これらの方法を通じて、鳥インフルエンザは迅速に広がる可能性があります。特に、高病原性鳥インフルエンザ（HPAI）は、感染した鳥類に急速に致死的な影響を与える可能性があるため、注意が必要です。

S3 Vectorsを使って、この回答を改善します。
検索結果として返すものはチャンクである必要がなくなりました。
summaryとして該当論文の2000字程度の要約を作成して、その要約を検索結果として返します。

結果、同じクエリ「鳥インフルエンザはどのように伝播しますか」に対して、以下の応答を返すようになりました。

鳥インフルエンザは主に以下の方法で伝播します：

1. **野鳥から家畜への伝播**：
   - 野鳥、特に水鳥は鳥インフルエンザウイルス（AIV）の自然宿主であり、感染を拡散する役割を果たします。
   - 野鳥の水鳥飛来と家畜の分布が重なる地域では、鳥インフルエンザの伝播リスクが高まります。

2. **家畜間の伝播**：
   - 家畜間の密接な接触や、汚染された環境（例：餌、水、器具）を通じて感染が広がります。
   - 特に、高密度の家畜飼育地域では、感染の拡大が速く起こる可能性が高いです。

3. **人為的な要因**：
   - 人々の移動、飼料や器具の移動、野鳥との接触など、人為的な要因も伝播に影響を与えます。

### 時空間的な要因
- 春と秋の渡りの時期には、野鳥の移動が活発になるため、鳥インフルエンザの伝播リスクが特に高くなります。

### モデルによる予測
- 研究では、野鳥の水鳥飛来と家畜の分布を考慮した伝播リスクモデルが開発され、時空間的なリスクパターンを予測しています。このモデルは、特に渡りの時期に高いリスクを示し、早期の監視と対策の重要性を強調しています。

### 結論
鳥インフルエンザの伝播は複雑な要因に影響され、野鳥と家畜の相互作用、季節的な移動パターン、家畜の分布などが重要な役割を果たします。効果的な監視と対策は、これらの要因を考慮して実施される必要があります。

修正前は400文字のチャンク（論文のごく一部を切り出したもの）から回答していたため、ほとんど一般論で回答が埋まっていました。修正後はチャンクではなく論文全体から回答しているため、内容をふまえた回答を返しています。

明らかに回答が良くなります。