Amazon S3 Vectors 概要・仕様まとめ
本書は、Amazon S3 のサーバーレスベクトルストレージ機能である Amazon S3 Vectors の全体概要、アーキテクチャ、全API仕様、メタデータフィルタリング、制約事項、料金、AWS Lambda対応実装例、および公式参照リンクをまとめたドキュメントです。
目次
- サービス概要
- アーキテクチャと階層構造
- 主要コンポーネント詳細
- 全API & SDK リファレンス一覧（全19種完全網羅）
- メタデータフィルタリング仕様
- 制限と制約（Quotas & Limits）
- 料金体系（Pricing）
- AWS Lambda 実装コード（ベクトルデータ操作 API 5種完全対応）
- ユースケースと選定基準
- 参照URL・公式リンク集
1. サービス概要
▲ 目次に戻る
Amazon S3 Vectors は、AIエージェント、推論、RAG（Retrieval-Augmented Generation / 検索拡張生成）、セマンティック検索向けに最適化された完全サーバーレスのベクトルストレージ機能です。
主な特徴とメリット
- 完全サーバーレス・ゼロ管理: インスタンスやクラスタのプロビジョニング、キャパシティプランニング、インデックス再構築のインフラ管理が不要。
- S3同等の堅牢性と可用性: 99.999999999%（11ナイン）のデータ耐久性と強い整合性（Read-after-write consistency）を提供。
- レイテンシ特性: アクセス頻度の低いクエリでサブ秒台（1秒未満）、頻度の高いクエリでは最速約100ミリ秒台のレイテンシを実現。
- 優れたコスト効率: クエリ頻度が中・低頻度なRAG用途において、常時起動型のベクトルDB（Amazon OpenSearch Service、Pinecone等）と比較して最大90%のコスト削減が可能。
- 一貫したセキュリティモデル: S3既存のバケットポリシーや IAM ポリシーをそのまま適用可能。
2. アーキテクチャと階層構造
▲ 目次に戻る
S3 Vectors は以下の階層構造でリソースを管理します。
AWS Account / Region
└── Vector Bucket (ベクトル専用バケット)
└── Vector Index (次元数・距離指標・データ型を設定)
└── Vectors (Key + float32 Embedding + Metadata)
3. 主要コンポーネント詳細
▲ 目次に戻る
(1) Vector Bucket（ベクトルバケット）
- ベクトルデータおよびインデックスを格納するための専用バケットタイプ。
- 通常の S3 オブジェクトバケットとは区別され、専用のバケット管理 API / ポリシーで管理されます。
- アカウント・リージョンごとに最大 10,000 個まで作成可能（作成・維持費は無料）。
(2) Vector Index（ベクトルインデックス）
- ベクトルバケット内に作成される検索インデックス。
- インデックス作成時に以下のパラメーターを設定します：
- Dimension（次元数）: 1 〜 4,096
- DataType（データ型）:
float32
- DistanceMetric（距離指標）:
COSINE（コサイン類似度）、EUCLIDEAN（ユークリッド距離）、DOT_PRODUCT（内積）
- NonFilterableMetadataKeys（非フィルタメタデータキー）: 検索フィルタには使用しないが、ドキュメント本文等を保存するためのキー（最大10個）。
(3) Vectors（ベクトルデータ）
- 各ベクトルは以下の3要素で構成されます：
- Key: インデックス内で一意な識別子（文字列）。
- Data:
float32 配列の数値埋め込みベクトル。
- Metadata: 属性情報（最大40KB、キー数は最大50個）。
4. 全API & SDK リファレンス一覧（全19種完全網羅）
▲ 目次に戻る
S3 Vectors は専用の名前空間（API: s3vectors:* / boto3: s3vectors）を使用します。
提供されている 全19種のAPI をカテゴリ別に完全に網羅しています。
(1) ベクトルデータ操作 API（5種）
(2) バケット管理 API（4種）
(3) バケットポリシー管理 API（3種）
(4) インデックス管理 API（4種）
(5) タグ管理 API（3種）
5. メタデータフィルタリング仕様
▲ 目次に戻る
🔗 メタデータフィルタリング (Metadata filtering) - ユーザーガイド
(1) 同時評価方式（In-Tandem Filtering）
S3 Vectors は類似度計算とメタデータフィルタリングを同時に実行します。検索後にフィルタで除外する「Post-filtering」とは異なり、条件を満たす上位 $K$ 件を確実に抽出できます。
(2) フィルタ可能 vs フィルタ不可メタデータ
| 区分 |
フィルタ可能（Filterable） |
フィルタ不可（Non-Filterable） |
| 主な用途 |
カテゴリ、日付、ユーザーID等のクエリ絞り込み条件 |
ドキュメント本文、長文サマリー等のデータ保持 |
| サイズ上限 |
最大 2 KB (2,048 bytes) / ベクトル |
合計40KBの残枠（最大約38KB） / ベクトル |
| キー数上限 |
合計50キー以内 |
最大 10 キー（インデックス作成時に指定） |
| クエリ利用 |
filter 句で指定可能 |
filter 句では指定不可（取得のみ） |
2KB超過エラーへの対策: Amazon Bedrock 等と連携する際、自動付与されるメタデータが 2KB を超えてエラーになるケースがあります。本文等の長文テキストは必ずインデックス作成時に NonFilterableMetadataKeys に指定してください。
(3) サポートされる演算子（MongoDB互換構文）
| 演算子 |
説明 |
正しい構文例 |
$eq |
完全一致 |
{"category": {"$eq": "technical"}} |
$ne |
不一致 |
{"status": {"$ne": "deleted"}} |
$gt / $gte |
より大きい / 以上 |
{"year": {"$gte": 2025}} |
$lt / $lte |
より小さい / 以下 |
{"price": {"$lte": 1000}} |
$in |
配列内のいずれかに一致 |
{"tag": {"$in": ["aws", "cloud"]}} |
$nin |
配列内のいずれにも一致しない |
{"dept": {"$nin": ["hr", "legal"]}} |
$exists |
フィールドの存在確認 |
{"author": {"$exists": true}} |
$and / $or |
論理AND / 論理OR（複数条件指定時は必須） |
{"$and": [{"category": {"$eq": "technical"}}, {"year": {"$gte": 2026}}]} |
[!WARNING]
複数条件指定時の重要ルール（Invalid filter エラーの防止）:
S3 Vectors では、トップレベルに複数のキーを並べる暗黙的 AND（例: {"category": {...}, "year": {...}}）は構文エラー（Invalid filter）となります。
複数のフィルタ条件を指定する場合は、必ず {"$and": [ {条件1}, {条件2} ]} または {"$or": [ ... ]} の配列形式でラップしてください。
必要な IAM 権限: メタデータフィルタを指定する場合、または検索結果にメタデータを含めて取得（returnMetadata=True）する場合は、s3vectors:QueryVectors に加えて s3vectors:GetVectors 権限が必要です。
6. 制限と制約（Quotas & Limits）
▲ 目次に戻る
🔗 制限と制約 (Limitations and restrictions) - ユーザーガイド
| 項目 |
上限・仕様値 |
| ベクトルバケット数 |
10,000 / リージョン / アカウント |
| ベクトルインデックス数 |
10,000 / ベクトルバケット |
| ベクトル格納数 |
最大 20億 (2 Billion) / インデックス |
| 次元数 (Dimension) |
1 〜 4,096 次元 |
| サポートデータ型 |
float32 |
| 距離指標 |
COSINE, EUCLIDEAN, DOT_PRODUCT |
| 総メタデータサイズ |
最大 40 KB / ベクトル |
| フィルタ可能メタデータサイズ |
最大 2 KB (2,048 bytes) / ベクトル |
| メタデータキー数 |
合計最大 50 キー / ベクトル |
| 非フィルタメタデータキー数 |
最大 10 キー / インデックス |
| PutVectors バッチ件数 |
最大 500 件 / 1リクエスト |
| QueryVectors topK 取得件数 |
最大 100 件 / 1クエリ |
| リクエストレート制限 |
約 1,000 write req/sec/index（超過時は 429 TooManyRequestsException） |
7. 料金体系（Pricing）
▲ 目次に戻る
🔗 Amazon S3 料金表 - AWS 公式
S3 Vectors は完全な従量課金制です（プロビジョンド料金なし）。
| 課金項目 |
単価の目安（us-east-1等） |
課金対象 |
| ベクトルバケット作成・維持 |
無料 ($0.00) |
バケット自体の維持コストなし |
| 論理ストレージ料金 |
約 $0.06 / GB-月 |
ベクトルデータおよびメタデータの合計容量 |
| データ書き込み (PUT) |
約 $0.20 / GB |
アップロードされたデータ転送量 |
| クエリ API 料金 |
約 $2.50 / 100万クエリ |
API リクエスト回数 |
| クエリデータ処理料金 |
インデックスサイズに応じた従量課金 |
1,000万件超の大規模インデックス向けに最大80%割引 |
8. AWS Lambda 実装コード（ベクトルデータ操作 API 5種完全対応）
▲ 目次に戻る
S3 Vectors の 「ベクトルデータ操作 API（5種）」 である PutVectors, QueryVectors, GetVectors, DeleteVectors, ListVectors の全操作に対応した AWS Lambda 実装です。
boto3 の仕様（引数名が lowerCamelCase: vectorBucketName, indexName 等）に完全準拠しており、AWS Lambda コンソールに直接貼り付けて即座に実行できます。
(1) Lambda 実行に必要な IAM ポリシー（IAM Policy）
Lambda 関数の実行ロール（Execution Role）にアタッチする IAM ポリシーです。
※ AWS マネージドポリシー AWSLambdaBasicExecutionRole に加え、以下の S3 Vectors 権限が必要です。
① 最小権限ポリシー（推奨・特定バケット/インデックス限定）
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "S3VectorsDataPlanePermissions",
"Effect": "Allow",
"Action": [
"s3vectors:PutVectors",
"s3vectors:QueryVectors",
"s3vectors:GetVectors",
"s3vectors:DeleteVectors",
"s3vectors:ListVectors"
],
"Resource": [
"arn:aws:s3vectors:*:*:bucket/my-knowledge-vector-bucket",
"arn:aws:s3vectors:*:*:bucket/my-knowledge-vector-bucket/index/*"
]
},
{
"Sid": "CloudWatchLogsLogging",
"Effect": "Allow",
"Action": [
"logs:CreateLogGroup",
"logs:CreateLogStream",
"logs:PutLogEvents"
],
"Resource": "arn:aws:logs:*:*:log-group:/aws/lambda/*"
}
]
}
[!IMPORTANT]
QueryVectors 実行時の必須権限:
s3vectors:QueryVectors でメタデータフィルタを使用する場合、または検索結果にメタデータを含めて取得（returnMetadata=True）する場合は、s3vectors:QueryVectors だけでなく s3vectors:GetVectors の権限も必須となります。
② 開発・検証用ポリシー（全リソース対象）
{
"Version": "2012-10-17",
"Statement": [
{
"Sid": "S3VectorsDataPlaneAllAccess",
"Effect": "Allow",
"Action": [
"s3vectors:PutVectors",
"s3vectors:QueryVectors",
"s3vectors:GetVectors",
"s3vectors:DeleteVectors",
"s3vectors:ListVectors"
],
"Resource": "*"
}
]
}
(2) AWS Lambda ハンドラーコード (Python)
"""
AWS Lambda 関数: Amazon S3 Vectors ベクトルデータ操作ハンドラー
- 対象API（5種）:
1. PutVectors : ベクトルの一括登録・更新（最大500件/リクエスト）
2. QueryVectors : 類似度検索 (ANN) ＋ In-Tandem メタデータフィルタリング
3. GetVectors : キー指定によるベクトルデータ・メタデータの取得
4. DeleteVectors : キー指定によるベクトルデータの一括削除
5. ListVectors : インデックス内のベクトルキー一覧取得（ページネーション対応）
- 全API呼び出しおよびLambdaリクエスト/レスポンスの全量ログ出力を実装
- 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 クライアント初期化
# ※ コールドスタート対策としてハンドラー外で一度だけ初期化し接続を再利用
# ※ boto3 >= 1.39.16 が必要
s3vectors_client = boto3.client("s3vectors", region_name=DEFAULT_REGION)
def build_response(status_code: int, body: Dict[str, Any]) -> Dict[str, Any]:
"""
API Gateway プロキシ統合互換のレスポンス辞書を構築する共通ヘルパー関数。
返却するレスポンスの全量（ステータスコード、ヘッダー、ボディ）をログに出力する。
Args:
status_code (int): HTTPステータスコード (200, 400, 500等)
body (Dict[str, Any]): レスポンスボディとして返却する辞書データ
Returns:
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: ベクトルのバッチ登録・更新
# ==============================================================================
def handle_put_vectors(bucket_name: str, index_name: str, vectors: List[Dict[str, Any]]) -> Dict[str, Any]:
"""
指定されたインデックスへベクトルデータを一括登録（UPSERT）する。
実装意図:
- S3 Vectors の仕様上、1リクエストあたりの最大登録数は500件のため事前にバリデーションを実施。
- 各ベクトルの float32 埋め込みデータが不正な形式でないかを検査し、純粋な float 型リストへ変換。
- boto3 引数名は lowerCamelCase (vectorBucketName, indexName, vectors) を指定。
- S3 Vectors API 送信リクエスト全量および受信レスポンス全量をログ出力。
"""
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 型へキャスト
float_values = [float(val) for val in data["float32"]]
formatted_vectors.append({
"key": str(key),
"data": {"float32": float_values},
"metadata": metadata
})
# APIリクエスト全量の構築とログ出力
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)}")
# API呼び出し
response = s3vectors_client.put_vectors(**api_request_payload)
# APIレスポンス全量のログ出力
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]:
"""
クエリベクトルに基づく近似最近傍探索 (ANN) を実行する。
実装意図:
- S3 Vectors は探索とフィルタリングを同時に実行する In-Tandem Filtering を採用。
- topK は 1〜100 の範囲に正規化して API 制約違反を防止。
- メタデータフィルタやメタデータ返却には s3vectors:GetVectors 権限が必要。
- S3 Vectors API 送信リクエスト全量および受信レスポンス全量をログ出力。
"""
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:
# S3 Vectors の仕様上、トップレベルに複数条件がある場合は {"$and": [...]} 配列でラップが必要
# 例: {"category": {"$eq": "technical"}, "year": {"$gte": 2026}} -> {"$and": [{"category": ...}, {"year": ...}]}
if (
isinstance(metadata_filter, dict)
and len(metadata_filter) > 1
and "$and" not in metadata_filter
and "$or" not in metadata_filter
):
logger.info("[QueryVectors] Auto-wrapping top-level multi-key filter into '$and' array for S3 Vectors compliance.")
normalized_filter = {
"$and": [{k: v} for k, v in metadata_filter.items()]
}
else:
normalized_filter = metadata_filter
query_params["filter"] = normalized_filter
# APIリクエスト全量のログ出力
logger.info(f"[QueryVectors API Request] Payload: {json.dumps(query_params, ensure_ascii=False, default=str)}")
# API呼び出し
response = s3vectors_client.query_vectors(**query_params)
# APIレスポンス全量のログ出力
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: キー指定によるベクトル・メタデータ取得
# ==============================================================================
def handle_get_vectors(
bucket_name: str,
index_name: str,
keys: List[str],
return_data: bool = True,
return_metadata: bool = True
) -> Dict[str, Any]:
"""
キーを指定してベクトルデータおよびメタデータを直接取得する。
実装意図:
- returnData / returnMetadata フラグにより、ベクトル数値データやメタデータの取得有無を制御可能。
- 取得成功したベクトルリストと、インデックス内に存在しなかった notFoundKeys の双方を返却。
- S3 Vectors API 送信リクエスト全量および受信レスポンス全量をログ出力。
"""
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
}
# APIリクエスト全量のログ出力
logger.info(f"[GetVectors API Request] Payload: {json.dumps(api_request_payload, ensure_ascii=False, default=str)}")
# API呼び出し
response = s3vectors_client.get_vectors(**api_request_payload)
# APIレスポンス全量のログ出力
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: キー指定によるベクトル一括削除
# ==============================================================================
def handle_delete_vectors(bucket_name: str, index_name: str, keys: List[str]) -> Dict[str, Any]:
"""
指定されたキーに一致するベクトルデータをインデックスから削除する。
実装意図:
- 削除対象キー配列を受け取り、一括削除を実行して結果ステータスを返却。
- S3 Vectors API 送信リクエスト全量および受信レスポンス全量をログ出力。
"""
if not keys:
raise ValueError("Parameter 'keys' list is required for deletion.")
api_request_payload = {
"vectorBucketName": bucket_name,
"indexName": index_name,
"keys": keys
}
# APIリクエスト全量のログ出力
logger.info(f"[DeleteVectors API Request] Payload: {json.dumps(api_request_payload, ensure_ascii=False, default=str)}")
# API呼び出し
response = s3vectors_client.delete_vectors(**api_request_payload)
# APIレスポンス全量のログ出力
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: ベクトル一覧の取得（ページネーション完全対応）
# ==============================================================================
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]:
"""
インデックス内に登録されているベクトル一覧をページネーション取得する。
実装意図:
- S3 Vectors の ListVectors レスポンス仕様は `vectors` キーにリスト（[{"key": "..."}, ...]）が格納される。
- maxResults は 1〜500 の範囲に正規化（S3 Vectors の単一リクエスト上限は 500 件）。
- returnData / returnMetadata フラグにより、キーだけでなくベクトル数値やメタデータも取得可能（要 s3vectors:GetVectors 権限）。
- fetch_all=True を指定した場合、nextToken を自動走査して最大 max_total_results 件まで一括取得可能。
"""
page_size = min(max(1, int(max_results)), 500)
# 1. 全件自動取得モード (fetch_all=True)
if fetch_all:
all_vectors = []
current_token = next_token
page_count = 0
logger.info(f"[ListVectors (fetchAll)] Starting auto-pagination on {bucket_name}/{index_name} (max_total={max_total_results})")
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
# APIリクエスト全量のログ出力
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)
# APIレスポンス全量のログ出力
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")
# ページごとの累積件数と進行状況をログ出力
logger.info(f"[ListVectors (fetchAll Progress)] Page {page_count}: received {len(page_vectors)} vectors, totalCount: {len(all_vectors)}, nextToken: {current_token}")
# 終了条件: 次ページトークンが存在しない、または安全上限件数に到達
if not current_token or len(all_vectors) >= max_total_results:
break
logger.info(f"[ListVectors (fetchAll Completed)] totalPages: {page_count}, totalCount: {len(all_vectors)}, nextToken: {current_token}")
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)
}
# 2. 単一ページ取得モード (fetch_all=False)
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
# APIリクエスト全量のログ出力
logger.info(f"[ListVectors API Request] Payload: {json.dumps(params, ensure_ascii=False, default=str)}")
# API呼び出し
response = s3vectors_client.list_vectors(**params)
# APIレスポンス全量のログ出力
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
}
# ==============================================================================
# メインエントリポイント: lambda_handler
# ==============================================================================
def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
"""
AWS Lambda メインハンドラー
- API Gateway プロキシペイロードおよびダイレクト呼び出しイベントを自動識別・パース
- 要求された action に応じて各ベクトルデータ操作ハンドラーへルーティング
"""
# Lambda呼び出しイベント全量のログ出力
logger.info(f"[Lambda Event Input] {json.dumps(event, ensure_ascii=False, default=str)}")
try:
# 1. リクエストボディのパース処理
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"]
# クエリパラメータが存在する場合はマージ（GETリクエスト対応）
if "queryStringParameters" in event and event["queryStringParameters"]:
payload.update(event["queryStringParameters"])
# 2. アクションパラメータの検証
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"
]
})
# 3. 対象バケット名・インデックス名の決定
bucket_name = payload.get("vectorBucketName", DEFAULT_BUCKET_NAME)
index_name = payload.get("indexName", DEFAULT_INDEX_NAME)
# 4. 各アクションへのディスパッチ
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:
# クライアント側の入力値バリデーションエラー (400 Bad Request)
logger.error(f"[Validation Error] {ve}")
return build_response(400, {"error": str(ve)})
except ParamValidationError as pve:
# boto3 のパラメータ型/キー名不正エラー (400 Bad Request)
logger.error(f"[boto3 ParamValidationError] {pve}")
return build_response(400, {
"error": "Parameter validation failed",
"detail": str(pve)
})
except ClientError as ce:
# AWS S3 Vectors サービス側のエラー (403/404/429/500 等)
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}, FullResponse: {json.dumps(ce.response, ensure_ascii=False, default=str)}")
return build_response(status_code, {
"error": "AWS Service Error",
"code": error_code,
"message": error_message,
"rawErrorResponse": ce.response
})
except Exception as e:
# 予期せぬ内部サーバーエラー (500 Internal Server Error)
logger.error(f"[Internal Server Error] {e}", exc_info=True)
return build_response(500, {"error": "Internal server error", "detail": str(e)})
(3) Lambda テストイベント例（JSON: ベクトルデータ操作 API 5種）
AWS Lambda コンソールの「テストイベント」作成時にそのまま使用できる5種類すべてのJSON定義です。
① 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年度の財務レポート概要です。"
}
}
]
}
② 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 } }
]
}
}
③ get_vectors（キー指定によるベクトル取得）
{
"action": "get_vectors",
"vectorBucketName": "my-knowledge-vector-bucket",
"indexName": "document-embeddings-index",
"keys": ["doc-001", "doc-002"],
"returnData": true,
"returnMetadata": true
}
④ list_vectors（ベクトル一覧取得・ページネーション例）
パターンA: 単一ページ取得（初回リクエスト）
{
"action": "list_vectors",
"vectorBucketName": "my-knowledge-vector-bucket",
"indexName": "document-embeddings-index",
"maxResults": 10,
"returnData": false,
"returnMetadata": false
}
パターンB: 次ページ取得（レスポンスの nextToken を指定）
{
"action": "list_vectors",
"vectorBucketName": "my-knowledge-vector-bucket",
"indexName": "document-embeddings-index",
"maxResults": 10,
"nextToken": "eyJleGNsdXNpdmVTdGFydEtleSI6ICJkb2MtMDEwIn0="
}
パターンC: 全件自動取得モード（fetchAll: true）
{
"action": "list_vectors",
"vectorBucketName": "my-knowledge-vector-bucket",
"indexName": "document-embeddings-index",
"fetchAll": true,
"maxTotalResults": 500
}
⑤ delete_vectors（キー指定によるベクトル一括削除）
{
"action": "delete_vectors",
"vectorBucketName": "my-knowledge-vector-bucket",
"indexName": "document-embeddings-index",
"keys": ["doc-001", "doc-002"]
}
9. ユースケースと選定基準
▲ 目次に戻る
| 比較項目 |
Amazon S3 Vectors |
Amazon OpenSearch Service / 専用ベクトルDB |
| アーキテクチャ |
完全サーバーレス |
クラスタ / インスタンスプロビジョニング |
| 維持費用 |
従量課金のみ ($0〜) |
常時起動インスタンス費用が発生 |
| レイテンシ |
100ms 〜 サブ秒台 |
数ミリ秒 (Single-digit ms) |
| 全文検索連携 |
ベクトル検索に特化 |
BM25全文検索 + ベクトルのハイブリッド検索 |
| 主な適合ケース |
社内文書RAG、ナレッジベース、中低頻度検索、アーカイブ |
リアルタイムレコメンド、高QPS EC検索、高度な複合検索 |
10. 参照URL・公式リンク集
▲ 目次に戻る
(1) ユーザーガイド & サービス仕様
(2) Python SDK (boto3) リファレンス
(3) AWS REST API リファレンス
