# backend/app/services/explainer.py import numpy as np from app.core.ml_manager import ml_manager class RecommendationExplainer: """ Decoupled service to generate human-readable explanations for ML recommendations. Can be swapped or upgraded independently of the core recommender. """ @staticmethod def generate_reason(target_index: int, history_indices: list, history_ratings: list, sources: list) -> str: if not history_indices: return "Recommended based on overall popularity." pos_indices = [idx for idx, r in zip(history_indices, history_ratings) if r >= 3.0] if not pos_indices: return "A global trending recommendation." # Route to the appropriate explainer based on the primary source if "Collaborative" in sources: return RecommendationExplainer._explain_ease(target_index, pos_indices) elif "Semantic" in sources: return RecommendationExplainer._explain_semantic(target_index, pos_indices) elif "Deep Two-Tower" in sources: return RecommendationExplainer._explain_tt(target_index, pos_indices) return "Matches your general cinematic aesthetic." @staticmethod def _explain_ease(target_index: int, pos_indices: list) -> str: """Finds the historical items with the highest weight in the EASE B-matrix.""" # B_ease is shape (N, N). We slice the rows of the user's history for the target column. weights = ml_manager.B_ease[pos_indices, target_index] # Get top 2 contributing movies top_local_idx = np.argsort(weights)[-2:][::-1] anchors = [] for loc_idx in top_local_idx: if weights[loc_idx] > 0: # Only count positive correlations try: title = ml_manager.movie_meta.loc[pos_indices[loc_idx]]["title"] anchors.append(str(title)) except KeyError: pass if len(anchors) == 2: return f"Because you enjoyed '{anchors[0]}' and '{anchors[1]}'." elif len(anchors) == 1: return f"Because you enjoyed '{anchors[0]}'." return "People with similar taste profiles loved this." @staticmethod def _explain_semantic(target_index: int, pos_indices: list) -> str: """Finds overlap in Tag Genome (Flavor Fingerprint) or Plot Embeddings.""" if ml_manager.tag_scores is not None: user_agg = ml_manager.tag_scores[pos_indices].mean(axis=0) target_tags = ml_manager.tag_scores[target_index] overlap = user_agg * target_tags top_cols = np.argsort(overlap)[-2:][::-1] tags = [] for col in top_cols: tag_id = ml_manager.tag_col_to_id.get(int(col), "") tag_name = ml_manager.tag_id_to_name.get(str(tag_id), "").strip() # Filter out generic tags (using your existing logic from profile_analytics) if tag_name and tag_name.lower() not in {"drama", "comedy", "action", "thriller"}: tags.append(tag_name) if tags: return f"Matches your taste for {', '.join(tags)}." # Fallback to plot similarity target_emb = ml_manager.plot_embeddings[target_index] hist_embs = ml_manager.plot_embeddings[pos_indices] sims = hist_embs @ target_emb best_idx = pos_indices[np.argmax(sims)] try: best_title = ml_manager.movie_meta.loc[best_idx]["title"] return f"Shares deep thematic and plot elements with '{best_title}'." except KeyError: return "Shares themes with movies you've liked." @staticmethod def _explain_tt(target_index: int, pos_indices: list) -> str: """Finds the historical movie closest to the target in the Two-Tower latent space.""" target_emb = ml_manager.tt_item_corpus[target_index] hist_embs = ml_manager.tt_item_corpus[pos_indices] sims = hist_embs @ target_emb best_idx = pos_indices[np.argmax(sims)] try: best_title = ml_manager.movie_meta.loc[best_idx]["title"] return f"Deep structural match based on your timeline, structurally similar to '{best_title}'." except KeyError: return "Analyzed your full timeline and found a deep structural match."