Personalize v2.0 完全ガイド（Machine Learning & AI）

概要

Amazon Personalize は、ユーザー行動履歴から個人化されたレコメンデーション・ユーザーセグメンテーション・ランキング機能を ML 専門知識なしで構築できるマネージドサービスです。Amazon.com の 20+ 年のレコメンデーション技術を基盤とし、協調フィルタリング・Matrix Factorization・深層学習などの複雑な ML 手法を自動適用して高精度な推奨を実現します。リアルタイム・バッチ・フィルター・トレンド・次の最適アクション推奨など複数のユースケースに対応し、EC サイト・動画ストリーミング・音楽配信・ニュースアプリ・SaaS プラットフォームなど幅広い業界で活用されています。A/B テスト機能で推奨戦略の効果測定も容易です。

課題と特徴

従来型レコメンデーション構築の課題

• ML エンジニア不足: 協調フィルタリング・リカレントニューラルネットワークの実装・チューニングに数ヶ月
• 大規模データ処理: 数百万ユーザー × 数百万アイテムの行列分解計算に高価な GPU クラスタ
• 冷スタート問題: 新規ユーザー・新規アイテムへの推奨が困難
• スケーリング・保守: リアルタイム更新・バッチ処理の両立に複雑なアーキテクチャ
• A/B テスト: 推奨アルゴリズム比較に独自実装が必須

Personalize の特徴

アーキテクチャ

graph TB
    subgraph DataIngestion["データ取り込み"]
        A["インタラクションデータ<br/>ユーザー・アイテム・イベント"]
        B["アイテムメタデータ<br/>カテゴリ・価格・説明"]
        C["ユーザーメタデータ<br/>年齢・地域・セグメント"]
        D["リアルタイムイベント<br/>PutEvents API"]
    end
    
    subgraph DataPrep["データ準備"]
        E["Dataset Import"]
        F["Event Tracking"]
    end
    
    subgraph Solutions["Solutions（ソリューション）"]
        G["Recommenders<br/>Domain Use Case"]
        H["Custom Solutions<br/>Recipe Selection"]
    end
    
    subgraph Training["モデル学習"]
        I["Solution Versions<br/>自動アルゴリズム選択"]
    end
    
    subgraph APIs["推奨 API"]
        J["GetRecommendations<br/>リアルタイム"]
        K["GetPersonalizedRanking<br/>ランキング"]
        L["GetRecommendations<br/>Similar Items"]
    end
    
    subgraph Campaigns["キャンペーン"]
        M["キャンペーン管理"]
        N["A/B テスト"]
        O["メトリクス追跡"]
    end
    
    A --> E
    B --> E
    C --> E
    D --> F
    E --> G
    E --> H
    G --> I
    H --> I
    I --> J
    I --> K
    I --> L
    J --> M
    K --> N
    L --> O

コアコンポーネント

1. Recommenders（推奨ソリューション）

import boto3
import time

personalize = boto3.client('personalize', region_name='ap-northeast-1')
personalize_runtime = boto3.client('personalize-runtime', region_name='ap-northeast-1')

# ドメイン特化ソリューション（推奨：セットアップが簡単）
# - Retail: 商品レコメンデーション
# - Media & Entertainment: 動画・音楽推奨
# - Travel: ホテル・航空券推奨
# - etc.

# Step 1: Recommender を作成（トレーニング自動実行）
response = personalize.create_recommender(
    name='retail-product-recommender',
    datasetGroupArn='arn:aws:personalize:ap-northeast-1:123456789:dataset-group/my-dataset',
    recipeArn='arn:aws:personalize:ap-northeast-1:123456789:recipe/retail-products-default'
)

recommender_arn = response['recommenderArn']

# トレーニング完了を待機
while True:
    status = personalize.describe_recommender(recommenderArn=recommender_arn)
    if status['recommender']['status'] == 'ACTIVE':
        print("✓ Recommender が利用可能")
        break
    elif status['recommender']['status'] == 'CREATE_FAILED':
        print("✗ 作成失敗")
        break
    time.sleep(30)

# Step 2: リアルタイム推奨を取得
result = personalize_runtime.get_recommendations(
    recommenderArn=recommender_arn,
    userId='user-123',
    numResults=10
)

for item in result['itemList']:
    print(f"  {item['itemId']:20} スコア: {item['score']:.4f}")

