Amazon S3でベクトル検索！「ベクトルバケット」の作成とAWS Lambdaからの操作をサクッと試してみる

概要～Amazon S3 Vectorsでベクトル検索をAWS Lambdaからサクッと試す～

「S3でベクトル検索ができるって本当？」「AIやRAGアプリを作りたいけれど、専用のベクトルデータベースを導入・管理するのはコストも運用もハードルが高い…」
そんな方にぜひ試していただきたいのが、Amazon S3 Vectors（ベクトルバケット） です！

これまでAIアプリケーションやRAG（検索拡張生成）の構築では、高価な専用ベクトルDBやクラスタを起動・管理するのが一般的でした。しかし、S3のベクトルバケットを活用すれば、完全サーバーレス＆S3ならではの圧倒的な低コスト・高耐久性 で、手軽にベクトルデータの保存と高速な類似度検索（ANN）が実現できます。

今回は、AWSコンソールでのベクトルバケット・インデックスの作成から、AWS Lambda (Python 3.14) を使ったベクトルの登録・更新・検索・取得・一覧・削除の全操作までを、実際に手を動かしながらサクッと体験できるハンズオン形式で解説します！

---

なぜベクトルストアがAIやRAGの心臓部になるのか？

まずLLM（大規模言語モデル）を活用したRAG（Retrieval-Augmented Generation）において、なぜベクトルストアが不可欠な基盤となるのかを簡単に整理します。

1. テキストの意味を数値化（Embeddings / 埋め込み）:
   社内文書やマニュアルなどのテキストデータをAIモデル（Embeddingモデル）に入力すると、数百〜千次元を超える数値の配列（ベクトル）に変換されます。意味が近い文章同士は、ベクトル空間上で「向きや距離が近く」配置されます。
   （※このテキストからベクトルへの変換処理は、Amazon Bedrock（Amazon Titan Text Embeddings や Cohere Embed 等）のAPIを呼び出すだけで非常に簡単・高速に行うことができます）
2. 意味による高速な類似検索（ANN検索）:
   ユーザーが質問を入力した際、その質問文も同様にベクトル化します。ベクトルストアは、膨大なドキュメントの中から「質問ベクトルと最も意味が近い（距離が近い）ドキュメント」をミリ秒単位で瞬時に見つけ出します。従来のキーワード完全一致検索では難しかった「言葉の揺らぎ」や「文脈の意味」を捉えた高精度な検索が可能になります。
3. LLMへのコンテキスト注入と回答生成:
   検索で見つかった関連性の高い文章をLLMのプロンプトに「参考情報」として渡すことで、最新の社内データに基づいた正確な回答を生成させ、ハルシネーション（嘘の回答）を防止します。

この「大量のベクトルデータを安全・安価に永続化し、瞬時に類似検索する」役割を担うのが、今回構築する Amazon S3 Vectors です。

Bedrockを含めたイメージ図は以下が非常に分かりやすいです。

https://aws.amazon.com/jp/blogs/aws/introducing-amazon-s3-vectors-first-cloud-storage-with-native-vector-support-at-scale/

---

想定読者

• Amazon S3を使って手軽かつ安価にベクトルデータを管理・検索したい方
• AWS Lambda (Python + boto3) から S3 Vectors API を使ってベクトル操作（登録・ANN検索・フィルタリング・削除）を実装したい方
• RAGや生成AIアプリの開発で、サーバーレスなベクトルストレージを検討している方
• AWSコンソール操作およびAPI連携を通じて、ベクトルバケットおよびインデックスの理解を深めたい方

料金について

今回のサンプルは非常に小さいデータを使用しているためほぼ 無料 です。

ただし、S3のベクトルバケットやExpress One Zoneなどのストレージ利用には、標準のS3ストレージ料金に加えてAPI呼び出しやプロビジョニングに応じた課金が発生します。
個人で試される場合は、不要になったバケットやインデックス、Lambda関数を検証後に削除することを推奨します。詳細な最新料金プランについては、AWS公式のAmazon S3 料金ページをご確認ください。

作業時間

約1時間

