Interpretabilidade - Explicador SHAP tabular

Utilize o Kernel SHAP (SHapley Additive exPlanations) para explicar um modelo tabular de classificação. Kernel SHAP é um método independente do modelo que estima a contribuição de cada característica para a previsão de um modelo. Treina um modelo de regressão logística no conjunto de dados Adult Census Income e depois usa o transformador SynapseML TabularSHAP para calcular explicações ao nível das características.

Pré-requisitos

• Obtenha uma assinatura do Microsoft Fabric. Ou inscreva-se para uma avaliação gratuita do Microsoft Fabric.

• Faça login no Microsoft Fabric.

• Mude para o Fabric usando o seletor de experiência no canto inferior esquerdo da sua página inicial.

  

• Crie um novo caderno no seu espaço de trabalho e anexe-o a uma casa no lago. Para mais informações, consulte Criar um caderno.

SynapseML, PySpark, pandas e plotly estão pré-instalados em ambientes de notebooks Fabric. Não é necessária instalação de pacotes adicionais.

Importar pacotes e definir UDFs auxiliares

No seu caderno Fabric, cole o código seguinte numa célula e execute-o. Esta etapa importa as bibliotecas necessárias e define duas funções definidas pelo utilizador (UDFs) para extração posterior dos elementos vetoriais.

import pyspark
from synapse.ml.explainers import TabularSHAP
from pyspark.ml import Pipeline
from pyspark.ml.classification import LogisticRegression
from pyspark.ml.feature import StringIndexer, OneHotEncoder, VectorAssembler
from pyspark.sql.types import FloatType, ArrayType
from pyspark.sql.functions import col, lit, rand, broadcast, udf
import pandas as pd

vec_access = udf(lambda v, i: float(v[i]), FloatType())
vec2array = udf(lambda vec: vec.toArray().tolist(), ArrayType(FloatType()))

Verificar: Execute o código seguinte numa nova célula. Deverás ver o resultado TabularSHAP imported successfully.

print("TabularSHAP imported successfully")
print(f"PySpark version: {pyspark.__version__}")

Carregar dados e treinar um modelo de classificação

Carregue o conjunto de dados Adult Census Income a partir do Armazenamento de Blobs do Azure, indexe a etiqueta alvo e treine um pipeline de regressão logística.

df = spark.read.parquet(
    "wasbs://publicwasb@mmlspark.blob.core.windows.net/AdultCensusIncome.parquet"
)

labelIndexer = StringIndexer(
    inputCol="income", outputCol="label", stringOrderType="alphabetAsc"
).fit(df)
print("Label index assignment: " + str(set(zip(labelIndexer.labels, [0, 1]))))

training = labelIndexer.transform(df).cache()

categorical_features = [
    "workclass",
    "education",
    "marital-status",
    "occupation",
    "relationship",
    "race",
    "sex",
    "native-country",
]
categorical_features_idx = [feat + "_idx" for feat in categorical_features]
categorical_features_enc = [feat + "_enc" for feat in categorical_features]
numeric_features = [
    "age",
    "education-num",
    "capital-gain",
    "capital-loss",
    "hours-per-week",
]

strIndexer = StringIndexer(
    inputCols=categorical_features, outputCols=categorical_features_idx
)
onehotEnc = OneHotEncoder(
    inputCols=categorical_features_idx, outputCols=categorical_features_enc
)
vectAssem = VectorAssembler(
    inputCols=categorical_features_enc + numeric_features, outputCol="features"
)
lr = LogisticRegression(featuresCol="features", labelCol="label", weightCol="fnlwgt")
pipeline = Pipeline(stages=[strIndexer, onehotEnc, vectAssem, lr])
model = pipeline.fit(training)

Verificar: Executar a célula seguinte. Deverás ver o número de linhas dos dados de treino e a confirmação das etapas do fluxo de processamento.

print(f"Training rows: {training.count()}")
print(f"Pipeline stages: {[type(s).__name__ for s in model.stages]}")
assert training.count() > 30000, "Dataset should contain over 30,000 rows"
print("Model trained successfully")

# Expected output:
#Training rows: 32561
#Pipeline stages: ['StringIndexerModel', 'OneHotEncoderModel', #'VectorAssembler', 'LogisticRegressionModel']
#Model trained successfully

Selecionar observações para explicar

Selecione aleatoriamente cinco observações dos dados de treino pontuados. Estas observações são as instâncias para as quais geras explicações SHAP.

explain_instances = (
    model.transform(training).orderBy(rand()).limit(5).repartition(200).cache()
)
display(explain_instances)

Verificar: Confirmar o tamanho da amostra.

