[AWS]ベクトル対応した DynamoDB のエンベディング戦略を考えてみる

DynamoDB が 2026年8月5日にベクトル検索へ対応しました。商品データとそのベクトルを同じテーブルに置けるのがありがたいところです。ただしEmmbeding処理は自前でする必要があります。この記事ではその埋め込みを誰がいつ作るのか？という話を考えてみます。

📌 TL;DR

• 登録側はDynamoDB Streamsに任せれば、既存のコードを触らずにベクトルが付与される
• 検索側はベクトル検索のAPIを呼ぶ都合でどのみちアプリに手が入るのでついでにそこで埋め込む
• 登録も共通DAOに寄せる手もあり
• 今後のアプデに期待

📦 データとベクトルを一元管理できる

これまでは商品データをDynamoDBにベクトルをOpenSearchやS3Vectorsに置いていました。同じものの片割れが別のサービスに載っている状態で削除のたびに両方へ消しにいく必要があります。片方が失敗したり消し忘れたりすれば孤児ベクトルが残り、消したはずの商品が検索結果にひょっこり顔を出します

一元管理だとこれが要りません。ライフサイクルを共にするからです。整合性を保つためのコードがまるごと不要になるのが、個人的にはいちばん大きいと感じています。

📖 テーブル作成から検索までの基本的な流れ

こちらの素晴らしい記事をご参照ください。

✍️ 埋め込み(Embedding)は自分で作る

DynamoDB はベクトルを保存して検索しますが、生成はしてくれません。

DynamoDB stores and searches vector embeddings, but it does not generate them.

出典: Creating and searching vector indexes

OpenSearch なら ingest pipeline を挟んでおけば書き込み時に自動で埋め込んでくれます。DynamoDB にその仕組みはありません。素直に実装するとデータ登録実装すべてにemmbeding処理を足して回ることになります。何か所あるか分かりませんし他のチームが持っているものもあります。ここは触らずに済ませたいところです。

公式のベストプラクティスには Streams を使えと書いてありました。

Consider capturing content changes with DynamoDB Streams and using a downstream process to regenerate and rewrite affected embeddings.

出典: Best practices for vector indexes

ただ、具体的な実装例も公式にもブログにも見当たらなかったので自分で組んでみました。

🌊 DynamoDB Streams で Embedding 処理を外に切り出す

Lambda のイベントソースマッピングにフィルタを設定して、INSERT のイベントだけを拾います。

{"eventName":["INSERT"]}

Lambda がやることは3つです。新しく入ったアイテムから ProductName と Description をつないで埋め込みの元になる本文を組み立て、それを Bedrock の Titan Text Embeddings に投げてベクトルを受け取り、同じアイテムへ Embedding 属性として UpdateItem で書き戻します。

書き戻しも Streams には流れますが、こちらは MODIFY なのでフィルタに弾かれて Lambda までは届きません。だからループしません。

embedding_enricher.py

import json
import os

import boto3

TABLE_NAME = os.environ.get("TABLE_NAME", "Products")
MODEL_ID = os.environ.get("MODEL_ID", "amazon.titan-embed-text-v1")
EXPECTED_DIMENSIONS = int(os.environ["DIMENSIONS"])  # インデックスの Dimensions と一致させる

dynamodb = boto3.client("dynamodb")
bedrock = boto3.client("bedrock-runtime")


def lambda_handler(event, context):
    for record in event["Records"]:
        image = record["dynamodb"].get("NewImage")
        if not image:
            continue

        product_id = image["ProductId"]["S"]
        name = image.get("ProductName", {}).get("S", "")
        description = image.get("Description", {}).get("S", "")

        # 両方とも空だと Bedrock が ValidationException を返すので、呼ぶ前に抜ける
        if not name and not description:
            continue

        vector = embed(f"{name}。{description}")

        dynamodb.update_item(
            TableName=TABLE_NAME,
            Key={"ProductId": {"S": product_id}},
            UpdateExpression="SET Embedding = :vector",
            ExpressionAttributeValues={
                ":vector": {"L": [{"N": fmt_num(v)} for v in vector]}
            },
        )
        print(f"{product_id}: Embedding を書き戻しました（{len(vector)} 次元）")


