はじめに
画像をベクトル化して類似画像検索を実現するというテーマで、
今回はAmazon BedrockのAmazon Titan Multimodal Embeddingsを使った
画像ベクトル化と類似画像検索の実装について解説します。
Amazon Titan Multimodal Embeddings とは
概要
- 提供形態: Amazon Bedrock（マネージドサービス）
- 認証方式: IAM（AWSの標準認証）
- 対象ユーザー: AWS利用企業、エンタープライズ
- 特徴: AWSエコシステムとのシームレスな統合
モデルID
model_id = 'amazon.titan-embed-image-v1'
ベクトル次元
1024次元 – 高精度な画像特徴表現が可能
セットアップ方法
ステップ1: AWS IAMユーザーの作成と権限設定
# AWS CLIでIAMユーザーに必要な権限を付与 aws iam attach-user-policy \ --user-name your-user-name \ --policy-arn arn:aws:iam::aws:policy/AmazonBedrockFullAccess # S3アクセス権限も必要な場合 aws iam attach-user-policy \ --user-name your-user-name \ --policy-arn arn:aws:iam::aws:policy/AmazonS3ReadOnlyAccess
ステップ2: Bedrock APIの有効化
1. AWSコンソールにログイン 2. Amazon Bedrockサービスに移動 3. リージョンを選択（例: us-east-1） 4. Model Accessから「Titan Multimodal Embeddings」を有効化
ステップ3: 環境の整備
# .envファイルまたはcompose.yml AWS_DEFAULT_REGION=us-east-1 AWS_ACCESS_KEY_ID=your_access_key AWS_SECRET_ACCESS_KEY=your_secret_key AWS_S3_BUCKET=your-bucket-name
Docker環境での構成
# compose.yml
services:
backend:
build: .
environment:
# Amazon Bedrock設定
- AWS_DEFAULT_REGION=us-east-1
- AWS_ACCESS_KEY_ID=${AWS_ACCESS_KEY_ID}
- AWS_SECRET_ACCESS_KEY=${AWS_SECRET_ACCESS_KEY}
- AWS_S3_BUCKET=[$bucket_name]
# PostgreSQL設定
- DB_HOST=postgres
- DB_NAME=image_search
- DB_USER=postgres
- DB_PASSWORD=postgres
- DB_PORT=5432
depends_on:
- postgres
ports:
- "5000:5000"
postgres:
image: pgvector/pgvector:pg16
environment:
- POSTGRES_USER=postgres
- POSTGRES_PASSWORD=postgres
- POSTGRES_DB=image_search
volumes:
- postgres_data:/var/lib/postgresql/data
ports:
- "5432:5432"
volumes:
postgres_data:
ステップ4: 依存関係のインストール
requirements.txtファイルを作成し、必要なパッケージを記載します：
# requirements.txt boto3>=1.28.0 psycopg2-binary>=2.9.9 pillow>=10.0.0 flask>=3.0.0
各パッケージの役割
boto3: AWS SDK（Bedrock、S3クライアント）psycopg2-binary: PostgreSQL接続pillow: 画像処理（リサイズ、エンコード）flask: Webフレームワーク（API実装用）
Dockerfileでの構成
Dockerfileを作成して、依存関係のインストールとアプリケーションのセットアップを自動化します：
APIをFlaskで作成想定
FROM python:3.12-slim WORKDIR /app # requirements.txtをコピーして依存関係をインストール COPY requirements.txt . RUN pip install --no-cache-dir -r requirements.txt # アプリケーションコードをコピー COPY . . # アップロード用ディレクトリを作成 RUN mkdir -p uploads # 環境変数を設定 ENV FLASK_APP=app.py ENV FLASK_ENV=production # ポートを公開 EXPOSE 5000 # アプリケーションを起動 CMD ["python", "app.py"]
Dockerイメージのビルドと起動:
# Dockerイメージをビルド docker compose build # コンテナを起動 docker compose up -d
ステップ5: PostgreSQLでベクトル拡張を有効化
-- pgvector拡張をインストール
CREATE EXTENSION IF NOT EXISTS vector;
-- 画像テーブルの作成（1024次元）
CREATE TABLE IF NOT EXISTS images (
id SERIAL PRIMARY KEY,
file_path TEXT NOT NULL,
file_name TEXT NOT NULL,
description TEXT,
embedding VECTOR(1024), -- Titan Multimodalは1024次元
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- ベクトル検索用のインデックス作成（高速化）
CREATE INDEX ON images USING ivfflat (embedding vector_cosine_ops);
実装コード詳解
BedrockImageVectorizerクラス
import base64
import os
import boto3
import psycopg2
from io import BytesIO
from PIL import Image
import json
import logging
class BedrockImageVectorizer:
def __init__(self, aws_region=None):
"""初期化: Bedrockクライアントとデータベース接続を設定"""
self.aws_region = aws_region or os.environ.get('AWS_DEFAULT_REGION')
# Bedrock Runtimeクライアントの初期化
self.bedrock_runtime = boto3.client(
service_name='bedrock-runtime',
region_name=self.aws_region
)
# S3クライアントの初期化
self.s3_client = boto3.client(
service_name='s3',
region_name=self.aws_region
)
self.s3_bucket = os.environ.get('AWS_S3_BUCKET')
self.model_id = 'amazon.titan-embed-image-v1'
# PostgreSQL接続情報
self.db_config = {
'host': os.environ.get('DB_HOST'),
'database': os.environ.get('DB_NAME'),
'user': os.environ.get('DB_USER'),
'password': os.environ.get('DB_PASSWORD'),
'port': os.environ.get('DB_PORT', 5432)
}
画像エンコード処理
def _encode_image(self, image_source):
"""画像をBase64エンコードする（ファイルパスまたはBytesIOに対応）"""
try:
with Image.open(image_source) as img:
# 画像が大きすぎる場合はリサイズ
max_size = 1024
if max(img.width, img.height) > max_size:
ratio = max_size / max(img.width, img.height)
new_size = (int(img.width * ratio), int(img.height * ratio))
img = img.resize(new_size, Image.Resampling.LANCZOS)
# JPEG形式でBase64エンコード
buffer = BytesIO()
img.save(buffer, format="JPEG")
return base64.b64encode(buffer.getvalue()).decode('utf-8')
except Exception as e:
logger.error(f"画像のエンコードに失敗しました: {e}")
raise
ベクトル化処理
def vectorize_image(self, image_source):
"""Bedrock Titan Multimodalを使用して画像をベクトル化する"""
try:
base64_image = self._encode_image(image_source)
# Bedrock Titan Multimodalモデルへのリクエスト
request_body = json.dumps({
"inputImage": base64_image
})
response = self.bedrock_runtime.invoke_model(
modelId=self.model_id,
body=request_body
)
response_body = json.loads(response['body'].read())
embedding = response_body['embedding'] # 1024次元のベクトル
return embedding
except Exception as e:
logger.error(f"画像のベクトル化に失敗しました: {e}")
raise
S3のパスをテーブルに格納（統合処理）
def store_image_embedding_from_s3(self, file_key, image_data, bucket_name, description=None):
"""S3から取得した画像データをベクトル化してPostgreSQLに格納"""
conn = None
cursor = None
try:
# 画像をベクトル化
embedding = self.vectorize_image(image_data)
# S3のフルパスを作成
s3_file_path = f"s3://{bucket_name}/{file_key}"
file_name = os.path.basename(file_key)
# PostgreSQLに格納
conn = psycopg2.connect(**self.db_config)
cursor = conn.cursor()
cursor.execute("""
INSERT INTO images (file_path, file_name, description, embedding)
VALUES (%s, %s, %s, %s)
RETURNING id;
""", (s3_file_path, file_name, description, embedding))
image_id = cursor.fetchone()[0]
conn.commit()
return image_id
finally:
if cursor:
cursor.close()
if conn:
conn.close()
類似画像検索
def find_similar_images(self, query_image_source, limit=5):
"""類似画像を検索する（コサイン距離ベース）"""
conn = None
cursor = None
try:
# クエリ画像をベクトル化
query_embedding = self.vectorize_image(query_image_source)
# PostgreSQLで類似度検索（コサイン距離）
conn = psycopg2.connect(**self.db_config)
cursor = conn.cursor()
cursor.execute("""
SELECT
id,
file_path,
file_name,
description,
embedding <-> %s::vector(1024) AS distance
FROM images
ORDER BY distance
LIMIT %s;
""", (query_embedding, limit))
results = cursor.fetchall()
similar_images = []
for row in results:
image_data = {
"id": row[0],
"file_path": row[1],
"file_name": row[2],
"description": row[3],
"distance": row[4]
}
# S3パスの場合は署名付きURLを生成
if row[1].startswith('s3://'):
s3_path = row[1].replace('s3://', '')
parts = s3_path.split('/')
bucket = parts[0]
key = '/'.join(parts[1:])
image_data['presigned_url'] = self.get_presigned_url(bucket, key)
similar_images.append(image_data)
return similar_images
finally:
if cursor:
cursor.close()
if conn:
conn.close()
ポイント
- pgvectorの
<->オペレーターでユークリッド距離（L2距離）を計算 - 距離が小さいほど類似度が高い
- S3画像には自動的に署名付きURL生成
- セキュアなアクセス制御
バッチ処理スクリプト
S3バケット内の大量の画像を一括でベクトル化するスクリプト:
#!/usr/bin/env python
import boto3
from io import BytesIO
import logging
from services.bedrock_image_service import BedrockImageVectorizer
S3_BUCKET_NAME = '[$bucket_name]'
S3_FILES_PREFIX = 'files/'
def get_s3_image_files(bucket_name, prefix, patterns):
"""S3バケットから画像ファイル一覧を取得"""
s3_client = boto3.client('s3')
image_files = []
response = s3_client.list_objects_v2(Bucket=bucket_name, Prefix=prefix)
if 'Contents' not in response:
return []
import fnmatch
for obj in response['Contents']:
file_key = obj['Key']
file_name = os.path.basename(file_key)
if not file_name:
continue
for pattern in patterns:
if fnmatch.fnmatch(file_name.lower(), pattern.lower()):
image_files.append(file_key)
break
return image_files
def main():
vectorizer = BedrockImageVectorizer()
# S3から画像ファイル一覧を取得
patterns = ['*.jpg', '*.jpeg', '*.png']
image_files = get_s3_image_files(S3_BUCKET_NAME, S3_FILES_PREFIX, patterns)
# 各画像を処理
for idx, file_key in enumerate(image_files, 1):
print(f"[{idx}/{len(image_files)}] 処理中: {file_key}")
# S3から画像をメモリにダウンロード
s3_client = boto3.client('s3')
response = s3_client.get_object(Bucket=S3_BUCKET_NAME, Key=file_key)
image_data = BytesIO(response['Body'].read())
# ベクトル化してデータベースに格納
vectorizer.store_image_embedding_from_s3(
file_key=file_key,
image_data=image_data,
bucket_name=S3_BUCKET_NAME
)
if __name__ == '__main__':
main()
API処理(今回はFlaskを想定)
from flask import request, jsonify
from io import BytesIO
from services.bedrock_image_service import BedrockImageVectorizer
bedrock_vectorizer = BedrockImageVectorizer()
@app.route('/api/search-bedrock', methods=['POST'])
@login_required
def search_similar_images_bedrock():
"""Bedrockを使った類似画像検索API"""
try:
if 'image' not in request.files:
return jsonify({'error': '画像ファイルが提供されていません'}), 400
image_file = request.files['image']
# 画像データをメモリに読み込む
image_data = BytesIO(image_file.read())
# 検索数を取得（デフォルト6件）
limit = int(request.form.get('limit', 6))
if limit < 6 or limit > 10:
limit = 6
# 類似画像を検索
similar_images = bedrock_vectorizer.find_similar_images(
image_data,
limit=limit
)
return jsonify({
'success': True,
'similar_images': similar_images,
'count': len(similar_images)
})
except Exception as e:
return jsonify({
'success': False,
'error': f'検索エラー: {str(e)}'
}), 500
料金とコスト
Amazon Titan Multimodal Embeddingsの料金
2025年10月時点の料金:
画像エンベディング: $0.020 / 1,000画像
月間処理量ごとのコスト試算
| 月間処理画像数 | 月額コスト |
|---|---|
| 10,000枚 | $0.20 |
| 100,000枚 | $2.00 |
| 1,000,000枚 | $20.00 |
| 10,000,000枚 | $200.00 |
追加コスト
- S3ストレージ: $0.023 / GB / 月（スタンダードストレージ）
- S3データ転送: 無料（同一リージョン内）
- PostgreSQL（RDS使用時）: インスタンスタイプによる
まとめ
Amazon Bedrockの Amazon Titan Multimodal Embeddings を使えば、画像の類似検索システムを簡単に構築できます。
画像をBase64エンコードしてAmazon Bedrockに送信するだけで、1024次元の高精度ベクトルが取得でき、pgvectorを使ったPostgreSQLで効率的に類似度検索が可能です。
S3に保存された大量の画像も、バッチ処理スクリプトで一括ベクトル化できます。
既にAWSを利用している場合は、IAM認証やS3ストレージとシームレスに統合でき、追加の認証設定やインフラ構築が不要です。
マネージドサービスなので、スケーリングやメンテナンスの手間もかかりません。
本記事で紹介した実装コードをベースに、ぜひ画像検索システムの構築にチャレンジしてみてください。
簡単30秒