DynamoDBネイティブベクトル検索をAWSコンソールとPythonで構築する

はじめに

前編では、Amazon DynamoDBのネイティブベクトル検索の仕組み、主な用途、他のAWSサービスとの使い分け、設計上の注意点を解説しました。

本記事では、スポーツ用品の商品カタログを題材に、DynamoDBテーブルとベクトルインデックスをAWSマネジメントコンソールから作成します。その後、PythonからAmazon Bedrockで商品説明の埋め込みを生成してDynamoDBへ登録し、SearchVectors APIで通常検索とフィルター付き検索を実行します。
AWSリソースの設定をコンソールで一つずつ確認し、アプリケーションから利用する部分だけをPythonで実装することで、インフラとAPIの双方から機能を理解することを目的とします。

今回構築するセマンティック商品検索

今回のデモでは、5件のスポーツ用品をDynamoDBへ登録します。各アイテムには、商品ID、商品名、カテゴリ、説明文、説明文から生成した1,024次元のベクトルを保存します。
検索時は「暑い季節に長距離を走るための軽い靴」という文章をAmazon Titan Text Embeddings V2でベクトル化します。そのベクトルをDynamoDBへ渡し、商品説明の意味が近い順に上位3件を取得します。さらに、インラインフィルターでカテゴリを「シューズ」に限定した検索も実行します。
処理の流れは次のとおりです。

1. AWSマネジメントコンソールでDynamoDBテーブルを作成する
2. 商品説明のベクトル属性を対象とするベクトルインデックスを作成する
3. PythonからBedrockを呼び出し、商品説明の埋め込みを生成する
4. 商品属性と埋め込みをDynamoDBへ登録する
5. 検索文の埋め込みを生成し、SearchVectors APIを呼び出す
6. フィルターなしとカテゴリ指定ありの検索結果を比較する

本記事では、ベクトルインデックスのルーティングキーを設定しません。5件だけを登録する学習用デモであり、検索時の必須条件を減らすためです。本番環境や大規模データでは、テナントやマーケットプレイスなど、アクセスパターンに合うルーティングキーを検討してください。

なお、AWSマネジメントコンソールでは「ルーティングキー」と表示されますが、AWS公式ブログや開発者ガイドでは「ベクトルインデックスのパーティションキー」と説明されている場合があります。どちらも検索対象を特定の属性値に限定し、大規模なデータセットで検索処理を分散するための同じ設定を指します。本記事のコンソール操作では、画面表示に合わせて「ルーティングキー」と表記します。

事前準備

今回使用する設定は以下のとおりです。

Pythonコードの実行には、少なくともBedrockのInvokeModel、DynamoDBのPutItemとSearchVectorsを許可するIAM権限が必要です。コンソールでテーブルとインデックスを作成・削除する権限も用意します。本番環境では、対象リージョン、テーブル、モデルを限定した最小権限のIAMポリシーを設計してください。
Amazon Titan Text Embeddings V2を対象リージョンで呼び出せることも事前に確認します。モデルの提供リージョンやアクセス方法は更新される可能性があるため、実行時点のAmazon Bedrock公式ドキュメントを確認してください。

DynamoDBテーブルをコンソールから作成する

DynamoDBコンソールを開き、ナビゲーションペインの「テーブル」から「テーブルの作成」を選択します。
テーブル名にProductCatalog、パーティションキーにproductIdを入力し、キーの型はStringを選択します。ソートキーは設定しません。
ベクトルインデックスはオンデマンド容量モードのテーブルでのみ利用できます。「テーブル設定」を確認し、読み込み/書き込みキャパシティーモードにオンデマンドを設定してからテーブルを作成します。その他の設定は検証環境に合わせて選択してください。

テーブルのステータスが「アクティブ」になったことを確認してから、ベクトルインデックスを作成します。

ベクトルインデックスをコンソールから作成する

ProductCatalogテーブルを開き、「インデックス」タブから「ベクトルインデックスを作成」を選択します。コンソールの表記は更新によって変わる場合がありますが、設定する内容は次のとおりです。