def embed(text):
    response = bedrock.invoke_model(
        modelId=MODEL_ID,
        body=json.dumps({"inputText": text}, ensure_ascii=False),
        accept="application/json",
        contentType="application/json",
    )
    vector = json.loads(response["body"].read())["embedding"]

    # 次元数が違うベクトルは DynamoDB が拒否する。原因が分かりやすいよう手前で落とす
    if len(vector) != EXPECTED_DIMENSIONS:
        raise ValueError(f"次元数が {len(vector)} でした（期待値 {EXPECTED_DIMENSIONS}）")

    return vector


def fmt_num(value):
    """N 型に渡す文字列へ変換する。指数表記になった場合は固定小数点へ直す。"""
    text = f"{value:.9g}"
    if "e" in text or "E" in text:
        text = f"{value:.12f}".rstrip("0") or "0"
    return text

これで登録側は動きます。アプリは Embedding の存在を知らないまま PutItem するだけで済みます。実際に商品を1件入れてみたところ、体感ではわからないレベルの速さでベクトルが付いて検索に乗りました。

🔁 更新に追従させると無限ループ

登録がうまくいくと欲が出て更新にも追従させたくなります。ならば MODIFY も拾えばいい。ところが拾った瞬間、Lambda自身の書き戻しがLambdaを呼び戻して無限ループになります。Streams から見れば「Lambda が埋め込みを書いた」も「人が説明文を直した」も同じ MODIFY で、区別がつかないわけです。

止める手はあります。埋め込みの元にしたテキストの目印をアイテムに残しておけば、自分が書き戻したレコードかどうかをLambda側で判定できます。ここでは深掘りしません。

MODIFY に手を出すなら何らかの対処が必須になります。放っておくと Lambda が自分の書き込みを延々と追いかけ、気づいたときには Bedrock の請求だけが立派に育っています。

🔍 検索はどのみち手が入る

DynamoDB のベクトル検索に自然言語の文字列は渡せません。渡せるのはベクトルだけ。

aws dynamodb search-vectors \
  --table-name Products \
  --index-name ProductEmbeddingIndex \
  --search-vector file://query-vector.json \
  --top-k 5 \
  --search-condition-expression "Category = :cat" \
  --expression-attribute-values '{":cat":{"S":"Electronics"}}'

--search-vector に渡している query-vector.json の中身は、数値がひたすら並んだ配列です。「音楽を聴くための機器」という文字列ではありません。人間が開いても何が書いてあるのかさっぱり分かりませんが、検索のたびにこれを用意する必要があります。しかも登録時と同じモデルで作らないと別の空間にある座標同士を比べることになって結果が意味をなさなくなります。

OpenSearchなら、ここもneural queryが面倒を見てくれます。自然文をそのまま投げれば裏で埋め込みモデルを呼んでベクトル検索まで通してくれる仕組みがあります。DynamoDBにはそれがありません。書き込み側のingest pipelineに当たるものが無いのと同じで検索側にも無いわけです。

登録と同じようにStreamsを使えば？

→　DynamoDBへの書き込みが発生するわけではないのでStreamsは使えないのです

検索機能そのものが新規実装でSearchVectorsという新しいAPIを呼ぶコードをこれから書くのですからアプリに手が入るのは織り込み済みです。埋め込みの1行が増えたところで影響はたかが知れています。どうせ書くならそのついでに埋め込んでしまえばいい！ということでアプリに組み込むという手法を選択しました。

	既存コードへの影響	埋め込みをどこでやるか
登録	Streams ならゼロ	Lambda
検索	API が変わるのでどのみち発生	アプリの中

登録もまとめてしまう手もある

ここまで来るともう一つの選択肢が見えてきます。検索側で埋め込みモジュールを書くことはもう確定しているのですからそれを登録処理からも呼んでしまえばいいわけです。

まず埋め込みを独立したモジュールに切り出します。保存用と検索用でベクトルの形が違うのでその差もここに閉じ込めておきます。

embedding.py