2. Solutions（カスタムソリューション）+ Recipes

# より細かい制御が必要な場合は Custom Solution を作成

# 利用可能な Recipe:
# - User-Personalization: ユーザー個人への最良推奨
# - SIMS: 類似アイテム検索
# - Personalized-Ranking: ユーザー向けランキング
# - Ranking: 全ユーザー向けランキング
# - Trending-Now: トレンド + 個人化
# - Next-Best-Action: 次アクション予測

# Step 1: Dataset Group 作成
dataset_response = personalize.create_dataset_group(
    name='ecommerce-dataset',
    domain='ECOMMERCE'
)

dataset_group_arn = dataset_response['datasetGroupArn']

# Step 2: Interaction Dataset を作成
interaction_dataset_response = personalize.create_dataset(
    datasetGroupArn=dataset_group_arn,
    datasetType='Interactions',
    schemaArn='arn:aws:personalize:ap-northeast-1:123456789:schema/interaction-schema'
)

interaction_dataset_arn = interaction_dataset_response['datasetArn']

# Step 3: CSV データを S3 にアップロード
"""
s3://my-bucket/interactions.csv
USER_ID, ITEM_ID, EVENT_TYPE, TIMESTAMP
user-001, item-100, CLICK, 1706000000
user-001, item-100, PURCHASE, 1706010000
user-002, item-200, CLICK, 1706005000
"""

# Step 4: Data Import を実行
import_response = personalize.create_dataset_import_job(
    jobName='import-interactions-001',
    datasetArn=interaction_dataset_arn,
    dataSource={'dataLocation': 's3://my-bucket/interactions.csv'},
    roleArn='arn:aws:iam::123456789:role/PersonalizeRole'
)

job_arn = import_response['datasetImportJobArn']

# インポート完了待機
while True:
    status = personalize.describe_dataset_import_job(datasetImportJobArn=job_arn)
    if status['datasetImportJob']['status'] == 'ACTIVE':
        print("✓ データインポート完了")
        break
    time.sleep(30)

# Step 5: Solution を作成（アルゴリズム学習）
solution_response = personalize.create_solution(
    name='user-personalization-solution',
    datasetGroupArn=dataset_group_arn,
    recipeArn='arn:aws:personalize:ap-northeast-1:123456789:recipe/aws-user-personalization'
)

solution_arn = solution_response['solutionArn']

# Step 6: Solution Version（学習モデル）を作成
solution_version_response = personalize.create_solution_version(
    solutionArn=solution_arn
)

solution_version_arn = solution_version_response['solutionVersionArn']

# 学習完了待機（数時間）
while True:
    status = personalize.describe_solution_version(solutionVersionArn=solution_version_arn)
    if status['solutionVersion']['status'] == 'ACTIVE':
        print(f"✓ ソリューション学習完了")
        metrics = status['solutionVersion']['metrics']
        print(f"  Precision@25: {metrics.get('precision@25', 'N/A')}")
        print(f"  Normalized RMSE: {metrics.get('normalized_rmse', 'N/A')}")
        break
    elif status['solutionVersion']['status'] == 'CREATE_FAILED':
        print("✗ 学習失敗")
        break
    time.sleep(60)

3. リアルタイムイベントトラッキング

import boto3
from datetime import datetime

personalize_events = boto3.client('personalize-events', region_name='ap-northeast-1')

# Recommender で EventTrackingArn を取得
# または Solution Version で trackingId を取得

# ユーザーのリアルタイム行動を Personalize に送信
def track_user_event(user_id, item_id, event_type, tracking_id):
    """
    event_type: VIEW, CLICK, PURCHASE, ADD_TO_CART, REMOVE_FROM_CART, LIKE, etc.
    """
    
    current_timestamp = int(datetime.now().timestamp())
    
    personalize_events.put_events(
        trackingId=tracking_id,  # EventTrackingArn から生成
        userId=user_id,
        sessionId=f'session-{user_id}-{int(datetime.now().timestamp())}',
        eventList=[{
            'sentAt': current_timestamp,
            'eventType': event_type,
            'itemId': item_id,
            'properties': '{"discount": "true", "price": "9999"}'  # カスタム属性
        }]
    )
    
    print(f"✓ イベント記録: {user_id} - {event_type} - {item_id}")

