yash-b18
feat: explainability — SHAP for LightGBM and metadata-grounded language explanations for Two-Tower
fbc171a | """ | |
| 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}.' | |