Upload operators.py with huggingface_hub

operators.py

@@ -20,6 +20,7 @@ Some operators are specielized in specific task such as:
 
 - :class:`loaders<unitxt.loaders>` for loading data.
 - :class:`splitters<unitxt.splitters>` for fixing data splits.
+- :class:`struct_data_operators<unitxt.struct_data_operators>` for structured data operators.
 
 Other specelized operators are used by unitxt internally:
 
@@ -32,7 +33,7 @@ General Operaotrs List:
 ------------------------
 """
 import collections
-import importlib
+import copy
 import operator
 import os
 import uuid
@@ -41,7 +42,6 @@ from abc import abstractmethod
 from collections import Counter
 from copy import deepcopy
 from dataclasses import field
-from importlib import import_module
 from itertools import zip_longest
 from random import Random
 from typing import (
@@ -64,6 +64,7 @@ from .dict_utils import dict_delete, dict_get, dict_set, is_subpath
 from .operator import (
     MultiStream,
     MultiStreamOperator,
+    PackageRequirementsMixin,
     PagedStreamOperator,
     SequentialOperator,
     SideEffectOperator,
@@ -782,7 +783,7 @@ class Apply(StreamInstanceOperator):
         elif module_name in globals():
             obj = globals()[module_name]
         else:
-            obj = importlib.import_module(module_name)
+            obj = __import__(module_name)
         for part in function_name.split("."):
             obj = getattr(obj, part)
         return obj
@@ -963,7 +964,16 @@ class CopyFields(FieldOperator):
     """
 
     def process_value(self, value: Any) -> Any:
-        return value
+        return copy.deepcopy(value)
+
+
+class GetItemByIndex(FieldOperator):
+    """Get from the item list by the index in the field."""
+
+    items_list: List[Any]
+
+    def process_value(self, value: Any) -> Any:
+        return self.items_list[value]
 
 
 class AddID(StreamInstanceOperator):
@@ -1230,10 +1240,13 @@ class ComputeExpressionMixin(Artifact):
     expression: str
     imports_list: List[str] = OptionalField(default_factory=list)
 
+    def verify(self):
+        PackageRequirementsMixin.check_missing_requirements(self, self.imports_list)
+
     def prepare(self):
         # can not do the imports here, because object does not pickle with imports
         self.globals = {
-            module_name: import_module(module_name) for module_name in self.imports_list
+            module_name: __import__(module_name) for module_name in self.imports_list
         }
 
     def compute_expression(self, instance: dict) -> Any:
@@ -1574,7 +1587,7 @@ class ApplyMetric(SingleStreamOperator, ArtifactFetcherMixin):
     calc_confidence_intervals: bool
 
     def process(self, stream: Stream, stream_name: Optional[str] = None) -> Generator:
-        from .metrics import Metric, MetricPipeline, MetricWithConfidenceInterval
+        from .metrics import Metric
 
         first_instance = stream.peek()
 
@@ -1593,6 +1606,16 @@ class ApplyMetric(SingleStreamOperator, ArtifactFetcherMixin):
         # by the first listed metric (as desired).
         metric_names = list(reversed(metric_names))
 
+        # Workaround: The metric/MetricPipeline modifies the stream itself, sometines making it incompatible
+        # for further metrics' processing, instead of just modifying the score field.
+        # Here we keep all the fields besides the score, and restore them after the metric finishes.
+        first_instance = stream.peek()
+        keys_to_restore = set(first_instance.keys()).difference({"score"})
+        multi_stream = MultiStream({"tmp": stream})
+        multi_stream = CopyFields(
+            field_to_field={k: f"{k}_orig" for k in keys_to_restore}
+        )(multi_stream)
+
         for metric_name in metric_names:
             metric = self.get_artifact(metric_name)
             assert isinstance(
@@ -1600,15 +1623,17 @@ class ApplyMetric(SingleStreamOperator, ArtifactFetcherMixin):
             ), f"Operator {metric_name} must be a Metric"
 
             if not self.calc_confidence_intervals:
-                if isinstance(metric, MetricWithConfidenceInterval):
-                    metric.disable_confidence_interval_calculation()
-                elif isinstance(metric, MetricPipeline) and isinstance(
-                    metric.metric, MetricWithConfidenceInterval
-                ):
-                    metric.metric.disable_confidence_interval_calculation()
+                metric.disable_confidence_interval_calculation()
 
-            stream = metric(MultiStream({"tmp": stream}))["tmp"]
+            multi_stream = metric(multi_stream)
+            multi_stream = CopyFields(
+                field_to_field={f"{k}_orig": k for k in keys_to_restore}
+            )(multi_stream)
 
+        multi_stream = RemoveFields(fields=[f"{k}_orig" for k in keys_to_restore])(
+            multi_stream
+        )
+        stream = multi_stream["tmp"]
         yield from stream