Upload metric.py with huggingface_hub

metric.py

@@ -1,16 +1,13 @@
-from dataclasses import field
-from typing import Any, Dict, Generator, Iterable, List, Optional, Union
+from typing import Dict, Iterable, List
 
-import datasets
 import evaluate
-from datasets import Features, Sequence, Value
+from datasets import Features, Value
 
 from .artifact import __file__ as _
 from .blocks import __file__ as _
 from .card import __file__ as _
 from .catalog import __file__ as _
 from .collections import __file__ as _
-from .common import __file__ as _
 from .dataclass import __file__ as _
 from .dict_utils import __file__ as _
 from .file_utils import __file__ as _
@@ -23,20 +20,12 @@ from .load import __file__ as _
 from .loaders import __file__ as _
 from .metrics import __file__ as _
 from .normalizers import __file__ as _
-from .operator import (
- MultiStreamOperator,
- SequntialOperator,
- SequntialOperatorInitilizer,
- StreamInitializerOperator,
-)
+from .operator import (MultiStreamOperator, SequentialOperator,
+ SequentialOperatorInitilizer, StreamInitializerOperator)
 from .operator import __file__ as _
-from .operators import (
- ApplyOperatorsField,
- ApplyStreamOperatorsField,
- FlattenInstances,
- MergeStreams,
- SplitByValue,
-)
+from .operators import (Apply, ApplyMetric, ApplyOperatorsField,
+ ApplyStreamOperatorsField, FlattenInstances,
+ MergeStreams, SplitByValue)
 from .operators import __file__ as _
 from .processors import __file__ as _
 from .random_utils import __file__ as _
@@ -44,6 +33,7 @@ from .recipe import __file__ as _
 from .register import __file__ as _
 from .register import _reset_env_local_catalogs, register_all_artifacts
 from .renderers import __file__ as _
+from .schema import UNITXT_DATASET_SCHEMA
 from .schema import __file__ as _
 from .split_utils import __file__ as _
 from .splitters import __file__ as _
@@ -75,12 +65,31 @@ class MultiStreamScoreMean(MultiStreamOperator):
  instance["score"]["global"]["groups_mean_score"] = score
  yield instance
 
+ def spread_results_one_stream(self, stream: Stream):
+ for instance in stream:
+ instance["score"]["global"]["groups_mean_score"] = instance["score"][
+ "global"
+ ]["score"]
+ yield instance
+
  def process(self, multi_stream: MultiStream) -> MultiStream:
- mean_score = self.aggegate_results(multi_stream)
+ result = {}
+
+ # optimization in to avoid double calculation of metrics
+ # when aggregating results, if there is only one stream.
+ if len(multi_stream) == 1:
+ for stream_name, stream in multi_stream.items():
+ result[stream_name] = Stream(
+ self.spread_results_one_stream, gen_kwargs={"stream": stream}
+ )
+ return MultiStream(result)
 
+ mean_score = self.aggegate_results(multi_stream)
  result = {}
  for stream_name, stream in multi_stream.items():
- result[stream_name] = Stream(self.spread_results, gen_kwargs={"stream": stream, "score": mean_score})
+ result[stream_name] = Stream(
+ self.spread_results, gen_kwargs={"stream": stream, "score": mean_score}
+ )
 
  return MultiStream(result)
 
@@ -90,20 +99,41 @@ class FromPredictionsAndOriginalData(StreamInitializerOperator):
  for prediction, original in zip(predictions, references):
  yield {**original, "prediction": prediction}
 
- def process(self, predictions: List[str], references: Iterable, split_name: str = "all") -> MultiStream:
+ def process(
+ self, predictions: List[str], references: Iterable, split_name: str = "all"
+ ) -> MultiStream:
  return MultiStream(
- {split_name: Stream(self.zip, gen_kwargs={"predictions": predictions, "references": references})}
+ {
+ split_name: Stream(
+ self.zip,
+ gen_kwargs={"predictions": predictions, "references": references},
+ )
+ }
  )
 
 
