S3 Vectors でお手軽なマルチモーダル楽曲検索を試す

AWS S3 Vectors を使うと，ベクトル DB を別途立てずに S3 だけでベクトルを用いた類似度検索が可能とのことで。

今回は楽曲データセット MusicCaps の音声 (キャプション) を CLAP でエンベディングして， 「テキストで楽曲を検索する」マルチモーダル検索システムを，aws-cdk と uv で一瞬で組みます。

gradioによるデモアプリでは，テキストでの検索が可能になっています。

実装のコードは以下のリポジトリに。

atsukoba/MusicCap-S3VectorsSearch-CLAP

S3 Vectors

Amazon S3 Vectors はS3 ネイティブのベクトルストレージ機能で 2025年7月 に GA（一般公開）になったとのこと。

Amazon S3 Vectors is the first cloud storage with native vector support at scale. — AWS Blog: Introducing Amazon S3 Vectors

ベクトル検索をやろうとすると Pinecone・Weaviate・pgvector などを別途用意する必要がありましたが，S3 Vectors では「ベクトルバケット（vector bucket）」という新しいバケット種別が追加され，オブジェクトストレージと同じ感覚でベクトルを保存・検索できるそうです。

• コスト: 専用ベクトル DB に比べ最大 90% のコスト削減が可能らしい
• スケール: 1 バケットあたり最大 10,000 のベクトルインデックス，各インデックスに数千万ベクトルを格納可能らしい
• サーバーレス: インフラのプロビジョニング不要，従量課金
• AWS エコシステム統合: Amazon Bedrock Knowledge Bases や SageMaker Unified Studio とのネイティブ連携 (今回は使わない)

boto3 での操作

boto3 の新しいクライアント s3vectors で操作でき，ベクトルの投入は put_vectors，検索は query_vectors を使う。

1import boto3
2
3s3vectors = boto3.client("s3vectors", "ap-northeast-1")
4
5s3vectors.put_vectors(
6    vectorBucketName="music-cap-vectors",
7    indexName="music-embeddings",
8    vectors=[
9        {
10            "key": "track_001",
11            "data": {"float32": [0.12, -0.34, ...]},
12            "metadata": {"caption": "A soft piano melody ...", "ytid": "abc123"},  # 任意のメタデータ
13        },
14    ],
15)
16
17response = s3vectors.query_vectors(
18    vectorBucketName="music-cap-vectors",
19    indexName="music-embeddings",
20    queryVector={"float32": [0.05, -0.28, ...]},
21    topK=10,
22    returnMetadata=True,
23    returnDistance=True,
24)

MusicCaps

MusicCaps は Google が公開した音楽キャプションデータセットで，5,521 件の音楽クリップとテキスト説明のペアからなる。

• 各クリップは YouTube から取得した 10 秒間の音楽
• 説明文はプロの音楽家が書いた平均 4 文程度の自由記述（ジャンル・楽器・ムード・テンポなどを記述）
• アスペクトリスト（"mellow piano melody", "fast-paced drums" などのキーワード群）も付属
• ライセンス: CC-BY-SA 4.0

データセットのメタデータには ytid（YouTube ID），start_s，end_s，caption，aspect_list 等のカラムがある。実際の音声は YouTube から yt-dlp で取得する。

YouTubeの利用規約上、動画のダウンロードは原則禁止なので研究・学習目的に限定してください

1from datasets import load_dataset
2
3ds = load_dataset("google/MusicCaps", split="train", token=HF_TOKEN)
4# 5521 サンプル
5# Features: ytid, start_s, end_s, audioset_positive_labels, aspect_list, caption, ...

音声ファイルの取得は yt-dlp で，--download-sections オプションで必要な区間だけを切り出す。

yt-dlp のバージョンやYoutube側の状況によってはダウンロードができないケースがいくつかあるようです

1yt-dlp --quiet --no-warnings -x --audio-format wav -f bestaudio \
2    -o "{ytid}.wav" \
3    --download-sections "*{start_s}-{end_s}" \
4    "https://www.youtube.com/watch?v={ytid}"

注意: 取得できないクリップも一定数存在するのと，datasetのダウンロード時にHF_TOKEN を設定しないと 32 サンプルに制限されるため，.env に設定しておく

CLAP：音声とテキストの共同埋め込みモデル

CLAP (Contrastive Language-Audio Pretraining) は，LAION が公開したオープンソースのマルチモーダルモデルで，OpenAI の CLIP（画像×テキスト）の音声版と理解。

