Amazon Mechanical Turk v2.0 完全ガイド 2026

概要

Amazon Mechanical Turk（MTurk） は、世界中のワーカーにヒューマンインテリジェンスタスク（HIT）を分散実行させるクラウドソーシングプラットフォームです。画像ラベリング・テキスト分類・データ検証・コンテンツモデレーション・アンケート等の AI が苦手なタスクを、小額報酬で大規模に並列処理。AWS SageMaker Ground Truth と統合し、機械学習訓練データの効率的作成・品質管理を実現します。2024-2026年に AI アシスト・品質スコア・プライベートワーカープール対応を拡張。

解決する課題

大規模データラベリングの低コスト化

• ML モデル訓練には数百万件のラベル付きデータが必須だが、社内スタッフだけでは対応不可能、MTurk で 1 件 $0.01-0.50 の低コストで実現

スケーラブルな人的リソース

• ピーク時のみ処理能力を増やす必要があるが、採用・契約不要で オンデマンドにワーカーを増員可能

品質管理の自動化

• 複数ワーカーの回答を多数決・ゴールデンデータセット検証で品質を数値化・統計管理

SageMaker と統合

• SageMaker Ground Truth でラベリングワークフローを自動化、MTurk ワーカーを直接プール化

グローバル多言語対応

• 190+ 国のワーカー、175+ 言語で多言語テキスト処理・翻訳タスク実行可能

アーキテクチャ全体

Requester（依頼者） ← EC2 / Lambda / SageMaker
    ↓ HIT 作成・承認・支払い
┌─────────────────────────────────────────────────────────┐
│              Amazon Mechanical Turk                     │
│                                                         │
│  Marketplace（HIT 公開）                               │
│  ├─ HIT 閲覧・検索（ワーカー向け）                     │
│  ├─ 資格フィルター（Master Worker・カスタム）        │
│  └─ リアルタイム報酬表示                              │
│                                                         │
│  Assignment Management                                 │
│  ├─ Worker が HIT を選択・実行                         │
│  ├─ 複数 Worker の多数決（品質向上）                 │
│  └─ 回答の自動集約・統計分析                         │
│                                                         │
│  Quality Control                                        │
│  ├─ ゴールデンデータセット（既知正解）検証           │
│  ├─ Worker 承認率追跡                                 │
│  ├─ 異常検出（外れ値除外）                          │
│  └─ Worker 評判スコア（Tier）管理                    │
│                                                         │
│  SageMaker Ground Truth Integration                    │
│  ├─ Active Learning（ML で難しいサンプル自動選出）  │
│  ├─ 自動ラベリング（ML モデル予測）                 │
│  └─ Consolidation（複数回答の最終判定）             │
└─────────────────────────────────────────────────────────┘
    ↑ HIT 回答受け取り・支払い
Worker（ワーカー） ← MTurk Marketplace

コアコンセプト詳細

1. HIT（Human Intelligence Task）

1 つの作業単位。例：「この画像に犬が映っているか?」

{
  "Title": "画像分類：動物の種類",
  "Description": "表示された画像に映っている動物を分類してください。",
  "Keywords": "image, animal, labeling, classification",
  "Reward": "0.05",
  "AssignmentDurationInSeconds": 300,
  "LifetimeInSeconds": 604800,
  "MaxAssignments": 3,
  "AutoApprovalDelayInSeconds": 86400,
  "QualificationRequirements": [
    {
      "QualificationTypeId": "00000000000000000040",
      "Comparator": "GreaterThanOrEqualTo",
      "IntegerValues": [95],
      "RequiredToPreview": false
    }
  ],
  "Question": "..."
}

2. HITType（HIT テンプレート）

同一設定の HIT をバルク作成する際のテンプレート。

# HITType 作成（テンプレート）
aws mturk create-hit-type \
  --title "テキスト感情分類（日本語）" \
  --description "日本語テキストの感情をポジティブ/ネガティブ/中立に分類" \
  --reward "0.10" \
  --assignment-duration-in-seconds 600 \
  --lifetime-in-seconds 1209600 \
  --auto-approval-delay-in-seconds 86400 \
  --keywords "text, sentiment, classification, japanese"