# 使用例
track_user_event(
    user_id='user-123',
    item_id='product-456',
    event_type='CLICK',
    tracking_id='tracking-xxx'
)

# リアルタイムイベント送信直後に推奨を取得すると、
# 最新の行動を反映した個人化推奨が返される（セッション内レコメンデーション）

4. Personalized Ranking（メールマーケティングの最適化）

# メール受信者ごとに商品リストを個人最適化

def optimize_email_content(user_id, solution_version_arn, candidate_items):
    """
    user_id: user-123
    candidate_items: ['item-100', 'item-200', 'item-300', ...]
    返値: ユーザー個人に最適化されたランキング順序
    """
    
    personalize_runtime = boto3.client('personalize-runtime')
    
    response = personalize_runtime.get_personalized_ranking(
        campaignArn=solution_version_arn,
        userId=user_id,
        inputList=candidate_items
    )
    
    ranked_items = response['personalizedRanking']
    
    print(f"メール {user_id} の最適ランキング:")
    for rank, item in enumerate(ranked_items, 1):
        print(f"  {rank}. {item['itemId']} (スコア: {item['score']:.4f})")
    
    # メール本文に最適なランキング順で商品表示
    return [item['itemId'] for item in ranked_items]

# 使用例: 毎週月曜に全ユーザーへメール送信
def send_weekly_recommendation_email():
    users = ['user-123', 'user-456', 'user-789']
    available_products = ['item-100', 'item-200', 'item-300', 'item-400']
    
    for user_id in users:
        ranked_items = optimize_email_content(
            user_id,
            'arn:aws:personalize:ap-northeast-1:123456789:campaign/ranking-campaign',
            available_products
        )
        # メール送信（Braze・Segment 等と統合）
        # send_email(user_id, ranked_items)

5. フィルター機能

# ルールベースフィルターで推奨結果を制御

# フィルター Expression の例:
# - 在庫切れ商品を除外: EXCLUDE Items WHERE Items.inventory < 1
# - 高価格商品のみ: INCLUDE Items WHERE Items.price > 10000
# - 新着商品を優先: BOOST Items WHERE Items.days_since_publication < 7
# - カテゴリ指定: INCLUDE Items WHERE Items.category IN ('Electronics')

response = personalize_runtime.get_recommendations(
    recommenderArn='arn:aws:personalize:ap-northeast-1:...',
    userId='user-123',
    numResults=10,
    filterArn='arn:aws:personalize:ap-northeast-1:123456789:filter/exclude-expensive'
    # または filterExpression を直接指定
)

6. Batch レコメンデーション（大量ユーザーへの推奨一括生成）

# 数百万ユーザーへの推奨を非同期バッチ生成

def batch_generate_recommendations(solution_version_arn):
    """毎日深夜にバッチで全ユーザーの推奨を事前計算"""
    
    response = personalize.start_batch_inference_job(
        jobName='batch-recommendations-001',
        solutionVersionArn=solution_version_arn,
        jobInput={
            'itemIds': ['item-100', 'item-200', 'item-300', ...]  # または userIds
        },
        jobOutput={'s3OutputPath': 's3://my-bucket/batch-output/'},
        roleArn='arn:aws:iam::123456789:role/PersonalizeRole',
        numResults=10
    )
    
    job_arn = response['batchInferenceJobArn']
    
    # 完了待機（1-2 時間）
    while True:
        status = personalize.describe_batch_inference_job(batchInferenceJobArn=job_arn)
        if status['batchInferenceJob']['status'] == 'COMPLETED':
            print("✓ バッチ推奨生成完了")
            print(f"  出力: {status['batchInferenceJob']['jobOutput']['s3OutputPath']}")
            break
        elif status['batchInferenceJob']['status'] == 'FAILED':
            print("✗ バッチ処理失敗")
            break
        time.sleep(60)
    
    # 出力例:
    """
    s3://my-bucket/batch-output/user-123.csv
    item-100, 0.95
    item-200, 0.87
    item-300, 0.75
    """

batch_generate_recommendations('arn:aws:personalize:ap-northeast-1:123456789:solution-version/...')

7. Campaign（キャンペーン） + A/B テスト

# キャンペーン: Solution Version をエンドポイント化（リアルタイムAPI）

