def create_clarify_xai_job()

in code/workflow/implementations/autopilot/bp_xai_analysis_stage.py [0:0]

• 38 lines of code
• 1 McCabe index (conditional complexity)

def create_clarify_xai_job(event) :
    
    role = event["Input"]["Payload"]["security-config"]["iam_role"]
    data_params = event["Input"]["Payload"]["data-config"]
    ws_params = event["Input"]["Payload"]["workspace-config"]
    model_params = event["Input"]["Payload"]["model-config"]
    automl_params = event["Input"]["Payload"]["automl-config"]
    xai_params = event["Input"]["Payload"]["xai-config"]
    
    session = Session()
    clarify_processor = clarify.SageMakerClarifyProcessor(role=role,
                                                        instance_count=xai_params["instance_count"],
                                                        instance_type=xai_params["instance_type"],
                                                        sagemaker_session=session)
                                                          
    output_uri = "s3://{}/{}/{}".format(ws_params["s3_bucket"],
                                        ws_params["s3_prefix"],
                                        xai_params["output_prefix"])
    
    shap_params = xai_params["shap-config"]
    num_samples = shap_params["num_samples"]
    df = get_samples(automl_params["data_uri"],shap_params["num_samples"])
    
    samples = df.iloc[:,:-1].values.tolist()
    columns = df.columns.to_list()
    
    shap_config = clarify.SHAPConfig(baseline=samples,
                                    num_samples=num_samples,
                                    agg_method=shap_params["agg_method"])
    
    data_config = clarify.DataConfig(s3_data_input_path=automl_params["data_uri"], 
                                    s3_output_path=output_uri,
                                    label=automl_params["target_name"],
                                    headers=columns,
                                    dataset_type='text/csv')
        
    model_config = clarify.ModelConfig(model_name=model_params["model_name"],
                                    instance_type=model_params["instance_type"],
                                    instance_count=model_params["instance_count"],
                                    content_type='text/csv')
                                        
    clarify_processor.run_explainability(job_name=xai_params["job_name"],
                                    data_config=data_config,
                                    model_config=model_config,
                                    explainability_config=shap_config,
                                    wait=False,
                                    logs=False)