目次

Rekognition v2.0 完全ガイド(Machine Learning & AI)

概要

Amazon Rekognition は、画像・動画のコンテンツ分析をコード不要で提供するコンピュータビジョン AI サービスです。顔検出・顔認識・物体検出・シーン分析・テキスト抽出・不適切コンテンツ検出・PPE 検出・Face Liveness(生体認証)・動画内の人物トラッキングなどの機能を API 呼び出しで利用でき、ML の専門知識なしに高精度な画像分析を実装できます。Amazon.com の物体検出・顔認識技術を基礎として構築され、毎日数十億の画像・動画を分析する本番実績を備えています。セキュリティゲート・入退室管理・本人確認・コンテンツモデレーション・監視カメラ分析・メディアアーカイブの自動インデックス化に活用されます。


課題と特徴

従来型コンピュータビジョンの課題

  • モデル構築の高度な専門知識が必要: PyTorch・TensorFlow による CNNアーキテクチャ設計・ハイパーパラメータチューニング・大規模 GPU クラスタが必須
  • 大量の ラベル付きデータが必須: 高精度モデルには数百~数万の注釈付きサンプルが必要
  • スケーリング・デプロイの複雑さ: 推論エンジンの最適化・コンテナ化・負荷分散を手動構築
  • 継続的な精度メンテナンス: 新しい環境・照明・角度への適応が手動対応

Rekognition の特徴

特徴 効果
Pre-trained Models Amazon が 20 年以上の実績で構築した高精度モデル、セットアップ不要
Image Analysis(物体・シーン・概念検出) 数千ラベルで自動分類、信頼度スコア付き
Face Detection & Analysis 顔検出・顔属性(年齢・性別・表情・感情)・顔向き・ランドマーク
Face Recognition(顔コレクション) IndexFaces で顔登録、SearchFacesByImage で一致検索
Face Liveness スプーフィング検出(印刷写真・動画・3D マスク)で本人確認
Celebrity Recognition 数万セレブリティを自動認識、カテゴリ分類(政治家・スポーツ選手・俳優)
Text Detection 印刷・手書きテキスト認識(OCR、複数言語対応)
Content Moderation 成人向け・暴力・不適切コンテンツを階層的に分類
PPE Detection 作業服・ヘルメット・ゴーグルなど安全装備の検出
Custom Labels ノーコードで独自分類モデルを構築(最小 10 枚/ラベル)
Video Analysis(非同期) S3 ビデオの人物トラッキング・顔検出・シーン分析・テキスト抽出
Streaming Video Events リアルタイムビデオストリーム(Kinesis Video)の並列分析

アーキテクチャ

graph TB
    subgraph Input["入力"]
        A["S3 静止画像"]
        B["S3 動画ファイル"]
        C["Kinesis Video<br/>ライブストリーム"]
        D["LocalFile<br/>バイナリ"]
    end
    
    subgraph RecognitionCore["Rekognition Core"]
        E["Image Analysis<br/>DetectLabels"]
        F["Face Detection<br/>& Analysis"]
        G["Text Detection<br/>DetectText"]
        H["Content Moderation<br/>DetectModerationLabels"]
    end
    
    subgraph Advanced["Advanced Features"]
        I["Face Recognition<br/>IndexFaces/Search"]
        J["Face Liveness<br/>Anti-Spoofing"]
        K["Celebrity<br/>Recognition"]
        L["Custom Labels"]
        M["Video Analysis<br/>StartPersonTracking"]
    end
    
    subgraph Output["出力"]
        N["JSON<br/>信頼度・ラベル"]
        O["顔ID・座標・属性"]
        P["テキスト・位置情報"]
    end
    
    A --> E
    A --> F
    A --> G
    A --> H
    B --> M
    C --> M
    D --> E
    E --> N
    F --> O
    I --> O
    G --> P

コアコンポーネント

1. DetectLabels(物体・シーン検出)

import boto3

rekognition = boto3.client('rekognition', region_name='ap-northeast-1')

# S3 画像から物体・シーン検出
response = rekognition.detect_labels(
    Image={'S3Object': {'Bucket': 'my-bucket', 'Name': 'scene.jpg'}},
    MaxLabels=10,
    MinConfidence=70  # 信頼度閾値
)