def create_ab_test_campaign():
    """2 つの Solution Version を A/B テスト"""
    
    # Campaign A: User-Personalization
    campaign_a = personalize.create_campaign(
        name='user-personalization-campaign-a',
        solutionVersionArn='arn:aws:personalize:ap-northeast-1:123456789:solution-version/v1',
        minProvisionedTPS=1
    )
    
    # Campaign B: SIMS（比較対象）
    campaign_b = personalize.create_campaign(
        name='sims-campaign-b',
        solutionVersionArn='arn:aws:personalize:ap-northeast-1:123456789:solution-version/v2',
        minProvisionedTPS=1
    )
    
    # フロントエンド: ユーザーを 50:50 で分割、異なるキャンペーンから推奨取得
    def get_recommendation(user_id, treatment):
        runtime = boto3.client('personalize-runtime')
        
        if treatment == 'A':
            campaign_arn = campaign_a['campaignArn']
        else:
            campaign_arn = campaign_b['campaignArn']
        
        result = runtime.get_recommendations(
            campaignArn=campaign_arn,
            userId=user_id,
            numResults=5
        )
        
        return result['itemList']
    
    # メトリクス: Personalize は自動的に CTR・Conversion を追跡
    # → ダッシュボードで A/B テスト結果を比較

主要ユースケース（10+）

1. EC サイト - 商品推奨

class ECommerceRecommendation:
    def __init__(self, recommender_arn):
        self.runtime = boto3.client('personalize-runtime')
        self.recommender_arn = recommender_arn
    
    def get_homepage_recommendations(self, user_id):
        """ホームページの「あなたへのおすすめ」セクション"""
        
        response = self.runtime.get_recommendations(
            recommenderArn=self.recommender_arn,
            userId=user_id,
            numResults=5
        )
        
        return [
            {
                'product_id': item['itemId'],
                'score': item['score']
            }
            for item in response['itemList']
        ]
    
    def get_similar_products(self, user_id, product_id):
        """「この商品を見た人はこちらも見ています」"""
        # SIMS Recipe を使用
        pass
    
    def personalize_cart_abandonment_email(self, user_id, abandoned_items):
        """カート放置メールで個人化商品推奨"""
        # Ranking Recipe でメール内商品を最適化
        pass

2. 動画ストリーミング - コンテンツ推奨

class VideoStreamingRecommendation:
    def __init__(self, recommender_arn):
        self.runtime = boto3.client('personalize-runtime')
        self.events = boto3.client('personalize-events')
        self.recommender_arn = recommender_arn
    
    def get_homepage_content(self, user_id):
        """ホーム画面の個人化コンテンツセクション"""
        
        response = self.runtime.get_recommendations(
            recommenderArn=self.recommender_arn,
            userId=user_id,
            numResults=15
        )
        
        return response['itemList']
    
    def track_view_event(self, user_id, episode_id, watch_time):
        """視聴イベントをリアルタイム記録"""
        
        self.events.put_events(
            trackingId='tracking-xxx',
            userId=user_id,
            sessionId=f'session-{user_id}-{int(time.time())}',
            eventList=[{
                'sentAt': int(time.time()),
                'eventType': 'WATCH',
                'itemId': episode_id,
                'properties': json.dumps({'watch_time': watch_time})
            }]
        )
    
    def get_next_episode_recommendation(self, user_id, current_episode):
        """エピソード終了後の「次に見る」推奨"""
        # リアルタイムイベント送信直後に推奨取得
        response = self.runtime.get_recommendations(
            recommenderArn=self.recommender_arn,
            userId=user_id,
            numResults=1
        )
        
        return response['itemList'][0]

3. ニュース・メディア - 記事推奨

class NewsRecommendation:
    def __init__(self, recommender_arn):
        self.runtime = boto3.client('personalize-runtime')
    
    def get_feed_recommendations(self, user_id):
        """ニュースフィード個人化"""
        
        response = self.runtime.get_recommendations(
            recommenderArn='arn:aws:personalize:ap-northeast-1:...',
            userId=user_id,
            numResults=20
        )
        
        # ユーザー読書傾向に基づいたフィードを生成
        return response['itemList']

4. SaaS - 機能推奨

