Upload task.py with huggingface_hub

task.py

@@ -13,10 +13,65 @@ class FormTask(Tasker, StreamInstanceOperator):
     metrics: List[str]
 
     def process(self, instance: Dict[str, Any], stream_name: str = None) -> Dict[str, Any]:
-        inputs = {key: instance[key] for key in self.inputs}
-        outputs = {key: instance[key] for key in self.outputs}
+        try:
+            inputs = {key: instance[key] for key in self.inputs}
+        except KeyError as e:
+            raise KeyError(
+                f"Unexpected input column names: {list(key for key in self.inputs if key not in instance)}"
+                f"\n available names:{list(instance.keys())}\n given input names:{self.inputs}"
+            )
+        try:
+            outputs = {key: instance[key] for key in self.outputs}
+        except KeyError as e:
+            raise KeyError(
+                f"Unexpected output column names: {list(key for key in self.inputs if key not in instance)}"
+                f" \n available names:{list(instance.keys())}\n given output names:{self.outputs}"
+            )
+
         return {
             "inputs": inputs,
             "outputs": outputs,
             "metrics": self.metrics,
         }
+
+
+class MultipleChoiceTask(FormTask):
+    choices_field: str = "choices"
+    choices_separator: str = "\n"
+    enumeration_suffix: str = ". "
+    use_text_in_target: bool = False
+    alphabet: str = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+
+    def process_single_choice(self, choice: str, index: int, use_text: bool = True) -> str:
+        try:
+            processed_choice = f"{self.alphabet[index]}"
+        except IndexError:
+            raise ValueError(
+                f"Too many choices, the length of alphabet '{self.alphabet}': {len(self.alphabet)} is the limit"
+            )
+        if use_text:
+            processed_choice += f"{self.enumeration_suffix}{choice}"
+        return processed_choice
+
+    def process_choices(self, choices: List[str]) -> str:
+        processed_choices = []
+        for index, choice in enumerate(choices):
+            processed_choices.append(self.process_single_choice(choice, index))
+        return self.choices_separator.join(processed_choices)
+
+    def process_target(self, choices, target_index):
+        return self.process_single_choice(choices[target_index], target_index, use_text=self.use_text_in_target)
+
+    def process(self, instance: Dict[str, Any], stream_name: str = None) -> Dict[str, Any]:
+        result = super().process(instance, stream_name)
+        target_key, target_value = next(iter(result["outputs"].items()))
+        choices = result["inputs"][self.choices_field]
+        target_index_in_choices = choices.index(target_value)
+
+        processed_choices = self.process_choices(choices)
+        processed_target = self.process_target(choices, target_index_in_choices)
+
+        result["inputs"][self.choices_field] = processed_choices
+        result["outputs"][target_key] = processed_target
+
+        return result