print("検出ラベル:")
for label in response['Labels']:
    print(f"  {label['Name']:20} 信頼度: {label['Confidence']:.1f}%")
    
    # 子ラベル(階層構造)
    for child in label.get('Categories', []):
        print(f"    → {child['Name']}")

# 出力例
"""
検出ラベル:
  Car                   信頼度: 99.2%
  Street                信頼度: 97.8%
  Urban                 信頼度: 95.3%
    → City
    → Built-up area
"""

2. DetectFaces(顔検出・分析)

# 顔の検出・属性分析
response = rekognition.detect_faces(
    Image={'S3Object': {'Bucket': 'my-bucket', 'Name': 'people.jpg'}},
    Attributes=['ALL']  # 全属性取得(感情・年齢・性別等)
)

for face_detail in response['FaceDetails']:
    print(f"信頼度: {face_detail['Confidence']:.1f}%")
    
    # 顔属性
    print(f"  年齢: {face_detail['AgeRange']['Low']}-{face_detail['AgeRange']['High']}")
    print(f"  性別: {face_detail['Gender']['Value']} ({face_detail['Gender']['Confidence']:.1f}%)")
    
    # 感情
    print("  感情スコア:")
    for emotion in face_detail['Emotions']:
        print(f"    {emotion['Type']:15} {emotion['Confidence']:.1f}%")
    
    # 顔ランドマーク(目・鼻・口等の座標)
    landmarks = face_detail['Landmarks']
    print(f"  ランドマーク数: {len(landmarks)}")
    
    # 顔向き(ロール・ピッチ・ヤー)
    print(f"  向き: Roll={face_detail['Pose']['Roll']:.1f}°")

3. Face Recognition - Face Collection

# 顔コレクション(最大 2000 万顔を保存可能)

# Step 1: コレクション作成
rekognition.create_collection(CollectionId='employees')

# Step 2: 従業員の顔を登録
response = rekognition.index_faces(
    CollectionId='employees',
    Image={'S3Object': {'Bucket': 'my-bucket', 'Name': 'emp-001.jpg'}},
    ExternalImageId='emp-001',
    MaxFaces=1,
    QualityFilter='HIGH'  # 高品質のみ
)

print(f"登録された顔: {len(response['FaceRecords'])}")
for face_record in response['FaceRecords']:
    print(f"  FaceId: {face_record['Face']['FaceId']}")

# Step 3: 入退室カメラから顔を検索
def identify_employee(camera_image_uri):
    response = rekognition.search_faces_by_image(
        CollectionId='employees',
        Image={'S3Object': camera_image_uri},
        FaceMatchThreshold=95,  # 95% 以上の類似度
        MaxFaces=1
    )
    
    matches = response['FaceMatches']
    if matches:
        match = matches[0]
        employee_id = match['Face']['ExternalImageId']
        similarity = match['Similarity']
        print(f"✓ 従業員 {employee_id} を認識({similarity:.1f}% 類似度)")
        return employee_id
    else:
        print("✗ 顔が一致しません")
        return None

# Step 4: コレクション削除
rekognition.delete_collection(CollectionId='employees')

4. Face Liveness(生体認証・スプーフィング検出)

# 本人確認時にライブ人物かどうかを判定

response = rekognition.start_face_liveness_session()

session_id = response['SessionId']

# フロントエンド: WebSocket で Kinesis Video に音声映像を送信
# サーバー: Liveness 分析結果を取得

result = rekognition.get_face_liveness_session_results(SessionId=session_id)

liveness_confidence = result['Confidence']
is_live = result['IsLive']

print(f"ライブ判定: {is_live}")
print(f"信頼度: {liveness_confidence:.1f}%")

# 応用例: パスポート + Face Liveness で身分確認
if is_live and liveness_confidence > 99:
    print("✓ 本人確認完了(本人がカメラの前に居る)")
else:
    print("✗ スプーフィング検出(印刷写真・動画・3D マスク等)")

5. DetectModerationLabels(不適切コンテンツ検出)

# ユーザー投稿画像のコンテンツモデレーション

response = rekognition.detect_moderation_labels(
    Image={'S3Object': {'Bucket': 'uploads', 'Name': 'post.jpg'}},
    MinConfidence=60
)