class SaaSOncboardingRecommendation:
    def __init__(self, recommender_arn):
        self.runtime = boto3.client('personalize-runtime')
    
    def recommend_next_feature(self, user_id):
        """新規ユーザー向けの「次に試すべき機能」推奨"""
        
        response = self.runtime.get_recommendations(
            recommenderArn=self.recommender_arn,
            userId=user_id,
            numResults=1  # 次の 1 つの機能を推奨
        )
        
        if response['itemList']:
            feature_id = response['itemList'][0]['itemId']
            print(f"✓ ユーザー {user_id} には {feature_id} を推奨")
            return feature_id
        
        return None

CLI・SDK・IaC 例（5+）

CLI 例

# Recommender 作成（簡単）
aws personalize create-recommender \
  --name my-recommender \
  --dataset-group-arn arn:aws:personalize:ap-northeast-1:123456789:dataset-group/my-group \
  --recipe-arn arn:aws:personalize:ap-northeast-1:123456789:recipe/retail-products-default \
  --region ap-northeast-1

# Dataset Group 作成（カスタムソリューション）
aws personalize create-dataset-group \
  --name my-dataset \
  --domain ECOMMERCE \
  --region ap-northeast-1

# Interaction データをインポート
aws personalize create-dataset-import-job \
  --job-name import-interactions \
  --dataset-arn arn:aws:personalize:ap-northeast-1:123456789:dataset/... \
  --data-source dataLocation=s3://my-bucket/interactions.csv \
  --role-arn arn:aws:iam::123456789:role/PersonalizeRole \
  --region ap-northeast-1

# Solution を作成（モデル学習）
aws personalize create-solution \
  --name my-solution \
  --dataset-group-arn arn:aws:personalize:ap-northeast-1:123456789:dataset-group/... \
  --recipe-arn arn:aws:personalize:ap-northeast-1:123456789:recipe/aws-user-personalization \
  --region ap-northeast-1

# リアルタイム推奨を取得
aws personalize-runtime get-recommendations \
  --recommender-arn arn:aws:personalize:ap-northeast-1:123456789:recommender/... \
  --user-id user-123 \
  --num-results 10 \
  --region ap-northeast-1

SDK 例（Python）

import boto3
import json

personalize = boto3.client('personalize')
runtime = boto3.client('personalize-runtime')

# イベント一括インポート
def import_interactions_csv(dataset_arn, csv_s3_uri):
    response = personalize.create_dataset_import_job(
        jobName='batch-import-001',
        datasetArn=dataset_arn,
        dataSource={'dataLocation': csv_s3_uri},
        roleArn='arn:aws:iam::123456789:role/PersonalizeRole'
    )
    return response['datasetImportJobArn']

# 推奨取得
def get_recommendations(recommender_arn, user_id, num_results=5):
    response = runtime.get_recommendations(
        recommenderArn=recommender_arn,
        userId=user_id,
        numResults=num_results
    )
    
    return [
        {
            'item_id': item['itemId'],
            'score': item['score'],
            'reason': item.get('promotionName', 'Default')
        }
        for item in response['itemList']
    ]

IaC 例（Terraform）

resource "aws_personalize_dataset_group" "ecommerce" {
  name   = "ecommerce-dataset"
  domain = "ECOMMERCE"
}

resource "aws_personalize_dataset" "interactions" {
  name             = "interactions-dataset"
  dataset_group_id = aws_personalize_dataset_group.ecommerce.id
  dataset_type     = "Interactions"
  schema_arn       = aws_personalize_schema.interactions.arn
}

resource "aws_personalize_recommender" "products" {
  name               = "product-recommender"
  dataset_group_id   = aws_personalize_dataset_group.ecommerce.id
  recipe_arn         = "arn:aws:personalize:${var.region}::recipe/retail-products-default"
}

resource "aws_personalize_campaign" "online" {
  name                    = "online-campaign"
  solution_version_arn    = aws_personalize_solution_version.main.arn
  min_provisioned_tps     = 1
}

比較表

ベストプラクティス

✅ DO

• 十分なインタラクションデータを用意: 最低 1,000 イベント、推奨 10,000+ イベント
• リアルタイムイベント統合: PutEvents で最新の行動を反映して精度向上
• フィルター活用: ビジネスルール（在庫・価格等）を推奨に適用
• A/B テスト実施: キャンペーン機能で複数ソリューション比較
• メトリクス監視: 推奨の CTR・コンバージョン率を定期確認