---

システム概念図とメタデータの解説

S3 Vectors のデータ構造と処理フロー

主要な設定項目

本ハンズオンで使用する主要な設定パラメータは以下の通りです。

設定パラメータ一覧

• ベクトルバケット名
  • my-knowledge-vector-bucket （一意のバケット名）
• ベクトルインデックス名
  • document-embeddings-index
• ディメンション数
  • 8 （テスト用に入力ベクトルの次元数を8次元に設定）
• 距離メトリック
  • コサイン （テキストの長さに依存せず、ベクトルの「向き」だけで純粋な類似度を測定できるため、自然言語処理・RAGで最も一般的に利用されます）
• Lambda ランタイム
  • Python 3.14 （最新のboto3でS3 Vectors APIを呼び出します）

S3 Vectors におけるメタデータ（metadata）の役割

S3 Vectors では、ベクトルデータ（float32の数値配列）と一緒に、任意のJSON形式の メタデータ（metadata） を保存できます。

• 属性情報の保持: ドキュメントのカテゴリ（category）、作成部門（department）、年（year）、元テキスト（raw_text）などをベクトルと一体で管理できます。
• In-Tandem Filtering（同時フィルタリング）: 類似度検索（QueryVectors）を実行する際、近似最近傍探索（ANN）とメタデータ条件（例: category == "technical" かつ year >= 2026）による絞り込みを同時に高速実行できます。
• 検索レスポンスでの活用: returnMetadata=True を指定することで、検索結果にメタデータを含めて取得できるため、外部DBを参照することなくそのままLLMへのプロンプトに組み込むことが可能です。

---

全体の作業手順

本記事は大きく2つのパートに分かれています。

パート1：AWSコンソールでの環境構築

• ステップ1：S3ナビゲーションメニューからベクトルバケットを選択
• ステップ2：ベクトルバケット一覧画面の確認
• ステップ3：ベクトルバケットの作成設定（バケット名指定）
• ステップ4：ベクトルバケット一覧での作成完了確認
• ステップ5：ベクトルインデックスの作成開始
• ステップ6：ベクトルインデックスのパラメータ設定と作成

パート2：AWS Lambda (Python 3.14) によるベクトルデータ操作

• ステップ1：Lambda関数の作成とプログラム実装（Python 3.14）
• ステップ2：ベクトルの登録・更新テスト（put_vectors）
• ステップ3：類似度検索とメタデータフィルタテスト（query_vectors）
• ステップ4：ベクトルのkey指定取得テスト（get_vectors）
• ステップ5：ベクトルキー一覧の取得テスト（list_vectors）
• ステップ6：ベクトルのkey指定削除テスト（delete_vectors）

---

パート1：AWSコンソールでの環境構築

まずはAWSマネジメントコンソールから、ベクトルデータを格納するためのバケットとインデックスを作成します。

---

ステップ 1：S3ナビゲーションメニューからベクトルバケットを選択

AWSマネジメントコンソールでAmazon S3サービスにアクセスした直後のナビゲーションメニュー画面です。
左ペインの「バケット」配下には「汎用バケット」「ディレクトリバケット」「テーブルバケット」に並び、「ベクトルバケット」という選択肢が用意されています。

左メニューの 「ベクトルバケット」 をクリックして、ベクトルバケットの管理画面へ遷移します。

---

ステップ 2：ベクトルバケット一覧画面

「ベクトルバケット」管理画面の初期状態です。まだベクトルバケットが一つも作成されていないため、画面中央には「ベクトルバケットなし」と表示されています。

画面右上にあるオレンジ色の 「ベクトルバケットを作成」 ボタンをクリックして作成ウィザードを開始します。

---

ステップ 3：ベクトルバケットの作成設定（バケット名指定）

新しいベクトルバケットを作成するための設定画面です。一般的な設定、暗号化、タグ・オプション等の項目が並んでいます。

「一般的な設定」エリア内の 「ベクトルバケット名」 入力欄に、バケット名（例として my-knowledge-vector-bucket ）を入力します。その他の設定（暗号化やタグなど）はデフォルトのままでOKです。
画面右下のオレンジ色の 「ベクトルバケットを作成」 ボタンをクリックしてバケットの作成を実行します。

