Systemy rekomendacji to narzędzia, które pomagają użytkownikom odkrywać nowe treści, takie jak filmy, książki czy produkty, na podstawie ich wcześniejszych preferencji. W tym artykule pokażę Ci, jak krok po kroku zbudować prosty system rekomendacji filmów przy użyciu PyTorch, jednej z najpopularniejszych bibliotek do uczenia maszynowego. Dzięki temu artykułowi dowiesz się, jak działają takie systemy i jak samodzielnie stworzyć model, który będzie sugerował filmy użytkownikom.
Czym jest system rekomendacji?
System rekomendacji to algorytm, który analizuje dane dotyczące użytkowników (na przykład jakie filmy oglądali lub jak je oceniali) i sugeruje im nowe treści. Istnieje kilka głównych metod budowania takich systemów:
- Filtrowanie oparte na treści (Content-based filtering): Rekomenduje elementy podobne do tych, które użytkownik już lubi.
- Filtrowanie oparte na współpracy (Collaborative filtering): Rekomenduje elementy, które inni użytkownicy o podobnych gustach ocenili wysoko.
- Podejścia hybrydowe (Hybrid approaches): Łączy różne metody, aby uzyskać lepsze wyniki.
My skupimy się na filtrowaniu opartym na współpracy, które wykorzystuje macierze ocen użytkowników i filmów.
Dlaczego PyTorch?
PyTorch to elastyczna i łatwa w użyciu biblioteka do uczenia maszynowego. Pozwala na szybkie tworzenie modeli i eksperymentowanie z różnymi podejściami. Idealnie nadaje się do budowy systemów rekomendacji.
Przygotowanie środowiska
Zanim zaczniemy kodować, musimy przygotować nasze środowisko. Potrzebne będą następujące biblioteki:
- PyTorch
- Pandas
- NumPy
- Scikit-learn
Możesz zainstalować je za pomocą komendy:
pip install torch pandas numpy scikit-learn
Przetwarzanie danych
Na początek potrzebujemy danych o ocenach filmów przez użytkowników. Skorzystamy ze zbioru danych MovieLens, który zawiera takie informacje. Po pobraniu danych załadujemy je do Pandas i przetworzymy.
import pandas as pd
# Ładowanie danych o ocenach
ratings = pd.read_csv('ratings.csv')
# Ładowanie danych o filmach
movies = pd.read_csv('movies.csv')
# Podgląd kilku pierwszych wierszy
print(ratings.head())
print(movies.head())
Zbiór ratings zawiera informacje o tym, jaki użytkownik (userId) ocenił jaki film (movieId) i jaką ocenę wystawił (rating). Z kolei zbiór movies zawiera tytuły i gatunki filmów.
Podział danych na zbiór treningowy i testowy
Aby rzetelnie ocenić model, dzielimy dane na zbiór treningowy i testowy:
from sklearn.model_selection import train_test_split
train_data, test_data = train_test_split(ratings, test_size=0.2, random_state=42)
train_matrix = train_data.pivot(index='userId', columns='movieId', values='rating').fillna(0)
test_matrix = test_data.pivot(index='userId', columns='movieId', values='rating').fillna(0)Tworzenie macierzy użytkowników i filmów
Przekształcimy nasze dane w macierz, w której wiersze będą odpowiadały użytkownikom, a kolumny filmom. Wartości w tej macierzy to oceny, jakie użytkownicy przyznali filmom.
user_movie_matrix = train_matrixModelowanie przy użyciu PyTorch
W naszym modelu użyjemy techniki zwanej funkcyjnym rozkładem macierzy (Matrix Factorization). Chcemy podzielić macierz użytkowników i filmów na dwie mniejsze macierze, które po pomnożeniu dadzą nam przybliżone wartości ocen.
import torch
import torch.nn as nn
import torch.optim as optim
class MatrixFactorization(nn.Module):
def __init__(self, num_users, num_items, num_factors):
super(MatrixFactorization, self).__init__()
self.user_factors = nn.Embedding(num_users, num_factors)
self.item_factors = nn.Embedding(num_items, num_factors)
def forward(self, user, item):
return (self.user_factors(user) * self.item_factors(item)).sum(1)
num_users, num_items = user_movie_matrix.shape
model = MatrixFactorization(num_users, num_items, num_factors=20)
loss_fn = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)Trenowanie modelu z batchowaniem
Teraz przystępujemy do trenowania modelu, czyli uczymy go przewidywać oceny użytkowników na podstawie dotychczasowych danych.
from torch.utils.data import DataLoader, TensorDataset
user_tensor = torch.tensor(user_movie_matrix.index.values)
movie_tensor = torch.tensor(user_movie_matrix.columns.values)
rating_tensor = torch.tensor(user_movie_matrix.values)
dataset = TensorDataset(user_tensor, movie_tensor, rating_tensor)
data_loader = DataLoader(dataset, batch_size=64, shuffle=True)
def train_with_batches(model, epochs, optimizer, loss_fn, data_loader):
for epoch in range(epochs):
model.train()
epoch_loss = 0
for batch in data_loader:
user, movie, rating = batch
optimizer.zero_grad()
prediction = model(user, movie)
loss = loss_fn(prediction, rating)
loss.backward()
optimizer.step()
epoch_loss += loss.item()
print(f'Epoch {epoch+1}, Loss: {epoch_loss}')
train_with_batches(model, epochs=10, optimizer=optimizer, loss_fn=loss_fn, data_loader=data_loader)Model uczy się przewidywać oceny, minimalizując różnicę między przewidywaną a rzeczywistą oceną. Trening trwa przez kilka „epok”, czyli iteracji po całym zbiorze danych.
Ocena modelu za pomocą RMSE
Po zakończeniu treningu oceniamy model za pomocą RMSE:
Generowanie rekomendacji
Po zakończeniu treningu możemy zacząć generować rekomendacje dla użytkowników.
def recommend_movies(model, user_id, user_movie_matrix, top_n=10):
model.eval()
user = torch.tensor([user_id])
scores = []
for movie_id in range(num_items):
if user_movie_matrix.iloc[user_id, movie_id] == 0:
movie = torch.tensor([movie_id])
prediction = model(user, movie)
scores.append((movie_id, prediction.item()))
scores.sort(key=lambda x: x[1], reverse=True)
recommended_movie_ids = [x[0] for x in scores[:top_n]]
recommended_movies = movies[movies['movieId'].isin(recommended_movie_ids)]
return recommended_movies
user_id = 1 # Przykładowy użytkownik
recommended_movies = recommend_movies(model, user_id, user_movie_matrix)
print(recommended_movies)Funkcja recommend_movies przewiduje oceny dla wszystkich filmów, które użytkownik jeszcze nie ocenił, a następnie wybiera te o najwyższych przewidywanych ocenach.
Pełny kod z opisem
import pandas as pd
import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
from torch.utils.data import DataLoader, TensorDataset
from sklearn.model_selection import train_test_split
# Krok 1: Ładowanie danych
ratings = pd.read_csv('ratings.csv')
movies = pd.read_csv('movies.csv')
# Krok 2: Podział danych na zbiory treningowe i testowe
train_data, test_data = train_test_split(ratings, test_size=0.2, random_state=42)
train_matrix = train_data.pivot(index='userId', columns='movieId', values='rating').fillna(0)
test_matrix = test_data.pivot(index='userId', columns='movieId', values='rating').fillna(0)
# Krok 3: Tworzenie macierzy użytkowników i filmów
user_movie_matrix = train_matrix
# Konwersja danych do tensorów
user_tensor = torch.tensor(user_movie_matrix.index.values)
movie_tensor = torch.tensor(user_movie_matrix.columns.values)
rating_tensor = torch.tensor(user_movie_matrix.values)
# Krok 4: Tworzenie modelu
class MatrixFactorization(nn.Module):
def __init__(self, num_users, num_items, num_factors):
super(MatrixFactorization, self).__init__()
self.user_factors = nn.Embedding(num_users, num_factors)
self.item_factors = nn.Embedding(num_items, num_factors)
def forward(self, user, item):
return (self.user_factors(user) * self.item_factors(item)).sum(1)
num_users, num_items = user_movie_matrix.shape
model = MatrixFactorization(num_users, num_items, num_factors=20)
loss_fn = nn.MSELoss()
optimizer = optim.Adam(model.parameters(), lr=0.001)
# Krok 5: Trenowanie modelu z batchowaniem
dataset = TensorDataset(user_tensor, movie_tensor, rating_tensor)
data_loader = DataLoader(dataset, batch_size=64, shuffle=True)
def train_with_batches(model, epochs, optimizer, loss_fn, data_loader):
for epoch in range(epochs):
model.train()
epoch_loss = 0
for batch in data_loader:
user, movie, rating = batch
optimizer.zero_grad()
prediction = model(user, movie)
loss = loss_fn(prediction, rating)
loss.backward()
optimizer.step()
epoch_loss += loss.item()
print(f'Epoch {epoch+1}, Loss: {epoch_loss}')
train_with_batches(model, epochs=10, optimizer=optimizer, loss_fn=loss_fn, data_loader=data_loader)
# Krok 6: Ocena modelu za pomocą RMSE
def rmse(predictions, targets):
return np.sqrt(((predictions - targets) ** 2).mean())
def evaluate_rmse(model, test_matrix):
model.eval()
predictions = []
targets = []
with torch.no_grad():
for user_id in range(num_users):
for movie_id in range(num_items):
if test_matrix.iloc[user_id, movie_id] > 0:
user = torch.tensor([user_id])
movie = torch.tensor([movie_id])
rating = test_matrix.iloc[user_id, movie_id]
prediction = model(user, movie).item()
predictions.append(prediction)
targets.append(rating)
return rmse(np.array(predictions), np.array(targets))
test_rmse = evaluate_rmse(model, test_matrix)
print(f'Test RMSE: {test_rmse}')
# Krok 7: Generowanie rekomendacji
def recommend_movies(model, user_id, user_movie_matrix, top_n=10):
model.eval()
user = torch.tensor([user_id])
scores = []
for movie_id in range(num_items):
if user_movie_matrix.iloc[user_id, movie_id] == 0:
movie = torch.tensor([movie_id])
prediction = model(user, movie)
scores.append((movie_id, prediction.item()))
scores.sort(key=lambda x: x[1], reverse=True)
recommended_movie_ids = [x[0] for x in scores[:top_n]]
recommended_movies = movies[movies['movieId'].isin(recommended_movie_ids)]
return recommended_movies
user_id = 1 # Przykładowy użytkownik
recommended_movies = recommend_movies(model, user_id, user_movie_matrix)
print(recommended_movies)