File size: 4,386 Bytes
fbc171a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 | """
SHAP-based explainability for the LightGBM reranker.
Computes per-prediction SHAP values to identify which features contributed
most to a given recommendation. This is used by the API to generate
feature-level explanations returned to the frontend.
Reference: Lundberg & Lee, "A Unified Approach to Interpreting Model Predictions",
NeurIPS 2017. https://arxiv.org/abs/1705.07874
"""
from __future__ import annotations
from typing import NamedTuple
import numpy as np
import pandas as pd
import shap
from src.models.classical import LightGBMReranker
from src.utils.logging import get_logger
logger = get_logger(__name__)
class FeatureExplanation(NamedTuple):
"""Container for a single feature's SHAP contribution."""
feature: str
shap_value: float
feature_value: float
direction: str # "positive" or "negative"
class SHAPExplainer:
"""
Wraps a LightGBMReranker with SHAP TreeExplainer for per-prediction explanations.
Args:
model: Trained LightGBMReranker.
"""
def __init__(self, model: LightGBMReranker) -> None:
if model.model is None:
raise RuntimeError("Model must be fitted before creating a SHAPExplainer.")
self.model = model
self.explainer = shap.TreeExplainer(model.model)
logger.info("SHAPExplainer initialised for %d features", len(model.feature_names))
def explain_row(
self,
feature_row: pd.DataFrame,
top_n: int = 5,
) -> list[FeatureExplanation]:
"""
Compute SHAP values for a single user-item feature row.
Args:
feature_row: Single-row DataFrame with model feature columns.
top_n: Number of top contributing features to return.
Returns:
List of FeatureExplanation sorted by |shap_value| descending.
"""
X = feature_row[self.model.feature_names].values.astype(np.float32)
shap_values = self.explainer.shap_values(X)
# For binary classification, shap_values may be a list [neg_class, pos_class]
if isinstance(shap_values, list):
sv = shap_values[1][0] # positive class SHAP values for the single row
else:
sv = shap_values[0]
explanations = []
for feat_name, sv_val, feat_val in zip(self.model.feature_names, sv, X[0]):
explanations.append(FeatureExplanation(
feature=feat_name,
shap_value=float(sv_val),
feature_value=float(feat_val),
direction="positive" if sv_val > 0 else "negative",
))
explanations.sort(key=lambda e: abs(e.shap_value), reverse=True)
return explanations[:top_n]
def explain_as_text(
self,
feature_row: pd.DataFrame,
item_title: str,
top_n: int = 3,
) -> str:
"""
Generate a human-readable explanation string for a recommendation.
Args:
feature_row: Single-row feature DataFrame.
item_title: Title of the recommended item.
top_n: Number of features to mention in the explanation.
Returns:
Natural language explanation string.
"""
explanations = self.explain_row(feature_row, top_n=top_n)
# Map feature names to human-readable descriptions
feature_descriptions = {
"item_avg_rating": "high average rating",
"item_num_ratings": "many reviews",
"item_popularity": "high popularity",
"item_positive_rate": "high positive review rate",
"category_match": "matches your preferred category",
"text_similarity": "similar to items you liked",
"user_avg_rating": "your rating history",
"user_num_ratings": "your activity level",
"category_idx": "category relevance",
"brand_idx": "brand relevance",
"price_norm": "price range match",
}
pos_features = [
feature_descriptions.get(e.feature, e.feature)
for e in explanations
if e.direction == "positive"
][:top_n]
if not pos_features:
return f"Recommended based on general popularity patterns."
features_str = ", ".join(pos_features)
return f'"{item_title[:60]}" is recommended because: {features_str}.'
|