""" Recommender Agent — Product recommendation with hybrid LightFM. The Recommender activates when the user clicks on a product in the showcase after a search. The selected product becomes the anchor and LightFM generates the most similar products via dot product on the hybrid embeddings. Hybrid approach that combines two signals: - Content-based: 519 product categories + 5 price bands as item features. Products of the same category have closer embeddings. - Collaborative filtering: 485k user-item interactions (reviews with rating >= 4). Products liked by the same users have closer embeddings. LightFM merges the two signals into a single vector per product via get_item_representations(), which multiplies the features matrix by the embeddings. How it works: 1. Receives the ASIN of the product selected by the user (recommend_asin in the state) 2. Converts ASIN → position in the array of LightFM vectors 3. Vectorized dot product between the anchor and all products → top 8 4. Returns the top-8 most similar products with name, brand and price 0 LLM calls — all computation is offline (hybrid embeddings + dot product). Uses the factory pattern: create_recommender_node(rec_model, rec_idx_map, metadata_lookup, item_features) → node function. """ import random import numpy as np from langchain_core.messages import AIMessage from utils.config import RECOMMENDER_K from utils.i18n import t def create_recommender_node(rec_model, rec_idx_map, metadata_lookup, item_features=None): """Creates the Recommender Agent node with injected dependencies. Args: rec_model: trained LightFM model rec_idx_map: dictionary {internal_index: ASIN} metadata_lookup: dictionary {ASIN: {title, brand, price, ...}} item_features: sparse CSR matrix of features (categories + price), or None """ # --- Pre-computations at init (run only once) --- # Filter only the items that have both a LightFM embedding and metadata (just as a precaution, training is actually already filtered) indices_with_metadata = [ idx for idx, asin in rec_idx_map.items() if asin in metadata_lookup ] # Inverse map to convert ASIN → internal LightFM index asin_to_idx = {asin: idx for idx, asin in rec_idx_map.items()} # Set for O(1) lookup valid_set = set(indices_with_metadata) # Compute the item vectors: hybrid (features) or pure (collaborative only) if rec_model and indices_with_metadata: if item_features is not None: # Hybrid: multiply features matrix × embeddings → vector per item _, precomputed_vectors = rec_model.get_item_representations(features=item_features) else: # Collaborative only: each item has an independent embedding precomputed_vectors = rec_model.item_embeddings # Extract only the vectors of the items with metadata (submatrix) valid_indices = np.array(indices_with_metadata) # list → NumPy array for advanced indexing valid_vectors = precomputed_vectors[valid_indices] # submatrix: only rows of valid products # Map LightFM index → position in the valid_vectors array (e.g. {idx_lightfm: 0, ...}) idx_to_pos = {idx: pos for pos, idx in enumerate(indices_with_metadata)} else: valid_indices = None valid_vectors = None idx_to_pos = {} # --- Graph node --- def recommender_node(state): print("--- [RECOMMENDER AGENT] LightFM engine active... ---") if not rec_model: return {"messages": [AIMessage(content=t("recommender_offline"))]} if valid_vectors is None or len(indices_with_metadata) == 0: return {"messages": [AIMessage(content=t("recommender_no_data"))]} try: # Receive the ASIN selected by the user (from the click on the card) anchor_asin = state.get("recommend_asin") if anchor_asin and anchor_asin in asin_to_idx: # Direct anchor: the user clicked on a specific product idx = asin_to_idx[anchor_asin] if idx in valid_set: anchor_pos = idx_to_pos[idx] anchor_title = metadata_lookup.get(anchor_asin, {}).get("title", anchor_asin) print(f" Selected anchor: {anchor_title[:60]}") else: # ASIN exists but not in the LightFM model (rare) print(f" ASIN {anchor_asin} not in the LightFM model, using a random anchor") anchor_pos = random.randrange(len(valid_indices)) anchor_asin = rec_idx_map[valid_indices[anchor_pos]] else: # Fallback: no ASIN specified (should not happen in the normal flow) anchor_pos = random.randrange(len(valid_indices)) anchor_asin = rec_idx_map[valid_indices[anchor_pos]] print(" Anchor: random (no ASIN specified)") target_vector = valid_vectors[anchor_pos] # Vectorized dot product (scalar product): similarity score with all items all_scores = valid_vectors @ target_vector # Exclude the anchor from the results all_scores[anchor_pos] = -np.inf # Select the top-k by score (argpartition is O(n) instead of O(n log n)) top_count = RECOMMENDER_K top_pos = np.argpartition(all_scores, -top_count)[-top_count:] top_pos = top_pos[np.argsort(all_scores[top_pos])[::-1]] # Build structured product list (shown visually in the cards) products = [] for pos in top_pos: idx = valid_indices[pos] # position → LightFM index (e.g. 3) asin = rec_idx_map[idx] # LightFM index → ASIN (e.g. "B00XYZ123") p = metadata_lookup[asin] # ASIN → metadata dictionary {title, brand, price, ...} brand = p.get('brand', '') price = p.get('price', '') if brand == 'Generic': brand = '' if price == 'N/A': price = '' title = p.get('title', asin) products.append({ "asin": asin, "title": title, "brand": brand, "price": price, "similarity": None, }) print(f" -> {title[:60]}") # Build anchor info for the frontend anchor_meta = metadata_lookup.get(anchor_asin, {}) anchor_title = anchor_meta.get("title", anchor_asin) anchor_info = {"asin": anchor_asin, "title": anchor_title} msg = t("recommender_results") except Exception as e: msg = t("recommender_error", error=e) products = [] anchor_info = None return {"messages": [AIMessage(content=msg)], "products": products, "anchor": anchor_info} return recommender_node