---

ステップ 4：ベクトルバケット一覧（作成完了）

ベクトルバケットの作成が完了し、バケット一覧画面に戻った状態です。
作成したバケット my-knowledge-vector-bucket が、ステータス（アクティブ）や作成日、ARN（Amazonリソースネーム）と共に一覧に表示されます。

一覧表内のバケット名 「my-knowledge-vector-bucket」 をクリックして、詳細画面およびインデックス作成画面へ進みます。

---

ステップ 5：ベクトルインデックスの作成開始

選択したベクトルバケット my-knowledge-vector-bucket の詳細画面です。「ベクトルインデックス」タブが選択されており、現在はインデックスが一つも存在しないため「ベクトルインデックスなし」と表示されています。

画面右側にあるオレンジ色の 「ベクトルインデックスを作成」 ボタンをクリックして、インデックス作成ウィザードに進みます。

---

ステップ 6：ベクトルインデックスのパラメータ設定と作成

ベクトルインデックスの具体的なパラメータを指定して作成を実行する画面です。インデックス名、ディメンション数、距離メトリック（コサイン、ユーグリッドなど）、暗号化、タグなどの詳細設定項目があります。

「プロパティ」エリア内の 「ベクトルインデックス名」 入力欄に document-embeddings-index と入力し、「ディメンション」 入力欄に 8 と入力します。
（※ディメンション数は実際の埋め込みモデルに合わせて設定しますが、今回は後述のLambdaサンプルプログラムのテストデータに合わせて 8 を指定しています）

最後に、画面右下のオレンジ色の 「ベクトルインデックスを作成」 ボタンをクリックしてインデックスの作成を完了させます。

---

パート2：AWS Lambda (Python 3.14) によるベクトルデータ操作

基盤となるインデックスが完成したので、ここからはAWS Lambdaからプログラム経由でベクトルデータを操作していきます。

---

ステップ 1：Lambda関数の作成とプログラム実装

AWS Lambda コンソールで新しい関数を作成します。

Lambda 関数の基本設定

• 関数名: s3-vectors-handler（任意の名前）
• ランタイム: Python 3.14
• 実行ロールのIAMポリシー:
  Lambda実行ロールに以下の S3 Vectors 操作権限を付与します。
  • s3vectors:PutVectors
  • s3vectors:QueryVectors
  • s3vectors:GetVectors
  • s3vectors:DeleteVectors
  • s3vectors:ListVectors

IAMポリシーサンプル

{
    "Version": "2012-10-17",
    "Statement": [
        {
            "Sid": "S3VectorsDataPlanePermissions",
            "Effect": "Allow",
            "Action": [
                "s3vectors:PutVectors",
                "s3vectors:QueryVectors",
                "s3vectors:GetVectors",
                "s3vectors:DeleteVectors",
                "s3vectors:ListVectors"
            ],
            "Resource": [
                "arn:aws:s3vectors:ap-northeast-1:*:bucket/my-knowledge-vector-bucket",
                "arn:aws:s3vectors:ap-northeast-1:*:bucket/my-knowledge-vector-bucket/index/*"
            ]
        },
        {
            "Effect": "Allow",
            "Action": "logs:CreateLogGroup",
            "Resource": "arn:aws:logs:ap-northeast-1:*:*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "logs:CreateLogStream",
                "logs:PutLogEvents"
            ],
            "Resource": [
                "arn:aws:logs:ap-northeast-1:*:log-group:/aws/lambda/s3-vectors-handler:*"
            ]
        }
    ]
}

以下のコードを Lambda 関数のエディタ（lambda_function.py）に貼り付けて 「Deploy」 をクリックします。

lambda_function.py

