code: def ComputeWithConfidenceIntervals(  # pylint: disable=invalid-name
      sliced_extracts: beam.pvalue.PCollection[
          Tuple[slicer.SliceKeyType, types.Extracts]
      ],
      computations_combine_fn: beam.CombineFn,
      derived_metrics_ptransform: beam.PTransform,
      num_jackknife_samples: int,
      skip_ci_metric_keys: Optional[Set[metric_types.MetricKey]] = None,
      random_seed_for_testing: Optional[int] = None,
  ) -> beam.pvalue.PCollection[
      Tuple[
          slicer.SliceKeyOrCrossSliceKeyType, metric_types.MetricsDict
      ]
  ]:
      """Computes base metrics and derived metrics and adds std error estimates.
  
      Args:
        sliced_extracts: Incoming PCollection consisting of slice key and extracts.
        computations_combine_fn: a beam.CombineFn instance that takes input elements
          of type Extracts and returns a MetricsDict. This will be invoked as part
          of a CombinePerKey in which the key is a slice key.
        derived_metrics_ptransform: A PTransform which adds derived metrics to the
          results of the computations_combine_fn. This PTransform should both input
          and output a single PCollection with elements of type MetricsDict where
          the output MetricsDict includes additional derived metrics.
        num_jackknife_samples: The number of jackknife replicates to use in
          computing the jackknife standard error.
        skip_ci_metric_keys: Set of metric keys for which to skip confidence
          interval computation. For metric keys in this set, just the unsampled
          value will be returned.
        random_seed_for_testing: Seed to use for unit testing, because
          nondeterministic tests stink. Each partition will use this value + i.
  
      Returns:
        A PCollection of sliced metrics containing standard error estimates for
        each numeric metric.
      """
  
      random_state = np.random.RandomState(random_seed_for_testing)
  
      def partition_fn(_, num_partitions):
          return random_state.randint(num_partitions)
  
      # Partition the data
      # List[PCollection[Tuple[slicer.SliceKeyType, types.Extracts]]]
      partitions = (
          sliced_extracts
          | f"Partition({num_jackknife_samples})"
          >> beam.Partition(partition_fn, num_jackknife_samples)
      )
  
      # Within each partition, partially combine per slice key to get accumulators
      # and partition sizes; add partition_id for determinism.
      # List[PCollection[Tuple[slicer.SliceKeyType, AccumulatorType]]]
      partition_accumulators = []
      for i, partition in enumerate(partitions):
          partition_accumulators.append(
              partition
              | f"CombinePartitionPerSlice[{i}]"
              >> beam.CombinePerKey(
                  _AccumulateOnlyCombineFn(computations_combine_fn)
              )
          )
  
      unsampled_metrics = (
          partition_accumulators
          | "FlattenPartitions" >> beam.Flatten()
          | "MergePartitionsPerSlice"
          >> beam.CombinePerKey(
              _AccumulatorCombineFn(computations_combine_fn)
          )
          | "AddDerivedMetrics" >> derived_metrics_ptransform
          | "AddSampleIdToValue"
          >> beam.MapTuple(_add_sample_id, sample_id=_FULL_SAMPLE_ID)
      )
  
      # Compute the combine_fn output for the delete-d samples by merging all but
      # one partitions.
      # List[PCollection[Tuple[slicer.SliceKeyType, SampleMetrics]]]
      delete_d_samples = []
      for sample_id in range(num_jackknife_samples):
          # TODO(b/194732335): Push filter and Flatten into _ComputeJackknifeSample.
          # TODO(b/130032676): Replace the 'ExcludePartition' step with for-loop
          # exclusion after cl/435922775 (or equivalent) is submitted.
          sample_accumulators = [
              acc
              | f"ExcludePartition[{sample_id}]"
              >> beam.Filter(lambda _: False)
              if i == sample_id
              else acc
              for i, acc in enumerate(partition_accumulators)
          ]
          delete_d_samples.append(
              sample_accumulators
              | f"FlattenPartitions[{sample_id}]" >> beam.Flatten()
              | f"ComputeJackknifeSample[{sample_id}]"
              >> _ComputeJackknifeSample(  # pylint: disable=no-value-for-parameter
                  sample_id=sample_id,
                  computations_combine_fn=computations_combine_fn,
                  derived_metrics_ptransform=derived_metrics_ptransform,
              )
          )
  
      # PCollection[Tuple[slicer.SliceKeyType, metric_types.MetricsDict]]
      return (
          delete_d_samples + [unsampled_metrics]
          | "FlattenSamples" >> beam.Flatten()
          | "CombineJackknifeSamplesPerSlice"
          >> beam.CombinePerKey(
              _JackknifeSampleCombineFn(
                  num_jackknife_samples, skip_ci_metric_keys
              )
          )
      )
  
imports: from typing import Optional, Set, Tuple, TypeVar
  import apache_beam
  import numpy
  from tensorflow_model_analysis.api import types
  from tensorflow_model_analysis.evaluators import confidence_intervals_util
  from tensorflow_model_analysis.metrics import metric_types
  from tensorflow_model_analysis.slicer import slicer_lib
  from tensorflow_model_analysis.utils import beam_util
label_desc: Use image node_id 3 for calling a global function with example usage: ComputeWithConfidenceIntervals(sliced_extracts, computations_combine_fn, derived_metrics_ptransform, num_jackknife_samples, skip_ci_metric_keys, random_seed_for_testing) and returns: unknown, random_state