Amazon Web Services ブログ

教師あり学習に Amazon SageMaker の Amazon Mechanical Turk を使用する

教師付き学習には、プロジェクトのトレーニング段階で、何が正しい答えかをアルゴリズムに教えるラベルまたはアノテーションが必要です。事実、MXNet 、TensorFlow 、および PyTorch を使った例の多くは、これらのフレームワークが持ついろんな機能を調べるのに利用するアノテーション付きデータセットから始めています。しかし残念ながら、例からアプリケーションに移行する際に、完全なアノテーションが付随するデータセットを簡単に入手できることはまずありません。このチュートリアルでは、Amazon SageMaker ノートブックの中から Amazon Mechanical Turk (MTurk) を使って、データセットのアノテーションを取得し、トレーニングに使用する方法を解説します。

TensorFlow では、ニューラルネットワーク識別子を使ってアイリス ( アヤメ ) を識別するのに、エスティメーターを利用するという例があります。この例も、SageMaker サンプルノートブックライブラリ、または GitHub の SageMaker Examples プロジェクトの中にあります。チュートリアルとサンプルノートの両方でアイリスデータセットを使用しています。これには、3 つの関連するアイリス種および関連する種の測定が含まれます。これらのデータポイントは、4 つの測定値に基づいてアイリスの種を予測するモデルをトレーニングするのに使用します。

左から右へ、Iris setosa ( 画像は Radomil 氏、CC BY-SA 3.0) 、Iris versicolor ( 画像は Dlanglois 氏、CC BY-SA 3.0) 、および Iris virginica ( 画像は Frank Mayfield 氏、CC BY-SA 2.0) 。

これは TensorFlow エスティメーターを使って開始するにはよい例ですが、各アイリスの画像と関連する測定値しかない場合、はるかに難しい例となります。アノテーション付きデータがなければ、手動で画像にアノテーションを付けることはできますが、モデルの開発に費やせるはずの時間を無駄にすることになるでしょう。

MTurk は API を介してヒューマンインテリジェンスを備えているため、トレーニングで使用できる多様なアノテーションを提供するのに理想的です。MTurk を使用すると、データセット用のアノテーションを提供できる労働力に、世界中にそして365日休みなくアクセスできるのです。ワーカーを利用して、各データポイントで完了し、市場に流通し、数分または数時間で結果を得たい、というタスクを定義するだけです。MTurk を使えば、チームを雇うことも自分の時間を費やすこともなく、必要なアノテーションを迅速に取得できるのです。

SageMaker ノートブックの中から、アノテーションのタスクを MTurk ワーカーに迅速に送信し、結果を確認し、トレーニングへと進みます。このチュートリアルでは、種に関する情報の代わりに、画像を含むアイリスデータセットのバージョンを作成し、MTurk ワーカーにそれらの画像に基づいて種を識別してもらいます。最後に、Amazon SageMaker Python SDK で、TensorFlow の tf.estimator を使ったニューラルネットワーク識別子を構築およびトレーニングを行います。このチュートリアルのノートブックは、こちらよりダウンロードできます。Amazon SageMaker ノートブックをスピンアップし、MTurk でアノテーションを収集し、モデルをトレーニングおよびデプロイすると、料金が発生することに留意ください。トレーニング完了時には、必ずリソースを削除してください。そうすれば、引き続き請求されることはありません。

データにアノテーションを付けてモデルをトレーニングするには、8　つの手順があります。

  1. Amazon Mechanical Turk Requester アカウントを設定し、それを AWS アカウントにリンクします。
  2. アイリスデータセットをロードし、画像を含むように変更します。
  3. データセットにアノテーションを付ける MTurk タスクを定義します。
  4. アノテーション用のタスクを MTurk ワーカーに送信します。
  5. MTurk ワーカーが提供した結果を検索し、調整します。
  6. トレーニングデータを使用して、モデルを構築し、トレーニングします。
  7. モデルをエンドポイントにデプロイし、新しいサンプルを分類するのにそれを使用します。
  8. リソースをクリーンアップします。

このプロジェクトでは、上記の画像を、データセット内および MTurk のタスク自体の両方で使用しています。これらの画像は、Creative Commons Attribution-ShareAlike 2.0 および 3.0 のライセンスの下で使用しています。画像は Radomil 氏 (Iris setosa 、CC BY-SA 3.0) 、 Dlanglois 氏 (Iris versicolor 、CC BY-SA 3.0) 、および Frank Mayfield 氏(Iris virginica 、CC BY-SA 2.0) より。

ステップ 1.Amazon Mechanical Turk アカウントの設定

