Recommendation Algorithms

Comprehensive collection of recommendation algorithms including collaborative filtering, content-based filtering, graph-based methods, and sequential recommendations.

Core Classes

ALSRecommender

Description: Alternating Least Squares collaborative filtering for implicit feedback

Parameters:

• factors (int): Number of latent factors (default: 30)
• alpha (float): Confidence weighting parameter (default: 100.0)
• regularization (float): L2 regularization parameter (default: 0.1)
• iterations (int): Number of training iterations (default: 10)
• seed (int, optional): Random seed for reproducibility

Returns: ALSRecommender instance

Example:

from recoagent.recommendations.algorithms.collaborative import ALSRecommender, ALSConfig

# Create configuration
config = ALSConfig(
    factors=50,
    alpha=150.0,
    regularization=0.05,
    iterations=20
)

# Create recommender
als_recommender = ALSRecommender(config=config)

# Train on interaction data
interactions = [
    {"user_id": "user1", "item_id": "item1", "rating": 1},
    {"user_id": "user1", "item_id": "item2", "rating": 1},
    {"user_id": "user2", "item_id": "item1", "rating": 1}
]

als_recommender.fit(interactions)

# Generate recommendations
recommendations = als_recommender.recommend("user1", n_recommendations=10)

MatrixFactorizationRecommender

Description: Matrix factorization for explicit feedback recommendations

Parameters:

• factors (int): Number of latent factors
• learning_rate (float): Learning rate for optimization
• regularization (float): Regularization parameter
• epochs (int): Number of training epochs

Returns: MatrixFactorizationRecommender instance

Example:

from recoagent.recommendations.algorithms.collaborative import MatrixFactorizationRecommender

# Create recommender
mf_recommender = MatrixFactorizationRecommender(
    factors=40,
    learning_rate=0.01,
    regularization=0.1,
    epochs=100
)

# Train on ratings data
ratings = [
    {"user_id": "user1", "item_id": "movie1", "rating": 5.0},
    {"user_id": "user1", "item_id": "movie2", "rating": 4.0},
    {"user_id": "user2", "item_id": "movie1", "rating": 3.0}
]

mf_recommender.fit(ratings)

# Predict ratings
predicted_rating = mf_recommender.predict("user1", "movie3")

GraphBasedRecommender

Description: Graph-based recommendation using network analysis

Parameters:

• algorithm (str): Graph algorithm ("deepwalk", "line", "metapath")
• walk_length (int): Random walk length
• num_walks (int): Number of random walks
• embedding_dim (int): Embedding dimension

Returns: GraphBasedRecommender instance

Example:

from recoagent.recommendations.algorithms.graph_based import GraphBasedRecommender

# Create graph-based recommender
graph_recommender = GraphBasedRecommender(
    algorithm="deepwalk",
    walk_length=10,
    num_walks=20,
    embedding_dim=128
)

# Build graph from interactions
interactions = [
    {"user_id": "user1", "item_id": "item1"},
    {"user_id": "user1", "item_id": "item2"},
    {"user_id": "user2", "item_id": "item1"}
]

graph_recommender.build_graph(interactions)
graph_recommender.train()

# Generate recommendations
recommendations = graph_recommender.recommend("user1", n_recommendations=5)

SequentialRecommender

Description: Sequential recommendation using deep learning models

Parameters:

• model_type (str): Model type ("gru4rec", "sasrec", "nextitnet")
• sequence_length (int): Maximum sequence length
• embedding_dim (int): Embedding dimension
• hidden_dim (int): Hidden layer dimension

Returns: SequentialRecommender instance

Example:

from recoagent.recommendations.algorithms.sequential import SequentialRecommender

# Create sequential recommender
seq_recommender = SequentialRecommender(
    model_type="sasrec",
    sequence_length=50,
    embedding_dim=64,
    hidden_dim=128
)

# Train on user sequences
user_sequences = [
    {"user_id": "user1", "sequence": ["item1", "item2", "item3"]},
    {"user_id": "user2", "sequence": ["item2", "item4", "item1"]}
]

seq_recommender.fit(user_sequences)

# Predict next items
next_items = seq_recommender.predict_next("user1", ["item1", "item2"])

Usage Examples