print("不適切コンテンツ検出:")
for label in response['ModerationLabels']:
    print(f"  {label['Name']:25} {label['Confidence']:.1f}%")
    
    # 親ラベル(トップレベルカテゴリ)
    if 'ParentLabels' in label:
        for parent in label['ParentLabels']:
            print(f"    親: {parent['Name']}")

# 検出可能なラベル例:
"""
Explicit Nudity / Suggestive / Violence / Visually Disturbing
Rude Gestures / Weapons / Drugs / Tobacco / Self Injury
Hate Symbols
"""

# 自動ブロック例
for label in response['ModerationLabels']:
    if label['Name'] in ['Explicit Nudity', 'Violence'] and label['Confidence'] > 80:
        print(f"⚠️  投稿をブロック: {label['Name']}")
        # delete_from_platform(post_id)

6. DetectText(テキスト抽出)

# 看板・書類・標識のテキスト認識

response = rekognition.detect_text(
    Image={'S3Object': {'Bucket': 'my-bucket', 'Name': 'sign.jpg'}}
)

print("検出テキスト:")
for text_detection in response['TextDetections']:
    # TYPE: LINE(行)または WORD(単語)
    if text_detection['Type'] == 'LINE':
        print(f"  {text_detection['DetectedText']:30} 信頼度: {text_detection['Confidence']:.1f}%")

# 用途: メニュー OCR、看板読み込み、証券書類の自動読取
# 注: Textract の方が高精度(段落・テーブル構造対応)

7. Custom Labels(カスタムモデル)

# 独自の分類モデルを構築(例: 工場の欠陥検出)

# Step 1: S3 にラベル付き画像をアップロード
"""
s3://my-bucket/training/
  defective/
    image-001.jpg
    image-002.jpg
    ...  (最小 10 枚)
  normal/
    image-001.jpg
    image-002.jpg
    ...
"""

# Step 2: Dataset 作成
response = rekognition.create_dataset(
    DatasetType='TRAIN',
    DatasetSource={'GroundTruthManifestS3Object': {
        'Bucket': 'my-bucket',
        'Name': 'manifest.json'
    }}
)

dataset_arn = response['DatasetArn']

# Step 3: Project・Model 作成・トレーニング
project_response = rekognition.create_project(ProjectName='defect-detector')
project_arn = project_response['ProjectArn']

model_response = rekognition.create_model(
    ProjectArn=project_arn,
    ModelDescription='Manufacturing defect detector',
    TrainingDatasetArn=dataset_arn
)

model_arn = model_response['ModelArn']

# Step 4: モデルをトレーニング(数時間)
rekognition.train_model(
    ProjectArn=project_arn,
    ModelType='OBJECT_DETECTION'  # または CLASSIFICATION
)

# Step 5: パフォーマンス確認
while True:
    describe = rekognition.describe_model(ModelArn=model_arn)
    status = describe['ModelDescription']['Status']
    if status == 'TRAINED':
        print(f"トレーニング完了")
        print(f"精度: {describe['ModelDescription']['EvaluationResult']}")
        break
    elif status == 'TRAINING_FAILED':
        print("トレーニング失敗")
        break
    time.sleep(30)

# Step 6: モデルをデプロイ($4/推論ユニット-時間)
rekognition.start_model(
    ProjectArn=project_arn,
    ModelArn=model_arn,
    MinInferenceUnits=1
)

# Step 7: 推論実行
inference_response = rekognition.detect_custom_labels(
    ProjectVersionArn=model_arn,
    Image={'S3Object': {'Bucket': 'my-bucket', 'Name': 'test-image.jpg'}}
)

for detection in inference_response['CustomLabels']:
    print(f"{detection['Name']:20} 信頼度: {detection['Confidence']:.1f}%")

8. Video Analysis(非同期動画分析)

# 動画ファイル(S3)の非同期分析

# Step 1: 人物トラッキング開始
response = rekognition.start_person_tracking(
    Video={'S3Object': {'Bucket': 'videos', 'Name': 'surveillance.mp4'}},
    ClientRequestToken='tracking-token-001'
)

job_id = response['JobId']

