import numpy as np
from corerec.api.base_recommender import BaseRecommender
from corerec.api.exceptions import ModelNotFittedError
from typing import Union, List, Dict, Any, Optional
from pathlib import Path
import pickle
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
[docs]
class TFIDFRecommender(BaseRecommender):
"""
TF-IDF based content recommender.
Uses TF-IDF vectorization to find similar items based on text content.
Can recommend items by text query or by item similarity.
Supports two initialization modes:
1. Old API: TFIDFRecommender(feature_matrix) - for backward compatibility
2. New API: TFIDFRecommender() then fit(items, docs)
"""
def __init__(self, feature_matrix=None, name: Optional[str] = None, verbose: bool = False):
super().__init__(name=name or "TFIDFRecommender", trainable=True, verbose=verbose)
self.vectorizer = None
self.tfidf_matrix = None
self.item_to_index = {}
self.index_to_item = {}
self.docs = {}
self.is_fitted = False
self._similarity_matrix = None
self.feature_matrix = None # For old API
# Support old API: if feature_matrix is provided, use it directly
if feature_matrix is not None:
if hasattr(feature_matrix, "toarray"):
self.feature_matrix = feature_matrix.toarray()
else:
self.feature_matrix = np.array(feature_matrix)
self._similarity_matrix = self._compute_similarity_matrix()
self.is_fitted = True
self.num_items = self.feature_matrix.shape[0]
def _compute_similarity_matrix(self):
"""Compute cosine similarity matrix from feature matrix."""
if self.feature_matrix is None:
return None
norm = np.linalg.norm(self.feature_matrix, axis=1, keepdims=True)
norm[norm == 0] = 1 # Avoid division by zero
normalized_features = self.feature_matrix / norm
similarity = np.dot(normalized_features, normalized_features.T)
return similarity
@property
def similarity_matrix(self):
"""Get or compute the similarity matrix."""
if self._similarity_matrix is not None:
return self._similarity_matrix
elif self.feature_matrix is not None:
self._similarity_matrix = self._compute_similarity_matrix()
return self._similarity_matrix
elif self.tfidf_matrix is not None:
# Compute from TF-IDF matrix if available
self.feature_matrix = self.tfidf_matrix.toarray() if hasattr(self.tfidf_matrix, 'toarray') else np.array(self.tfidf_matrix)
self._similarity_matrix = self._compute_similarity_matrix()
return self._similarity_matrix
else:
raise ValueError("No feature matrix available. Fit the model first or provide feature_matrix to constructor.")
[docs]
def fit(self, items: List[Any], docs: Dict[Any, str], **kwargs) -> "TFIDFRecommender":
"""
Fit the TF-IDF model on item documents.
Args:
items: List of item IDs
docs: Dictionary mapping item IDs to text documents
**kwargs: Additional arguments (unused)
Returns:
Self for method chaining
"""
if not items:
raise ValueError("items must be a non-empty list")
if not docs:
raise ValueError("docs must be a non-empty dictionary")
# Build item to index mapping
self.item_to_index = {item: idx for idx, item in enumerate(items)}
self.index_to_item = {idx: item for item, idx in self.item_to_index.items()}
self.docs = docs
# Extract texts in order of items
texts = [docs.get(item, "") for item in items]
# Fit TF-IDF vectorizer
self.vectorizer = TfidfVectorizer(
lowercase=True,
stop_words='english',
max_features=1000,
ngram_range=(1, 2)
)
self.tfidf_matrix = self.vectorizer.fit_transform(texts)
self.is_fitted = True
self.num_items = len(items)
if self.verbose:
print(f"{self.name} fitted on {len(items)} items")
return self
[docs]
def predict(self, user_id: Any, item_id: Any, **kwargs) -> float:
"""
Predict relevance score for a user-item pair.
For TF-IDF, this returns a default score since we don't have user preferences.
Consider using recommend_by_text or recommend instead.
Args:
user_id: User identifier (unused for content-based)
item_id: Item identifier
**kwargs: Additional arguments (unused)
Returns:
Default score of 0.5 (neutral)
"""
if not self.is_fitted:
self._check_fitted()
if item_id not in self.item_to_index:
return 0.0
# Return neutral score since we don't have user preferences
return 0.5
[docs]
def recommend(
self,
user_id_or_indices: Any = None,
top_k: int = 10,
top_n: int = None,
exclude_items: Optional[List[Any]] = None,
**kwargs
) -> Union[List[Any], np.ndarray]:
"""
Generate recommendations.
Supports two modes:
1. Old API: recommend(item_indices, top_n) - returns numpy array of indices
2. New API: recommend(user_id, top_k) - returns list of item IDs
Args:
user_id_or_indices: User ID (new API) or list of item indices (old API)
top_k: Number of recommendations (new API, alias for top_n)
top_n: Number of recommendations (old API, takes precedence)
exclude_items: Items to exclude (new API only)
**kwargs: Additional arguments (unused)
Returns:
List of recommended item IDs (new API) or numpy array of indices (old API)
"""
if not self.is_fitted:
self._check_fitted()
# Determine which API is being used
top_n = top_n if top_n is not None else top_k
# Old API: if user_id_or_indices is a list/array, treat as item indices
if isinstance(user_id_or_indices, (list, np.ndarray)) or (hasattr(user_id_or_indices, '__iter__') and not isinstance(user_id_or_indices, str)):
# Old API: recommend based on item indices using similarity matrix
sim_matrix = self.similarity_matrix # This will compute if needed
if sim_matrix is None:
raise ValueError("Similarity matrix not available. Use new API or provide feature_matrix in constructor.")
item_indices = list(user_id_or_indices)
combined_scores = np.zeros(sim_matrix.shape[1])
for idx in item_indices:
if 0 <= idx < len(combined_scores):
combined_scores += sim_matrix[idx]
top_items = combined_scores.argsort()[-top_n:][::-1]
return top_items
# New API: recommend based on user (content-based without user history)
exclude_items = exclude_items or []
exclude_set = set(exclude_items)
# If we have similarity matrix (old API mode), use it
if self.similarity_matrix is not None:
# Return items sorted by average similarity (simple heuristic)
avg_similarities = self.similarity_matrix.mean(axis=0)
top_indices = avg_similarities.argsort()[-top_n:][::-1]
return top_indices.tolist()
# Return items sorted by document length (simple heuristic for new API)
items_with_scores = [
(item, len(self.docs.get(item, "")))
for item in self.item_to_index.keys()
if item not in exclude_set
]
items_with_scores.sort(key=lambda x: x[1], reverse=True)
return [item for item, _ in items_with_scores[:top_k]]
[docs]
def recommend_by_text(self, query_text: str, top_k: int = 10,
top_n: Optional[int] = None) -> List[Any]:
"""
Recommend items based on text query using TF-IDF similarity.
Args:
query_text: Text query to find similar items
top_k: Number of recommendations to return
top_n: old name for top_k, kept for existing callers
Returns:
List of recommended item IDs sorted by relevance
"""
if not self.is_fitted:
self._check_fitted()
if not query_text or not query_text.strip():
return []
# Vectorize query
query_vector = self.vectorizer.transform([query_text])
# Compute cosine similarity with all items
similarities = cosine_similarity(query_vector, self.tfidf_matrix).flatten()
if top_n is not None:
top_k = top_n
top_indices = np.argsort(similarities)[::-1][:top_k]
# Return top items even if similarity is low (content-based can have low scores)
recommendations = [
self.index_to_item[idx]
for idx in top_indices
if idx in self.index_to_item
]
return recommendations
[docs]
def save(self, path: Union[str, Path], safe: bool = True, **kwargs) -> None:
"""
Save model to disk.
Args:
path: File path to save model
safe: Use corerec_safe_v1 bundle (default). Set False for legacy pickle.
**kwargs: Additional arguments (unused)
"""
if not self.is_fitted:
self._check_fitted()
from corerec.api.bundle_helpers import pairs, save_map_state, sparse_to_dense
path_obj = Path(path)
config = {"name": self.name, "verbose": self.verbose}
state = {
"docs_pairs": pairs(self.docs),
"is_fitted": self.is_fitted,
**save_map_state(item_to_index=self.item_to_index, index_to_item=self.index_to_item),
}
arrays = {}
if self.tfidf_matrix is not None:
dense = sparse_to_dense(self.tfidf_matrix)
if dense is not None:
arrays["tfidf_matrix"] = np.asarray(dense, dtype=np.float64)
if self.feature_matrix is not None:
arrays["feature_matrix"] = np.asarray(self.feature_matrix, dtype=np.float64)
from corerec.api.torch_bundle import save_numpy_production
if save_numpy_production(self, path_obj, config=config, state=state, arrays=arrays, safe=safe):
if self.verbose:
print(f"{self.name} saved (safe bundle) to {path}")
return
path_obj.parent.mkdir(parents=True, exist_ok=True)
with open(path_obj, 'wb') as f:
pickle.dump(self, f, protocol=pickle.HIGHEST_PROTOCOL)
if self.verbose:
print(f"{self.name} saved to {path}")
[docs]
@classmethod
def load(cls, path: Union[str, Path], **kwargs) -> "TFIDFRecommender":
"""
Load model from disk.
Args:
path: File path to load model from
**kwargs: Additional arguments (unused)
Returns:
Loaded TFIDFRecommender instance
"""
from corerec.api.bundle_helpers import dict_from_pairs, load_map_state
from corerec.api.torch_bundle import load_numpy_production
from scipy import sparse as sp
def _restore(instance, config, state, arrays):
maps = load_map_state(
state, "item_to_index", "index_to_item", int_key_names=("index_to_item",)
)
instance.item_to_index = maps["item_to_index"]
instance.index_to_item = maps["index_to_item"]
instance.docs = dict_from_pairs(state.get("docs_pairs"))
instance.is_fitted = state.get("is_fitted", True)
instance.num_items = len(instance.item_to_index)
arrays = arrays or {}
if arrays.get("feature_matrix") is not None:
instance.feature_matrix = arrays["feature_matrix"]
if arrays.get("tfidf_matrix") is not None:
instance.tfidf_matrix = sp.csr_matrix(arrays["tfidf_matrix"])
if instance.docs and instance.item_to_index:
ordered_items = sorted(instance.item_to_index, key=lambda x: instance.item_to_index[x])
texts = [instance.docs.get(item, "") for item in ordered_items]
instance.vectorizer = TfidfVectorizer(
lowercase=True,
stop_words="english",
max_features=1000,
ngram_range=(1, 2),
)
instance.tfidf_matrix = instance.vectorizer.fit_transform(texts)
loaded = load_numpy_production(
cls,
path,
restore=_restore,
factory=lambda cfg: cls(name=cfg.get("name"), verbose=cfg.get("verbose", False)),
)
if loaded is not None:
if loaded.verbose:
print(f"{loaded.name} loaded (safe bundle) from {path}")
return loaded
with open(path, 'rb') as f:
model = pickle.load(f)
if not isinstance(model, TFIDFRecommender):
raise ValueError(f"Loaded object is not a TFIDFRecommender instance")
if model.verbose:
print(f"{model.name} loaded from {path}")
return model