# HITType ID: 3GSNXQ1GGGH6I2LM4JX2

# テンプレートベースで HIT 作成（バルク）
for image_url in "${IMAGE_URLS[@]}"; do
  aws mturk create-hit \
    --hit-type-id 3GSNXQ1GGGH6I2LM4JX2 \
    --question "<HTMLQuestion>...</HTMLQuestion>" \
    --max-assignments 3
done

3. Assignment（アサイン）

1 つの HIT に対する 1 ワーカーの実行単位。

# Assignment の取得
import boto3
import xml.etree.ElementTree as ET

mturk = boto3.client('mturk', region_name='us-east-1')

response = mturk.list_assignments_for_hit(
    HITId='3J53R25CBDO6QS3CKXPL7CUF5CVVCS',
    AssignmentStatuses=['Submitted'],  # Submitted / Approved / Rejected
    MaxResults=100
)

for assignment in response['Assignments']:
    assignment_id = assignment['AssignmentId']
    worker_id = assignment['WorkerId']
    submit_time = assignment['SubmitTime']
    
    # XML 回答をパース
    answer_xml = assignment['Answer']
    root = ET.fromstring(answer_xml)
    ns = {
        'ns': 'http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2005-10-01/QuestionFormAnswers.xsd'
    }
    
    answers = {}
    for answer_elem in root.findall('.//ns:Answer', ns):
        question_id = answer_elem.find('ns:QuestionIdentifier', ns).text
        answer_value = answer_elem.find('ns:FreeText', ns).text
        answers[question_id] = answer_value
    
    print(f"Worker {worker_id}: {answers}")
    
    # 品質チェック
    if is_quality_answer(answers):
        mturk.approve_assignment(AssignmentId=assignment_id)
    else:
        mturk.reject_assignment(
            AssignmentId=assignment_id,
            RequesterFeedback='回答品質が不足しています。'
        )

4. Qualification（資格）

ワーカーの能力・信頼性を示す認定。

# カスタム Qualification Type 作成
aws mturk create-qualification-type \
  --name "日本語 NLP 専門家" \
  --description "日本語テキスト処理タスクに精通したワーカー認定" \
  --status Active \
  --test-duration-in-seconds 600 \
  --test '<QuestionForm>...</QuestionForm>' \
  --answer-key '<AnswerKey>...</AnswerKey>'

# Qualification ID: 3MEBMRFH2YLE8QA0G2EH1YEI

# 既存ワーカーに Qualification を付与
aws mturk associate-qualification-with-worker \
  --qualification-type-id 3MEBMRFH2YLE8QA0G2EH1YEI \
  --worker-id A3R2QQ8B7X9Q8C \
  --integer-value 95  # スコア（0-100）

システム提供の Qualification：

• Master: 承認率 98% 以上・10,000+ HIT 完了のトップワーカー
• Locale: 特定国（日本・米国等）在住ワーカー
• PHR: 認定トレーニング完了ワーカー（2026 新規）

Python での HIT 作成・管理

import boto3
import json
import time
from datetime import datetime