まだ Amazon Mechanical Turk アカウントをお持ちでない場合は、Amazon SageMaker で使用している AWS アカウントにリンクする Amazon Mechanical Turk Requester アカウントを設定する必要があります。まずは、https://requester.mturk.com で新しいアカウントを作成しましょう。

MTurk アカウントを設定したら、Amazon SageMaker で使っている AWS アカウントにそのアカウントをリンクする必要があります。アカウントのルートのメールアドレスを使用して、ルートユーザーとして AWS アカウントにログインします。IAM ユーザーサインインページが表示された場合は、[Sign-in using root account credentials] を選択します。次に https://requester.mturk.com/developer に行き、アカウントを一緒にリンクします。

Amazon SageMaker ロールのアクセス許可を更新し、AmazonMechanicalTurkFullAccess ポリシーを含める必要があります。これを行うには、AWS Management Console で IAM コンソールを開き、ロールを表示します。SageMaker インスタンスに関連付けられている IAM ロールを選択し、[Attach policy] を選択します。AmazonMechanicalTurkFullAccess を検索し、ロールにポリシーを追加します。

MTurk for Workers にタスクを配置するには、タスクを完了したワーカーに報酬を与えるのに使用するプリペイドの HIT を最初に購入しておく必要があります。HIT (Human Intelligence Task) は、個々のタスクが MTurk で表示されます。https://requester.mturk.com/account で、購入できます。

注意 : 次のコードで定義されているように、このチュートリアルではワーカーの報酬と手数料で、18.00 USD から 19.00 USD の費用がかかります。

Xmltodict ライブラリを使って MTurk から返された結果を処理するため、SageMaker ノートブックインスタンスにこれをインストールします。

!pip install xmltodict

xmltodict と他に必要なライブラリをいくつかインポートします。アノテーション付きデータを保存するのに使用する Amazon S3 バケットを指定します。これはトレーニングに使用されます。

import xmltodict
import boto3
import json
import pandas as pd
import sagemaker
import io

s3 = boto3.resource('s3')
training_bucket_name = '<bucket-name>'
training_bucket = s3.Bucket(training_bucket_name)

MTurk では、プロダクションとサンドボックスの 2 つの環境を利用できます。プロダクション環境を使用すると、タスクは https://worker.mturk.com にあるワーカーに表示されます。テストに使用できるサンドボックス環境もあります。ワーカーはサンドボックス内では見えませんが、https://workersandbox.mturk.com では、自分で実行し、タスクインターフェイスをテストできます。サンドボックス環境を使用するためのコストはかかりません。まず、プロダクション環境に移行する前に、タスクが必要なデータに返されることを確認するテストを行うことをお勧めします。サンドボックスでテストする場合は、サンドボックスに追加のアカウントを作成し、AWS アカウントにリンクする必要があることに注意してください。

次の手順では、create_hits_in_production の値に応じて、2 つの環境のいずれかでクライアントを作成します。サンドボックスでテストする場合は、この値を False に変更します。

create_hits_in_production = True
environments = {
        "production": {
            "endpoint": "https://mturk-requester.us-east-1.amazonaws.com",
            "preview": "https://www.mturk.com/mturk/preview"
        },
        "sandbox": {
            "endpoint": "https://mturk-requester-sandbox.us-east-1.amazonaws.com",
            "preview": "https://workersandbox.mturk.com/mturk/preview"
        },
}
mturk_environment = environments["production"] if create_hits_in_production else environments["sandbox"]

mturk = boto3.client(
    service_name='mturk',
    region_name='us-east-1',
    endpoint_url=mturk_environment['endpoint'],
)

アカウントが正しく設定されていることを確認するには、アカウントの残高を呼び出し、取得してください。サンドボックスに接続している場合、残高は常に 10,000 USD です。このタスクは 18.00 USD から 19.00 USD の間で完了する必要がありますので、プロダクションアカウントに 19.00 USD を追加してください。ワーカーが完了するためのタスクを送信する前に、アカウントに資金あることが必要です。

print(mturk.get_account_balance()['AvailableBalance'])

ステップ 2.トレーニングデータセットをインポートする

まず、アイリスデータセットを UCI Machine Learning Repository から pandas DataFrame へインポートします。ありがたいことに、このデータセットを利用するためのサポートがあります。: Dua, D. and Karra Taniskidou, E. (2017).UCI Machine Learning Repository 「 http://archive.ics.uci.edu/ml 」。Irvine, CA: カリフォルニア大学、情報とコンピューターサイエンス学部。

