Model analysis using TFX Pipeline

Model analysis using TFX Pipeline and TensorFlow Model Analysis

 -  We will create and run a TFX pipeline which creates a simple classification model and analyzes its performance across multiple runs(여러번의 실행에서 성능을 분석함).
 -  As you tweak(조정, 약간수정) your model or train it with a new dataset, you need to check whether your model has improved or become worse.

 -  Just checking top-level metrics like accuracy might not be enough. Every trained model should be evaluated before it is pushed to production(production으로 푸시되기 전에 평가되어야 함).

We will add an Evaluator component to the pipeline created in the previous tutorial. 

→ The Evaluator component performs deep analysis for your models and compare the new model against a baseline to determine they are "good enough".

 

1. Create a Pipeline

We will add an Evaluator component to the pipeline.

 → 'Evaluator components'에는 exampleGen 구성 요소의 입력 데이터와 Trainer 구성 요소 및 tfma.EvalConfig 개체의 모델이 필요하다.

 

An evaluator creates two kinds of output artifacts, ModelEvaluation and ModelBlessing. 

 → ModelEvaluation : TFMA 라이브러리를 사용하여 추가로 조사하고 시각화할 수 있는 자세한 평가 결과가 포함됨

 → ModelBlessing : 모델이 주어진 기준을 통과했는지 여부에 대한 boolean result가 포함되어 있으며, Pusher 같은 이후 구성 요소에서 신호로 사용될 수 있다.

 

We will define a function to create a TFX pipeline. 

- 위에서 언급한 Evaluator 구성 요소 외에도 Resolver라는 노드를 하나 더 추가해야 한다.

- To check a new model is getting better than previous model, we need to compare it against a previous published model, called baseline.

- ML Metadata(MLMD) tracks all previous artifacts of the pipeline and Resolver can find what was the latest blessed model(;a model passed Evaluator successfully) from MLMD using a strategy class called LatestBlessedModelStrategy.

import tensorflow_model_analysis as tfma
def _create_pipeline(pipeline_name: str, pipeline_root: str, data_root: str,
                     module_file: str, serving_model_dir: str,
                     metadata_path: str) -> tfx.dsl.Pipeline:
  """Creates a three component penguin pipeline with TFX."""
  # Brings data into the pipeline.
  example_gen = tfx.components.CsvExampleGen(input_base=data_root)

  # Uses user-provided Python function that trains a model.
  trainer = tfx.components.Trainer(module_file=module_file,
                                   examples=example_gen.outputs['examples'],
                                   train_args=tfx.proto.TrainArgs(num_steps=100),
                                   eval_args=tfx.proto.EvalArgs(num_steps=5))

  # NEW: Get the latest blessed model for Evaluator.
  model_resolver = tfx.dsl.Resolver(strategy_class=tfx.dsl.experimental.LatestBlessedModelStrategy,
                                    model=tfx.dsl.Channel(type=tfx.types.standard_artifacts.Model),
                                    model_blessing=tfx.dsl.Channel(type=tfx.types.standard_artifacts.ModelBlessing)).with_id(
                                                                    'latest_blessed_model_resolver')

  # NEW: Uses TFMA to compute evaluation statistics over features of a model and
  #   perform quality validation of a candidate model (compared to a baseline).
  eval_config = tfma.EvalConfig(model_specs=[tfma.ModelSpec(label_key='species')], 
                                slicing_specs=[tfma.SlicingSpec(), tfma.SlicingSpec(feature_keys=['species'])],
                                metrics_specs=[tfma.MetricsSpec(per_slice_thresholds={'sparse_categorical_accuracy':
                                                                                      tfma.PerSliceMetricThresholds(thresholds=[
                      tfma.PerSliceMetricThreshold(
                          slicing_specs=[tfma.SlicingSpec()],
                          threshold=tfma.MetricThreshold(
                              value_threshold=tfma.GenericValueThreshold(
                                   lower_bound={'value': 0.6}),
                              # Change threshold will be ignored if there is no
                              # baseline model resolved from MLMD (first run).
                              change_threshold=tfma.GenericChangeThreshold(
                                  direction=tfma.MetricDirection.HIGHER_IS_BETTER,
                                  absolute={'value': -1e-10}))
                       )]),
          })],
      )
  evaluator = tfx.components.Evaluator(
      examples=example_gen.outputs['examples'],
      model=trainer.outputs['model'],
      baseline_model=model_resolver.outputs['model'],
      eval_config=eval_config)

  # Checks whether the model passed the validation steps and pushes the model
  # to a file destination if check passed.
  pusher = tfx.components.Pusher(
      model=trainer.outputs['model'],
      model_blessing=evaluator.outputs['blessing'], # Pass an evaluation result.
      push_destination=tfx.proto.PushDestination(
          filesystem=tfx.proto.PushDestination.Filesystem(
              base_directory=serving_model_dir)))

  components = [example_gen, trainer, model_resolver, evaluator, pusher]
  return tfx.dsl.Pipeline(pipeline_name=pipeline_name,
                          pipeline_root=pipeline_root,
                          metadata_connection_config=tfx.orchestration.metadata
                                                          .sqlite_metadata_connection_config(metadata_path),
                          components=components)

2. Get analysis result from output artifacts

we will define some utility functions to search for the output artifacts that were just produced.

from ml_metadata.proto import metadata_store_pb2
from tfx.orchestration.portable.mlmd import execution_lib

def get_latest_artifacts(metadata, pipeline_name, component_id):
  context = metadata.store.get_context_by_type_and_name('node', f'{pipeline_name}.{component_id}')
  executions = metadata.store.get_executions_by_context(context.id)
  latest_execution = max(executions,
                         key=lambda e:e.last_update_time_since_epoch)
  return execution_lib.get_output_artifacts(metadata, latest_execution.id)

We can find the latest execution(최신의 실행) of the Evaluator component and get output artifacts of it.

from tfx.orchestration.metadata import Metadata
from tfx.types import standard_component_specs

metadata_connection_config = tfx.orchestration.metadata.sqlite_metadata_connection_config(METADATA_PATH)

with Metadata(metadata_connection_config) as metadata_handler:
  evaluator_output = get_latest_artifacts(metadata_handler, PIPELINE_NAME,'Evaluator')
  eval_artifact = evaluator_output[standard_component_specs.EVALUATION_KEY][0]

Evaluator always returns one evaluation artifact, and we can visualize it using TensorFlow Model Analysis library. 

For example, following code will render the accuracy metrics for each penguin species. (각 펭귄 종에 대한 정확도 측정항목)

import tensorflow_model_analysis as tfma

eval_result = tfma.load_eval_result(eval_artifact.uri)
tfma.view.render_slicing_metrics(eval_result, slicing_column='species')

Slices Overview - sparse_categorical_accuracy

 

Metrics Histogram - Loss

'sparse_categorical_accuracy'를 선택하면 종별 정확도 값을 확인할 수 있다.

You might want to add more slices and check whether your model is good for all distribution and if there is any possible bias.

 

[Practice Code]

https://github.com/juooo1117/practice_AI_Learning/blob/main/MLOps/ModelAnalysis_TFXPipeline.ipynb