"""
AWS Lambda 関数: Amazon S3 Vectors ベクトルデータ操作ハンドラー
- ランタイム: Python 3.14
- 対象API（5種）:
    1. PutVectors    : ベクトルの一括登録・更新（最大500件/リクエスト、UPSERT仕様）
    2. QueryVectors  : 類似度検索 (ANN) ＋ In-Tandem メタデータフィルタリング
    3. GetVectors    : key指定によるベクトルデータ・メタデータの取得
    4. DeleteVectors : key指定によるベクトルデータの一括削除
    5. ListVectors   : インデックス内のベクトルkey一覧取得（ページネーション対応）
- boto3 S3Vectors 仕様準拠（引数名は lowerCamelCase）
- API Gateway プロキシ統合と Lambda 直接呼び出し（テストイベント）の両方に対応
"""

import json
import os
import logging
from typing import Any, Dict, List
import boto3
from botocore.exceptions import ClientError, ParamValidationError

# CloudWatch Logs へのログ出力設定
logger = logging.getLogger()
logger.setLevel(logging.INFO)

# デフォルト設定値（Lambda環境変数から取得、未指定時はフォールバック）
DEFAULT_REGION = os.environ.get("AWS_REGION", "ap-northeast-1")
DEFAULT_BUCKET_NAME = os.environ.get("VECTOR_BUCKET_NAME", "my-knowledge-vector-bucket")
DEFAULT_INDEX_NAME = os.environ.get("VECTOR_INDEX_NAME", "document-embeddings-index")

# boto3 s3vectors クライアント初期化（コールドスタート対策としてハンドラー外で再利用）
s3vectors_client = boto3.client("s3vectors", region_name=DEFAULT_REGION)


def build_response(status_code: int, body: Dict[str, Any]) -> Dict[str, Any]:
    """API Gateway プロキシ統合互換のレスポンス辞書を構築する共通ヘルパー関数"""
    response = {
        "statusCode": status_code,
        "headers": {
            "Content-Type": "application/json",
            "Access-Control-Allow-Origin": "*",
            "Access-Control-Allow-Headers": "Content-Type,Authorization",
            "Access-Control-Allow-Methods": "OPTIONS,POST,GET"
        },
        "body": json.dumps(body, ensure_ascii=False, default=str)
    }
    logger.info(f"[Lambda Response Output] {json.dumps(response, ensure_ascii=False, default=str)}")
    return response


# 1. PutVectors API: ベクトルのバッチ登録・更新 (UPSERT)
def handle_put_vectors(bucket_name: str, index_name: str, vectors: List[Dict[str, Any]]) -> Dict[str, Any]:
    if not vectors:
        raise ValueError("Parameter 'vectors' is required and must not be empty.")
    if len(vectors) > 500:
        raise ValueError(f"Vector batch size ({len(vectors)}) exceeds the maximum allowed limit of 500.")

    formatted_vectors = []
    for item in vectors:
        key = item.get("key")
        data = item.get("data", {})
        metadata = item.get("metadata", {})

        if not key or "float32" not in data:
            raise ValueError(f"Each vector item must contain 'key' and 'data.float32'. Invalid item: {item}")

        float_values = [float(val) for val in data["float32"]]
        formatted_vectors.append({
            "key": str(key),
            "data": {"float32": float_values},
            "metadata": metadata
        })

    api_request_payload = {
        "vectorBucketName": bucket_name,
        "indexName": index_name,
        "vectors": formatted_vectors
    }
    logger.info(f"[PutVectors API Request] Payload: {json.dumps(api_request_payload, ensure_ascii=False, default=str)}")
    response = s3vectors_client.put_vectors(**api_request_payload)
    logger.info(f"[PutVectors API Response] Payload: {json.dumps(response, ensure_ascii=False, default=str)}")

    return {
        "action": "put_vectors",
        "processedCount": len(formatted_vectors),
        "rawApiResponse": response
    }