属性の射影には、コンソール上でAll、Only keys、Includeの3種類があります。API上の指定値は、それぞれALL、KEYS_ONLY、INCLUDEです。

Only keysを選んだ場合、商品名や説明文などの非キー属性はベクトル検索から返せません。今回の構成では、埋め込みベクトルはデフォルトで検索結果に含まれないため、主にproductId、インラインフィルターに指定したcategory、類似度スコアが得られます。商品情報が必要な場合は、返されたproductIdを使ってGetItemまたはBatchGetItemを実行します。上位数件をまとめて取得する用途では、通常はBatchGetItemが適しています。
このチュートリアルでは、検索結果だけで商品名や説明文を確認できるようにAllを選びます。本番環境では、インデックスの保存量と検索で処理するデータ量を抑えるため、必要な属性だけを射影するIncludeや、ベーステーブルから別途取得するOnly keysも検討してください。
「ベクトルインデックスを作成」を選択すると、インデックスの作成が非同期に進みます。インデックスのステータスが「アクティブ」になり、バックフィルが完了してからデータを登録します。

Pythonで商品データと埋め込みを登録する

本記事のサンプルコードでは、Boto3 1.43.64以降とpython-dotenvを使用します。認証にはAWS CLIで設定したプロファイルを使用し、リージョンは環境変数から取得します。
今回のPython実装は、AWS公式記事「Amazon DynamoDBのネイティブベクトルサポートでセマンティック検索を構築する」と、そのDynamoDBベクトル検索サンプルを基準に、商品検索用の構成へ置き換えています。
register_products.pyは、5件の商品説明を1件ずつBedrockへ送り、1,024次元の埋め込みを生成します。その後、商品属性とベクトルを低レベルのDynamoDBクライアントで登録します。

import json
import os

import boto3
from dotenv import load_dotenv

load_dotenv()

REGION = os.environ["AWS_REGION"]
TABLE_NAME = "ProductCatalog"
MODEL_ID = "amazon.titan-embed-text-v2:0"
EMBEDDING_DIMENSIONS = 1024

PRODUCTS = [
    {
        "productId": "P001",
        "productName": "エアフローランニングシューズ",
        "category": "シューズ",
        "description": "通気性の高いメッシュ素材と軽量ソールを採用した、夏の長距離ランニング向けシューズです。",
    },
    {
        "productId": "P002",
        "productName": "ウォータープルーフトレッキングブーツ",
        "category": "シューズ",
        "description": "防水性と足首の安定性を備えた、雨天や岩場での登山向けブーツです。",
    },
    {
        "productId": "P003",
        "productName": "ライトランニングジャケット",
        "category": "ウェア",
        "description": "薄手で持ち運びやすく、早朝のランニングで風を防ぐ軽量ジャケットです。",
    },
    {
        "productId": "P004",
        "productName": "クッションヨガマット",
        "category": "アクセサリー",
        "description": "関節への負担を抑える厚手のクッションを備えた、室内運動向けのヨガマットです。",
    },
    {
        "productId": "P005",
        "productName": "スピードサッカーシューズ",
        "category": "シューズ",
        "description": "芝のグラウンドで素早く方向転換できる、グリップ力の高いサッカーシューズです。",
    },
]

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


def generate_embedding(text: str) -> list[float]:
    response = bedrock.invoke_model(
        modelId=MODEL_ID,
        contentType="application/json",
        accept="application/json",
        body=json.dumps(
            {"inputText": text, "dimensions": EMBEDDING_DIMENSIONS, "normalize": True}
        ),
    )
    return json.loads(response["body"].read())["embedding"]


for product in PRODUCTS:
    embedding = generate_embedding(product["description"])
    dynamodb.put_item(
        TableName=TABLE_NAME,
        Item={
            "productId": {"S": product["productId"]},
            "productName": {"S": product["productName"]},
            "category": {"S": product["category"]},
            "description": {"S": product["description"]},
            "embedding": {"L": [{"N": str(value)} for value in embedding]},
        },
    )
    print(f"登録完了: {product['productId']} {product['productName']}")