Collaborative Filtering Pipeline

from recoagent.recommendations.algorithms.collaborative import ALSRecommender, MatrixFactorizationRecommender
from recoagent.recommendations.evaluation import RecommendationEvaluator

# Create collaborative filtering recommenders
als_recommender = ALSRecommender(
    factors=50,
    alpha=100.0,
    regularization=0.1
)

mf_recommender = MatrixFactorizationRecommender(
    factors=40,
    learning_rate=0.01,
    regularization=0.1
)

# Train on interaction data
interactions = load_interaction_data()  # Your data loading function

als_recommender.fit(interactions)
mf_recommender.fit(interactions)

# Generate recommendations
user_id = "user123"
als_recs = als_recommender.recommend(user_id, n_recommendations=20)
mf_recs = mf_recommender.recommend(user_id, n_recommendations=20)

# Evaluate performance
evaluator = RecommendationEvaluator()
als_metrics = evaluator.evaluate(als_recs, ground_truth)
mf_metrics = evaluator.evaluate(mf_recs, ground_truth)

print(f"ALS Precision@10: {als_metrics['precision@10']:.3f}")
print(f"MF Precision@10: {mf_metrics['precision@10']:.3f}")

Graph-Based Recommendations

from recoagent.recommendations.algorithms.graph_based import GraphBasedRecommender
import networkx as nx

# Create graph-based recommender
graph_recommender = GraphBasedRecommender(
    algorithm="deepwalk",
    walk_length=15,
    num_walks=30,
    embedding_dim=64
)

# Build heterogeneous graph
interactions = load_interaction_data()
user_features = load_user_features()
item_features = load_item_features()

# Add user-item interactions
for interaction in interactions:
    graph_recommender.add_edge(
        f"user_{interaction['user_id']}",
        f"item_{interaction['item_id']}",
        weight=interaction.get('weight', 1.0)
    )

# Add user-user similarities
user_similarities = calculate_user_similarities(user_features)
for user1, user2, similarity in user_similarities:
    if similarity > 0.5:  # Threshold
        graph_recommender.add_edge(f"user_{user1}", f"user_{user2}", weight=similarity)

# Train embeddings
graph_recommender.train()

# Generate recommendations
recommendations = graph_recommender.recommend("user_123", n_recommendations=10)

# Get similar users
similar_users = graph_recommender.get_similar_users("user_123", n_users=5)

Sequential Recommendation System

from recoagent.recommendations.algorithms.sequential import SequentialRecommender
from recoagent.recommendations.preprocessing import SequencePreprocessor

# Create sequential recommender
seq_recommender = SequentialRecommender(
    model_type="sasrec",
    sequence_length=100,
    embedding_dim=128,
    hidden_dim=256
)

# Preprocess sequences
preprocessor = SequencePreprocessor(
    min_sequence_length=5,
    max_sequence_length=100
)

# Load and preprocess data
raw_sequences = load_user_sequences()
processed_sequences = preprocessor.process(raw_sequences)

# Train model
seq_recommender.fit(processed_sequences)

# Generate recommendations for active users
active_users = get_active_users()
for user_id in active_users:
    user_sequence = get_user_sequence(user_id)
    next_items = seq_recommender.predict_next(user_id, user_sequence)
    
    print(f"User {user_id} next recommendations:")
    for item_id, score in next_items:
        print(f"  {item_id}: {score:.3f}")

Hybrid Recommendation System

from recoagent.recommendations.algorithms.collaborative import ALSRecommender
from recoagent.recommendations.algorithms.graph_based import GraphBasedRecommender
from recoagent.recommendations.algorithms.sequential import SequentialRecommender

# Create multiple recommenders
als_recommender = ALSRecommender(factors=50)
graph_recommender = GraphBasedRecommender(algorithm="deepwalk")
seq_recommender = SequentialRecommender(model_type="sasrec")

# Train all models
interactions = load_interaction_data()
sequences = load_sequence_data()

als_recommender.fit(interactions)
graph_recommender.build_graph(interactions)
graph_recommender.train()
seq_recommender.fit(sequences)