# 2. QueryVectors API: 類似度検索 (ANN) ＋ メタデータフィルタ
def handle_query_vectors(
    bucket_name: str,
    index_name: str,
    query_vector: List[float],
    top_k: int = 5,
    metadata_filter: Dict[str, Any] = None,
    return_distance: bool = True,
    return_metadata: bool = True
) -> Dict[str, Any]:
    if not query_vector:
        raise ValueError("Parameter 'queryVector' is required for similarity search.")

    float_query = [float(val) for val in query_vector]
    top_k_clamped = min(max(1, int(top_k)), 100)

    query_params: Dict[str, Any] = {
        "vectorBucketName": bucket_name,
        "indexName": index_name,
        "queryVector": {"float32": float_query},
        "topK": top_k_clamped,
        "returnDistance": return_distance,
        "returnMetadata": return_metadata
    }

    if metadata_filter:
        if (
            isinstance(metadata_filter, dict)
            and len(metadata_filter) > 1
            and "$and" not in metadata_filter
            and "$or" not in metadata_filter
        ):
            normalized_filter = {"$and": [{k: v} for k, v in metadata_filter.items()]}
        else:
            normalized_filter = metadata_filter
        query_params["filter"] = normalized_filter

    logger.info(f"[QueryVectors API Request] Payload: {json.dumps(query_params, ensure_ascii=False, default=str)}")
    response = s3vectors_client.query_vectors(**query_params)
    logger.info(f"[QueryVectors API Response] Payload: {json.dumps(response, ensure_ascii=False, default=str)}")

    return {
        "action": "query_vectors",
        "resultsCount": len(response.get("vectors", [])),
        "vectors": response.get("vectors", []),
        "rawApiResponse": response
    }


# 3. GetVectors API: key指定によるベクトル・メタデータ取得
def handle_get_vectors(
    bucket_name: str,
    index_name: str,
    keys: List[str],
    return_data: bool = True,
    return_metadata: bool = True
) -> Dict[str, Any]:
    if not keys:
        raise ValueError("Parameter 'keys' list is required and must not be empty.")

    api_request_payload = {
        "vectorBucketName": bucket_name,
        "indexName": index_name,
        "keys": keys,
        "returnData": return_data,
        "returnMetadata": return_metadata
    }
    logger.info(f"[GetVectors API Request] Payload: {json.dumps(api_request_payload, ensure_ascii=False, default=str)}")
    response = s3vectors_client.get_vectors(**api_request_payload)
    logger.info(f"[GetVectors API Response] Payload: {json.dumps(response, ensure_ascii=False, default=str)}")

    return {
        "action": "get_vectors",
        "vectors": response.get("vectors", []),
        "notFoundKeys": response.get("notFoundKeys", []),
        "rawApiResponse": response
    }


# 4. DeleteVectors API: key指定によるベクトル一括削除
def handle_delete_vectors(bucket_name: str, index_name: str, keys: List[str]) -> Dict[str, Any]:
    if not keys:
        raise ValueError("Parameter 'keys' list is required for deletion.")

    api_request_payload = {
        "vectorBucketName": bucket_name,
        "indexName": index_name,
        "keys": keys
    }
    logger.info(f"[DeleteVectors API Request] Payload: {json.dumps(api_request_payload, ensure_ascii=False, default=str)}")
    response = s3vectors_client.delete_vectors(**api_request_payload)
    logger.info(f"[DeleteVectors API Response] Payload: {json.dumps(response, ensure_ascii=False, default=str)}")

    return {
        "action": "delete_vectors",
        "deletedKeys": keys,
        "rawApiResponse": response
    }