def embed(text):
    """テキストをベクトルへ変換する。登録側と検索側で必ずこれを共有する。"""
    response = bedrock.invoke_model(
        modelId=MODEL_ID,
        body=json.dumps({"inputText": text}, ensure_ascii=False),
        accept="application/json",
        contentType="application/json",
    )
    vector = json.loads(response["body"].read())["embedding"]
    if len(vector) != EXPECTED_DIMENSIONS:
        raise ValueError(f"次元数が {len(vector)} でした（期待値 {EXPECTED_DIMENSIONS}）")
    return vector


def to_ddb_vector(vector):
    """保存用。L 型のリストに N を並べる。"""
    return {"L": [{"N": fmt_num(v)} for v in vector]}


def to_search_vector(vector):
    """検索用。L で包まず素の配列にする。保存用と形が違うので注意。"""
    return [{"N": fmt_num(v)} for v in vector]

検索側はこのモジュールを呼ぶだけです。

search_handler.py

import embedding

def lambda_handler(event, context):
    condition = "Category = :cat"
    values = {":cat": {"S": event["category"]}}

    response = dynamodb.search_vectors(
        TableName=TABLE_NAME,
        IndexName=INDEX_NAME,
        # 登録側と同じモジュールで埋め込む。ここが共通 DAO の肝
        SearchVector=embedding.to_search_vector(embedding.embed(event["query"])),
        SearchConditionExpression=condition,
        ExpressionAttributeValues=values,
        ProjectionExpression="ProductId, ProductName, Brand, Price",
        TopK=event.get("topK", 5),
    )
    return [to_result(r) for r in response["SearchResults"]]

そして保存側も、同じモジュールを DAO の中から呼びます。

product_repository.py

class ProductRepository:
    def save(self, product):
        item = to_ddb_item(product)
        item["Embedding"] = embedding.to_ddb_vector(
            embedding.embed(f"{product.name}。{product.description}")
        )
        self.ddb.put_item(TableName="Products", Item=item)

実際に検索 Lambda を動かすとこうなります。自然言語を投げるだけで結果が返ってきます。

aws lambda invoke --function-name dynamodb-vector-search \
  --cli-binary-format raw-in-base64-out \
  --payload '{"query":"耳をふさがずに音を聞ける機器","category":"Electronics","topK":3}' \
  response.json

response.json

{
  "query": "耳をふさがずに音を聞ける機器",
  "count": 3,
  "results": [
    {"ProductId": "P-021", "ProductName": "骨伝導イヤホン",     "Score": 0.2206},
    {"ProductId": "P-001", "ProductName": "ワイヤレスイヤホン", "Score": 0.2842},
    {"ProductId": "P-002", "ProductName": "Bluetoothヘッドホン", "Score": 0.3071}
  ]
}

呼び出し側は repo.save(product) と書くだけで埋め込みの存在を知らずに済みます。登録と検索が同じモジュールを通るのでモデルを差し替えるときに片方だけ古いまま残る事故も防げます。更新もsave()を通るので、無限ループのガードも棚卸しも要りません。

ただしこれが効くのは、登録処理が限定されている場合に限ります。書き込みが複数のチームや複数の言語から直接来ていて共通のDAOを配れないならテーブルの手前で受け止められるStreamsのほうが現実的だと思います。ログや投稿のように一度書いたら本文を変えないデータも、INSERT だけで完結するのでStreamsが素直にハマります。

🍜 締め

登録はStreamsに任せれば既存のコードを1行も触らずに済みます。検索はベクトル検索のAPIを呼ぶ都合でどのみち手が入るのでそのついでに埋め込めば影響はたかが知れています。

そして検索側で埋め込みモジュールを書くことになるのでそれを登録からも呼ぶ共通DAOに寄せるという三つ目の選択肢が出てきます。どれが正解かは既存の書き込み経路を自分が握っているかどうか次第です。

いずれにせよ、検索側にneural query相当の仕組みが入れば話は変わります。検索文字列をそのまま渡せるようになればアプリから埋め込みが消えて登録側もStreamsに任せきりでよくなります。GAしたばかりの機能なのでそのあたりは期待して待ちたいところです。