# Step 2: ジョブ完了を待機(SNS 通知推奨)
while True:
    result = rekognition.get_person_tracking(JobId=job_id)
    
    if result['JobStatus'] == 'SUCCEEDED':
        print(f"✓ 分析完了")
        
        # フレームごとの人物トラッキング
        for person_tracking in result['Persons']:
            timestamp = person_tracking['Timestamp']  # ミリ秒
            person_index = person_tracking['Person']['Index']
            
            print(f"フレーム {timestamp}ms: 人物 {person_index}")
            
            # 顔詳細(存在する場合)
            if 'FaceDetails' in person_tracking['Person']:
                face = person_tracking['Person']['FaceDetails']
                print(f"  感情: {face['Emotions']}")
        
        break
    elif result['JobStatus'] == 'FAILED':
        print("分析失敗")
        break
    
    time.sleep(5)

# 他の動画分析 API:
# - start_face_detection: 顔検出
# - start_label_detection: シーン・物体検出
# - start_content_moderation: 不適切コンテンツ検出
# - start_text_detection: テキスト抽出

9. Streaming Video Events(リアルタイム処理)

# Kinesis Video Streams のリアルタイムビデオをアナライズ

response = rekognition.start_stream_processor(
    Name='real-time-face-detector',
    Input={
        'KinesisVideoStream': {
            'Arn': 'arn:aws:kinesisvideo:ap-northeast-1:123456789:stream/camera-feed'
        }
    },
    Output={
        'KinesisDataStream': {
            'Arn': 'arn:aws:kinesis:ap-northeast-1:123456789:stream/rekognition-output'
        }
    },
    RoleArn='arn:aws:iam::123456789:role/RekognitionRole',
    Settings={
        'FaceSearch': {
            'CollectionId': 'security-faces',
            'FaceMatchThreshold': 95
        }
    }
)

# Kinesis Data Streams から結果を取得(Lambda で処理)
# → リアルタイム警告・エスカレーション

主要ユースケース(10+)

1. 入退室管理・セキュリティゲート

class AccessControlSystem:
    def __init__(self, collection_id='employees'):
        self.rekognition = boto3.client('rekognition')
        self.collection_id = collection_id
        self.dynamodb = boto3.resource('dynamodb')
    
    def check_access(self, gate_camera_image_uri):
        """カメラ映像から従業員を認識してゲート開閉"""
        
        response = self.rekognition.search_faces_by_image(
            CollectionId=self.collection_id,
            Image={'S3Object': gate_camera_image_uri},
            FaceMatchThreshold=92
        )
        
        if response['FaceMatches']:
            employee_id = response['FaceMatches'][0]['Face']['ExternalImageId']
            similarity = response['FaceMatches'][0]['Similarity']
            
            # アクセスログを記録
            table = self.dynamodb.Table('AccessLogs')
            table.put_item(Item={
                'employee_id': employee_id,
                'timestamp': int(time.time()),
                'confidence': similarity,
                'gate': 'main-entrance'
            })
            
            return {'access': 'ALLOWED', 'employee_id': employee_id}
        else:
            # 不明人物 → セキュリティアラート
            return {'access': 'DENIED', 'alert': 'Unknown person detected'}

2. 本人確認(KYC / eKYC)

class IdentityVerificationPipeline:
    def __init__(self):
        self.rekognition = boto3.client('rekognition')
        self.textract = boto3.client('textract')
    
    def verify_identity(self, passport_image_uri, selfie_image_uri):
        """パスポート + セルフィで本人確認"""
        
        # Step 1: パスポートから顔抽出
        passport_response = self.rekognition.detect_faces(
            Image={'S3Object': passport_image_uri},
            Attributes=['ALL']
        )
        
        passport_face = passport_response['FaceDetails'][0] if passport_response['FaceDetails'] else None
        
        # Step 2: セルフィが本人か(Face Liveness + 顔比較)
        liveness_response = self.rekognition.start_face_liveness_session()
        # ... ライブ判定 ...
        
        # Step 3: 顔の一致度を検査
        compare_response = self.rekognition.compare_faces(
            SourceImage={'S3Object': passport_image_uri},
            TargetImage={'S3Object': selfie_image_uri},
            SimilarityThreshold=95
        )
        
        if compare_response['FaceMatches']:
            similarity = compare_response['FaceMatches'][0]['Similarity']
            print(f"✓ 本人確認完了(一致度: {similarity:.1f}%)")
            return True
        else:
            print("✗ 顔が一致しません")
            return False

