""" Recommender Agent — Raccomandazione prodotti con LightFM ibrido. Approccio ibrido che combina due segnali: - Content-based: 408 categorie prodotto + 5 fasce di prezzo come item features. Prodotti della stessa categoria hanno embedding piu' vicini. - Collaborative filtering: 131k interazioni user-item (recensioni con rating >= 4). Prodotti apprezzati dagli stessi utenti hanno embedding piu' vicini. LightFM fonde i due segnali in un unico vettore per prodotto tramite get_item_representations(), che moltiplica la matrice features per gli embedding. Funzionamento: 1. Riceve la query dell'utente (es. "consigliami scarpe estive") 2. ANCHOR CONTESTUALE: cerca via FAISS un prodotto attinente alla query 3. Calcola il dot product tra l'anchor e tutti gli altri prodotti (embedding ibridi) 4. Restituisce i 5 prodotti piu' simili con nome, brand e prezzo Se FAISS non trova un prodotto attinente, usa un anchor casuale (fallback). 0 chiamate LLM — tutto il calcolo e' offline (1 query FAISS per l'anchor). Usa il factory pattern: create_nodo_recommender(rec_model, rec_idx_map, metadata_lookup, item_features, product_retriever) → funzione nodo. """ import random import numpy as np from langchain_core.messages import AIMessage def create_nodo_recommender(rec_model, rec_idx_map, metadata_lookup, item_features=None, product_retriever=None, llm=None): """Crea il nodo Recommender Agent con le dipendenze iniettate. Args: rec_model: modello LightFM addestrato rec_idx_map: dizionario {indice_interno: ASIN} metadata_lookup: dizionario {ASIN: {title, brand, price, ...}} item_features: matrice sparsa CSR delle feature (categorie + prezzo), o None product_retriever: retriever FAISS per l'anchor contestuale, o None (quindi anchor random) llm: istanza ChatOpenAI per traduzione IT→EN (opzionale, migliora anchor search) """ # --- Pre-computazioni all'init (eseguite una sola volta) --- # Filtra solo gli item che hanno sia embedding LightFM sia metadati (giusto per precauzione, training già filtrato in realta') indices_with_metadata = [ idx for idx, asin in rec_idx_map.items() if asin in metadata_lookup ] # Mappa inversa per convertire ASIN (da FAISS) → indice interno LightFM (serve per l'anchor) asin_to_idx = {asin: idx for idx, asin in rec_idx_map.items()} # Set per lookup O(1) nell'anchor search per miglior prestazioni di ricerca (lista → set) valid_set = set(indices_with_metadata) # Calcola i vettori item: ibridi (features) o puri (solo collaborative) if rec_model and indices_with_metadata: if item_features is not None: # Ibrido: moltiplica matrice features × embedding → vettore per item _, precomputed_vectors = rec_model.get_item_representations(features=item_features) else: # Solo collaborative: ogni item ha un embedding indipendente precomputed_vectors = rec_model.item_embeddings # Estrai solo i vettori degli item con metadati (sottomatrice) valid_indices = np.array(indices_with_metadata) # lista → array NumPy per indicizzazione avanzata valid_vectors = precomputed_vectors[valid_indices] # sottomatrice: solo righe dei prodotti validi # Mappa indice LightFM → posizione nell'array valid_vectors (es. {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 = {} # --- Funzioni interne --- def find_contextual_anchor(state): """Cerca un anchor contestuale: usa FAISS per trovare un prodotto attinente alla query dell'utente, poi lo converte in posizione nell'array dei vettori validi. Ritorna None se non trova nulla.""" if not product_retriever: return None # Usa la query risolta dal Router (con contesto), fallback all'ultimo messaggio user_query = state.get("resolved_query", state["messages"][-1].content) # Traduce la query IT → EN per migliorare il match FAISS (catalogo in inglese) if llm: try: translation_prompt = f"""Task: Translate search query to English keywords for Amazon. Input: "{user_query}" Output keywords only:""" english_query_msg = llm.invoke(translation_prompt) search_query = english_query_msg.content.strip() print(f" Anchor search: '{user_query}' -> '{search_query}'") except: search_query = user_query else: search_query = user_query try: # FAISS restituisce i prodotti piu' simili alla query tradotta results = product_retriever.invoke(search_query) # Cerca il primo risultato che esiste anche nel modello LightFM for doc in results: asin = doc.metadata.get("asin") if asin and asin in asin_to_idx: idx = asin_to_idx[asin] if idx in valid_set: print(f" Anchor contestuale: {doc.metadata.get('title', asin)[:60]}") return idx_to_pos[idx] except Exception: pass return None # --- Nodo del grafo --- def nodo_recommender(state): print("--- [AGENT RECOMMENDER] Motore LightFM Attivo... ---") if not rec_model: return {"messages": [AIMessage(content="Il sistema di raccomandazione e' offline.")]} if valid_vectors is None or len(indices_with_metadata) == 0: return {"messages": [AIMessage(content="Non ho abbastanza dati per fare raccomandazioni.")]} try: # Cerca anchor contestuale dalla query, fallback a random anchor_pos = find_contextual_anchor(state) if anchor_pos is None: anchor_pos = random.randrange(len(valid_indices)) print(" Anchor: random (nessun contesto trovato)") target_vector = valid_vectors[anchor_pos] # Dot product vettorizzato (prodotto scalare): score di similarita' con tutti gli item all_scores = valid_vectors @ target_vector # Escludi l'anchor dai risultati all_scores[anchor_pos] = -np.inf # Seleziona i top 5 per score (argpartition e' O(n) invece di O(n log n)) top_count = 5 top_pos = np.argpartition(all_scores, -top_count)[-top_count:] # per maggiore performance rispetto a argsort completo top_pos = top_pos[np.argsort(all_scores[top_pos])[::-1]] # Formatta la risposta con i dettagli dei prodotti suggestions_text = "" for i, pos in enumerate(top_pos, 1): idx = valid_indices[pos] # posizione → indice LightFM (es. 3) asin = rec_idx_map[idx] # indice LightFM → ASIN (es. "B00XYZ123") p = metadata_lookup[asin] # ASIN → dizionario metadati {title, brand, price, ...} price_str = f" - Prezzo: ${p['price']}" if p.get('price') and p['price'] != 'N/A' else "" brand_str = f" ({p['brand']})" if p.get('brand') and p['brand'] != 'Generic' else "" suggestions_text += f" {i}. {p['title']}{brand_str}{price_str}\n" msg = f"Ecco alcuni prodotti consigliati per te:\n\n{suggestions_text}" except Exception as e: msg = f"Errore nel calcolo dei consigli: {e}" return {"messages": [AIMessage(content=msg)]} return nodo_recommender