Large-scale Contrastive Language-Audio Pretraining with Feature Fusion and Keyword-to-Caption Augmentation — Wu et al., ICASSP 2023 (arXiv

.06687)

アーキテクチャ

CLAP は 音声エンコーダ（HTSAT ベース）とテキストエンコーダ（RoBERTa ベース）を持ち，両者の埋め込みを同一の潜在空間にマッピングするようにContrastive Learningする。

これにより，「テキストの埋め込み」と「音声の埋め込み」が同じ空間上で比較できる。

HuggingFace Transformers でのモデル利用

今回は laion/clap-htsat-unfused を HuggingFace Transformers 経由でロードする。

1import torch
2from transformers import AutoModel, AutoTokenizer, AutoProcessor, ClapModel
3
4MODEL_ID = "laion/clap-htsat-unfused"
5
6# Apple Silicon では float32，GPU 環境では float16 で動作させる
7dtype = torch.float32 if torch.backends.mps.is_available() else torch.float16
8model: ClapModel = AutoModel.from_pretrained(
9    MODEL_ID, dtype=dtype, device_map="auto"
10).eval()
11
12tokenizer = AutoTokenizer.from_pretrained(MODEL_ID)
13processor = AutoProcessor.from_pretrained(MODEL_ID)

Text Embeddings

1import numpy as np
2
3def get_text_embeddings(texts: list[str]) -> list[float]:
4    inputs = tokenizer(texts, padding=True, return_tensors="pt").to(model.device)
5    features = model.get_text_features(**inputs)
6    return (
7        features["pooler_output"][0, :]
8        .detach().cpu().numpy().astype(np.float32).tolist()
9    )

Audio Embeddings

1from datasets import Dataset, Audio
2
3def get_audio_embeddings(audio_path: str) -> list[float]:
4    input_features = processor(
5        audio=Dataset.from_dict({"audio": [audio_path]}).cast_column(
6            "audio", Audio(sampling_rate=48000)
7        )[0]["audio"]["array"],
8        sampling_rate=48000,
9        return_tensors="pt",
10    )["input_features"].to(model.device)
11    features = model.get_audio_features(input_features=input_features)
12    return (
13        features["pooler_output"][0, :]
14        .detach().cpu().numpy().astype(np.float32).tolist()
15    )

pooler_output が 512 次元の L2 正規化済みベクトルとなる。

実装の流れ

データセットのダウンロードと音声ファイルの取得

scripts/create_demo_datasets.py でデータセットを HuggingFace Hub からロードし，yt-dlp で音声ファイルを取得する。datasets.map を使ってマルチプロセスで並列ダウンロードする。

1from datasets import Audio, load_dataset
2
3ds = load_dataset(
4    "google/MusicCaps", split="train",
5    cache_dir=str(data_dir / "MusicCaps"),
6    token=HF_TOKEN,
7).select(range(samples_to_load))
8
9ds = ds.map(
10    lambda example: _process(example, data_dir / "audio"),
11    num_proc=4,
12).cast_column("audio", Audio(sampling_rate=44100))

CLAP で埋め込む

scripts/create_embeddings.py で，ダウンロード済みの音声ファイルにエンベディングを付与する。各サンプルの ytid ごとに .npy ファイルとして保存する。

1from tqdm import tqdm
2from torch import Tensor
3
4musiccaps_ds = (
5    load_dataset("google/MusicCaps", split="train", ...)
6    .map(link_audio_fn(str(data_dir / "audio")), batched=True)
7    .filter(lambda d: os.path.exists(d["audio"]))
8    .cast_column("audio", Audio(sampling_rate=48000))
9    .map(process_audio_fn(processor, sampling_rate=48000))
10)
11
12for _data in tqdm(musiccaps_ds):
13    _input = Tensor(_data["input_features"]).unsqueeze(0).to(model.device)
14    audio_embed = model.get_audio_features(_input)
15    np.save(
16        data_dir / "embeddings" / f"{_data['ytid']}.npy",
17        audio_embed["pooler_output"][0, :].detach().cpu().numpy(),
18    )

AWS CDK で S3 Vectors バケット・インデックスを作成

バケットとインデックスのプロビジョニングは AWS CDK（TypeScript）で行う。aws-cdk-lib/aws-s3vectors の L1 コンストラクトを使うと数行で書ける。