❌ DON’T

• 少量データで始める: コールドスタート問題が深刻、最低限のデータ量確保が必須
• リアルタイムイベント忘れ: ストリーム更新なしでは精度改善が困難
• 複雑なフィルター: パフォーマンス低下、シンプルなルールを優先
• 単一アルゴリズムに執着: Recommenders で複数ソリューション並列テスト推奨

2025-2026 最新動向

• Recommenders の拡大: ドメイン別プリセットテンプレート増加（Travel, Healthcare 等）
• Next-Best-Action の強化: マルチチャネル推奨（Email, SMS, Push）
• Bedrock 統合: 推奨理由の自然言語説明生成
• Real-time Personalization の低レイテンシー化: <100ms の推奨取得
• プライバシー強化: 差分プライバシー・FedML 対応

学習リソース

公式ドキュメント（8+）

1. Amazon Personalize Developer Guide
2. Recommenders API Reference
3. Recipes Overview
4. Event Tracking Setup
5. Batch Inference
6. A/B Testing
7. Pricing
8. GitHub Samples

比較・ベンダー情報（5+）

1. Google Cloud Discovery Engine - 検索・レコメンデーション統合
2. Microsoft Personalizer - 強化学習ベース
3. RecBole - OSS レコメンデーション
4. Pinecone - Vector DB + 埋め込み
5. Hugging Face Recommendations - Transformer ベース

採用判断チェックリスト

• [ ] EC サイト・動画配信・コンテンツプラットフォームで個人化推奨が必須か
• [ ] ユーザーの行動履歴データが 1,000+ イベント存在するか
• [ ] リアルタイムに行動を反映した推奨が必要か
• [ ] メールマーケティングで受信者ごとにコンテンツを個人化したいか
• [ ] A/B テストで推奨戦略の効果測定をしたいか
• [ ] ML 専門知識がなくレコメンデーションを実装したいか
• [ ] スケーラビリティ（数百万ユーザー）が必要か

まとめ

Personalize は 「Amazon.com のレコメンデーション技術を API で提供するパーソナライゼーションサービス」。ユーザーの行動履歴から個人化推奨を生成し、PutEvents でリアルタイムに更新し、複数ソリューション + A/B テストで最適化できます。EC・動画配信・ニュース・SaaS など幅広い業界に対応し、Recommenders（ドメイン特化）でセットアップが簡単、Custom Solutions（細かい制御）で柔軟性も備え、ML 専門知識不要でエンタープライズグレードのレコメンデーション機能を実現します。Bedrock・SageMaker との統合で、推奨理由説明や高度なセグメンテーションも可能です。

トラブルシューティング

詳細な診断手順

データセット問題の診断

def diagnose_dataset_issues():
    """データセットの問題を診断"""
    
    personalize = boto3.client('personalize')
    
    # Step 1: インポートジョブの詳細確認
    import_jobs = personalize.list_dataset_import_jobs(
        datasetArn='arn:aws:personalize:...'
    )
    
    for job in import_jobs['datasetImportJobs']:
        print(f"Job: {job['jobName']}")
        print(f"  Status: {job['status']}")
        print(f"  Failure Reason: {job.get('failureReason', 'N/A')}")
        
        if job['status'] == 'CREATE_FAILED':
            # 原因を特定
            if 'encoding' in job['failureReason'].lower():
                print("  → ファイル エンコーディングを UTF-8 に修正")
            elif 'schema' in job['failureReason'].lower():
                print("  → CSV カラム構成をスキーマと一致させる")
    
    # Step 2: Dataset の統計情報確認
    dataset_groups = personalize.list_dataset_groups()
    
    for dg in dataset_groups['datasetGroups']:
        describe = personalize.describe_dataset_group(
            datasetGroupArn=dg['datasetGroupArn']
        )
        
        print(f"Dataset Group: {describe['datasetGroup']['name']}")
        print(f"  Creation Time: {describe['datasetGroup']['creationDateTime']}")
        print(f"  Last Update: {describe['datasetGroup']['lastUpdatedDateTime']}")

モデル精度の詳細評価

