Skip to main content

Tasks : Create, Get, Update, Delete

Group your labeling work into tasks that can be delegated

Task is the work that needs to be done for labeling the inputs in an app. It's a powerful way to help your team annotate inputs fast and efficiently.

These are some parameters you can specify when working with tasks:

  • Task type — It can be:
    • CONCEPTS_CLASSIFICATION — Concepts classification tasks annotate concepts for the overall image, frame of video, or section of text.
    • BOUNDING_BOX_DETECTION — Bounding box detection tasks annotate rectangular bounding box regions around each concept in an image, frame of video, or section of text.
    • POLYGON_DETECTION — Polygon detection tasks annotate free-form regions around concepts in an image, frame of video, or section of text.
    • TYPE_NOT_SET — This is the default task type. It should be used when creating an auto-annotation task.
  • Worker — Task worker includes information about the workers who will work on the task. For manual labeling tasks, the workers can only be users; no limitation on number of workers. For auto-annotation tasks, the worker can be either a model or a workflow; currently only supports 1 worker.
  • Concepts — List of concept IDs used in the work on the task. The concepts should already be existing in your app.
  • Task worker strategy — It can be:
    • DYNAMIC — Each worker will dynamically get 10 inputs assigned at a time. No inputs are assigned at task creation. It's the recommended way to set a task worker strategy.
    • PARTITIONED — The inputs will be partitioned in several partitions. Each worker will label one or more input partitions. All inputs are assigned at task creation.
    • FULL — Each worker will label all inputs from the input source. All inputs are assigned at task creation.
  • Input source — It can be:
    • ALL_INPUTS — Use all inputs in the app.
    • DATASET — Use inputs from a dataset.
  • sample_ms — Used in video model predictions. It specifies the sample delay for video predictions (1 frame per N milliseconds).
  • Review strategy — It can be:
    • NONE — No review is needed.
    • MANUAL — Manual review strategy.
    • CONSENSUS — Consensus review strategy.
  • Partitioned strategy info — It can be:
    • EVENLY — Each worker will label (approximately) the same number of inputs.
    • WEIGHTED — Each worker will have an assigned weight.
  • Workers per input — The number of workers who will label each input.
  • Auto-annotation config - The concepts configurations for setting up an auto-annotation labeling task using a model or a workflow. You can set:
    • annotation_data_types — An integer for filtering annotations by their annotation data type. It's a bit-mask field that holds multiple annotation data type values that are combined in an OR fashion. For example, if annotation_data_types = 34, then we filter annotations that appear as a mask or a bounding box, because MASK = 32 and BOUNDING_BOX = 2. You can look for the various annotation data types values here. For example, annotation_data_types=1 corresponds to AnnotationDataType_TAG.
    • threshold_range — It specifies a range of predictions values based on the lower and upper bounds, and it defines whether these bounds are inclusive or exclusive. For example, if you set is_lower_inclusive = true, is_upper_inclusive = true, lower = 0.7 and upper = 1.0, it is interpreted as the prediction range includes all values from 0.7 to 1.0, including both 0.7 and 1.0.
    • status_code — It specifies the code related to the status of the annotation.
info

The initialization code used in the following examples is outlined in detail on the client installation page.

Create​

To create a new task in your app, you POST the task information to the v2/task endpoint.

Assigned Task​

A task should be assigned to a list of users. These users will do the labeling work, so they're also called workers. A task may also be assigned to a list of users for review purposes.

#################################################################################
# In this section, we set the user authentication, app ID, and details for
# assigning a task. Change these strings to run your own example.
#################################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'
# Change these to assign your own task
CONCEPT_ID = 'water'
WORKER_USER_ID = 'WORKER_USER_ID_HERE'
REVIEWER_USER_ID = 'REVIEWER_USER_ID_HERE' # User who will review this task
DATASET_ID = 'DATASET_ID_HERE'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID)

post_tasks_response = stub.PostTasks(
    service_pb2.PostTasksRequest(
        user_app_id=userDataObject,  # The userDataObject is created in the overview and is required when using a PAT
        tasks=[
            resources_pb2.Task(
                type='CONCEPTS_CLASSIFICATION',
                name='Annotate ' + CONCEPT_ID,
                worker=resources_pb2.TaskWorker(
                    strategy='DYNAMIC',
                    workers=[
                        resources_pb2.Worker(
                            user=resources_pb2.User(
                                id=WORKER_USER_ID
                            )
                        )
                    ]
                ),
                concepts=[
                    resources_pb2.TaskConcept(
                        concept=resources_pb2.Concept(
                            id=CONCEPT_ID
                        )
                    )
                ],
                input_source=resources_pb2.TaskInputSource(
                    type='DATASET',
                    id=DATASET_ID
                ),
                sample_ms=1000,
                review=resources_pb2.TaskReview(
                    strategy='MANUAL',
                    manual_strategy_info=resources_pb2.TaskReviewManualStrategyInfo(
                        sample_percentage=0.5
                    ),
                    users=[
                        resources_pb2.User(
                            id=REVIEWER_USER_ID
                        )
                    ]
                )
            )
        ]
    ),
    metadata=metadata
)

