import labelbox as lb
from labelbox.schema.workflow import (
NodeType,
NodeOutput,
ProjectWorkflowFilter,
LabelingConfig,
ReworkConfig,
labeled_by,
metadata,
sample,
labeled_at,
mp_condition,
m_condition,
labeling_time,
review_time,
issue_category,
batch,
dataset,
annotation,
consensus_average,
model_prediction,
natural_language,
feature_consensus_average
)
from labelbox.schema.workflow.enums import IndividualAssignment, MatchFilters
from datetime import datetime
# Initialize client and project
client = lb.Client(api_key="<YOUR_API_KEY>")
project_id = "<project_id>"
project = client.get_project(project_id)
# Get workflow
workflow = project.get_workflow()
# Reset config to start a fresh workflow
initial_nodes = workflow.reset_to_initial_nodes(
labeling_config=LabelingConfig(
instructions="This is the entry point",
max_contributions_per_user=10
),
rework_config=ReworkConfig(
individual_assignment=[IndividualAssignment.LabelCreator]
)
)
initial_labeling = initial_nodes.labeling
initial_rework = initial_nodes.rework
initial_review = workflow.add_node(
type=NodeType.Review,
name="Initial review task",
group_assignment=["<user_group_id_1>", "<user_group_id_2>"]
)
logic = workflow.add_node(
type=NodeType.Logic,
name="Logic node",
match_filters=MatchFilters.Any,
filters=ProjectWorkflowFilter([
labeled_by.is_one_of(["<user_id_1>", "<user_id_2>", "<user_id_3>"]),
metadata([m_condition.contains("<metadata_schema_id>", ["test"])]),
sample(23),
labeled_at.between(
datetime(2024, 3, 9, 5, 5, 42),
datetime(2025, 4, 28, 13, 5, 42)
),
labeling_time.greater_than(1000),
review_time.less_than_or_equal(100),
issue_category.is_one_of(["<issue_category_id>"]),
batch.is_one_of(["<batch_id>"]),
dataset.is_one_of(["<dataset_id>"]),
annotation.is_one_of(["<schema_node_id>"]),
consensus_average(0.17, 0.61),
model_prediction([
mp_condition.is_one_of(["<model_id_1>"], 1),
mp_condition.is_not_one_of(["<model_id_2>"], 2, 6),
mp_condition.is_none()
]),
natural_language("Birds in the sky/Blue sky/clouds/0.5", 0.178, 0.768),
feature_consensus_average(0.17, 0.67, ["<schema_node_id>"])
])
)
# Terminal and step nodes
done = workflow.add_node(type=NodeType.Done)
rework = workflow.add_node(type=NodeType.Rework, name="To rework")
custom_rework_1 = workflow.add_node(
type=NodeType.CustomRework,
name="Custom Rework 1",
individual_assignment=IndividualAssignment.LabelCreator,
group_assignment=["<user_group_id_1>", "<user_group_id_2>"]
)
review_2 = workflow.add_node(
type=NodeType.Review,
name="Review 2"
)
custom_rework_2 = workflow.add_node(
type=NodeType.CustomRework,
name="Custom Rework 2",
instructions="Additional rework instructions"
)
done_2 = workflow.add_node(
type=NodeType.Done,
name="Ready for final review"
)
# Create edges between nodes
workflow.add_edge(initial_labeling, initial_review)
workflow.add_edge(initial_rework, initial_review)
workflow.add_edge(initial_review, logic, NodeOutput.Approved)
workflow.add_edge(initial_review, rework, NodeOutput.Rejected)
workflow.add_edge(logic, review_2, NodeOutput.If)
workflow.add_edge(logic, custom_rework_1, NodeOutput.Else)
workflow.add_edge(review_2, done, NodeOutput.Approved)
workflow.add_edge(review_2, custom_rework_2, NodeOutput.Rejected)
workflow.add_edge(custom_rework_2, done_2)
# Commit the workflow
try:
updated_workflow = workflow.update_config(reposition=True)
print("Workflow updated successfully!")
except ValueError as e:
print(f"Validation failed: {e}")