Upload recsys.py with huggingface_hub

recsys.py

@@ -0,0 +1,270 @@
+"""
+TripAlchemy - Recommendation Engine (embedding-agnostic core)
+================================================================
+
+This module is intentionally decoupled from any specific embedding model.
+It takes a precomputed embedding matrix (produced by embeddings_compare_hf.py,
+using the winning HF sentence-embedding model) and the experiences list, and
+exposes a hybrid recommender:
+
+    final_score = ALPHA * text_embedding_similarity + (1 - ALPHA) * category_slider_match
+
+Two ways a "query" reaches the text-embedding side:
+  1. Free-text query from the user ("cozy jazz bar with cocktails") -> embedded
+     with the same model used to embed the dataset -> cosine similarity.
+  2. No free text: sliders alone are converted into a synthetic natural-language
+     preference sentence (e.g. "An experience that is mostly culinary and
+     nightlife, with some art_culture") which is embedded instead. This keeps
+     the embedding step meaningful even in pure-slider mode, per the
+     assignment's "Recommendation with Embeddings" requirement.
+
+The category-slider side is always a straightforward weighted dot product
+against each experience's `category_scores`, so the sliders stay precise
+even if the text-embedding side is fuzzy.
+
+Usage (see bottom __main__ for a runnable sanity check):
+
+    engine = RecommendationEngine.load(
+        experiences_path="../data/experiences.json",
+        embeddings_path="../data/embeddings/winning_model.npz",
+    )
+    results = engine.recommend(
+        category_weights={"culinary": 0.9, "nightlife": 0.6},
+        free_text=None,
+        top_k=5,
+    )
+"""
+
+from __future__ import annotations
+
+import json
+import re
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable, Dict, List, Optional
+
+import numpy as np
+
+CATEGORIES = ["culinary", "historical", "shopping", "nature", "nightlife", "art_culture"]
+
+try:
+    import faiss
+    _HAS_FAISS = True
+except ImportError:
+    _HAS_FAISS = False
+
+
+def _l2_normalize(mat: np.ndarray) -> np.ndarray:
+    norms = np.linalg.norm(mat, axis=1, keepdims=True)
+    norms[norms == 0] = 1e-9
+    return mat / norms
+
+
+def sliders_to_sentence(category_weights: Dict[str, float]) -> str:
+    """Turn a slider dict into a natural-language preference sentence,
+    so pure-slider search still goes through the embedding model."""
+    ranked = sorted(category_weights.items(), key=lambda kv: -kv[1])
+    ranked = [(c, w) for c, w in ranked if w > 0.05]
+    if not ranked:
+        return "A well-rounded, broadly appealing travel experience."
+
+    strong = [c for c, w in ranked if w >= 0.66]
+    medium = [c for c, w in ranked if 0.33 <= w < 0.66]
+    labels = {
+        "culinary": "food and dining",
+        "historical": "history and heritage",
+        "shopping": "markets and shopping",
+        "nature": "nature and outdoors",
+        "nightlife": "nightlife and evening entertainment",
+        "art_culture": "art, museums and culture",
+    }
+    parts = []
+    if strong:
+        parts.append("mostly " + " and ".join(labels[c] for c in strong))
+    if medium:
+        parts.append("with some " + " and ".join(labels[c] for c in medium))
+    return "An experience that is " + ", ".join(parts) + "."
+
+
+@dataclass
+class RecommendationEngine:
+    experiences: List[dict]
+    embeddings: np.ndarray          # (N, D), L2-normalized
+    embed_fn: Optional[Callable[[List[str]], np.ndarray]] = None  # text -> (M, D)
+    model_name: str = "unknown"
+
+    def __post_init__(self):
+        assert len(self.experiences) == self.embeddings.shape[0], (
+            f"experiences ({len(self.experiences)}) and embeddings "
+            f"({self.embeddings.shape[0]}) must be the same length"
+        )
+        self.embeddings = _l2_normalize(self.embeddings.astype("float32"))
+        self._category_matrix = np.array(
+            [[e["category_scores"].get(c, 0.0) for c in CATEGORIES] for e in self.experiences],
+            dtype="float32",
+        )
+        if _HAS_FAISS:
+            self._index = faiss.IndexFlatIP(self.embeddings.shape[1])
+            self._index.add(self.embeddings)
+        else:
+            self._index = None
+
+    # ---- persistence -----------------------------------------------------
+
+    @classmethod
+    def load(cls, experiences_path: str, embeddings_path: str, embed_fn=None, model_name="unknown"):
+        with open(experiences_path, encoding="utf-8") as f:
+            data = json.load(f)
+        experiences = data["experiences"] if isinstance(data, dict) else data
+
+        npz = np.load(embeddings_path, allow_pickle=True)
+        embeddings = npz["embeddings"]
+        ids = list(npz["ids"]) if "ids" in npz else [e["id"] for e in experiences]
+
+        # re-order experiences to match embeddings row order, if ids provided
+        by_id = {e["id"]: e for e in experiences}
+        ordered = [by_id[i] for i in ids if i in by_id]
+        if len(ordered) != len(experiences):
+            missing = len(experiences) - len(ordered)
+            print(f"⚠️  {missing} experiences had no matching embedding row and were dropped")
+
+        return cls(experiences=ordered, embeddings=embeddings, embed_fn=embed_fn, model_name=model_name)
+
+    # ---- core recommendation ----------------------------------------------
+
+    def _text_similarity(self, query_text: str) -> np.ndarray:
+        if self.embed_fn is None:
+            raise RuntimeError(
+                "No embed_fn attached — pass one in (must match the model used to "
+                "build `embeddings`), or call recommend(..., use_text_similarity=False)."
+            )
+        q = self.embed_fn([query_text])
+        q = _l2_normalize(np.asarray(q, dtype="float32"))
+
+        if self._index is not None:
+            # Exact search (IndexFlatIP) over the whole collection via FAISS -
+            # same result as the brute-force dot product, but this is the path
+            # that scales once the dataset grows to 10k+ experiences.
+            n = self.embeddings.shape[0]
+            sims_sorted, idxs = self._index.search(q, n)
+            sims = np.empty(n, dtype="float32")
+            sims[idxs[0]] = sims_sorted[0]
+            return sims
+
+        return (self.embeddings @ q.T).ravel()  # cosine sim since both L2-normalized
+
+    def _category_match(self, category_weights: Dict[str, float]) -> np.ndarray:
+        w = np.array([category_weights.get(c, 0.0) for c in CATEGORIES], dtype="float32")
+        if w.sum() == 0:
+            return np.ones(len(self.experiences), dtype="float32") * 0.5
+        w = w / (np.linalg.norm(w) + 1e-9)
+        cat = self._category_matrix / (np.linalg.norm(self._category_matrix, axis=1, keepdims=True) + 1e-9)
+        return cat @ w
+
+    def recommend(
+        self,
+        category_weights: Optional[Dict[str, float]] = None,
+        free_text: Optional[str] = None,
+        city_id: Optional[str] = None,
+        max_budget_usd: Optional[float] = None,
+        energy: Optional[str] = None,
+        alpha: float = 0.5,
+        top_k: int = 5,
+        use_text_similarity: bool = True,
+        diversify_by_category: bool = True,
+    ) -> List[dict]:
+        """
+        alpha: weight on text-embedding similarity vs. category-slider match.
+               alpha=1.0 -> pure embedding search, alpha=0.0 -> pure slider match.
+        """
+        category_weights = category_weights or {}
+        n = len(self.experiences)
+        text_sim = np.zeros(n, dtype="float32")
+
+        if use_text_similarity:
+            query_text = free_text or sliders_to_sentence(category_weights)
+            text_sim = self._text_similarity(query_text)
+            # normalize to 0..1 for blending
+            text_sim = (text_sim - text_sim.min()) / (text_sim.max() - text_sim.min() + 1e-9)
+
+        cat_match = self._category_match(category_weights)
+        cat_match = (cat_match - cat_match.min()) / (cat_match.max() - cat_match.min() + 1e-9)
+
+        final_score = alpha * text_sim + (1 - alpha) * cat_match
+
+        # hard filters
+        mask = np.ones(n, dtype=bool)
+        if city_id:
+            mask &= np.array([e["city_id"] == city_id for e in self.experiences])
+        if max_budget_usd is not None:
+            mask &= np.array([e["cost_usd"] <= max_budget_usd for e in self.experiences])
+        if energy:
+            mask &= np.array([e.get("energy_required") == energy for e in self.experiences])
+
+        final_score = np.where(mask, final_score, -1.0)
+
+        order = np.argsort(-final_score)
+        results = []
+        seen_categories = set()
+        for idx in order:
+            if final_score[idx] < 0:
+                continue
+            exp = self.experiences[idx]
+            if diversify_by_category and len(results) < top_k:
+                primary = max(exp["category_scores"], key=exp["category_scores"].get)
+                if primary in seen_categories and len(seen_categories) < len(CATEGORIES):
+                    continue  # skip to encourage variety across the top_k
+                seen_categories.add(primary)
+            results.append({**exp, "_match_score": round(float(final_score[idx]), 4)})
+            if len(results) >= top_k:
+                break
+
+        # backfill if diversify filtering left us short
+        if len(results) < top_k:
+            have_ids = {r["id"] for r in results}
+            for idx in order:
+                if len(results) >= top_k:
+                    break
+                if final_score[idx] < 0:
+                    continue
+                exp = self.experiences[idx]
+                if exp["id"] in have_ids:
+                    continue
+                results.append({**exp, "_match_score": round(float(final_score[idx]), 4)})
+
+        return results
+
+
+# =====================================================
+# Sanity check (runs with ANY embedding matrix, real or placeholder)
+# =====================================================
+
+if __name__ == "__main__":
+    import argparse
+
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--experiences", default="../data/experiences.json")
+    parser.add_argument("--embeddings", default="../data/embeddings/winning_model.npz")
+    args = parser.parse_args()
+
+    engine = RecommendationEngine.load(args.experiences, args.embeddings)
+    print(f"Loaded {len(engine.experiences)} experiences, embedding dim={engine.embeddings.shape[1]}")
+
+    print("\n--- Pure slider test: heavy culinary + nightlife ---")
+    for r in engine.recommend(
+        category_weights={"culinary": 0.9, "nightlife": 0.7},
+        use_text_similarity=False,
+        alpha=0.0,
+        top_k=5,
+    ):
+        print(f"  [{r['_match_score']}] {r['title']} ({r['city_name']}) - {r['category_scores']}")
+
+    print("\n--- Budget filter test: nature, max $30 ---")
+    for r in engine.recommend(
+        category_weights={"nature": 1.0},
+        use_text_similarity=False,
+        alpha=0.0,
+        max_budget_usd=30,
+        top_k=5,
+    ):
+        print(f"  [{r['_match_score']}] {r['title']} ({r['city_name']}) - ${r['cost_usd']}")