if post_tasks_response.status.code != status_code_pb2.SUCCESS:
    print(post_tasks_response.status)
    raise Exception('Post tasks failed, status: ' + post_tasks_response.status.description)

Task With Consensus Review​

You can also create tasks with CONSENSUS review strategy. When enough workers label an input in the same way, it will automatically be approved, with no need for the reviewer to spend time to check. In this way, the reviewer will be able to focus on the inputs where the workers don't agree.

Note that an approval threshold must be set. It is the number of labelers that need to agree in order to automatically approve an annotation.

For example, in case of 3 workers and approval_threshold set to 2, if an input is labeled in the same way by 2 workers, they form a majority and the group reaches a consensus.

#################################################################################
# In this section, we set the user authentication, app ID, and details for
# creating a task. Change these strings to run your own example.
#################################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'
# Change these to create your own task with consensus review
CONCEPT_ID = 'water'
USER_ID_1 = 'USER_ID_1_HERE'
USER_ID_2 = 'USER_ID_2_HERE'
USER_ID_3 = 'USER_ID_3_HERE'
USER_ID_4 = 'USER_ID_4_HERE'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2
from google.protobuf.struct_pb2 import Struct

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

params = Struct()
params.update({
    USER_ID_1: 1,
    USER_ID_2: 1,
    USER_ID_3: 1
})

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID)

post_tasks_response = stub.PostTasks(
    service_pb2.PostTasksRequest(
        user_app_id=userDataObject,  # The userDataObject is created in the overview and is required when using a PAT
        tasks=[
            resources_pb2.Task(
                type='CONCEPTS_CLASSIFICATION',
                name='Annotate ' + CONCEPT_ID,
                worker=resources_pb2.TaskWorker(
                    strategy='PARTITIONED',
                    workers=[
                        resources_pb2.Worker(user=resources_pb2.User(id=USER_ID_1)),
                        resources_pb2.Worker(user=resources_pb2.User(id=USER_ID_2)),
                        resources_pb2.Worker(user=resources_pb2.User(id=USER_ID_3))
                    ],
                    partitioned_strategy_info=resources_pb2.TaskWorkerPartitionedStrategyInfo(
                        type='WEIGHTED',
                        workers_per_input=3,
                        weights=params
                    )
                ),
                concepts=[
                    resources_pb2.TaskConcept(
                        concept=resources_pb2.Concept(id=CONCEPT_ID)
                    )
                ],
                input_source=resources_pb2.TaskInputSource(type='ALL_INPUTS'),
                sample_ms=1000,
                review=resources_pb2.TaskReview(
                    strategy='CONSENSUS',
                    consensus_strategy_info=resources_pb2.TaskReviewConsensusStrategyInfo(
                        approval_threshold=2
                    ),
                    users=[resources_pb2.User(id=USER_ID_4)]
                )
            )
        ]
    ),
    metadata=metadata
)

if post_tasks_response.status.code != status_code_pb2.SUCCESS:
    print(post_tasks_response.status)
    raise Exception('Post tasks failed, status: ' + post_tasks_response.status.description)

Auto-Annotation Task​

You can create an auto-annotation task and automatically label the inputs in your dataset. You need to specify a model or a workflow you want to use its predictions to automatically generate annotations or labels for your data.

tip

You can learn how to perform auto-annotation via the UI here.

##########################################################################################
# In this section, we set the user authentication, app ID, and auto-annotation details.
# Change these strings to run your own example.
##########################################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'
# Change these to create your own auto-annotation task 
MODEL_ID = 'MODEL_ID_HERE'
MODEL_VERSION_ID = 'MODEL_VERSION_ID_HERE'
CONCEPT_ID = 'CONCEPT_ID_HERE'
DATASET_ID = 'DATASET_ID_HERE'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID)