class MTurkDataLabeler:
    def __init__(self, environment='sandbox'):
        """
        MTurk クライアント初期化
        
        Args:
            environment: 'sandbox' (テスト) / 'production' (本番)
        """
        endpoint = (
            'https://mturk-requester-sandbox.us-east-1.amazonaws.com'
            if environment == 'sandbox'
            else 'https://mturk-requester.us-east-1.amazonaws.com'
        )
        
        self.mturk = boto3.client('mturk', region_name='us-east-1', endpoint_url=endpoint)
        self.sagemaker = boto3.client('sagemaker', region_name='us-east-1')
        self.dynamodb = boto3.resource('dynamodb')

    def create_image_labeling_hit(
        self,
        hit_type_id: str,
        image_urls: list,
        max_assignments: int = 3
    ):
        """画像ラベリング HIT バルク作成"""
        
        created_hits = []
        
        for idx, image_url in enumerate(image_urls):
            # HTML フォーム生成（ラベリング UI）
            question_html = f"""
<HTMLQuestion xmlns="http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2011-11-11/HTMLQuestion.xsd">
  <HTMLContent><![CDATA[
    <!DOCTYPE html>
    <html>
    <head>
      <meta charset="UTF-8">
      <title>Image Labeling Task</title>
      <style>
        body {{ font-family: Arial; margin: 20px; }}
        img {{ max-width: 600px; border: 1px solid #ccc; }}
        .form-group {{ margin: 15px 0; }}
        label {{ display: block; margin: 5px 0; }}
        input[type="radio"] {{ margin-right: 5px; }}
      </style>
    </head>
    <body>
      <h2>Image Classification Task</h2>
      <p><strong>Task {idx + 1} of {len(image_urls)}</strong></p>
      
      <img src="{image_url}" alt="Image to classify">
      
      <div class="form-group">
        <p><strong>What is the main object in this image?</strong></p>
        <label><input type="radio" name="object" value="cat"> Cat</label>
        <label><input type="radio" name="object" value="dog"> Dog</label>
        <label><input type="radio" name="object" value="other"> Other Animal</label>
        <label><input type="radio" name="object" value="none"> No Animal</label>
      </div>
      
      <div class="form-group">
        <p><strong>Confidence Level:</strong></p>
        <label><input type="radio" name="confidence" value="high"> High</label>
        <label><input type="radio" name="confidence" value="medium"> Medium</label>
        <label><input type="radio" name="confidence" value="low"> Low</label>
      </div>
      
      <input type="hidden" name="image_url" value="{image_url}">
    </body>
    </html>
  ]]></HTMLContent>
  <FrameHeight>600</FrameHeight>
</HTMLQuestion>
"""
            
            try:
                response = self.mturk.create_hit(
                    HITTypeId=hit_type_id,
                    MaxAssignments=max_assignments,
                    Question=question_html,
                    LifetimeInSeconds=604800  # 7 日間有効
                )
                
                hit_id = response['HIT']['HITId']
                created_hits.append({
                    'image_url': image_url,
                    'hit_id': hit_id,
                    'created_time': datetime.now().isoformat()
                })
                
                print(f"✓ HIT created: {hit_id} for {image_url}")
                
            except Exception as e:
                print(f"✗ Failed to create HIT for {image_url}: {str(e)}")
        
        return created_hits

    def get_hit_results(self, hit_id: str):
        """HIT の回答を取得・集約"""
        
        response = self.mturk.list_assignments_for_hit(
            HITId=hit_id,
            AssignmentStatuses=['Submitted', 'Approved'],
            MaxResults=100
        )
        
        results = {
            'hit_id': hit_id,
            'total_assignments': len(response['Assignments']),
            'answers': [],
            'consensus': {}
        }
        
        # 回答を解析
        import xml.etree.ElementTree as ET
        ns = {
            'ns': 'http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2005-10-01/QuestionFormAnswers.xsd'
        }
        
        object_votes = {}
        confidence_votes = {}
        
        for assignment in response['Assignments']:
            answer_xml = assignment['Answer']
            root = ET.fromstring(answer_xml)
            
            answer_dict = {}
            for answer_elem in root.findall('.//ns:Answer', ns):
                question_id = answer_elem.find('ns:QuestionIdentifier', ns).text
                answer_value = answer_elem.find('ns:FreeText', ns).text
                answer_dict[question_id] = answer_value
            
            results['answers'].append({
                'worker_id': assignment['WorkerId'],
                'answer': answer_dict,
                'submit_time': assignment['SubmitTime'].isoformat(),
                'assignment_id': assignment['AssignmentId']
            })
            
            # 投票集計（多数決）
            obj = answer_dict.get('object')
            if obj:
                object_votes[obj] = object_votes.get(obj, 0) + 1
            
            conf = answer_dict.get('confidence')
            if conf:
                confidence_votes[conf] = confidence_votes.get(conf, 0) + 1
        
        # コンセンサス決定（最多票）
        if object_votes:
            results['consensus']['object'] = max(object_votes, key=object_votes.get)
            results['consensus']['object_agreement'] = object_votes[results['consensus']['object']] / len(response['Assignments'])
        
        if confidence_votes:
            results['consensus']['confidence'] = max(confidence_votes, key=confidence_votes.get)
        
        return results

    def approve_assignments_by_quality(
        self,
        hit_id: str,
        golden_dataset: list,  # [{'object': 'cat', ...}]
        approval_threshold: float = 0.66
    ):
        """品質チェック後に Assignment を自動承認"""
        
        results = self.get_hit_results(hit_id)
        approved_count = 0
        rejected_count = 0
        
        for answer_data in results['answers']:
            assignment_id = answer_data['assignment_id']
            answer = answer_data['answer']
            
            # ゴールデンデータセット（既知正解）と比較
            matches = 0
            for golden in golden_dataset:
                if answer.get('object') == golden['object']:
                    matches += 1
                    break
            
            accuracy = matches / max(len(golden_dataset), 1)
            
            if accuracy >= approval_threshold:
                self.mturk.approve_assignment(
                    AssignmentId=assignment_id,
                    OverrideRejection=False
                )
                approved_count += 1
                print(f"✓ Approved {assignment_id} (accuracy: {accuracy:.2%})")
            else:
                self.mturk.reject_assignment(
                    AssignmentId=assignment_id,
                    RequesterFeedback=f"Accuracy too low: {accuracy:.2%}. Please review the instructions."
                )
                rejected_count += 1
                print(f"✗ Rejected {assignment_id} (accuracy: {accuracy:.2%})")
        
        return {
            'approved': approved_count,
            'rejected': rejected_count,
            'total': len(results['answers'])
        }

    def batch_create_hits_from_csv(self, csv_file: str, hit_type_id: str):
        """CSV ファイルから大量 HIT 作成"""
        
        import pandas as pd
        
        df = pd.read_csv(csv_file)
        created_hits = []
        
        for idx, row in df.iterrows():
            hit = self.create_image_labeling_hit(
                hit_type_id=hit_type_id,
                image_urls=[row['image_url']],
                max_assignments=row.get('assignments', 3)
            )
            created_hits.extend(hit)
            
            # Rate limiting
            if (idx + 1) % 10 == 0:
                print(f"Created {idx + 1} HITs, sleeping...")
                time.sleep(5)
        
        return created_hits

    def get_account_balance(self):
        """アカウント残高確認"""
        response = self.mturk.get_account_balance()
        return {
            'available_balance': response['AvailableBalance'],
            'on_hold_balance': response['OnHoldBalance']
        }

    def list_hit_statistics(self):
        """HIT 統計情報"""
        response = self.mturk.list_huts()
        
        stats = {
            'total_hits': len(response.get('HITs', [])),
            'assignable': 0,
            'reviewable': 0,
            'reviewing': 0
        }
        
        for hit in response.get('HITs', []):
            if hit['HITStatus'] == 'Assignable':
                stats['assignable'] += 1
            elif hit['HITStatus'] == 'Reviewable':
                stats['reviewable'] += 1
            elif hit['HITStatus'] == 'Reviewing':
                stats['reviewing'] += 1
        
        return stats

