Examples

Prioritizing tasks based on file names

This example shows how to have an agent node early in the workflow which gives every task a priority before sending it on to the annotation stage.

For a project with a workflow similar to this:

Create a new file called agent.py with the following content:

agent.py

from encord.objects.ontology_labels_impl import LabelRowV2
from encord_agents.tasks import Runner

runner = Runner(project_hash="<your_project_hash>")


@runner.stage("<your_agent_stage_uuid>")
def by_file_name(lr: LabelRowV2) -> str | None:
    # Assuming the data_title is of the format "%d.jpg"
    # and in the range [0; 100]
    priority = int(lr.data_title.split(".")[0]) / 100
    lr.set_priority(priority=priority)
    return "<your_pathway_uuid>"


if __name__ == "__main__":
    runner.run()

The code will

Instantiate a runner.
Define a stage implementation that looks for the data title of a given task and parses the stem of the data title as an int.
Set the priority as a number between zero and one.
Pass the task to the annotation stage by returning the appropriate pathway uuid.

To run the agent, follow these steps: To make this example run in practice, you need to

Ensure that you've exported your private key, as described in the authentication section, and that you've installed the encord_agents package.
Create a workflow similar to the above
Copy the code to an agent.py file
Adjust these ids <your_project_hash>, <your_agent_stage_uuid>, and <your_pathway_uuid> to correspond to those ids of your workflow.
Execute the file python agent.py

Transferring labels to a twin project

This example shows how to take checklist labels annotated in "Project A" and translate them into yes/no radio labels in "Project B".

Assume you have an ontology like this one in Project A:

and you want to ensure that everytime a task in Project A is compelte, it gets translated to a "model friendly version" with radio classifications in Project B. A good way to do it is make an agent do the translation to the ontology of Project B:

Notice how there are now three classifications (with the exact same names!) with two options each.

To build an agent that automatically translates between the two, the dep_twin_label_row dependency is a good place to start. For every label row from Project A that the agent is called with, it will automatically get the corresponding label row (and potentially workflow task) from project B.

Disclaimer: Project A and Project B must be attached to the same datasets.

The code that defines such an agent might look similar to this:

twin_project.py

from encord.objects.ontology_labels_impl import LabelRowV2
from encord.objects.options import Option
from encord.workflow.stages.agent import AgentTask
from encord_agents.tasks import Depends, Runner
from encord_agents.tasks.dependencies import Twin, dep_twin_label_row
from typing_extensions import Annotated

# 1. Setup the runner
runner = Runner(project_hash="<project_hash_a>")

# 2. Get the classification attribute used to query answers
checklist_classification = (
    runner.project.ontology_structure.classifications[0]  # type: ignore
)
checklist_attribute = checklist_classification.attributes[0]


# 3. Define the agent
@runner.stage(stage="<transfer_agent_stage_uuid>")
def copy_labels(
    manually_annotated_lr: LabelRowV2,
    twin: Annotated[
        Twin, Depends(dep_twin_label_row(twin_project_hash="<project_hash_b>"))
    ],
) -> str | None:
    # 4. Reading the checkboxes that have been set
    instance = manually_annotated_lr.get_classification_instances()[0]
    answers = instance.get_answer(attribute=checklist_attribute)
    if answers is None or isinstance(answers, (str, Option)):
        return None

    set_options = {o.title for o in answers}  # Use title to match

    # 5. Set answer on the sink labels
    for radio_clf in twin.label_row.ontology_structure.classifications:
        ins = radio_clf.create_instance()

        attr = radio_clf.attributes[0]
        if radio_clf.title in set_options:
            ins.set_answer(attr.options[0])
        else:
            ins.set_answer(attr.options[1])

        ins.set_for_frames(frames=0)
        twin.label_row.add_classification_instance(ins)

    # 6. Save labels and proceed tasks
    twin.label_row.save()
    if twin.task and isinstance(twin.task, AgentTask):
        twin.task.proceed(pathway_uuid="<twin_completion_pathway_uuid>")

    return "<labeling_completion_pathway_uuid>"


if __name__ == "__main__":
    runner.run()

For this code to work, the project workflows could look like this:

With this setup, all the manual work happens in Project A and Project B just becomes a mirror with transformed labels.

Which would mean that the agents would be defined with the following decorator to make the workflow stage association explicit.