# Hybrid recommendation function
def hybrid_recommend(user_id, n_recommendations=20):
    # Get recommendations from each model
    als_recs = als_recommender.recommend(user_id, n_recommendations)
    graph_recs = graph_recommender.recommend(user_id, n_recommendations)
    seq_recs = seq_recommender.recommend(user_id, n_recommendations)
    
    # Combine with weights
    weights = {"als": 0.4, "graph": 0.3, "sequential": 0.3}
    
    combined_scores = {}
    for item_id, score in als_recs:
        combined_scores[item_id] = combined_scores.get(item_id, 0) + weights["als"] * score
    
    for item_id, score in graph_recs:
        combined_scores[item_id] = combined_scores.get(item_id, 0) + weights["graph"] * score
    
    for item_id, score in seq_recs:
        combined_scores[item_id] = combined_scores.get(item_id, 0) + weights["sequential"] * score
    
    # Sort and return top recommendations
    sorted_items = sorted(combined_scores.items(), key=lambda x: x[1], reverse=True)
    return sorted_items[:n_recommendations]

# Use hybrid system
recommendations = hybrid_recommend("user_123", n_recommendations=10)

Real-time Recommendation Updates

from recoagent.recommendations.algorithms.collaborative import ALSRecommender
import asyncio

# Create recommender with real-time updates
als_recommender = ALSRecommender(
    factors=30,
    enable_incremental=True,
    update_frequency=100  # Update every 100 new interactions
)

# Initial training
interactions = load_historical_data()
als_recommender.fit(interactions)

# Real-time recommendation service
async def recommend_realtime(user_id, context=None):
    # Get current recommendations
    recommendations = als_recommender.recommend(user_id, n_recommendations=10)
    
    # Apply real-time context filtering
    if context:
        recommendations = filter_by_context(recommendations, context)
    
    return recommendations

# Incremental learning
async def update_model(new_interactions):
    """Update model with new interactions."""
    als_recommender.incremental_fit(new_interactions)
    
    # Log model update
    logger.info(f"Model updated with {len(new_interactions)} new interactions")

# Simulate real-time updates
async def main():
    # Start recommendation service
    recommendations = await recommend_realtime("user_123")
    
    # Simulate new interactions
    new_interactions = [
        {"user_id": "user_123", "item_id": "new_item_1", "rating": 1},
        {"user_id": "user_123", "item_id": "new_item_2", "rating": 1}
    ]
    
    # Update model
    await update_model(new_interactions)
    
    # Get updated recommendations
    updated_recommendations = await recommend_realtime("user_123")

# Run async service
asyncio.run(main())

API Reference

ALSRecommender Methods

fit(interactions: List[Dict]) -> None

Train the ALS model on interaction data

Parameters:

• interactions (List[Dict]): List of user-item interactions

recommend(user_id: str, n_recommendations: int = 10) -> List[Tuple[str, float]]

Generate recommendations for a user

Parameters:

• user_id (str): User identifier
• n_recommendations (int): Number of recommendations

Returns: List of (item_id, score) tuples

predict(user_id: str, item_id: str) -> float

Predict rating for user-item pair

Parameters:

• user_id (str): User identifier
• item_id (str): Item identifier

Returns: Predicted rating

GraphBasedRecommender Methods

build_graph(interactions: List[Dict]) -> None

Build graph from interaction data

Parameters:

• interactions (List[Dict]): Interaction data

train() -> None

Train graph embeddings

recommend(user_id: str, n_recommendations: int = 10) -> List[Tuple[str, float]]

Generate recommendations using graph embeddings

Parameters:

• user_id (str): User identifier
• n_recommendations (int): Number of recommendations

Returns: List of (item_id, score) tuples

SequentialRecommender Methods

fit(sequences: List[Dict]) -> None

Train sequential model on user sequences

Parameters:

• sequences (List[Dict]): User interaction sequences

predict_next(user_id: str, sequence: List[str]) -> List[Tuple[str, float]]

Predict next items in sequence

Parameters:

• user_id (str): User identifier
• sequence (List[str]): Current sequence

Returns: List of (item_id, score) tuples

See Also

• Recommendation Bandits - Multi-armed bandit algorithms
• Recommendation Agents - Agent-based recommendations
• Recommendation Evaluation - Evaluation metrics
• Recommendation Industries - Industry-specific implementations