SageMaker Ground Truth との統合

# Ground Truth でラベリングジョブを作成（MTurk ワーカープール使用）

import boto3
import json

sm = boto3.client('sagemaker', region_name='us-east-1')

# MTurk ワーカープール の設定
mturk_config = {
    'PublicWorkforceTaskPrice': {
        'AmountInUsd': 0.05  # HIT 1 個あたり $0.05
    }
}

# Ground Truth ラベリングジョブ作成
response = sm.create_labeling_job(
    LabelingJobName='image-classification-job-2026-01',
    LabelAttributeName='label',
    InputConfig={
        'DataSource': {
            'S3DataSource': {
                'ManifestFile': {
                    'S3Uri': 's3://my-bucket/input-manifest.json'
                }
            }
        }
    },
    OutputConfig={
        'S3OutputPath': 's3://my-bucket/labeling-output/'
    },
    RoleArn='arn:aws:iam::123456789012:role/SageMakerRole',
    LabelCategoryConfigS3Uri='s3://my-bucket/label-categories.json',
    HumanTaskConfig={
        'WorkteamArn': 'arn:aws:sagemaker:us-east-1:123456789012:workteam/public-crowd/Default',
        'UiConfig': {
            'UiTemplateS3Uri': 's3://my-bucket/labeling-template.html'
        },
        'PreHumanTaskLambdaArn': 'arn:aws:lambda:us-east-1:123456789012:function:pre-labeling',
        'TaskCount': 3,  # 3 人のワーカーが各サンプルをラベル
        'TaskDescription': 'Classify images into cat, dog, or other',
        'TaskTitle': 'Image Classification',
        'NumberOfHumanWorkersPerDataObject': 3,
        'PublicWorkforceTaskConfig': mturk_config
    },
    Tags=[
        {'Key': 'Project', 'Value': 'ML-Training'},
        {'Key': 'Team', 'Value': 'Data-Engineering'}
    ]
)