このチュートリアルでは、このデータセットにすでに含まれている種の情報をそれぞれの花の画像に置き換えます。この例では、データポイント全てに対して同じ 3 つの画像を使用することに注意してください。現実的には、各画像のは異なっても、ラベルの結果は同じでしょう。

# Load the iris data set
training_df = pd.read_csv('https://archive.ics.uci.edu/ml/machine-learning-databases/iris/iris.data', header=None)

# Name the columns
training_df.columns = ['sepal_length', 'sepal_width', 'petal_length', 'petal_width', 'species']

# Add an image_url column and remove the species column
def species_to_url(species):
    if (species == 'Iris-setosa'): return 'https://upload.wikimedia.org/wikipedia/commons/5/56/Kosaciec_szczecinkowaty_Iris_setosa.jpg'
    elif (species == 'Iris-versicolor'): return 'https://upload.wikimedia.org/wikipedia/commons/4/41/Iris_versicolor_3.jpg'
    else: return 'https://upload.wikimedia.org/wikipedia/commons/3/38/Iris_virginica_-_NRCS.jpg'
image_urls = [ species_to_url(row.species) for index, row in training_df.iterrows() ]
training_df['image_url'] = image_urls
del training_df['species']

training_df

ステップ 3.MTurk タスクを定義する

MTurk では、ワーカーに表示される HTML を含んだ XML ドキュメントを使用できます。ワーカーは、公開している各アイテムの HTML を見ることができます。このタスクの HTML ファイルは、こちらよりダウンロードできます。

html_layout = open('./IrisAnnotation.html', 'r').read()
QUESTION_XML = """<HTMLQuestion xmlns="http://mechanicalturk.amazonaws.com/AWSMechanicalTurkDataSchemas/2011-11-11/HTMLQuestion.xsd">
        <HTMLContent><![CDATA[{}]]></HTMLContent>
        <FrameHeight>1000</FrameHeight>
        </HTMLQuestion>"""
question_xml = QUESTION_XML.format(html_layout)

この手順でロードする HTML には、タスクが作成される時に各データポイントの画像に置き換えられる変数 $ {content} が定義されています。

MTurk では、各タスクは Human Intelligence Task (HIT) によって表されます。HIT とは 1 人以上のワーカーがアノテーションを付ける個々のアイテムのことです。2 人のワーカーで各アイテムを完了したいので、最大アサインメントを 2 つに設定します。アサインメントは、それぞれのワーカーの応答を表します。以下の定義では、HIT が worker.mturk.com のウェブサイトに4時間だけいることと、ワーカーが 5 分未満で各アイテムに応答をするよう要求しています。各応答には 0.05 USD の料金がかかりますので、各 HIT のワーカーの料金合計に、MTurk 費用で 0.10 USD + 0.02 USD が加算されます。このタスクに関するものをワーカーに知らせるために、適切なタイトル、説明、およびキーワードも提供されます。

task_attributes = {
    'MaxAssignments': 2,                 
    'LifetimeInSeconds': 60*60*4,         # How long the task will be available on the MTurk website (4 hours)
    'AssignmentDurationInSeconds': 60*5,  # How long will Workers have to complete each item (5 minutes)
    'Reward': '0.05',                     # The reward you will offer Workers for each response
    'Title': 'Classify images of flowers',
    'Description': 'Provide the species of iris in each image',
    'Keywords': 'classification, image'
}

トレーニングに使用する前に、アノテーションに同意する少なくとも 2 人のワーカーを確保したいので、各アイテムを 2 人のワーカーに割り当てます。2 人のワーカーがデータポイントに同意しない場合は、少なくとも 2 人のワーカーが同意するまで、アサインメントを追加します。これは後に、Get Results のステップで定義されます。

ステップ 4.MTurk にタスクを公開する

タスクとデータセットを定義したので、これを MTurk ワーカーに送り、アノテーションを付ける準備が整いました。データセットの各行について、 image_url を質問に入れ替え、MTurk 上で HIT を作成します。HIT が作成されると、 HITId をキャプチャします。 これはこのアイテムに割り当てられているので、後で結果を取得できます。

hit_type_id = ''
results = []

for index, row in training_df.iterrows():
    response = mturk.create_hit(
        **task_attributes,
        Question=question_xml.replace('${image_url}',row['image_url'])
    )
    hit_type_id = response['HIT']['HITTypeId']
    results.append({
        'hit_id': response['HIT']['HITId']
    })

print("You can view the HITs here:")
print(mturk_environment['preview'] + "?groupId={}".format(hit_type_id))

ステップ 5.アノテーション結果を取得する

