collect.py

import argparse
from os import path

import boto3
from boto.mturk.question import ExternalQuestion
from config import MTURK_KEY, MTURK_SECRET

parser = argparse.ArgumentParser()
parser.add_argument("--mturk_region", default="us-east-1", help="The region for mturk (default: us-east-1)")
parser.add_argument(
    "--space_name",
    default="Tristan/static-rlhf-interface",
    help="Name of the accompanying Hugging Face space (default: huggingface/rlhf-interface)",
)
parser.add_argument("--num_hits", type=int, default=5, help="The number of HITs.")
parser.add_argument(
    "--num_assignments", type=int, default=1, help="The number of times that the HIT can be accepted and completed."
)
parser.add_argument(
    "--live_mode",
    action="store_true",
    help="""
    Whether to run in live mode with real turkers. This will charge your account money.
    If you don't use this flag, the HITs will be deployed on the sandbox version of mturk,
    which will not charge your account money.
    """,
)
parser.add_argument(
    "--refresh_qualification_test",
    action="store_true",
    help="""
    Whether to refresh the qualification test. If you've made edits to the test
    xml files, it is necessary to do this.
    """,
)
parser.add_argument(
    "--custom_qualification_test",
    action="store_true",
    help="""
    Whether to require the custom qualification test.
    """,
)
parser.add_argument(
    "--master_turkers",
    action="store_true",
    help="""
    Whether to only use turkers with the master qualification.
    """,
)
parser.add_argument(
    "--us_turkers",
    action="store_true",
    help="""
    Whether to only use US-based turkers.
    """,
)

args = parser.parse_args()

MTURK_URL = f"https://mturk-requester{'' if args.live_mode else '-sandbox'}.{args.mturk_region}.amazonaws.com"

mturk = boto3.client(
    "mturk",
    aws_access_key_id=MTURK_KEY,
    aws_secret_access_key=MTURK_SECRET,
    region_name=args.mturk_region,
    endpoint_url=MTURK_URL,
)

# This is the URL that makes the space embeddable in an mturk iframe
question = ExternalQuestion(f"https://hf.space/embed/{args.space_name}/+?__theme=light", frame_height=600)

qualification_requirements=[]

if args.master_turkers:
    qualification_requirements.append({
        QualificationTypeId: '2F1QJWKUDD8XADTFD2Q0G6UTO95ALH',
        Comparator: 'Exists'
      })

if args.us_turkers:
    qualification_requirements.append({
        QualificationTypeId: '00000000000000000071',
        Comparator: 'In',
        LocaleValues: [
          { Country: "US" },
        ]
      })

if args.custom_qualification_test:
    qualification_type_id = (
        open("qualification_type_id.txt", "r").read() if path.exists("qualification_type_id.txt") else None
    )
    if args.refresh_qualification_test or qualification_type_id is None:
        if qualification_type_id is not None:
            mturk.delete_qualification_type(QualificationTypeId=qualification_type_id)
        response = mturk.create_qualification_type(
            Name="rlhf--qualification",
            Keywords="RLHF qualification",
            Description="Qualification test for RLHF task.",
            QualificationTypeStatus="Active",
            Test=open("qualification_questions.xml", mode="r").read(),
            AnswerKey=open("qualification_answers.xml", mode="r").read(),
            TestDurationInSeconds=3600,
            AutoGranted=False,
        )
        qualification_type_id = response["QualificationType"]["QualificationTypeId"]
        open("qualification_type_id.txt", "w+").write(qualification_type_id)
    qualification_requirements.append({
            "QualificationTypeId": qualification_type_id,
            "Comparator": "Exists",
            "RequiredToPreview": False,
            "ActionsGuarded": "Accept",
        })

for i in range(args.num_hits):
    new_hit = mturk.create_hit(
        Title="RLHF HIT",
        Description="Interact with an AI",
        Keywords="chatbot",
        Reward="0.25",
        MaxAssignments=args.num_assignments,
        LifetimeInSeconds=172800,
        AssignmentDurationInSeconds=600,
        AutoApprovalDelayInSeconds=14400,
        Question=question.get_as_xml(),
        QualificationRequirements=qualification_requirements,
    )

print(
    f"HIT Group Link: https://worker{'' if args.live_mode else 'sandbox'}.mturk.com/mturk/preview?groupId="
    + new_hit["HIT"]["HITGroupId"]
)