post_tasks_response = stub.PostTasks(
    service_pb2.PostTasksRequest(
        user_app_id=userDataObject,  # The userDataObject is created in the overview and is required when using a PAT
        tasks=[
            resources_pb2.Task(
                type='TYPE_NOT_SET',
                name='Auto-Annotate ' + CONCEPT_ID,
                worker=resources_pb2.TaskWorker(
                    strategy='FULL',
                    workers=[
                        resources_pb2.Worker(
                            model=resources_pb2.Model(
                                id=MODEL_ID,
                                model_version=resources_pb2.ModelVersion(
                                    id=MODEL_VERSION_ID
                                )
                            )
                        )
                    ]
                ),
                concepts=[
                    resources_pb2.TaskConcept(
                        concept=resources_pb2.Concept(
                            id=CONCEPT_ID
                        ),
                        auto_annotation_config=resources_pb2.TaskConceptAutoAnnotationConfig(
                            annotation_data_types=1,
                            threshold_range=resources_pb2.ThresholdRange(
                                is_lower_inclusive=True,
                                is_upper_inclusive=True,
                                lower=0.7,
                                upper=0.999
                            ),
                            status_code=24150
                        )
                    )
                ],
                input_source=resources_pb2.TaskInputSource(
                    type='DATASET',
                    id=DATASET_ID
                ),
                sample_ms=1000,
                review=resources_pb2.TaskReview(
                    strategy='NONE'
                )
            )
        ]
    ),
    metadata=metadata
)

if post_tasks_response.status.code != status_code_pb2.SUCCESS:
    print(post_tasks_response.status)
    raise Exception("Post tasks failed, status: " + post_tasks_response.status.description)

Get​

Get Task by ID​

You can get the details of a single task by its ID, which is automatically generated upon task creation. You can output the API response to obtain this task ID.

Alternatively, you can also locate the task ID within the Clarifai platform's user interface.

#################################################################################
# In this section, we set the user authentication, app ID, and task ID.
# Change these strings to run your own example.
#################################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'
# Change this to get a task by its ID
TASK_ID = 'c454edb9446c4de58d4fe3a66c286e55'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID)

get_tasks_response = stub.GetTask(
    service_pb2.GetTaskRequest(
        user_app_id=userDataObject,  # The userDataObject is created in the overview and is required when using a PAT
        task_id=TASK_ID
    ),
    metadata=metadata
)

if get_tasks_response.status.code != status_code_pb2.SUCCESS:
    print(get_tasks_response.status)
    raise Exception(f'Get task failed, status: {get_tasks_response.status.description}')

print(get_tasks_response)

List All Tasks​

You can get a list of tasks within your app with a GET call. This call supports pagination.

#######################################################################
# In this section, we set the user authentication and app ID.
# Change these strings to run your own example.
#######################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID)

list_tasks_response = stub.ListTasks(
    service_pb2.ListTasksRequest(
        user_app_id=userDataObject,  # The userDataObject is created in the overview and is required when using a PAT       
    ),
    metadata=metadata
)

if list_tasks_response.status.code != status_code_pb2.SUCCESS:
    print(list_tasks_response.status)
    raise Exception(f'Get task failed, status: {list_tasks_response.status.description}')

print(list_tasks_response)

List Tasks Assigned to User​

Get only the tasks assigned to a specific user for work.

#######################################################################
# In this section, we set the user authentication, app ID, and
# worker user ID. Change these strings to run your own example.
#######################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'
WORKER_USER_ID = 'WORKER_USER_ID_HERE'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID)

list_tasks_response = stub.ListTasks(
    service_pb2.ListTasksRequest(
        user_app_id=userDataObject,  # The userDataObject is created in the overview and is required when using a PAT  
        worker_user_ids=[WORKER_USER_ID]     
    ),
    metadata=metadata
)

if list_tasks_response.status.code != status_code_pb2.SUCCESS:
    print(list_tasks_response.status)
    raise Exception(f'Get task failed, status: {list_tasks_response.status.description}')

print(list_tasks_response)

List Tasks Assigned to User for Review​

Get only the tasks assigned to a specific user for review.

#######################################################################
# In this section, we set the user authentication, app ID, and
# review user ID. Change these strings to run your own example.
#######################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'
REVIEW_USER_ID = 'REVIEW_USER_ID_HERE'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID)

list_tasks_response = stub.ListTasks(
    service_pb2.ListTasksRequest(
        user_app_id=userDataObject,  # The userDataObject is created in the overview and is required when using a PAT  
        review_user_ids=[REVIEW_USER_ID]     
    ),
    metadata=metadata
)

if list_tasks_response.status.code != status_code_pb2.SUCCESS:
    print(list_tasks_response.status)
    raise Exception(f'Get task failed, status: {list_tasks_response.status.description}')

