File size: 12,475 Bytes
6d9c101 09fcfe1 6d9c101 09fcfe1 6d9c101 2647d97 6d9c101 2647d97 6d9c101 2647d97 6d9c101 2647d97 6d9c101 2647d97 6d9c101 2647d97 6d9c101 2647d97 6d9c101 09fcfe1 6d9c101 09fcfe1 6d9c101 09fcfe1 6d9c101 09fcfe1 6d9c101 d439004 6d9c101 d439004 6d9c101 d439004 de2efef d439004 6d9c101 2647d97 6d9c101 d439004 6d9c101 d439004 6d9c101 2647d97 6d9c101 09fcfe1 6d9c101 09fcfe1 6d9c101 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 | """
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
# Allow running from repo root
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
# Pre-compute set of item_idxs that have no title — exclude from all recommendations
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:
# New user: just return global top-K from titled items only
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 []
# Get candidate pool from naive model (titled items only)
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)
# feat rows are in same order as candidate_items; build index map
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:
# Cold-start: if liked_items provided, use their mean embedding as proxy user vector
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
# No liked items either — fall back to naive
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
|