アノテーションタスクの結果は数分で得られることがほとんどですが、大規模なタスクには数時間以上かかることがあります。各 HIT のステータスを読み取り、各アサインメントの結果を取得することで、いつでもプロジェクトの進捗状況を確認することができます。

各アイテムについては、HIT のステータスを取得することから始めます。最も一般的な設定は、以下の通りです。

  • 割り当て可能 : ワーカーがタスクを受注できる時
  • 割り当て不可能 : ワーカーがタスクを処理しており、追加のワーカーによってタスクがもはや受注できない時
  • レビュー可能 : 全ての応答を受け取った時

次に、アサインメントを取得して、このアイテムのワーカーの応答を全て取得します。それぞれのアサインメントからワーカーの応答を抽出し、一連の回答に保存します。全てのアサインメントを直ちに承認し、ワーカーがすぐに報酬を受け取るようにします。

アイテムの回答を全て取得した後、少なくとも 2 人のワーカーが同意しているかどうかを確認することができます。彼らが同意すれば、それをアノテーションとして使用します。そうでない場合は、ワーカー入力を加えるためのアサインメントを追加します。

species_count = 0
for item in results:
    
    # Get the status of the HIT
    hit = mturk.get_hit(HITId=item['hit_id'])
    item['status'] = hit['HIT']['HITStatus']

    # Get a list of the Assignments that have been submitted by Workers
    assignmentsList = mturk.list_assignments_for_hit(
        HITId=item['hit_id'],
        AssignmentStatuses=['Submitted', 'Approved'],
        MaxResults=10
    )

    # Get the assignments that have been submitted and capture a count in the results
    assignments = assignmentsList['Assignments']
    item['assignments_submitted_count'] = len(assignments)

    answers = []
    for assignment in assignments:
    
        # Retrieve the attributes for each Assignment
        worker_id = assignment['WorkerId']
        assignment_id = assignment['AssignmentId']
        
        # Retrieve the value submitted by the Worker from the XML
        answer_dict = xmltodict.parse(assignment['Answer'])
        answer = answer_dict['QuestionFormAnswers']['Answer']['FreeText']
        answers.append(int(answer))
        
        # Approve the Assignment (if it hasn't already been approved)
        if assignment['AssignmentStatus'] == 'Submitted':
            mturk.approve_assignment(
                AssignmentId=assignment_id,
                OverrideRejection=False
            )
    
    # Add the answers that have been retrieved for this item to the results
    item['answers'] = answers
    
    # If we've received at least 2 answers for the same category, use that answer
    if len(answers) > 1:
        for species in [0,1,2]:
            if answers.count(species) >= 2:
                item['species'] = species
                species_count += 1
                
    # If none of the Workers agree after all answers have been provided, 
    # add an additional Assignment to break the tie
    if len(answers) == hit['HIT']['MaxAssignments'] and 'species' not in item:
        extend_result = mturk.create_additional_assignments_for_hit(HITId=item['hit_id'], NumberOfAdditionalAssignments=1)
        print("Extended HIT {} to {} assignments".format(item['hit_id'], hit['HIT']['MaxAssignments'] + 1))

print("Irises annotated: {}".format(species_count))

アノテーションを全て取得したので、結果を DataFrame に統合することができます。さらに、DataFrame をモデルのトレーニングに使用するトレーニングデータセットとテストデータセットに分割します。

# merge the results back into the dataframe
results_df = pd.merge(training_df, pd.DataFrame(results), left_index=True, right_index=True)[['sepal_length','sepal_width','petal_length','petal_width','species']]

# ensure the species column is defined as an int
results_df['species'] = results_df['species'].astype(int)

# split the frame into training and test
test_df = results_df.sample(n=30)
test_df.columns  = [30,4,'setosa','versicolor','virginica']
train_df = results_df.drop(test_df.index)
train_df.columns = [120,4,'setosa','versicolor','virginica']

train_df

これで Amazon S3 を CSV ファイルとして保存し、トレーニングを開始することができます。

csv_buffer = io.StringIO()
train_df.to_csv(csv_buffer, index=False)
s3.Object(training_bucket_name, 'trainingdata/iris_training.csv').put(Body=csv_buffer.getvalue())

csv_buffer = io.StringIO()
test_df.to_csv(csv_buffer, index=False)
s3.Object(training_bucket_name, 'trainingdata/iris_test.csv').put(Body=csv_buffer.getvalue())

ステップ 6.モデルのトレーニング