print(list_tasks_response)

Update​

Currently, we only support updating a task by providing all information at once.

Update Task​

#################################################################################
# In this section, we set the user authentication, app ID, and details for
# updating a task. Change these strings to run your own example.
#################################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'
# Change these to update your own task
CONCEPT_ID = 'water'
TASK_ID = 'd0f2fa2b61234d1cb6b66983ea021b5b'
USER_ID_1 = 'USER_ID_1_HERE'
USER_ID_2 = 'USER_ID_2_HERE'
USER_ID_3 = 'USER_ID_3_HERE'
USER_ID_4 = 'USER_ID_4_HERE'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2, status_pb2
from google.protobuf.struct_pb2 import Struct

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

params = Struct()
params.update({
    USER_ID_1: 1,
    USER_ID_2: 1,
    USER_ID_3: 1
})

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID)

patch_tasks_response = stub.PatchTasks(
    service_pb2.PatchTasksRequest(
        user_app_id=userDataObject,  # The userDataObject is created in the overview and is required when using a PAT
        action="overwrite",
        tasks=[
            resources_pb2.Task(
                id=TASK_ID,
                type='CONCEPTS_CLASSIFICATION',
                name='Annotate ' + CONCEPT_ID,
                worker=resources_pb2.TaskWorker(
                    strategy='PARTITIONED',
                    users=[
                        resources_pb2.User(id=USER_ID_1),
                        resources_pb2.User(id=USER_ID_2),
                        resources_pb2.User(id=USER_ID_3)
                    ],
                    partitioned_strategy_info=resources_pb2.TaskWorkerPartitionedStrategyInfo(
                        type='WEIGHTED',
                        workers_per_input=3,
                        weights=params
                    )
                ),
                concepts=[
                    resources_pb2.TaskConcept(
                        concept=resources_pb2.Concept(id=CONCEPT_ID)
                    )
                ],
                input_source=resources_pb2.TaskInputSource(type='ALL_INPUTS'),
                sample_ms=1000,
                review=resources_pb2.TaskReview(
                    strategy='CONSENSUS',
                    consensus_strategy_info=resources_pb2.TaskReviewConsensusStrategyInfo(
                        approval_threshold=2
                    ),
                    users=[resources_pb2.User(id=USER_ID_4)]
                ),
                status=status_pb2.Status(
                    code='TASK_DONE'
                )
            )
        ]
    ),
    metadata=metadata
)

if patch_tasks_response.status.code != status_code_pb2.SUCCESS:
    print(patch_tasks_response.status)
    raise Exception('Patch tasks failed, status: ' + patch_tasks_response.status.description)

Delete​

You can delete a task by specifying its ID. You can also delete multiple tasks by specifying a list of their IDs.

#################################################################################
# In this section, we set the user authentication, app ID, and task IDs. 
# Change these strings to run your own example.
#################################################################################

USER_ID = 'YOUR_USER_ID_HERE'
# Your PAT (Personal Access Token) can be found in the Account's Security section
PAT = 'YOUR_PAT_HERE'
APP_ID = 'YOUR_APP_ID_HERE'
# Change these to delete your own tasks
TASK_ID_1 = 'd1f2f891da9844ccafed93a18574b027'
TASK_ID_2 = 'd1f2f891da9844ccafed93a18574b027'

##########################################################################
# YOU DO NOT NEED TO CHANGE ANYTHING BELOW THIS LINE TO RUN THIS EXAMPLE
##########################################################################

from clarifai_grpc.channel.clarifai_channel import ClarifaiChannel
from clarifai_grpc.grpc.api import resources_pb2, service_pb2, service_pb2_grpc
from clarifai_grpc.grpc.api.status import status_code_pb2

channel = ClarifaiChannel.get_grpc_channel()
stub = service_pb2_grpc.V2Stub(channel)

metadata = (('authorization', 'Key ' + PAT),)

userDataObject = resources_pb2.UserAppIDSet(user_id=USER_ID, app_id=APP_ID)

delete_tasks_response = stub.DeleteTasks(
    service_pb2.DeleteTasksRequest(
        user_app_id=userDataObject,  # The userDataObject is created in the overview and is required when using a PAT       
        ids=[TASK_ID_1, TASK_ID_2] 

    ),
    metadata=metadata
)

if delete_tasks_response.status.code != status_code_pb2.SUCCESS:
    print(delete_tasks_response.status)
    raise Exception('Delete tasks failed, status: ' + delete_tasks_response.status.description)
© 2024 Clarifai, Inc. All rights reserved