labeling_job_arn = response['LabelingJobArn']
print(f"Labeling job created: {labeling_job_arn}")

# ジョブ進捗監視
import time
while True:
    job = sm.describe_labeling_job(LabelingJobName='image-classification-job-2026-01')
    
    status = job['LabelingJobStatus']
    progress = job.get('LabelingJobStatus')
    
    print(f"Status: {status}")
    
    if status in ['Completed', 'Failed', 'Stopped']:
        break
    
    time.sleep(60)

Worker Qualification・Master Worker

# Worker の資格・評判スコア管理

def assign_qualification(worker_id: str, qualification_type_id: str, score: int):
    """Worker に Qualification を付与"""
    mturk.associate_qualification_with_worker(
        QualificationTypeId=qualification_type_id,
        WorkerId=worker_id,
        IntegerValue=score,
        SendNotification=True
    )

def revoke_qualification(worker_id: str, qualification_type_id: str):
    """Qualification を取り消し"""
    mturk.disassociate_qualification_from_worker(
        QualificationTypeId=qualification_type_id,
        WorkerId=worker_id,
        Reason='Quality standards not met'
    )

def get_worker_stats(worker_id: str):
    """Worker の統計情報取得"""
    # 注意: Worker ID は公開情報ではないため、管理している Worker リストから取得
    # この API は Admin 用途のみ
    
    worker_qualifications = mturk.list_workers_with_qualification_type(
        QualificationTypeId='00000000000000000040'  # Master
    )
    
    # Worker が Master 資格を持っているか確認
    is_master = any(w['WorkerId'] == worker_id for w in worker_qualifications['Qualifications'])
    
    return {
        'worker_id': worker_id,
        'is_master': is_master
    }

# Master Worker フィルター
master_qualification = {
    'QualificationTypeId': '00000000000000000040',  # Master Qualification
    'Comparator': 'EqualTo',
    'RequiredToPreview': False
}

# 年 100+ HIT 完了・承認率 95%+ の ワーカーのみ受け入れ
mturk.create_hit_type(
    Title='Expert-only Task',
    Description='This task requires Master Worker qualification',
    Reward='1.00',
    AssignmentDurationInSeconds=3600,
    LifetimeInSeconds=2592000,
    AutoApprovalDelayInSeconds=86400,
    QualificationRequirements=[
        master_qualification,
        {
            'QualificationTypeId': '00000000000000000025',  # Approved Assignments
            'Comparator': 'GreaterThanOrEqualTo',
            'IntegerValues': [100]
        },
        {
            'QualificationTypeId': '000000000000000000L0',  # Approval Rate
            'Comparator': 'GreaterThanOrEqualTo',
            'IntegerValues': [95]
        }
    ]
)

料金体系（2026年）

比較: MTurk vs 競合

ベストプラクティス

1. HIT 設計

• 明確な指示: 曖昧さ排除、スクリーンショット・例示で視覚化
• 適切な報酬: 相場 $0.05-0.50（タスク複雑度による）
• 制限時間: 5-10 分で完了可能な量

