|
| 1 | +"""AutoQA node for automated quality assessment with pass/fail routing. |
| 2 | +
|
| 3 | +This module contains the AutoQANode class which performs automated quality assessment |
| 4 | +using configured evaluators and score thresholds. |
| 5 | +""" |
| 6 | + |
| 7 | +from typing import Dict, List, Any, Optional, Literal |
| 8 | +from pydantic import Field, model_validator, field_validator |
| 9 | + |
| 10 | +from labelbox.schema.workflow.base import BaseWorkflowNode |
| 11 | +from labelbox.schema.workflow.enums import ( |
| 12 | + WorkflowDefinitionId, |
| 13 | + NodeOutput, |
| 14 | +) |
| 15 | + |
| 16 | +# Constants for this module |
| 17 | +DEFAULT_FILTER_LOGIC_AND = "and" |
| 18 | + |
| 19 | + |
| 20 | +class AutoQANode(BaseWorkflowNode): |
| 21 | + """ |
| 22 | + Automated Quality Assessment node with pass/fail routing. |
| 23 | +
|
| 24 | + This node performs automated quality assessment using configured evaluators |
| 25 | + and score thresholds. Work that meets the quality criteria is routed to the |
| 26 | + "if" output (passed), while work that fails is routed to the "else" output. |
| 27 | +
|
| 28 | + Attributes: |
| 29 | + label (str): Display name for the node (default: "Label Score (AutoQA)") |
| 30 | + filters (List[Dict[str, Any]]): Filter conditions for the AutoQA node |
| 31 | + filter_logic (str): Logic for combining filters ("and" or "or", default: "and") |
| 32 | + custom_fields (Dict[str, Any]): Additional custom configuration |
| 33 | + definition_id (WorkflowDefinitionId): Node type identifier (read-only) |
| 34 | + node_config (List[Dict[str, Any]]): API configuration for evaluator settings |
| 35 | + evaluator_id (Optional[str]): ID of the evaluator for AutoQA assessment |
| 36 | + scope (Optional[str]): Scope setting for AutoQA ("any" or "all") |
| 37 | + score_name (Optional[str]): Name of the score metric for evaluation |
| 38 | + score_threshold (Optional[float]): Threshold score for pass/fail determination |
| 39 | +
|
| 40 | + Inputs: |
| 41 | + Default: Must have exactly one input connection |
| 42 | +
|
| 43 | + Outputs: |
| 44 | + If: Route for work that passes quality assessment (score >= threshold) |
| 45 | + Else: Route for work that fails quality assessment (score < threshold) |
| 46 | +
|
| 47 | + AutoQA Configuration: |
| 48 | + - evaluator_id: Specifies which evaluator to use for assessment |
| 49 | + - scope: Determines evaluation scope ("any" or "all" annotations) |
| 50 | + - score_name: The specific score metric to evaluate |
| 51 | + - score_threshold: Minimum score required to pass |
| 52 | + - Automatically syncs configuration with API format |
| 53 | +
|
| 54 | + Validation: |
| 55 | + - Must have exactly one input connection |
| 56 | + - Both passed and failed outputs can be connected |
| 57 | + - AutoQA settings are automatically converted to API configuration |
| 58 | + - Evaluator and scoring parameters are validated |
| 59 | +
|
| 60 | + Example: |
| 61 | + >>> autoqa = AutoQANode( |
| 62 | + ... label="Quality Gate", |
| 63 | + ... evaluator_id="evaluator-123", |
| 64 | + ... scope="all", |
| 65 | + ... score_name="accuracy", |
| 66 | + ... score_threshold=0.85 |
| 67 | + ... ) |
| 68 | + >>> # Route high-quality work to done, low-quality to review |
| 69 | + >>> workflow.add_edge(autoqa, done_node, NodeOutput.If) |
| 70 | + >>> workflow.add_edge(autoqa, review_node, NodeOutput.Else) |
| 71 | +
|
| 72 | + Quality Assessment: |
| 73 | + AutoQA nodes enable automated quality control by evaluating work |
| 74 | + against trained models or rule-based evaluators. This reduces manual |
| 75 | + review overhead while maintaining quality standards. |
| 76 | +
|
| 77 | + Note: |
| 78 | + AutoQA requires properly configured evaluators and score thresholds. |
| 79 | + The evaluation results determine automatic routing without human intervention. |
| 80 | + """ |
| 81 | + |
| 82 | + label: str = Field(default="Label Score (AutoQA)") |
| 83 | + filters: List[Dict[str, Any]] = Field( |
| 84 | + default_factory=lambda: [], |
| 85 | + description="Contains the filters for the AutoQA node", |
| 86 | + ) |
| 87 | + filter_logic: Literal["and", "or"] = Field( |
| 88 | + default=DEFAULT_FILTER_LOGIC_AND, alias="filterLogic" |
| 89 | + ) |
| 90 | + custom_fields: Dict[str, Any] = Field( |
| 91 | + default_factory=lambda: {}, |
| 92 | + alias="customFields", |
| 93 | + ) |
| 94 | + definition_id: WorkflowDefinitionId = Field( |
| 95 | + default=WorkflowDefinitionId.AutoQA, |
| 96 | + frozen=True, |
| 97 | + alias="definitionId", |
| 98 | + ) |
| 99 | + node_config: List[Dict[str, Any]] = Field( |
| 100 | + default_factory=lambda: [], |
| 101 | + description="Contains evaluator_id, scope, score_name, score_threshold etc.", |
| 102 | + alias="config", |
| 103 | + ) |
| 104 | + |
| 105 | + # AutoQA-specific fields |
| 106 | + evaluator_id: Optional[str] = Field( |
| 107 | + default=None, |
| 108 | + description="ID of the evaluator for AutoQA", |
| 109 | + ) |
| 110 | + scope: Optional[str] = Field( |
| 111 | + default=None, |
| 112 | + description="Scope setting for AutoQA (any/all)", |
| 113 | + ) |
| 114 | + score_name: Optional[str] = Field( |
| 115 | + default=None, |
| 116 | + description="Name of the score for AutoQA", |
| 117 | + ) |
| 118 | + score_threshold: Optional[float] = Field( |
| 119 | + default=None, |
| 120 | + description="Threshold score for AutoQA", |
| 121 | + ) |
| 122 | + |
| 123 | + @model_validator(mode="after") |
| 124 | + def sync_autoqa_config_with_node_config(self) -> "AutoQANode": |
| 125 | + """Sync AutoQA-specific fields with node_config.""" |
| 126 | + |
| 127 | + # Clear existing AutoQA config |
| 128 | + self.node_config = [ |
| 129 | + config |
| 130 | + for config in self.node_config |
| 131 | + if config.get("field") |
| 132 | + not in ["evaluator_id", "scope", "score_name", "score_threshold"] |
| 133 | + ] |
| 134 | + |
| 135 | + # Add evaluator_id if present |
| 136 | + if self.evaluator_id is not None: |
| 137 | + self.node_config.append( |
| 138 | + { |
| 139 | + "field": "evaluator_id", |
| 140 | + "value": self.evaluator_id, |
| 141 | + "metadata": None, |
| 142 | + } |
| 143 | + ) |
| 144 | + |
| 145 | + # Add scope if present |
| 146 | + if self.scope is not None: |
| 147 | + self.node_config.append( |
| 148 | + {"field": "scope", "value": self.scope, "metadata": None} |
| 149 | + ) |
| 150 | + |
| 151 | + # Add score_name if present |
| 152 | + if self.score_name is not None: |
| 153 | + self.node_config.append( |
| 154 | + { |
| 155 | + "field": "score_name", |
| 156 | + "value": self.score_name, |
| 157 | + "metadata": None, |
| 158 | + } |
| 159 | + ) |
| 160 | + |
| 161 | + # Add score_threshold if present |
| 162 | + if self.score_threshold is not None: |
| 163 | + self.node_config.append( |
| 164 | + { |
| 165 | + "field": "score_threshold", |
| 166 | + "value": self.score_threshold, |
| 167 | + "metadata": None, |
| 168 | + } |
| 169 | + ) |
| 170 | + |
| 171 | + return self |
| 172 | + |
| 173 | + @field_validator("inputs") |
| 174 | + @classmethod |
| 175 | + def validate_inputs(cls, v) -> List[str]: |
| 176 | + """Validate that AutoQA node has exactly one input.""" |
| 177 | + if len(v) != 1: |
| 178 | + raise ValueError("AutoQA node must have exactly one input") |
| 179 | + return v |
| 180 | + |
| 181 | + @property |
| 182 | + def supported_outputs(self) -> List[NodeOutput]: |
| 183 | + """Returns the list of supported output types for this node.""" |
| 184 | + return [NodeOutput.If, NodeOutput.Else] # Passed (if) and Failed (else) |
0 commit comments