-from .schema import UNITXT_DATASET_SCHEMA
+# The additional_inputs field in the schema is defined as
+# Sequence({"key": Value(dtype="string"), "value": Value("string")})
+# When receiving instances from this scheme, the keys and values are returned as two separate
+# lists, and are converted to a dictionary.
+
+
+def _from_key_value_pairs(key_value_list: Dict[str, list]) -> Dict[str, str]:
+ return dict(zip(key_value_list["key"], key_value_list["value"]))
 
 
-class MetricRecipe(SequntialOperatorInitilizer):
+class MetricRecipe(SequentialOperatorInitilizer):
+ calc_confidence_intervals: bool = True
+
  def prepare(self):
  register_all_artifacts()
  self.steps = [
  FromPredictionsAndOriginalData(),
+ Apply(
+ "additional_inputs",
+ function=_from_key_value_pairs,
+ to_field="additional_inputs",
+ ),
  ApplyOperatorsField(
  inputs_fields=["prediction", "references"],
  fields_to_treat_as_list=["references"],
@@ -111,37 +141,48 @@ class MetricRecipe(SequntialOperatorInitilizer):
  default_operators=["processors.to_string_stripped"],
  ),
  SplitByValue(["group"]),
- ApplyStreamOperatorsField(
+ ApplyMetric(
  "metrics",
- reversed=True,
+ calc_confidence_intervals=self.calc_confidence_intervals,
  ),
  MultiStreamScoreMean(),
  MergeStreams(),
  ]
 
 
-UNITXT_METRIC_SCHEMA = Features({"predictions": Value("string"), "references": dict(UNITXT_DATASET_SCHEMA)})
+UNITXT_METRIC_SCHEMA = Features(
+ {"predictions": Value("string"), "references": dict(UNITXT_DATASET_SCHEMA)}
+)
 
 
-def _compute(predictions: List[str], references: Iterable, flatten: bool = False, split_name: str = "all"):
+def _compute(
+ predictions: List[str],
+ references: Iterable,
+ flatten: bool = False,
+ split_name: str = "all",
+ calc_confidence_intervals: bool = True,
+):
  _reset_env_local_catalogs()
  register_all_artifacts()
- recipe = MetricRecipe()
+ recipe = MetricRecipe(calc_confidence_intervals=calc_confidence_intervals)
 
- multi_stream = recipe(predictions=predictions, references=references, split_name=split_name)
+ multi_stream = recipe(
+ predictions=predictions, references=references, split_name=split_name
+ )
 
  if flatten:
  operator = FlattenInstances()
  multi_stream = operator(multi_stream)
 
  stream = multi_stream[split_name]
-
  return list(stream)
 
 
 # TODO: currently we have two classes with this name. metric.Metric and matrics.Metric...
 # @evaluate.utils.file_utils.add_start_docstrings(_DESCRIPTION, _KWARGS_DESCRIPTION)
 class Metric(evaluate.Metric):
+ calc_confidence_intervals: bool = True
+
  def _info(self):
  return evaluate.MetricInfo(
  description="_DESCRIPTION",
@@ -155,11 +196,16 @@ class Metric(evaluate.Metric):
  ],
  )
 
- def _compute(self, predictions: List[str], references: Iterable, flatten: bool = False, split_name: str = "all"):
+ def _compute(
+ self,
+ predictions: List[str],
+ references: Iterable,
+ flatten: bool = False,
+ split_name: str = "all",
+ ):
  try:
- from unitxt.dataset import (
- get_dataset_artifact as get_dataset_artifact_installed,
- )
+ from unitxt.dataset import \
+ get_dataset_artifact as get_dataset_artifact_installed
 
  unitxt_installed = True
  except ImportError:
@@ -169,7 +215,17 @@ class Metric(evaluate.Metric):
  from unitxt.metric import _compute as _compute_installed
 
  return _compute_installed(
- predictions=predictions, references=references, flatten=flatten, split_name=split_name
+ predictions=predictions,
+ references=references,
+ flatten=flatten,
+ split_name=split_name,
+ calc_confidence_intervals=self.calc_confidence_intervals,
  )
- else:
- return _compute(predictions=predictions, references=references, flatten=flatten, split_name=split_name)
+
+ return _compute(
+ predictions=predictions,
+ references=references,
+ flatten=flatten,
+ split_name=split_name,
+ calc_confidence_intervals=self.calc_confidence_intervals,
+ )