このデータを使用して、エスティメーターをトレーニングすることができます。ここでは、 Iris_dnn_classify.py を使います。これは、Amazon SageMaker の例で使用されるトレーニングスクリプトです。これは、こちらでダウンロードすることができます。このモデルの設定方法の詳細については、このサンプルに関連付けられているノートブックをご参照ください。

変数をいくつか初期化することから始めましょう。

from sagemaker import get_execution_role

#Bucket location to save your custom code in tar.gz format.
custom_code_upload_location = 's3://{}/customcode/tensorflow_iris'.format(training_bucket_name)

#Bucket location where results of model training are saved.
model_artifacts_location = 's3://{}/artifacts'.format(training_bucket_name)

#IAM execution role that gives SageMaker access to resources in your AWS account.
role = get_execution_role()

Amazon SageMaker Python SDK を使用して、事前に構築してある SageMaker TensorFlow コンテナ内のローカルトレーニングスクリプトを実行することができます。 これには、iris_dnn_classifier.py ファイルのパスを sagemaker.TensorFlow init メソッドに返します。これには、エスティメーターを定義する関数が含まれています。

モデルをトレーニングするために、アノテーション付きデータの場所を、 fit() メソッドに返します。

from sagemaker.tensorflow import TensorFlow

iris_estimator = TensorFlow(entry_point='iris_dnn_classifier.py',
                            role=role,
                            output_path=model_artifacts_location,
                            code_location=custom_code_upload_location,
                            train_instance_count=1,
                            train_instance_type='ml.c4.xlarge',
                            training_steps=1000,
                            evaluation_steps=100)

train_data_location = 's3://{}/trainingdata'.format(training_bucket_name)
iris_estimator.fit(train_data_location)

ステップ 7.トレーニングしたモデルをデプロイする

この deploy() メソッドは、predict() を使用して予測リクエストを呼び出すことができるように、予測リクエストをリアルタイムで処理するエンドポイントを作成します。

iris_predictor = iris_estimator.deploy(initial_instance_count=1,
                                       instance_type='ml.m4.xlarge')

iris_predictor.predict([6.4, 3.2, 4.5, 1.5])

ステップ 8. リソースをクリーンアップする

このチュートリアルを終了後、AWS Management Console を使用して、今回のエクササイズ用に作成したリソースを削除することで、不要な料金が発生するのを避けることができます。

  1. Amazon SageMaker コンソールを https://console.aws.amazon.com/sagemaker/ で開き、エンドポイント、エンドポイント構成、およびモデルを削除します。次に、ノートブックインスタンスを停止し、削除します。
  2. Amazon S3 コンソールを https://console.aws.amazon.com/s3/ で開き、モデル生成物とトレーニングデータセットを保存するために作成したバケットを削除します。
  3. IAM コンソールを https://console.aws.amazon.com/iam/ で開き、IAM ロールを削除します。アクセス許可ポリシーを作成した場合は、削除することもできます。
  4. Amazon CloudWatch コンソールを https://console.aws.amazon.com/cloudwatch/ で開き、 / aws / sagemaker / で始まる名前のロググループを全て削除します。

まとめ

ご覧のように、MTurk はプロジェクトに必要なアノテーション付きデータを作成することができます。アノテーションなしのデータセットから開始し、それにアノテーションを付ける MTurk タスクを作成し、ワーカーが提供した結果を取得し、モデルのトレーニングを始めました。これらは全て、SageMaker ノートブック内で行いました。次のステップとして、MTurk のワーカーを利用して、モデルが提供する結果を検証することができます。これは、モデルが一部のデータに対して信頼性の低い予測を返す場合には特に有効です。これらのアイテムを確認するのに MTurk を使用すると、モデルを再トレーニングできる新しいアノテーションを取得できます。アノテーションを提供するのに MTurk を繰り返し使うことで、新しいデータが利用できるようなモデルへと着実に改良することができます。様々なコンテキストにおいて MTurk を使用する例については、MTurk Blog にあるチュートリアルをご参照ください。

Amazon Mechanical Turk は、モデルをトレーニングするのに、データを収集、アノテーション、検証する必要があるデータサイエンティストにとって強力なツールとなります。Amazon SageMaker から MTurk に簡単にアクセスすることで、プロジェクトに必要なデータを迅速かつ経済的に入手するのにサポートとなる人材を、オンデマンドで確保できるのです。

今回のブログ投稿者について

Dave Schultz は Amazon Mechanical Turk のビジネスディベロップメントを指揮しています。機械学習やデータ管理における複雑な問題にヒューマンインテリジェンスを適用して、顧客をサポートしています仕事以外では、木工と一風変わったプログラミングを楽しんでいます。