3. UGC コンテンツモデレーション

class UGCModerationPipeline:
    def __init__(self):
        self.rekognition = boto3.client('rekognition')
        self.s3 = boto3.client('s3')
        self.sns = boto3.client('sns')
    
    def moderate_user_post(self, post_id, image_s3_uri):
        """ユーザー投稿の自動モデレーション"""
        
        # Step 1: コンテンツモデレーション実行
        moderation = self.rekognition.detect_moderation_labels(
            Image={'S3Object': image_s3_uri},
            MinConfidence=50
        )
        
        # Step 2: リスク判定
        violation_found = False
        for label in moderation['ModerationLabels']:
            if label['Confidence'] > 80:
                if label['Name'] in ['Explicit Nudity', 'Violence']:
                    violation_found = True
                    print(f"⚠️  重大違反: {label['Name']}")
                    
                    # 即座に削除 + ユーザー報告
                    self.s3.delete_object(Bucket='uploads', Key=image_s3_uri.split('/')[-1])
                    self.send_notification(post_id, 'REMOVED_POLICY_VIOLATION')
                    return False
        
        # Step 3: グレーゾーン(信頼度 60-80%)は人間審査キューに
        if any(label['Confidence'] > 60 for label in moderation['ModerationLabels']):
            print(f"📋 レビューキューに追加: {post_id}")
            self.notify_moderators(post_id)
            return None  # pending
        
        print(f"✓ 承認: {post_id}")
        return True

4. 監視カメラの異常検知

class SurveillanceAnalytics:
    def __init__(self):
        self.rekognition = boto3.client('rekognition')
        self.cloudwatch = boto3.client('cloudwatch')
    
    def analyze_security_footage(self, video_s3_uri):
        """監視ビデオから異常(人物侵入など)を検知"""
        
        # 人物トラッキング実行
        response = self.rekognition.start_person_tracking(
            Video={'S3Object': video_s3_uri}
        )
        
        job_id = response['JobId']
        
        # 分析完了待機
        result = self.rekognition.get_person_tracking(JobId=job_id)
        
        # 時間帯外の人物検出 → アラート
        for tracking in result['Persons']:
            timestamp_ms = tracking['Timestamp']
            timestamp = datetime.fromtimestamp(timestamp_ms / 1000)
            
            # 営業時間外(21:00-6:00)の人物を検出
            if timestamp.hour < 6 or timestamp.hour >= 21:
                print(f"🚨 不正侵入の可能性: {timestamp}")
                self.cloudwatch.put_metric_alarm(
                    AlarmName='Unauthorized-Access-Detected',
                    MetricName='UnauthorizedPersonCount'
                )

5. メディア・アーカイブの自動インデックス

class MediaLibraryIndexer:
    def __init__(self):
        self.rekognition = boto3.client('rekognition')
        self.opensearch = boto3.client('opensearchserverless')
    
    def index_video_archive(self, video_s3_uri, video_id):
        """テレビ番組・スポーツ映像を自動インデックス"""
        
        # 複数の分析を並列実行
        jobs = {}
        
        # 1. 顔検出(出演者識別)
        jobs['faces'] = self.rekognition.start_face_detection(
            Video={'S3Object': video_s3_uri}
        )
        
        # 2. シーン検出("Goal", "Interview", "Ceremony" 等)
        jobs['scenes'] = self.rekognition.start_label_detection(
            Video={'S3Object': video_s3_uri}
        )
        
        # 3. テキスト検出(テロップ・字幕)
        jobs['text'] = self.rekognition.start_text_detection(
            Video={'S3Object': video_s3_uri}
        )
        
        # 分析完了待機・インデックス作成
        index_data = {}
        for job_type, response in jobs.items():
            job_id = response['JobId']
            # ... ジョブ完了待機 ...
            # 結果を OpenSearch にインデックス
        
        # クエリ例: "Goal scene with David Beckham"
        # → インデックスから自動推奨

6. 工場・農業の自動検査