count = explain_instances.count()
print(f"Explain instances: {count}")
assert count == 5, f"Expected 5 rows, got {count}"
print("Sample selected successfully")

Configurar e executar o TabularSHAP

Crie uma TabularSHAP explicação e aplique-a às observações selecionadas. Os principais parâmetros são:

shap = TabularSHAP(
    inputCols=categorical_features + numeric_features,
    outputCol="shapValues",
    numSamples=5000,
    model=model,
    targetCol="probability",
    targetClasses=[1],
    backgroundData=broadcast(training.orderBy(rand()).limit(100).cache()),
)

shap_df = shap.transform(explain_instances)

Verificar: Verifique se a coluna de saída SHAP existe.

assert "shapValues" in shap_df.columns, "shapValues column missing"
print(f"SHAP output columns: {shap_df.columns}")
print("TabularSHAP transform completed")

Extrair os valores SHAP

Extrair as probabilidades da classe 1 e os valores SHAP do DataFrame de resultados. Para cada observação, o vetor de valores SHAP começa com o valor base (saída média do conjunto de dados de fundo), seguido de um valor por característica.

shaps = (
    shap_df.withColumn("probability", vec_access(col("probability"), lit(1)))
    .withColumn("shapValues", vec2array(col("shapValues").getItem(0)))
    .select(
        ["shapValues", "probability", "label"] + categorical_features + numeric_features
    )
)

shaps_local = shaps.toPandas()
shaps_local.sort_values("probability", ascending=False, inplace=True, ignore_index=True)
pd.set_option("display.max_colwidth", None)
display(shaps_local)

Verificar: Confirmar a estrutura do DataFrame do pandas.

expected_cols = len(categorical_features) + len(numeric_features) + 3
print(f"DataFrame shape: {shaps_local.shape}")
print(f"Expected columns: {expected_cols}, Actual: {shaps_local.shape[1]}")
assert shaps_local.shape == (5, expected_cols), f"Unexpected shape: {shaps_local.shape}"
print("SHAP values extracted successfully")

Visualizar valores SHAP

Crie um gráfico de barras para cada observação que mostre como cada característica contribui para a probabilidade prevista.

from plotly.subplots import make_subplots
import plotly.graph_objects as go

features = categorical_features + numeric_features
features_with_base = ["Base"] + features

rows = shaps_local.shape[0]

fig = make_subplots(
    rows=rows,
    cols=1,
    subplot_titles="Probability: "
    + shaps_local["probability"].apply("{:.2%}".format)
    + "; Label: "
    + shaps_local["label"].astype(str),
)

for index, row in shaps_local.iterrows():
    feature_values = [0] + [row[feature] for feature in features]
    shap_values = row["shapValues"]
    list_of_tuples = list(zip(features_with_base, feature_values, shap_values))
    shap_pdf = pd.DataFrame(list_of_tuples, columns=["name", "value", "shap"])
    fig.add_trace(
        go.Bar(
            x=shap_pdf["name"],
            y=shap_pdf["shap"],
            hovertext="value: " + shap_pdf["value"].astype(str),
        ),
        row=index + 1,
        col=1,
    )

fig.update_yaxes(range=[-1, 1], fixedrange=True, zerolinecolor="black")
fig.update_xaxes(type="category", tickangle=45, fixedrange=True)
fig.update_layout(height=400 * rows, title_text="SHAP explanations")
fig.show()

Verificar: Confirmar que o objeto de gráfico foi criado.

print(f"Figure traces: {len(fig.data)}")
print(f"Figure height: {fig.layout.height}px")
assert len(fig.data) == 5, f"Expected 5 traces, got {len(fig.data)}"
print("Visualization created successfully")

Interpretar os resultados

Cada subplot representa uma observação. Os bares mostram:

• Base: A média de saída do modelo em todo o conjunto de dados de fundo (probabilidade base).
• Valores SHAP positivos: Características que impulsionam a previsão para a classe 1 (rendimento superior a 50K).
• Valores SHAP negativos: Características que empurram a previsão para a classe 0 (rendimento inferior ou igual a 50K).

A soma do valor base e de todos os valores SHAP das características é igual à probabilidade prevista pelo modelo para essa observação.

Troubleshooting

Limpeza

Se carregaste o conjunto de dados para um lakehouse neste tutorial, remove-o para libertar espaço de armazenamento:

# Remove cached DataFrames from memory
training.unpersist()
explain_instances.unpersist()
print("Cached DataFrames released")

Conteúdo relacionado

• Como usar SynapseML para deteção de anomalias multivariadas
• Como construir um motor de busca com o SynapseML