def evaluate_solution_version_metrics():
    """Solution Version のメトリクスを詳細評価"""
    
    personalize = boto3.client('personalize')
    
    solutions = personalize.list_solutions()
    
    for solution in solutions['solutions']:
        print(f"\nSolution: {solution['name']}")
        
        versions = personalize.list_solution_versions(
            solutionArn=solution['solutionArn']
        )
        
        for version in versions['solutionVersions']:
            print(f"  Version: {version['solutionVersionArn']}")
            
            describe = personalize.describe_solution_version(
                solutionVersionArn=version['solutionVersionArn']
            )
            
            metrics = describe['solutionVersion'].get('metrics', {})
            
            if metrics:
                print(f"    Metrics:")
                print(f"      Precision@25: {metrics.get('precision@25', 'N/A')}")
                print(f"      Recall@25: {metrics.get('recall@25', 'N/A')}")
                print(f"      Normalized RMSE: {metrics.get('normalized_rmse', 'N/A')}")
                print(f"      Mean Average Precision: {metrics.get('mean_average_precision', 'N/A')}")
            
            if metrics.get('precision@25', 0) < 0.1:
                print("    ⚠️ 低精度 - 別の Recipe を試してください")
            
            training_mode = describe['solutionVersion'].get('trainingMode', 'DEFAULT')
            print(f"    Training Mode: {training_mode}")

API パフォーマンス計測

import time

def benchmark_recommendation_api(recommender_arn, user_ids, num_calls=100):
    """推奨 API のレイテンシー計測"""
    
    runtime = boto3.client('personalize-runtime')
    latencies = []
    errors = 0
    
    for i in range(num_calls):
        user_id = user_ids[i % len(user_ids)]
        
        start = time.time()
        try:
            response = runtime.get_recommendations(
                recommenderArn=recommender_arn,
                userId=user_id,
                numResults=10
            )
            latency = (time.time() - start) * 1000  # ms
            latencies.append(latency)
        except Exception as e:
            errors += 1
            print(f"Error: {e}")
    
    print(f"\nAPI Performance ({num_calls} calls):")
    print(f"  Avg Latency: {sum(latencies)/len(latencies):.1f} ms")
    print(f"  Max Latency: {max(latencies):.1f} ms")
    print(f"  Min Latency: {min(latencies):.1f} ms")
    print(f"  P95 Latency: {sorted(latencies)[int(len(latencies)*0.95)]:.1f} ms")
    print(f"  Error Rate: {errors/num_calls*100:.1f}%")
    
    if sum(latencies)/len(latencies) > 500:
        print("  ⚠️ レイテンシー高い - minProvisionedTPS を増加")

セキュリティ・プライバシー考慮事項

✅ データ保護

• PII（個人識別情報）: ユーザーID・email・phone をハッシュ化して保存
• 暗号化: S3 暗号化・KMS キー活用
• アクセス制御: IAM ロール・ポリシーで最小権限設定
• 監査ログ: CloudTrail で API アクセス追跡

❌ プライバシー違反回避

• 個人情報の生データを直接送信しない
• ユーザーの同意なしに行動データ利用しない
• GDPR・個人情報保護法対応データ削除機能利用

ベストプラクティス補足

パフォーマンス最適化

• バッチ処理スケジューリング: 深夜オフピーク時間帯に実行
• 多地域展開: リージョン別キャンペーンで低レイテンシー
• キャッシング: 推奨結果を CDN（CloudFront）でキャッシュ
• 非同期処理: Lambda で EventBridge トリガー化

監視・アラート

• CloudWatch メトリクス: 推奨 API のレイテンシー・エラー率監視
• SQS DLQ: 失敗イベントをデッドレターキューで追跡
• SNS 通知: キャンペーン失敗時にアラート

エンドツーエンド実装チェックリスト

• [ ] Dataset Group 作成・Dataset スキーマ定義
• [ ] CSV データを S3 に準備（最低 1,000 rows）
• [ ] Data Import Job 実行・完了確認
• [ ] Recommender 作成・トレーニング完了待機
• [ ] リアルタイムイベントトラッキング統合
• [ ] API 呼び出しテスト（GetRecommendations）
• [ ] キャンペーン作成（リアルタイムエンドポイント化）
• [ ] フロントエンド統合（推奨表示）
• [ ] メトリクス計測（CTR・Conversion）
• [ ] A/B テスト実施・結果分析

最終更新：2026-04-26 バージョン：v2.0