1// cdk/lib/cdk-stack.ts
2import * as s3vectors from "aws-cdk-lib/aws-s3vectors";
3
4// ベクトルバケット
5const vectorBucket = new s3vectors.CfnVectorBucket(
6  this,
7  "MusicCapVectorBucket",
8  {
9    vectorBucketName: "music-cap-vectors",
10  },
11);
12
13// 512 次元 float32，コサイン類似度のインデックス
14const vectorIndex = new s3vectors.CfnIndex(this, "MusicEmbeddingsIndex", {
15  vectorBucketName: vectorBucket.vectorBucketName!,
16  indexName: "music-embeddings",
17  dataType: "float32",
18  dimension: 512,
19  distanceMetric: "cosine",
20});
21vectorIndex.addDependency(vectorBucket);

デプロイは CDK CLI で行う。

1cd cdk/
2pnpm install
3pnpm cdk bootstrap   # 初回のみ
4pnpm cdk deploy

成功すると以下のような出力が得られる。

1CdkStack.VectorBucketArn = arn:aws:s3vectors:ap-northeast-1:123456789012:bucket/music-cap-vectors
2CdkStack.VectorIndexArn  = arn:aws:s3vectors:ap-northeast-1:123456789012:bucket/music-cap-vectors/index/music-embeddings

boto3 でベクトルを PUT

scripts/upload_vectors.py でエンベディングを S3 Vectors に投入する。list_vectors で既存のキーを取得し，差分のみアップロードする増分投入に対応している。

1import boto3
2import numpy as np
3
4S3_BUCKET_NAME = "music-cap-vectors"
5S3_INDEX_NAME  = "music-embeddings"
6BATCH_SIZE     = 100
7
8s3vectors = boto3.client("s3vectors", region_name="ap-northeast-1")
9
10vectors_to_put = []
11
12for sample in musiccaps_ds:
13    ytid = sample["ytid"]
14    embedding = np.load(f"data/embeddings/{ytid}.npy").astype(np.float32)
15    vec = embedding.squeeze()
16
17    vectors_to_put.append({
18        "key": ytid,
19        "data": {"float32": vec.tolist()},
20        "metadata": {
21            "ytid": ytid,
22            "caption": str(sample["caption"]),
23            "aspect_list": ", ".join(sample["aspect_list"]),
24            "start_s": str(sample["start_s"]),
25            "end_s": str(sample["end_s"]),
26        },
27    })
28
29    if len(vectors_to_put) >= BATCH_SIZE:
30        s3vectors.put_vectors(
31            vectorBucketName=S3_BUCKET_NAME,
32            indexName=S3_INDEX_NAME,
33            vectors=vectors_to_put,
34        )
35        vectors_to_put = []

メタデータの型制約: S3 Vectors のメタデータ値はすべて文字列型で渡す必要あり

⑤ テキストクエリで楽曲を検索

demo/search.py でテキストを CLAP でエンベディングして query_vectors を呼ぶ。

1# demo/search.py
2from botocore.config import Config
3import boto3
4from demo.config import S3_BUCKET_NAME, S3_INDEX_NAME
5
6s3vectors_client = boto3.client(
7    "s3vectors", "ap-northeast-1",
8    config=Config(connect_timeout=10, read_timeout=10),
9)
10
11def search(embedding: list[float], top_k: int = 10):
12    response = s3vectors_client.query_vectors(
13        vectorBucketName=S3_BUCKET_NAME,
14        indexName=S3_INDEX_NAME,
15        queryVector={"float32": embedding},
16        topK=top_k,
17        returnMetadata=True,
18        returnDistance=True,
19    )
20    return response["vectors"]

検索結果の例

1{
2    'distance': 0.360,
3    'key': '2unse6chkMU',
4    'metadata': {
5        'caption': 'This is a piece that would be suitable as calming study music...',
6        'aspect_list': "calming piano music, soothing, bedtime music, sleep music, piano, reverb, violin",
7        'ytid': '2unse6chkMU',
8        ...
9    }
10}

Gradio でデモ作成

demo/app.py では gr.Blocks を使い，検索結果を DataFrame で表示する。行をクリックするとローカルに保存された音声を再生できる。