class ManufacturingQualityControl:
    def __init__(self):
        self.rekognition = boto3.client('rekognition')
    
    def detect_defects(self, production_line_image):
        """製造ラインの欠陥検出"""
        
        response = self.rekognition.detect_custom_labels(
            ProjectVersionArn='arn:aws:rekognition:ap-northeast-1:123456789:model/defect-detector/version/1',
            Image={'S3Object': production_line_image}
        )
        
        defects = []
        for detection in response['CustomLabels']:
            if detection['Name'] == 'DEFECT' and detection['Confidence'] > 85:
                defects.append({
                    'type': detection['Name'],
                    'confidence': detection['Confidence'],
                    'location': detection.get('Geometry', {})
                })
        
        if defects:
            print(f"⚠️  {len(defects)} 件の欠陥を検出 → 除却ラインに")
            return False  # reject
        else:
            print("✓ 品質OK")
            return True  # accept

7. スマートシティ・交通監視

class SmartCityTrafficMonitoring:
    def __init__(self):
        self.rekognition = boto3.client('rekognition')
    
    def analyze_intersection(self, camera_stream_arn):
        """交差点のリアルタイム分析"""
        
        # Kinesis Video → Rekognition Streaming の設定
        response = self.rekognition.start_stream_processor(
            Name='traffic-analyzer',
            Input={'KinesisVideoStream': {'Arn': camera_stream_arn}},
            Settings={
                'FaceSearch': {'CollectionId': 'traffic-violators'}  # 信号無視者検出
            }
        )
        
        # 結果: Kinesis Data Streams に出力
        # → 信号無視・迷惑行為を自動検出・記録

CLI・SDK・IaC 例(5+)

CLI 例

# 物体・シーン検出
aws rekognition detect-labels \
  --image 'S3Object={Bucket=my-bucket,Name=photo.jpg}' \
  --max-labels 10 \
  --region ap-northeast-1

# 顔検出
aws rekognition detect-faces \
  --image 'S3Object={Bucket=my-bucket,Name=people.jpg}' \
  --attributes ALL \
  --region ap-northeast-1

# テキスト検出
aws rekognition detect-text \
  --image 'S3Object={Bucket=my-bucket,Name=sign.jpg}' \
  --region ap-northeast-1

# コンテンツモデレーション
aws rekognition detect-moderation-labels \
  --image 'S3Object={Bucket=uploads,Name=post.jpg}' \
  --min-confidence 70 \
  --region ap-northeast-1

# 顔コレクション管理
aws rekognition create-collection \
  --collection-id employees \
  --region ap-northeast-1

aws rekognition index-faces \
  --collection-id employees \
  --image 'S3Object={Bucket=my-bucket,Name=emp.jpg}' \
  --external-image-id emp-001 \
  --region ap-northeast-1

aws rekognition search-faces-by-image \
  --collection-id employees \
  --image 'S3Object={Bucket=my-bucket,Name=camera.jpg}' \
  --face-match-threshold 95 \
  --region ap-northeast-1

SDK 例(Python)

import boto3
import json

rekognition = boto3.client('rekognition', region_name='ap-northeast-1')

# 複数画像バッチ処理
def batch_analyze_images(bucket, image_keys):
    results = {}
    for key in image_keys:
        response = rekognition.detect_labels(
            Image={'S3Object': {'Bucket': bucket, 'Name': key}},
            MaxLabels=5
        )
        results[key] = [l['Name'] for l in response['Labels']]
    return results

# 顔の比較(パスポート vs セルフィ)
def verify_face_match(source_image_key, target_image_key):
    response = rekognition.compare_faces(
        SourceImage={'S3Object': {'Bucket': 'docs', 'Name': source_image_key}},
        TargetImage={'S3Object': {'Bucket': 'uploads', 'Name': target_image_key}},
        SimilarityThreshold=90
    )
    
    if response['FaceMatches']:
        return {
            'match': True,
            'similarity': response['FaceMatches'][0]['Similarity']
        }
    return {'match': False}

IaC 例(CloudFormation)

AWSTemplateFormatVersion: '2010-09-09'
Description: Rekognition Face Collection for Access Control

