class_docstr: A `Metric` takes the results of execution and produces `Stat`s for a
  scenario.
  
  Note: `Metric` actually right now is a bit of misnomer because it produces many
  `Stat`s, that might be distinct but are computed together.  Eventually we
  might move to a world where there is one (or very few metrics that are domain-independent).
class_docstr_tok: ["A","`","Metric","`","takes","the","results","of","execution","and","produces","`","Stat","`","s","for","a","scenario",".","Note",":","`","Metric","`","actually","right","now","is","a","bit","of","misnomer","because","it","produces","many","`","Stat","`","s",",","that","might","be","distinct","but","are","computed","together",".","Eventually","we","might","move","to","a","world","where","there","is","one","(","or","very","few","metrics","that","are","domain-independent",")","."]
docstr: Main entry point for a `Metric`.  This function groups the single
  list of `RequestState` by training trial and instance, and invokes
  other functions to process those.  This should serve most purposes.
  
  Any logic that doesn't decompose along instances should go here, such
  as robustness.
docstr_tok: ["Main","entry","point","for","a","`","Metric","`",".","This","function","groups","the","single","list","of","`","RequestState","`","by","training","trial","and","instance",",","and","invokes","other","functions","to","process","those",".","This","should","serve","most","purposes",".","Any","logic","that","does","n't","decompose","along","instances","should","go","here",",","such","as","robustness","."]
code: def evaluate(
      self,
      scenario_state: ScenarioState,
      metric_service: MetricService,
      eval_cache_path: str,
      parallelism: int,
  ) -> MetricResult:
      """
      Main entry point for a `Metric`.  This function groups the single
      list of `RequestState` by training trial and instance, and invokes
      other functions to process those.  This should serve most purposes.
  
      Any logic that doesn't decompose along instances should go here, such
      as robustness.
      """
      if scenario_state.adapter_spec.method == ADAPT_LANGUAGE_MODELING:
          return self.evaluate_language_modeling(
              scenario_state, metric_service, eval_cache_path
          )
  
      adapter_spec = scenario_state.adapter_spec
      global_stats: Dict[MetricName, Stat] = {}  # MetricName -> Stat
      all_per_instance_stats: List[PerInstanceStats] = []
  
      for train_trial_index in range(adapter_spec.num_train_trials):
          # Construct inputs
          request_state_sets: List[RequestStateSet] = []
          for instance in scenario_state.instances:
              generation_states = scenario_state.get_request_states(
                  train_trial_index, instance, None
              )
              references_states = []
              for reference_index in range(len(instance.references)):
                  references_states.extend(
                      scenario_state.get_request_states(
                          train_trial_index, instance, reference_index
                      )
                  )
              request_state_set = RequestStateSet(
                  instance=instance,
                  generation_states=generation_states,
                  references_states=references_states,
              )
              request_state_sets.append(request_state_set)
  
          # Do it!
          processor = Processor(
              metric=self,
              metric_service=metric_service,
              eval_cache_path=eval_cache_path,
              adapter_spec=scenario_state.adapter_spec,
          )
          results: List[List[Stat]] = parallel_map(
              processor.process,
              request_state_sets,
              parallelism=parallelism,
          )
  
          # Compute per-instance stats
          per_instance_stats: List[PerInstanceStats] = []
          for instance, stats in zip(scenario_state.instances, results):
              assert (
                  instance.id is not None
              ), f"id was none for instance: {instance}"
              # Sometimes a metric (e.g., BiasMetric) doesn't produce any statistics
              if len(stats) > 0:
                  per_instance_stats.append(
                      PerInstanceStats(
                          instance.id,
                          instance.perturbation,
                          train_trial_index,
                          stats,
                      )
                  )
  
          # Aggregate these stats
          trial_stats: Dict[
              MetricName, Stat
          ] = {}  # Statistics just for this trial
          for instance_stats in results:
              for stat in instance_stats:
                  merge_stat(trial_stats, stat)
  
          # Derive new statistics based on existing stats by calling `derive_stats` (e.g., for perplexity).
          # Group stats according to the context (e.g., split, perturbation),
          # i.e., non-name part of the MetricName, and call `derive_stats` on
          # each grouping.
          grouped_trial_stats: Dict[
              MetricContext, Dict[MetricName, Stat]
          ] = defaultdict(dict)
          for metric_name, stat in trial_stats.items():
              grouped_trial_stats[
                  MetricContext.from_metric_name(metric_name)
              ][
                  metric_name
              ] = stat  # group by context
          for context, stats_dict in grouped_trial_stats.items():
              for stat in self.derive_stats(stats_dict):
                  # we could potentially allow derive_stats to overwrite context, but this feels more robust
                  merge_stat(
                      trial_stats, add_context(stat, context)
                  )  # add correct context
  
          # Derive new per-instance statistics by calling `derive_per_instance_stats` (e.g., for calibration).
          # Again, group stats according to the context before calling
          # `derive_per_instance_stats`.
          grouped_per_instance_stats: Dict[
              MetricContext, Dict[Instance, List[Stat]]
          ] = defaultdict(lambda: defaultdict(list))
          for instance, stats in zip(scenario_state.instances, results):
              for stat in stats:
                  grouped_per_instance_stats[
                      MetricContext.from_instance(instance)
                  ][instance].append(stat)
          for (
              context,
              instance_dict,
          ) in grouped_per_instance_stats.items():
              # Here, we assume that derive_per_instance_stats only computes trial_stats-level metrics
              # (instance-level metrics should be computed in the evaluate_{generation,references} anyway).
              for stat in self.derive_per_instance_stats(instance_dict):
                  merge_stat(trial_stats, add_context(stat, context))
  
          # Compute statistics that depend on all the `RequestStates` (e.g., bias metrics).
          # Aggregate request states and call evaluate_instances in case the metric needs it.
          grouped_request_states: Dict[
              MetricContext, List[RequestState]
          ] = defaultdict(list)
          for instance in scenario_state.instances:
              # TODO: do we need to support reference_index that is not None?
              grouped_request_states[
                  MetricContext.from_instance(instance)
              ].extend(
                  scenario_state.get_request_states(
                      train_trial_index, instance, None
                  )
              )
          for context, request_states in grouped_request_states.items():
              for stat in self.evaluate_instances(request_states):
                  merge_stat(trial_stats, add_context(stat, context))
  
          # Compute worst-case metrics.
          # This is here since we want these stats for all metrics and they
          # aggregate across contexts (perturbations).
          worst_case_stats = self.compute_worst_case_metrics(
              dict(zip(scenario_state.instances, results))
          )
          for stat in worst_case_stats:
              merge_stat(trial_stats, stat)
  
          # We take the mean value for each trial.
          for stat in trial_stats.values():
              merge_stat(global_stats, stat.take_mean())
  
          all_per_instance_stats.extend(per_instance_stats)
  
      # Wrap aggregated and per-instance stats in a MetricResult.
      return MetricResult(
          list(global_stats.values()), all_per_instance_stats
      )
  
imports: from abc import ABC
  from dataclasses import dataclass, replace
  from collections import defaultdict
  from typing import List, Dict, Tuple, Optional, Iterable, Set
  from helm.common.object_spec import ObjectSpec, create_object
  from helm.common.general import singleton, parallel_map
  from helm.benchmark.augmentations.perturbation_description import PerturbationDescription, PERTURBATION_ORIGINAL, PERTURBATION_WORST
  from helm.benchmark.adaptation.adapters.adapter_factory import ADAPT_LANGUAGE_MODELING
  from helm.benchmark.adaptation.scenario_state import ScenarioState
  from helm.benchmark.adaptation.request_state import RequestState
  from helm.benchmark.adaptation.adapter_spec import AdapterSpec
  from helm.benchmark.scenarios.scenario import Instance
  from .metric_name import MetricName, MetricContext
  from .metric_service import MetricService
  from .statistic import Stat, merge_stat
label_desc: Use image node_id 1 for calling the Metric obj's underlying member method code with example usage: obj.evaluate(scenario_state, metric_service, eval_cache_path, parallelism) and returns: MetricResult, self