Spaces:
Sleeping
Sleeping
File size: 2,940 Bytes
597500e | 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 | """
Content-Based Engine Module
---------------------------
Core recommendation pipeline leveraging a Deep Autoencoder for dimensionality
reduction and K-Means for latent space routing.
Refactored for low cyclomatic complexity using dictionary mapping.
"""
import numpy as np
import math
from sklearn.metrics.pairwise import cosine_similarity
from typing import List, Dict, Any
def get_priority_val(user_input: Dict[str, Any],
priority_list: List[str],
mapping_dicts: Dict[str, Dict[str, float]]) -> float:
"""
Resolves feature values heuristically based on a priority list.
Replaces nested if-statements with flat dictionary lookups to reduce CC.
Args:
user_input (Dict[str, Any]): Raw user preferences.
priority_list (List[str]): Keys to check in order of importance.
mapping_dicts (Dict[str, Dict[str, float]]): Mapping definitions.
Returns:
float: Quantitative feature value (defaults to 0.5).
"""
for source_key in priority_list:
user_choice = user_input.get(source_key)
if user_choice:
val = mapping_dicts.get(source_key, {}).get(user_choice)
if val is not None:
return val
return 0.5
def run_recommendation_pipeline(full_vector_raw: List[float],
valid_indices: List[int],
artifacts: Dict[str, Any]) -> List[str]:
"""
Executes the core inference pipeline: Autoencoder -> K-Means -> Cosine Similarity.
Args:
full_vector_raw (List[float]): Unmasked numerical user vector.
valid_indices (List[int]): Active indices for masked similarity.
artifacts (Dict[str, Any]): Loaded ML models and metadata.
Returns:
List[str]: Top recommended shoe IDs.
"""
df_data = artifacts['df_data']
encoder_model = artifacts['encoder_model']
kmeans_model = artifacts['kmeans_model']
X_combined_data = artifacts['X_combined_data']
# 1. Latent Space Projection
full_vector_np = np.array([full_vector_raw])
user_latent = encoder_model.predict(full_vector_np, verbose=0)
# 2. Cluster Routing
distances = kmeans_model.transform(user_latent)[0]
n_select = math.ceil(kmeans_model.n_clusters / 3)
closest_clusters = np.argsort(distances)[:n_select]
# 3. Filtering & Scoring
candidates = df_data[df_data['cluster'].isin(closest_clusters)].copy()
if candidates.empty:
return []
user_vec_masked = full_vector_np[:, valid_indices]
cand_vecs_masked = X_combined_data[candidates.index][:, valid_indices]
if np.all(user_vec_masked == 0):
candidates['match_score'] = 0.0
else:
candidates['match_score'] = cosine_similarity(user_vec_masked, cand_vecs_masked)[0]
return candidates.sort_values('match_score', ascending=False).head(10)['shoe_id'].tolist() |