# 5. ListVectors API: ベクトルkey一覧の取得（ページネーション対応）
def handle_list_vectors(
    bucket_name: str,
    index_name: str,
    next_token: str = None,
    max_results: int = 100,
    return_data: bool = False,
    return_metadata: bool = False,
    fetch_all: bool = False,
    max_total_results: int = 1000
) -> Dict[str, Any]:
    page_size = min(max(1, int(max_results)), 500)

    if fetch_all:
        all_vectors = []
        current_token = next_token
        page_count = 0

        while True:
            params: Dict[str, Any] = {
                "vectorBucketName": bucket_name,
                "indexName": index_name,
                "maxResults": page_size,
                "returnData": return_data,
                "returnMetadata": return_metadata
            }
            if current_token:
                params["nextToken"] = current_token

            logger.info(f"[ListVectors API Request (Page {page_count + 1})] Payload: {json.dumps(params, ensure_ascii=False, default=str)}")
            response = s3vectors_client.list_vectors(**params)
            logger.info(f"[ListVectors API Response (Page {page_count + 1})] Payload: {json.dumps(response, ensure_ascii=False, default=str)}")

            page_vectors = response.get("vectors", [])
            all_vectors.extend(page_vectors)
            page_count += 1
            current_token = response.get("nextToken")

            if not current_token or len(all_vectors) >= max_total_results:
                break

        extracted_keys = [item["key"] for item in all_vectors if isinstance(item, dict) and "key" in item]
        return {
            "action": "list_vectors",
            "mode": "fetchAll",
            "totalPages": page_count,
            "totalCount": len(all_vectors),
            "keys": extracted_keys,
            "vectors": all_vectors,
            "nextToken": current_token,
            "hasMore": bool(current_token)
        }

    params: Dict[str, Any] = {
        "vectorBucketName": bucket_name,
        "indexName": index_name,
        "maxResults": page_size,
        "returnData": return_data,
        "returnMetadata": return_metadata
    }
    if next_token:
        params["nextToken"] = next_token

    logger.info(f"[ListVectors API Request] Payload: {json.dumps(params, ensure_ascii=False, default=str)}")
    response = s3vectors_client.list_vectors(**params)
    logger.info(f"[ListVectors API Response] Payload: {json.dumps(response, ensure_ascii=False, default=str)}")

    vectors = response.get("vectors", [])
    extracted_keys = [item["key"] for item in vectors if isinstance(item, dict) and "key" in item]
    next_page_token = response.get("nextToken")

    return {
        "action": "list_vectors",
        "mode": "singlePage",
        "count": len(vectors),
        "keys": extracted_keys,
        "vectors": vectors,
        "nextToken": next_page_token,
        "hasNextPage": bool(next_page_token),
        "rawApiResponse": response
    }


# メインハンドラー
def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    logger.info(f"[Lambda Event Input] {json.dumps(event, ensure_ascii=False, default=str)}")

    try:
        payload = event
        if "body" in event and event["body"] is not None:
            if isinstance(event["body"], str):
                try:
                    payload = json.loads(event["body"])
                except json.JSONDecodeError:
                    return build_response(400, {"error": "Malformed JSON in request body."})
            elif isinstance(event["body"], dict):
                payload = event["body"]

        if "queryStringParameters" in event and event["queryStringParameters"]:
            payload.update(event["queryStringParameters"])

        action = payload.get("action")
        if not action:
            return build_response(400, {
                "error": "Missing 'action' parameter.",
                "supported_actions": ["put_vectors", "query_vectors", "get_vectors", "delete_vectors", "list_vectors"]
            })

        bucket_name = payload.get("vectorBucketName", DEFAULT_BUCKET_NAME)
        index_name = payload.get("indexName", DEFAULT_INDEX_NAME)

        if action == "put_vectors":
            vectors = payload.get("vectors", [])
            result = handle_put_vectors(bucket_name, index_name, vectors)

        elif action == "query_vectors":
            query_vector = payload.get("queryVector")
            top_k = payload.get("topK", 5)
            metadata_filter = payload.get("filter")
            return_distance = payload.get("returnDistance", True)
            return_metadata = payload.get("returnMetadata", True)
            result = handle_query_vectors(
                bucket_name=bucket_name,
                index_name=index_name,
                query_vector=query_vector,
                top_k=top_k,
                metadata_filter=metadata_filter,
                return_distance=return_distance,
                return_metadata=return_metadata
            )

        elif action == "get_vectors":
            keys = payload.get("keys", [])
            return_data = bool(payload.get("returnData", True))
            return_metadata = bool(payload.get("returnMetadata", True))
            result = handle_get_vectors(
                bucket_name=bucket_name,
                index_name=index_name,
                keys=keys,
                return_data=return_data,
                return_metadata=return_metadata
            )

        elif action == "delete_vectors":
            keys = payload.get("keys", [])
            result = handle_delete_vectors(bucket_name, index_name, keys)

        elif action == "list_vectors":
            next_token = payload.get("nextToken")
            max_results = int(payload.get("maxResults", 100))
            return_data = bool(payload.get("returnData", False))
            return_metadata = bool(payload.get("returnMetadata", False))
            fetch_all = bool(payload.get("fetchAll", False))
            max_total_results = int(payload.get("maxTotalResults", 1000))
            result = handle_list_vectors(
                bucket_name=bucket_name,
                index_name=index_name,
                next_token=next_token,
                max_results=max_results,
                return_data=return_data,
                return_metadata=return_metadata,
                fetch_all=fetch_all,
                max_total_results=max_total_results
            )

        else:
            return build_response(400, {
                "error": f"Unsupported action: '{action}'",
                "supported_actions": ["put_vectors", "query_vectors", "get_vectors", "delete_vectors", "list_vectors"]
            })

        return build_response(200, result)

    except ValueError as ve:
        logger.error(f"[Validation Error] {ve}")
        return build_response(400, {"error": str(ve)})

    except ParamValidationError as pve:
        logger.error(f"[boto3 ParamValidationError] {pve}")
        return build_response(400, {"error": "Parameter validation failed", "detail": str(pve)})

    except ClientError as ce:
        error_code = ce.response.get("Error", {}).get("Code", "UnknownClientError")
        error_message = ce.response.get("Error", {}).get("Message", str(ce))
        status_code = ce.response.get("ResponseMetadata", {}).get("HTTPStatusCode", 500)
        logger.error(f"[AWS ClientError] Code: {error_code}, Message: {error_message}")
        return build_response(status_code, {
            "error": "AWS Service Error",
            "code": error_code,
            "message": error_message
        })

    except Exception as e:
        logger.error(f"[Internal Server Error] {e}", exc_info=True)
        return build_response(500, {"error": "Internal server error", "detail": str(e)})

