Spaces:
Running
Running
| # 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. | |
| """ | |
| 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." | |
| 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." | |
| 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." | |
| 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." | |