code: def testEvaluateQueryBasedMetrics(self):
      temp_eval_export_dir = self._getEvalExportDir()
      (
          _,
          eval_export_dir,
      ) = fixed_prediction_estimator_extra_fields.simple_fixed_prediction_estimator_extra_fields(
          None, temp_eval_export_dir
      )
      eval_shared_model = self.createTestEvalSharedModel(
          eval_saved_model_path=eval_export_dir
      )
      extractors = [
          legacy_predict_extractor.PredictExtractor(eval_shared_model),
          slice_key_extractor.SliceKeyExtractor(),
      ]
  
      with beam.Pipeline() as pipeline:
          metrics = (
              pipeline
              | "Create" >> beam.Create(self._get_examples())
              | "InputsToExtracts" >> model_eval_lib.InputsToExtracts()
              | "Extract"
              >> tfma_unit.Extract(
                  extractors=extractors
              )  # pylint: disable=no-value-for-parameter
              | "EvaluateQueryBasedMetrics"
              >> query_based_metrics_evaluator.EvaluateQueryBasedMetrics(
                  prediction_key="",
                  query_id="fixed_string",
                  combine_fns=[
                      query_statistics.QueryStatisticsCombineFn(),
                      ndcg.NdcgMetricCombineFn(
                          at_vals=[1, 2],
                          gain_key="fixed_float",
                          weight_key="fixed_int",
                      ),
                      min_label_position.MinLabelPositionCombineFn(
                          label_key="", weight_key="fixed_int"
                      ),
                  ],
              )
          )
  
          def check_metrics(got):
              try:
                  self.assertEqual(1, len(got), "got: %s" % got)
                  got_slice_key, got_metrics = got[0]
                  self.assertEqual(got_slice_key, ())
                  self.assertDictElementsAlmostEqual(
                      got_metrics,
                      {
                          "post_export_metrics/total_queries": 3.0,
                          "post_export_metrics/total_documents": 6.0,
                          "post_export_metrics/min_documents": 1.0,
                          "post_export_metrics/max_documents": 3.0,
                          "post_export_metrics/ndcg@1": 0.9166667,
                          "post_export_metrics/ndcg@2": 0.9766198,
                          "post_export_metrics/average_min_label_position/__labels": 0.6666667,
                      },
                  )
  
              except AssertionError as err:
                  raise util.BeamAssertException(err)
  
          util.assert_that(
              metrics[constants.METRICS_KEY],
              check_metrics,
              label="metrics",
          )
  
imports: import os
  import apache_beam
  from apache_beam.testing import util
  import tensorflow
  from tensorflow_model_analysis import constants
  from tensorflow_model_analysis.api import model_eval_lib
  from tensorflow_model_analysis.api import tfma_unit
  from tensorflow_model_analysis.eval_saved_model import testutil
  from tensorflow_model_analysis.eval_saved_model.example_trainers import fixed_prediction_estimator_extra_fields
  from tensorflow_model_analysis.evaluators import legacy_query_based_metrics_evaluator
  from tensorflow_model_analysis.evaluators.query_metrics import min_label_position
  from tensorflow_model_analysis.evaluators.query_metrics import ndcg
  from tensorflow_model_analysis.evaluators.query_metrics import query_statistics
  from tensorflow_model_analysis.extractors import legacy_predict_extractor
  from tensorflow_model_analysis.extractors import slice_key_extractor
label_desc: Use image node_id 4 for calling the QueryBasedMetricsEvaluatorTest obj's underlying member method code with example usage: obj.testEvaluateQueryBasedMetrics() without return types