1# demo/app.py
2import gradio as gr
3import pandas as pd
4from demo.feature_extract import get_text_embeddings
5from demo.local_data import get_local_audio_by_ytid
6from demo.search import search
7
8def do_search(query: str, top_k: int) -> tuple[pd.DataFrame, dict]:
9    embedding = get_text_embeddings([query])
10    results = search(embedding, top_k=int(top_k))
11    rows = [
12        {
13            "rank": i + 1,
14            "ytid": r["key"],
15            "distance": round(r["distance"], 4),
16            "caption": r.get("metadata", {}).get("caption", ""),
17            "aspect_list": r.get("metadata", {}).get("aspect_list", ""),
18        }
19        for i, r in enumerate(results)
20    ]
21    return pd.DataFrame(rows), gr.update(value=None, visible=False)
22
23def on_select(df: pd.DataFrame, evt: gr.SelectData) -> dict:
24    ytid = str(df.iloc[evt.index[0]]["ytid"])
25    audio_path = get_local_audio_by_ytid(ytid)
26    return gr.update(value=audio_path, label=f"Preview: {ytid}", visible=True)
27
28with gr.Blocks(title="MusicCap Search") as demo:
29    gr.Markdown("# Text -> Audio Search on S3Vectors")
30    with gr.Row():
31        query_input = gr.Textbox(placeholder="ex: calming piano music with soft strings", label="Query", scale=4)
32        top_k_slider = gr.Slider(minimum=1, maximum=30, value=10, step=1, label="Top K", scale=1)
33    search_btn = gr.Button("検索", variant="primary")
34    results_df = gr.Dataframe(
35        headers=["rank", "ytid", "distance", "caption", "aspect_list"],
36        label="検索結果（行をクリックして再生）",
37        interactive=False, wrap=True,
38    )
39    audio_player = gr.Audio(label="Preview", visible=False)
40    search_btn.click(fn=do_search, inputs=[query_input, top_k_slider], outputs=[results_df, audio_player])
41    query_input.submit(fn=do_search, inputs=[query_input, top_k_slider], outputs=[results_df, audio_player])
42    results_df.select(fn=on_select, inputs=[results_df], outputs=[audio_player])
43
44demo.launch()

1uv run python -m demo.app

評価

検索システムが正常に動作しているかのバリデーションを目的に，簡単な評価をします。

現状のベクターデータはすべてAudioデータから計算されたものを突っ込んでいますが，元のデータ MusicCaps は人がつけたCaptionやタクソノミーがあります。 それを利用して， ground truth であるキャプション等を用いて検索してそのサンプルを引き出すことができるか？を検証します。

手元のデモでは5,521件のデータセットからランダムに抽出した1024件をターゲットにし実際にDLできた960件を用いてS3Vectorsを構築。 caption 情報と aspect_list (タグ) を用いて 検索します。

例: id=-7B9tPuIP-w

caption

1A male voice narrates a monologue to the rhythm of a song in the background. The song is fast tempo with enthusiastic drumming, groovy bass lines,cymbal ride, keyboard accompaniment ,electric guitar and animated vocals. The song plays softly in the background as the narrator speaks and burgeons when he stops. The song is a classic Rock and Roll and the narration is a Documentary.

aspect_list

1['r&b', 'soul', 'male vocal', 'melodic singing', 'strings sample', 'strong bass', 'electronic drums', 'sensual', 'groovy', 'urban']

検索は類似度 (cossim) の昇順 top_k=100 で取得し，Precision@k (というよりtop-k Acc.?) を算出しました。

結果としては960件のデータで検索した際に，4割程度のケースで上位10件にはクエリそのものをCaptionとして持つデータが含まれるとなりました。

所感とまとめ

S3 Vectors はかなり手軽

CDK でバケットとインデックスを数行で定義でき，boto3 でそのまま put_vectors / query_vectors を呼ぶだけという，追加の管理サービス不要な開発体験が非常にシンプルだった。

プロトタイプや小〜中規模（数百万ベクトル以内）のワークロードなら，Pinecone や Weaviate を採用する前にまず S3 Vectors を検討する価値は十分ある。

CLAP の検索精度

laion/clap-htsat-unfused を HuggingFace Transformers 経由で使ったが，楽器・ジャンル・テンポなどの音楽的特徴についてはかなり的確に検索できた。一方，「悲しい」「明るい」などの感情的なクエリは若干ブレが生じることがあった。CLAP の訓練データに MusicCaps のキャプションが含まれているケースもあるため，評価は慎重に行う必要がある。

Apple Silicon でも動く

torch.backends.mps.is_available() で MPS バックエンドを自動選択するようにしており，M4 Max でも問題なく動作した。float16 は MPS 環境でブロードキャストエラーが出るため，MPS 時は float32 に切り替えている。

参考リンク

• Introducing Amazon S3 Vectors - AWS Blog
• Amazon S3 Vectors 公式ドキュメント - Getting Started
• boto3 S3Vectors リファレンス
• LAION-AI/CLAP - GitHub
• laion/clap-htsat-unfused - Hugging Face
• Large-scale Contrastive Language-Audio Pretraining (arXiv
  .06687)
• google/MusicCaps - Hugging Face
• MusicCaps 解説記事 - Zenn (tatexh)