使用している主要な boto3 (s3vectors) API メソッド仕様

• ベクトルの登録・更新 (UPSERT)
  • s3vectors.Client.put_vectors
• 類似度検索 (ANN) ＋ メタデータフィルタ
  • s3vectors.Client.query_vectors
• ベクトルの key 指定取得
  • s3vectors.Client.get_vectors
• ベクトル key 一覧取得（ページネーション対応）
  • s3vectors.Client.list_vectors
• ベクトルの key 指定一括削除
  • s3vectors.Client.delete_vectors

---

ステップ 2：ベクトルの登録・更新テスト（put_vectors）

Lambdaコンソールの 「Test」 タブを開き、3件のベクトルデータとメタデータを一括登録するテストイベントを作成して 「Test」 ボタンをクリックします。

重要：UPSERT（登録・更新）の仕様
S3 Vectors の PutVectors は UPSERT仕様 です。すでに存在する 同一の key を指定して再度 put_vectors を実行すると、該当キーのベクトル値およびメタデータが最新の内容に上書き（更新）されます。

put_vectors（テストイベント）

{
  "action": "put_vectors",
  "vectorBucketName": "my-knowledge-vector-bucket",
  "indexName": "document-embeddings-index",
  "vectors": [
    {
      "key": "doc-001",
      "data": {
        "float32": [0.015, -0.023, 0.045, 0.012, 0.089, -0.054, 0.033, 0.011]
      },
      "metadata": {
        "category": "technical",
        "department": "engineering",
        "year": 2026,
        "raw_text": "Amazon S3 Vectors はサーバーレスのベクトルストレージです。"
      }
    },
    {
      "key": "doc-002",
      "data": {
        "float32": [0.020, -0.018, 0.040, 0.015, 0.075, -0.060, 0.028, 0.019]
      },
      "metadata": {
        "category": "financial",
        "department": "accounting",
        "year": 2025,
        "raw_text": "2025年度の財務レポート概要です。"
      }
    },
    {
      "key": "doc-003",
      "data": {
        "float32": [0.005, -0.023, 0.045, 0.012, -0.019, -0.054, 0.093, 0.011]
      },
      "metadata": {
        "category": "technical",
        "department": "engineering",
        "year": 2024,
        "raw_text": "Amazon S3 Vectors はまだありませんでした。"
      }
    }
  ]
}