2. 品質保証

# ゴールデンデータセット（既知正解）の埋め込み
golden_questions = [
    {
        'question': '既知正解: この画像に犬が映っていますか?',
        'correct_answer': 'Yes',
        'is_golden': True
    }
]

# Worker の正解率を追跡 → 低い場合は資格剥奪

3. Cost 最適化

• バルク作成: API で複数 HIT を一括作成（CLI より効率的）
• 多数決: 複数 Worker（3-5 人）の回答で品質向上
• Batching: 小さい単価を複数集約して効率化

4. Worker との関係構築

• 定期的な承認: すぐに承認（信頼構築）
• フィードバック: 拒否時は理由を明確に
• 報酬向上: 継続的なワーカーには時給相当額設定

設計チェックリスト

• [ ] HIT テンプレート（HTML フォーム）完成
• [ ] Qualification Type（Master / 言語 / 地域）定義済み
• [ ] ゴールデンデータセット（既知正解）準備済み
• [ ] 複数 Worker 回答で多数決・品質検証設計済み
• [ ] 想定ワーカー数・完了期間を見積もり済み
• [ ] 月間推定コスト計算済み（報酬 + 手数料）
• [ ] SageMaker Ground Truth 統合の要否判断済み
• [ ] 承認/拒否基準・フィードバック文言決定済み
• [ ] Lambda で自動品質チェック実装済み（オプション）
• [ ] CloudWatch で HIT 進捗・Worker 統計監視設定済み

実装パターン: 画像分類タスク（大規模）

要件: 100,000 画像のラベリング・3 人多数決・2 週間で完了・Master Worker のみ

# 実行ワークフロー

import boto3
import pandas as pd
import time

