| """ |
| Inference orchestrator for the API layer. |
| |
| Takes a ModelRegistry and handles recommendation logic for all three models, |
| including explanation generation. This is the single point of contact between |
| the FastAPI routes and the ML models. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import sys |
| from pathlib import Path |
| from typing import Optional |
|
|
| import numpy as np |
| import pandas as pd |
|
|
| |
| sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent)) |
|
|
| from src.api.schemas import RecommendedItem |
| from src.utils.logging import get_logger |
|
|
| logger = get_logger(__name__) |
|
|
|
|
| def _build_item_response( |
| item_idx: int, |
| score: float, |
| explanation: str, |
| item_features: pd.DataFrame, |
| ) -> RecommendedItem: |
| """Build a RecommendedItem response object for a given item_idx.""" |
| row = item_features[item_features["item_idx"] == item_idx] |
| if row.empty: |
| return RecommendedItem( |
| item_idx=item_idx, |
| title=f"Item #{item_idx}", |
| category="Unknown", |
| brand="Unknown", |
| price=None, |
| avg_rating=None, |
| num_ratings=None, |
| score=round(float(score), 4), |
| explanation=explanation, |
| ) |
| r = row.iloc[0] |
| title_val = r.get("title") |
| category_val = r.get("category") |
| brand_val = r.get("brand") |
| return RecommendedItem( |
| item_idx=item_idx, |
| title=str(title_val) if pd.notna(title_val) and str(title_val) != "nan" else f"Item #{item_idx}", |
| category=str(category_val) if pd.notna(category_val) and str(category_val) != "nan" else "Unknown", |
| brand=str(brand_val) if pd.notna(brand_val) and str(brand_val) != "nan" else "Unknown", |
| price=float(r["price"]) if pd.notna(r.get("price")) else None, |
| avg_rating=round(float(r["item_avg_rating"]), 2) if "item_avg_rating" in r and pd.notna(r.get("item_avg_rating")) else None, |
| num_ratings=int(r["item_num_ratings"]) if "item_num_ratings" in r and pd.notna(r.get("item_num_ratings")) else None, |
| score=round(float(score), 4), |
| explanation=explanation, |
| ) |
|
|
|
|
| class InferenceOrchestrator: |
| """ |
| Routes recommendation requests to the appropriate model and assembles responses. |
| |
| Args: |
| registry: Loaded ModelRegistry with all model artifacts. |
| cfg: Full Config object. |
| """ |
|
|
| def __init__(self, registry, cfg) -> None: |
| self.registry = registry |
| self.cfg = cfg |
| |
| if registry.item_features is not None: |
| df = registry.item_features |
| untitled_mask = df["title"].isna() | (df["title"].astype(str).str.strip() == "") | (df["title"].astype(str) == "nan") |
| self._untitled_items: set[int] = set(df.loc[untitled_mask, "item_idx"].tolist()) |
| else: |
| self._untitled_items = set() |
| logger.info("Untitled items excluded from recommendations: %d", len(self._untitled_items)) |
|
|
| def recommend_naive( |
| self, |
| user_idx: Optional[int], |
| top_k: int = 10, |
| ) -> list[RecommendedItem]: |
| """ |
| Recommend using the global popularity model. |
| |
| Falls back to top-K globally popular items if user_idx is unknown. |
| """ |
| reg = self.registry |
| if not reg.naive_available or reg.naive_model is None: |
| return [] |
|
|
| uid = user_idx if user_idx is not None else -1 |
|
|
| if uid in (reg.naive_model.user_seen or {}): |
| recs = reg.naive_model.recommend(uid, top_k=top_k + len(self._untitled_items)) |
| recs = [(iid, s) for iid, s in recs if iid not in self._untitled_items][:top_k] |
| else: |
| |
| pop_titled = reg.naive_model.popularity[ |
| ~reg.naive_model.popularity.index.isin(self._untitled_items) |
| ] |
| recs = list(pop_titled.head(top_k).items()) |
| recs = [(int(iid), float(score)) for iid, score in recs] |
|
|
| items = [] |
| for item_idx, score in recs: |
| explanation = "Recommended based on overall popularity — this item is frequently purchased." |
| items.append(_build_item_response(item_idx, score, explanation, reg.item_features)) |
| return items |
|
|
| def recommend_classical( |
| self, |
| user_idx: Optional[int], |
| top_k: int = 10, |
| ) -> list[RecommendedItem]: |
| """Recommend using the LightGBM reranker.""" |
| from src.features.builder import build_feature_matrix |
| from src.features.text_features import build_user_text_profiles |
|
|
| reg = self.registry |
| if not reg.classical_available or reg.lgbm_model is None: |
| return [] |
|
|
| |
| if reg.naive_available and user_idx is not None: |
| raw_candidates = reg.naive_model.recommend( |
| user_idx, top_k=self.cfg.classical.candidate_pool_size + len(self._untitled_items) |
| ) |
| candidates = [(iid, s) for iid, s in raw_candidates if iid not in self._untitled_items][:self.cfg.classical.candidate_pool_size] |
| else: |
| pop = reg.naive_model.popularity if reg.naive_available else None |
| if pop is None: |
| return [] |
| pop_titled = pop[~pop.index.isin(self._untitled_items)] |
| candidates = [(int(iid), float(s)) for iid, s in pop_titled.head(self.cfg.classical.candidate_pool_size).items()] |
|
|
| if not candidates: |
| return [] |
|
|
| candidate_items = [iid for iid, _ in candidates] |
| rows = pd.DataFrame({ |
| "user_idx": user_idx if user_idx is not None else 0, |
| "item_idx": candidate_items, |
| "label": 0, |
| }) |
|
|
| user_profiles = pd.DataFrame({"user_idx": [], "user_text_profile": []}) |
| vectorizer = reg.feature_artifacts.get("vectorizer") |
|
|
| if vectorizer is None: |
| return [] |
|
|
| feat, _ = build_feature_matrix( |
| rows, |
| reg.user_features if reg.user_features is not None else pd.DataFrame(), |
| reg.item_features if reg.item_features is not None else pd.DataFrame(), |
| user_profiles, |
| vectorizer, |
| feature_set="id_metadata_text_history", |
| ) |
|
|
| scores = reg.lgbm_model.predict_proba(feat) |
| |
| item_to_feat_idx = {iid: i for i, iid in enumerate(candidate_items)} |
| ranked = sorted(zip(candidate_items, scores.tolist()), key=lambda x: x[1], reverse=True)[:top_k] |
|
|
| items = [] |
| for item_idx, score in ranked: |
| try: |
| from src.explainability.shap_explainer import SHAPExplainer |
| explainer = SHAPExplainer(reg.lgbm_model) |
| feat_idx = item_to_feat_idx.get(item_idx, 0) |
| row = feat.iloc[[feat_idx]] |
| title_col = reg.item_features[reg.item_features["item_idx"] == item_idx]["title"].values |
| title = str(title_col[0]) if len(title_col) > 0 and pd.notna(title_col[0]) else f"Item #{item_idx}" |
| explanation = explainer.explain_as_text(row, title) |
| except Exception: |
| explanation = "Recommended based on your interaction patterns and item features." |
| items.append(_build_item_response(item_idx, score, explanation, reg.item_features)) |
| return items |
|
|
| def recommend_deep( |
| self, |
| user_idx: Optional[int], |
| top_k: int = 10, |
| liked_items: list[int] | None = None, |
| ) -> list[RecommendedItem]: |
| """Recommend using the Two-Tower neural model.""" |
| reg = self.registry |
| if not reg.deep_available or reg.deep_trainer is None: |
| return [] |
|
|
| seen = reg.user_seen.get(user_idx, set()) if user_idx is not None else set() |
|
|
| if user_idx is None or user_idx >= reg.n_users: |
| |
| if liked_items and reg.item_embeddings is not None: |
| valid = [i for i in liked_items if 0 <= i < len(reg.item_embeddings)] |
| if valid: |
| proxy = reg.item_embeddings[valid].mean(axis=0) |
| norm = np.linalg.norm(proxy) |
| if norm > 0: |
| proxy = proxy / norm |
| scores = reg.item_embeddings @ proxy |
| exclude = set(liked_items) | self._untitled_items |
| excl_arr = np.array([i for i in exclude if 0 <= i < len(scores)], dtype=np.int64) |
| if len(excl_arr): |
| scores[excl_arr] = -np.inf |
| top_indices = np.argpartition(scores, -top_k)[-top_k:] |
| top_indices = top_indices[np.argsort(scores[top_indices])[::-1]] |
| recs = [(int(i), float(scores[i])) for i in top_indices] |
| items = [] |
| for item_idx, score in recs: |
| if reg.language_explainer: |
| explanation = reg.language_explainer.explain_from_items( |
| liked_item_idxs=valid, |
| item_idx=item_idx, |
| score=score, |
| ) |
| else: |
| explanation = "Recommended based on the items you selected." |
| items.append(_build_item_response(item_idx, score, explanation, reg.item_features)) |
| return items |
| |
| return self.recommend_naive(user_idx=None, top_k=top_k) |
|
|
| recs = reg.deep_trainer.recommend( |
| user_idx, |
| reg.item_embeddings, |
| top_k=top_k, |
| exclude_items=seen | self._untitled_items, |
| ) |
|
|
| items = [] |
| for item_idx, score in recs: |
| if reg.language_explainer: |
| explanation = reg.language_explainer.explain(user_idx, item_idx, score) |
| else: |
| explanation = "Recommended based on your interaction history." |
| items.append(_build_item_response(item_idx, score, explanation, reg.item_features)) |
| return items |
|
|
| def recommend( |
| self, |
| model: str, |
| user_idx: Optional[int], |
| top_k: int = 10, |
| liked_items: list[int] | None = None, |
| ) -> list[RecommendedItem]: |
| """Route recommendation to the specified model.""" |
| if model == "naive": |
| return self.recommend_naive(user_idx, top_k) |
| elif model == "classical": |
| return self.recommend_classical(user_idx, top_k) |
| elif model == "deep": |
| return self.recommend_deep(user_idx, top_k, liked_items=liked_items) |
| else: |
| raise ValueError(f"Unknown model: {model!r}. Choose from naive, classical, deep.") |
|
|
| def get_popular_items(self, n: int = 50) -> list[dict]: |
| """Return the top-N globally popular items for the UI item selector.""" |
| reg = self.registry |
| if not reg.naive_available or reg.naive_model is None or reg.item_features is None: |
| return [] |
| pop_titled = reg.naive_model.popularity[~reg.naive_model.popularity.index.isin(self._untitled_items)] |
| top_items = list(pop_titled.head(n).index) |
| result = [] |
| for iid in top_items: |
| row = reg.item_features[reg.item_features["item_idx"] == iid] |
| if row.empty: |
| continue |
| r = row.iloc[0] |
| t = r.get("title") |
| cat = r.get("category") |
| brd = r.get("brand") |
| result.append({ |
| "item_idx": int(iid), |
| "title": str(t) if pd.notna(t) and str(t) != "nan" else f"Item #{iid}", |
| "category": str(cat) if pd.notna(cat) and str(cat) != "nan" else "Unknown", |
| "brand": str(brd) if pd.notna(brd) and str(brd) != "nan" else "Unknown", |
| "price": float(r["price"]) if pd.notna(r.get("price")) else None, |
| "avg_rating": round(float(r["item_avg_rating"]), 2) if "item_avg_rating" in r else None, |
| "num_ratings": int(r["item_num_ratings"]) if "item_num_ratings" in r else None, |
| }) |
| return result |
|
|