Resources:
  RekognitionRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: rekognition.amazonaws.com
            Action: 'sts:AssumeRole'
      ManagedPolicyArns:
        - 'arn:aws:iam::aws:policy/AmazonRekognitionFullAccess'

  RekognitionLambdaRole:
    Type: AWS::IAM::Role
    Properties:
      AssumeRolePolicyDocument:
        Version: '2012-10-17'
        Statement:
          - Effect: Allow
            Principal:
              Service: lambda.amazonaws.com
            Action: 'sts:AssumeRole'
      ManagedPolicyArns:
        - 'arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole'
      Policies:
        - PolicyName: RekognitionAccess
          PolicyDocument:
            Version: '2012-10-17'
            Statement:
              - Effect: Allow
                Action:
                  - 'rekognition:DetectLabels'
                  - 'rekognition:DetectFaces'
                  - 'rekognition:SearchFacesByImage'
                Resource: '*'
              - Effect: Allow
                Action:
                  - 's3:GetObject'
                Resource: 'arn:aws:s3:::my-bucket/*'

  AccessControlLambda:
    Type: AWS::Lambda::Function
    Properties:
      FunctionName: access-control-face-recognition
      Runtime: python3.11
      Role: !GetAtt RekognitionLambdaRole.Arn
      Handler: index.lambda_handler
      Code:
        ZipFile: |
          import boto3
          rekognition = boto3.client('rekognition')
          def lambda_handler(event, context):
              bucket = event['bucket']
              key = event['key']
              response = rekognition.search_faces_by_image(
                  CollectionId='employees',
                  Image={'S3Object': {'Bucket': bucket, 'Name': key}},
                  FaceMatchThreshold=92
              )
              if response['FaceMatches']:
                  return {'statusCode': 200, 'body': 'ACCESS_GRANTED'}
              return {'statusCode': 403, 'body': 'ACCESS_DENIED'}

  S3Bucket:
    Type: AWS::S3::Bucket
    Properties:
      BucketName: access-control-images
      VersioningConfiguration:
        Status: Enabled

比較表

特性 Rekognition Google Cloud Vision Azure Computer Vision Clarifai Hugging Face Vision
物体検出
顔認識(コレクション)
顔 Liveness
テキスト抽出(OCR)
コンテンツモデレーション
セレブ認識
PPE 検出
カスタムモデル
動画分析
リアルタイム・ストリーミング
無料枠 5,000 画像/月(12ヶ月) 1,000 画像/月(常時) 5,000 ユニット/月 無し セルフホスト(無料)
料金 $0.001/画像 $1.5-10/1,000 $1-6/1,000 $0.002-0.01/API セルフホスト

ベストプラクティス

✅ DO

  • 高品質画像を用意: 最小 100x100 px、顔は画像の 20% 以上
  • 照明・角度に配慮: 正面顔・左右 45° まで許容
  • バッチ処理で効率化: 大量画像は非同期 Video API を活用
  • キャッシング: 同一人物の顔検索は Rekognition より Face Collection が効率的
  • 信頼度閾値を調整: ユースケースに応じて段階的に(セキュリティ > 90%, コンテンツ分類 > 60%)
  • 人間確認ループ: A2I・Textract と組み合わせて高精度パイプライン実現

❌ DON’T

  • 低品質・小さい顔: 認識精度が著しく低下
  • 逆光・赤目・目隠し: Face Detection 失敗の可能性
  • フェアネス無視: バイアス監視(年齢・性別の推定が全グループで同等か確認)
  • 個人情報漏洩: 抽出画像を無暗に保存・共有しない
  • 無限スケーリング: Custom Labels は推論ユニット課金(テスト時に停止推奨)

トラブルシューティング

問題 原因 解決策
“InvalidImageFormatException” 形式が JPEG/PNG でない 画像を JPEG/PNG に変換(base64 or S3 アクセス)
顔検出失敗(0 件) 画像が小さい・背向き・顔半分 最小 100x100 px、正面顔を用意
“ResourceNotFoundException” Face Collection コレクション存在しない create_collection() を先に実行
高い False Positive(誤検出) 信頼度閾値が低すぎる MinConfidence を 70-80% に設定
Custom Labels トレーニング失敗 ラベルごとに 10 枚未満 各ラベルで最低 10 枚、推奨 100 枚
动影视频分析遅い 動画が大きい・長い(>4h) 複数チャンク分割、並列処理

2025-2026 最新動向

  • Face Liveness 拡張: より高度なスプーフィング検出(ディープフェイク対応を検討中)
  • マルチモーダル分析: Image + Text + Metadata 統合分析
  • Bedrock 統合: Rekognition の抽出結果を Claude・Llama で解釈
  • Real-time Streaming 高速化: Kinesis Video の低レイテンシー処理
  • Custom Labels の自動リトレーニング: Flywheel 的な継続学習機能の拡大
  • プライバシー強化: より細粒度な顔データ匿名化オプション