テストを実行すると statusCode: 200 と processedCount: 3 が返却され、インデックスへのベクトル登録が完了します。

---

ステップ 3：類似度検索とメタデータフィルタテスト（query_vectors）

登録したベクトルに対して、クエリベクトルを与えて近似最近傍探索（ANN）を実行します。同時にメタデータフィルタ（category == "technical" かつ year >= 2026）を指定して絞り込みを行います。

query_vectors（テストイベント）

{
  "action": "query_vectors",
  "vectorBucketName": "my-knowledge-vector-bucket",
  "indexName": "document-embeddings-index",
  "queryVector": [0.015, -0.023, 0.045, 0.012, 0.089, -0.054, 0.033, 0.011],
  "topK": 5,
  "returnDistance": true,
  "returnMetadata": true,
  "filter": {
    "$and": [
      { "category": { "$eq": "technical" } },
      { "year": { "$gte": 2026 } }
    ]
  }
}

実行すると、フィルタ条件を満たし最も距離が近い doc-001 がコサイン距離（distance）およびメタデータと共に上位結果として抽出されます。

---

ステップ 4：ベクトルのkey指定取得テスト（get_vectors）

特定のドキュメントのベクトル値やメタデータを再取得したい場合は、key指定 で get_vectors を呼び出します。

get_vector（テストイベント）

{
  "action": "get_vectors",
  "vectorBucketName": "my-knowledge-vector-bucket",
  "indexName": "document-embeddings-index",
  "keys": ["doc-001", "doc-002"],
  "returnData": true,
  "returnMetadata": true
}

指定した doc-001 と doc-002 のベクトルデータおよびメタデータが返却されます。存在しないkeyを指定した場合は notFoundKeys に格納されて返却されます。

---

ステップ 5：ベクトルキー一覧の取得テスト（list_vectors）

インデックス内に格納されているベクトルのキー一覧をページネーションで取得します。

list_vectors（テストイベント）

{
  "action": "list_vectors",
  "vectorBucketName": "my-knowledge-vector-bucket",
  "indexName": "document-embeddings-index",
  "maxResults": 10,
  "returnData": false,
  "returnMetadata": false
}

登録済みの doc-001, doc-002, doc-003 のキー配列が返却されます。件数が多い場合は nextToken が払い出され、ページネーションで順次取得できます。

---

ステップ 6：ベクトルのkey指定削除テスト（delete_vectors）

不要になったベクトルデータは、key指定 で一括削除します。

delete_vectors（テストイベント）

{
  "action": "delete_vectors",
  "vectorBucketName": "my-knowledge-vector-bucket",
  "indexName": "document-embeddings-index",
  "keys": ["doc-001", "doc-002"]
}

削除を実行すると deletedKeys として指定したキーが返却され、インデックスから該当のベクトルが削除されます。

---

まとめ

今回は Amazon S3 の「ベクトルバケット」および「ベクトルインデックス」のコンソール作成から、AWS Lambda (Python 3.14) を使った5つのベクトルデータ操作（登録・更新・ANN検索・key指定取得・一覧取得・削除） までを一気通貫で体験しました。

S3 Vectors を活用するメリット

1. 完全サーバーレス＆圧倒的な低コスト: 独立した専用ベクトルデータベースのクラスタ管理やノード常時起動コストが不要。
2. In-Tandem Filtering: ベクトル探索とメタデータ絞り込みが単一API呼び出しで高速に実行可能。
3. シンプルなライフサイクル管理: S3の使い慣れたバケット管理とboto3 SDKで、直感的にベクトルのCRUDを構築可能。

生成AIアプリケーションやRAGパイプラインのバックエンドとして、S3 VectorsとLambdaを組み合わせることで、セキュアでスケーラブルな検索システムを低コストで素早く立ち上げることができます。

ぜひご自身のAWS環境でも試してみてください！

ここまで読んで頂き、ありがとうございました！