Bedrockへのリクエストでは、インデックス設定と同じ1,024次元を指定しています。normalizeを有効にして単位ベクトルへ正規化していますが、今回のインデックスはCOSINEを使用するため、正規化前のベクトルでも方向に基づいて比較できます。
DynamoDBへは、埋め込みをList型、その各要素をNumber型として送信します。埋め込みの次元数がインデックス設定と一致しないアイテムは、ベクトルインデックスへ正しく追加できません。
register_products.pyを実行後、DynamoDBコンソールでProductCatalogテーブルを開き、5件の商品が登録されていることを確認します。

ベクトルインデックスはテーブルへの書き込み後に非同期で更新されます。続けて検索した際に一部の商品が現れない場合は、時間を置いてから再実行してください。

Pythonで通常検索とフィルター付き検索を実行する

search_products.pyでは、商品登録時と同じBedrockモデルを使って検索文をベクトル化し、SearchVectors APIへ渡します。最初にインデックス全体を検索し、次にcategory = シューズという検索条件を追加して結果を比較します。
SearchVectorsのリクエストとレスポンスの詳細は、DynamoDB APIリファレンスを参照してください。

import json
import os

import boto3
from boto3.dynamodb.types import TypeDeserializer
from dotenv import load_dotenv

load_dotenv()

REGION = os.environ["AWS_REGION"]
TABLE_NAME = "ProductCatalog"
INDEX_NAME = "ProductDescriptionIndex"
MODEL_ID = "amazon.titan-embed-text-v2:0"
EMBEDDING_DIMENSIONS = 1024

QUERY_TEXT = "暑い季節に長距離を走るための軽い靴"
FILTER_CATEGORY = "シューズ"

bedrock = boto3.client("bedrock-runtime", region_name=REGION)
dynamodb = boto3.client("dynamodb", region_name=REGION)
deserializer = TypeDeserializer()


def generate_embedding(text: str) -> list[float]:
    response = bedrock.invoke_model(
        modelId=MODEL_ID,
        contentType="application/json",
        accept="application/json",
        body=json.dumps(
            {"inputText": text, "dimensions": EMBEDDING_DIMENSIONS, "normalize": True}
        ),
    )
    return json.loads(response["body"].read())["embedding"]


def search_products(category: str | None = None) -> dict:
    query_embedding = generate_embedding(QUERY_TEXT)
    request = {
        "TableName": TABLE_NAME,
        "IndexName": INDEX_NAME,
        "SearchVector": [{"N": str(value)} for value in query_embedding],
        "TopK": 3,
        "ReturnConsumedCapacity": "INDEXES",
    }

    if category:
        request.update(
            {
                "SearchConditionExpression": "#category = :category",
                "ExpressionAttributeNames": {"#category": "category"},
                "ExpressionAttributeValues": {":category": {"S": category}},
            }
        )

    return dynamodb.search_vectors(**request)


def print_results(title: str, response: dict) -> None:
    print(f"\n{title}")
    for rank, result in enumerate(response["SearchResults"], 1):
        item = {
            name: deserializer.deserialize(value)
            for name, value in result["Item"].items()
        }
        print(
            f"{rank}. {item['productName']} / "
            f"{item['category']} / score={result['Score']:.6f}"
        )

    consumed = response.get("ConsumedCapacity", {})
    print("VectorSearchRequestBytes:", consumed.get("VectorSearchRequestBytes"))


print(f"検索文: {QUERY_TEXT}")
print_results("フィルターなし", search_products())
print_results(
    f"カテゴリーフィルター: {FILTER_CATEGORY}",
    search_products(FILTER_CATEGORY),
)

SearchVectorsのレスポンスに含まれる商品属性は、TypeDeserializerでDynamoDBのAttributeValue形式から通常のPython値へ変換してから表示しています。
フィルターなしの検索では、カテゴリを限定せず、検索文に意味が近い商品を上位3件取得します。続くフィルター付き検索では、インデックス作成時にインラインフィルターとして登録したcategoryをSearchConditionExpressionで指定し、検索対象を「シューズ」に限定します。