学習リソース

公式ドキュメント(8+)

  1. Amazon Rekognition Developer Guide
  2. Rekognition API Reference
  3. Face Recognition Best Practices
  4. Custom Labels Guide
  5. Video Analysis Documentation
  6. Streaming Video Events
  7. AWS SDK Examples (Python)
  8. Rekognition Pricing & Free Tier

比較・ベンダー情報(5+)

  1. Google Cloud Vision API - 汎用 CV API
  2. Azure AI Vision - Enterprise 向け
  3. Clarifai - 業界特化(ファッション・メディア等)
  4. Hugging Face Model Hub - OSS ビジョンモデル
  5. OpenCV - 自ホスト CV ライブラリ

実装例・チェックリスト

採用判断チェックリスト

  • [ ] 画像・動画内の物体・シーン・テキストを自動検出したいか
  • [ ] 顔認識・照合を使った本人確認・入退室管理が必要か
  • [ ] Face Liveness(スプーフィング検出)で本人確認を強化したいか
  • [ ] UGC プラットフォームでコンテンツを自動モデレーションしたいか
  • [ ] 独自の物体分類(工場の欠陥・農業の成熟度等)に Custom Labels を使いたいか
  • [ ] セキュリティカメラの 24/7 監視で異常検知をしたいか
  • [ ] ML 専門知識がなくコンピュータビジョンを実装したいか
  • [ ] HIPAA 対応が必要か(Rekognition は HIPAA 適格)

実装例:エンドツーエンド KYC パイプライン

class KYCPipeline:
    """本人確認・KYC のフルエンドツーエンド実装"""
    
    def __init__(self):
        self.rekognition = boto3.client('rekognition')
        self.textract = boto3.client('textract')
        self.dynamodb = boto3.resource('dynamodb')
    
    def verify_user(self, user_id, passport_image_uri, selfie_image_uri):
        """完全な KYC ワークフロー"""
        
        # Step 1: パスポートテキスト抽出(Textract)
        passport_text = self.textract.analyze_document(
            Document={'S3Object': passport_image_uri},
            FeatureTypes=['FORMS']
        )
        
        # Step 2: パスポート顔検出
        passport_face = self.rekognition.detect_faces(
            Image={'S3Object': passport_image_uri},
            Attributes=['ALL']
        )
        
        if not passport_face['FaceDetails']:
            return {'status': 'REJECTED', 'reason': 'No face in passport'}
        
        # Step 3: セルフィの生体認証(Liveness)
        liveness = self.rekognition.start_face_liveness_session()
        # ... ユーザーが自撮りビデオを送信 ...
        
        # Step 4: 顔一致度検証
        comparison = self.rekognition.compare_faces(
            SourceImage={'S3Object': passport_image_uri},
            TargetImage={'S3Object': selfie_image_uri},
            SimilarityThreshold=95
        )
        
        if not comparison['FaceMatches']:
            return {'status': 'REJECTED', 'reason': 'Face mismatch'}
        
        similarity = comparison['FaceMatches'][0]['Similarity']
        
        # Step 5: DynamoDB に保存
        table = self.dynamodb.Table('KYCVerifications')
        table.put_item(Item={
            'user_id': user_id,
            'status': 'VERIFIED',
            'similarity_score': similarity,
            'timestamp': int(time.time())
        })
        
        return {'status': 'APPROVED', 'similarity': similarity}

まとめ

Rekognition は 「画像・動画のコンピュータビジョン分析を API で即座に提供する AI サービス」。物体検出・顔認識・顔 Liveness・コンテンツモデレーション・動画分析を ML 専門知識なしで実装でき、Custom Labels で独自分類モデルもノーコード構築できます。セキュリティ(入退室・本人確認)・コンテンツ管理(UGC モデレーション)・監視(監視カメラ・スポーツ分析)・製造(品質管理)・メディア(アーカイブ自動インデックス)など幅広いユースケースに対応し、AWS AI/ML エコシステム(Textract・Comprehend・Bedrock・SageMaker)との統合で完全なパイプラインを実現します。


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