@runner.stage(stage="60d9f14f-755e-40fd-...")  # <- last bit omitted

Notice the match between the uuid in the "label transfer" agent stage of the workflow in Project A and the uuid in the decorator.

To prepare your projects:

Create two projects, one with each of the (classification) ontologies and workflows displayed above. For this particular example, it is not important which checklists the ontologies have as long as their names match between the two ontologies. Make sure that both projects are pointing to the same dataset(s).

To run the agent, follow these steps:

Ensure that you've exported your private key, as described in the authentication section, and that you've installed the encord_agents package.
Update the code to capture your own project hashes. (replace <project_hash_a> and <project_hash_b>.
Update the stage argument to the decorator to reflect the uuid of your actual agent stage.
Update the completion pathway uuids
Run the file: python twin_project.py

Now you should see tasks that have been approved by review starting to move to the Complete state and labels starting to show up

Pre-label video with fake predictions

Suppose yuou have a fake model like this one, which predicts labels, bounding boxes and, confidenceds.

Fake model predictions

@dataclass
class ModelPrediction:
    label: int
    coords: BoundingBoxCoordinates
    conf: float


def fake_predict(image: NDArray[np.uint8]) -> list[ModelPrediction]:
    return [
        ModelPrediction(
            label=random.choice(range(3)),
            coords=BoundingBoxCoordinates(
                top_left_x=random.random() * 0.5,
                top_left_y=random.random() * 0.5,
                width=random.random() * 0.5,
                height=random.random() * 0.5,
            ),
            conf=random.random() + 0.5,
        )
        for _ in range(10)
    ]


model = fake_predict

And you have an ontology looking like this:

If you then set up a workflow like this one with a pre-labeling agent node before the annotation stage, you will be able to pre-label the tasks with the model predictions.

Then you can make a pre-labeling agent that looks like this:

prelabel_video.py

import random
from dataclasses import dataclass
from typing import Iterable

import numpy as np
from encord.objects.coordinates import BoundingBoxCoordinates
from encord.objects.ontology_labels_impl import LabelRowV2
from encord.project import Project
from encord_agents.core.data_model import Frame
from encord_agents.tasks import Depends, Runner
from encord_agents.tasks.dependencies import dep_video_iterator
from numpy.typing import NDArray
from typing_extensions import Annotated

runner = Runner(project_hash="<project_hash>")


# === BEGIN FAKE MODEL === #
@dataclass
class ModelPrediction:
    label: int
    coords: BoundingBoxCoordinates
    conf: float


def fake_predict(image: NDArray[np.uint8]) -> list[ModelPrediction]:
    return [
        ModelPrediction(
            label=random.choice(range(3)),
            coords=BoundingBoxCoordinates(
                top_left_x=random.random() * 0.5,
                top_left_y=random.random() * 0.5,
                width=random.random() * 0.5,
                height=random.random() * 0.5,
            ),
            conf=random.random() + 0.5,
        )
        for _ in range(10)
    ]


model = fake_predict
# === END FAKE MODEL === #


@runner.stage(stage="pre-label")
def run_something(
    lr: LabelRowV2,
    project: Project,
    frames: Annotated[Iterable[Frame], Depends(dep_video_iterator)],
) -> str:
    ontology = project.ontology_structure

    for frame in frames:
        outputs = model(frame.content)
        for output in outputs:
            ins = ontology.objects[output.label].create_instance()
            ins.set_for_frames(
                frames=frame.frame, coordinates=output.coords, confidence=output.conf
            )

            lr.add_object_instance(ins)

    lr.save()
    return "annotate"  # Tell where the task should go


if __name__ == "__main__":
    runner.run()

Notice how we use the dep_video_iterator dependency to automatically load in an iterator of frames in the form or RGB np arrays from the video.

To run the agent, follow these steps:

Ensure that you've exported your private key, as described in the authentication section, and that you've installed the encord_agents package.
Ensure that your project has a stage names "pre-label" with a pathway names "annotate" and an ontology similar to the above.
Replace <project_hash> with your own project hash.
Run the file: python prelabel_video.py

Once you start annotating upon agent completion, you should see frames pre-populated with bounding boxes.

Examples that we're working on

Pre-labeling with YoloWorld
Transcribing with Whisper
Routing with Gemini
Prioritizing with GPT-4o mini
Evaluating Training projects
HF Image segmentation API
HF LLM API to classify frames