なお現在、ルーティングキーとインラインフィルターで使用できる演算子は等価演算子だけです。IN句や不等号演算子は使えないため注意しましょう。

コサイン距離ではスコアが小さいほど類似しており、0は同じ方向のベクトルを表します。
インラインフィルターは検索結果を条件に合うアイテムへ限定しますが、ルーティングキーのように検索範囲自体を分割するものではないため、検索処理量の削減を保証するものではありません。
レスポンスのVectorSearchRequestBytesは、検索で処理されたデータ量です。ルーティングキーで検索範囲を分割する、不要な属性を射影しない、必要以上に大きな次元を選ばないといった設計は、この値と検索コストの削減につながります。

検証後の確認とリソース削除

今回の検証では、次の点を確認します。

• キーワードが完全一致しなくても、商品説明の意味が近い順に取得できること
• コサイン距離のスコアが小さい順に結果が並ぶこと
• フィルター付き検索では、シューズ以外の商品が結果に含まれないこと
• 検索結果に商品名やカテゴリなどの通常属性が含まれること
• VectorSearchRequestBytesから検索で処理したデータ量を確認できること

ANN検索は近似検索であるため、同じデータでも結果の細かな順序が変わる可能性があります。単発の順位だけで品質を判断せず、実際の検索文と正解データを用意して再現率や業務上の適合率を評価してください。
検証が完了したら、DynamoDBコンソールでProductCatalogテーブルを選択し、テーブルを削除します。テーブルを削除すると、そのベクトルインデックスも削除されます。Bedrockのモデル呼び出しとDynamoDBのベクトル書き込み・検索には利用量に応じた料金が発生するため、不要なリソースと検証処理を残さないようにします。

まとめ

本記事では、DynamoDBテーブルとベクトルインデックスをAWSマネジメントコンソールから作成し、PythonからAmazon BedrockとDynamoDBを呼び出してセマンティック商品検索を構築しました。
実装上の要点は次のとおりです。

• テーブルはオンデマンド容量モードで作成する
• ベクトルインデックスの次元数を埋め込みモデルの出力と一致させる
• 商品登録と検索で同じ埋め込みモデルと次元数を使用する
• SearchVectors APIのSearchConditionExpressionで等価フィルターを指定する
• インデックスの非同期更新とANN検索の性質を考慮して結果を評価する
• 検証後はテーブルを削除して継続的な料金を防ぐ

今回の構成は、DynamoDBに保存する業務データへ意味検索を追加する最小構成です。本番導入時は、データ量とアクセスパターンに合わせたルーティングキー、必要最小限の射影属性、検索品質の評価方法、再試行や監視を追加してください。
ベクトル検索のアーキテクチャや他サービスとの使い分けは、前編で解説しています。

生成AI活用支援サービスのご紹介

Tech Funでは、お客様のフェーズに合わせ、生成AI活用に向けた様々な支援をご提供しています。

1. 無料診断パック：業務・プロセスの現状を無料で診断し、生成AI活用の可能性をレポートします。
2. 検証（PoC）パック：診断で有効性が確認された業務を対象に、プロトタイプ構築を支援します。
3. コンサルティングサービス：生成AI導入戦略の策定から運用体制構築までを包括的に支援します。
4. Plain RAG：社内文書をもとに回答し、社内問い合わせを効率化するAIチャットパッケージです。

生成AIに限らず、Web・業務システム開発やインフラ設計など、技術領域を問わずご相談を承っています。「何から始めれば良いか分からない」という段階でも構いませんので、ぜひお気軽にお問い合わせください。

執筆・編集

Tech Fun Magazine R&Dチーム

Tech Funの生成AI研究に携わるエンジニアが、最新のAIモデル動向やプロンプト設計、実業務への応用手法など、生成AIに特化した知見を執筆・編集しています。
モデル評価や業務シナリオに応じたAI活用設計など、日々のR&D活動で得られる実践的なノウハウをわかりやすく紹介します。