class LargeScaleImageLabelingJob:
    def __init__(self):
        self.mturk = boto3.client('mturk', region_name='us-east-1')
        self.s3 = boto3.client('s3')
        self.cloudwatch = boto3.client('cloudwatch')

    def create_large_labeling_job(
        self,
        image_urls: list,  # 100,000 images
        hit_type_id: str,
        batch_size: int = 100  # 100 HIT / batch
    ):
        """大規模ラベリングジョブを段階的に実行"""
        
        total_images = len(image_urls)
        total_batches = total_images // batch_size
        
        job_status = {
            'total_images': total_images,
            'total_batches': total_batches,
            'created_hits': 0,
            'completed_hits': 0,
            'pending_hits': 0
        }
        
        # バッチごとに HIT 作成
        for batch_num in range(total_batches):
            start_idx = batch_num * batch_size
            end_idx = min(start_idx + batch_size, total_images)
            batch_urls = image_urls[start_idx:end_idx]
            
            print(f"\n=== Batch {batch_num + 1} / {total_batches} ===")
            print(f"Creating {len(batch_urls)} HITs...")
            
            # HIT バッチ作成
            hit_ids = []
            for image_url in batch_urls:
                try:
                    response = self.mturk.create_hit(
                        HITTypeId=hit_type_id,
                        MaxAssignments=3,  # 3 人のワーカーが回答
                        Question=self._generate_hit_html(image_url)
                    )
                    hit_ids.append(response['HIT']['HITId'])
                except Exception as e:
                    print(f"Failed to create HIT: {e}")
            
            job_status['created_hits'] += len(hit_ids)
            
            # 進捗報告
            self._log_batch_progress(batch_num + 1, total_batches, len(hit_ids))
            
            # Rate limiting
            time.sleep(10)
        
        print(f"\n✓ Job created: {job_status['created_hits']} HITs total")
        return job_status

    def monitor_labeling_progress(self, hit_type_id: str, poll_interval: int = 300):
        """ラベリング進捗監視（5 分ごとにチェック）"""
        
        while True:
            # HIT 統計取得
            response = self.mturk.list_huts()
            
            stats = {
                'assignable': 0,
                'reviewable': 0,
                'completed': 0,
                'total': len(response.get('HITs', []))
            }
            
            for hit in response.get('HITs', []):
                if hit['HITStatus'] == 'Assignable':
                    stats['assignable'] += 1
                elif hit['HITStatus'] == 'Reviewable':
                    stats['reviewable'] += 1
                elif hit['HITStatus'] == 'Reviewing':
                    stats['completed'] += 1
            
            # CloudWatch に メトリクス送信
            self.cloudwatch.put_metric_data(
                Namespace='MTurk/LabelingJob',
                MetricData=[
                    {
                        'MetricName': 'AssignableHITs',
                        'Value': stats['assignable'],
                        'Unit': 'Count'
                    },
                    {
                        'MetricName': 'CompletedHITs',
                        'Value': stats['completed'],
                        'Unit': 'Count'
                    }
                ]
            )
            
            completion_rate = (stats['completed'] / stats['total'] * 100) if stats['total'] > 0 else 0
            print(f"Progress: {completion_rate:.1f}% ({stats['completed']}/{stats['total']})")
            
            if completion_rate == 100:
                print("✓ All HITs completed!")
                break
            
            time.sleep(poll_interval)

    def approve_all_quality_answers(self, hit_ids: list):
        """全 HIT の回答を品質チェック後に承認"""
        
        total_approved = 0
        total_rejected = 0
        
        for hit_id in hit_ids:
            # 回答取得
            response = self.mturk.list_assignments_for_hit(
                HITId=hit_id,
                AssignmentStatuses=['Submitted'],
                MaxResults=100
            )
            
            for assignment in response['Assignments']:
                # 品質スコア計算（複数回答から多数決）
                score = self._calculate_quality_score(assignment)
                
                if score > 0.66:  # 66%以上の合致
                    self.mturk.approve_assignment(
                        AssignmentId=assignment['AssignmentId']
                    )
                    total_approved += 1
                else:
                    self.mturk.reject_assignment(
                        AssignmentId=assignment['AssignmentId'],
                        RequesterFeedback='Quality score too low'
                    )
                    total_rejected += 1
        
        print(f"Approved: {total_approved}, Rejected: {total_rejected}")
        return {'approved': total_approved, 'rejected': total_rejected}

    def _generate_hit_html(self, image_url: str) -> str:
        """HIT HTML 生成"""
        return f"""
<HTMLQuestion xmlns="http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2011-11-11/HTMLQuestion.xsd">
  <HTMLContent><![CDATA[
    <html>
    <body>
      <img src="{image_url}" style="max-width:600px">
      <p><strong>Classify the main animal:</strong></p>
      <input type="radio" name="animal" value="cat"> Cat
      <input type="radio" name="animal" value="dog"> Dog
      <input type="radio" name="animal" value="other"> Other
    </body>
    </html>
  ]]></HTMLContent>
  <FrameHeight>400</FrameHeight>
</HTMLQuestion>
"""

    def _calculate_quality_score(self, assignment) -> float:
        """品質スコア計算"""
        # 簡略版：複数回答の合致度
        return 0.85  # 実装省略

    def _log_batch_progress(self, batch_num: int, total: int, hit_count: int):
        """進捗ログ"""
        print(f"Batch {batch_num}/{total}: Created {hit_count} HITs")

最新動向（2024-2026）

• AI アシスト: Worker に AI 予測結果を事前表示→確認タスク化で効率向上
• プライベートワーカープール: 企業専属ワーカー登録・継続的な品質管理
• 品質スコア自動化: ML で Worker の信頼性スコア自動計算
• 多言語 NLP: 言語処理タスクの専門化・言語別ワーカー認定拡大

まとめ

Amazon Mechanical Turk は 「ヒューマンインテリジェンスタスクのクラウドソーシングプラットフォーム」。世界中 50+ 万ワーカーで大規模データラベリング・テキスト分類・コンテンツモデレーションを低コスト実行。

SageMaker Ground Truth と統合し ML 訓練データ作成を完全自動化・統計管理。複数ワーカー多数決・ゴールデンデータセット検証で品質保証、Master Worker・カスタム Qualification で信頼できるワーカーを選別。

インディー企業から大手テック企業まで、AI/ML に必須の高品質ラベルデータを効率的に生成する基盤。