协同过滤：如何构建推荐系统

import os
import requests
import pandas as pd
from surprise import SVD
from surprise import Dataset, Reader

# we'll be downloading a few files for this example so here's a helper function
def fetch_dataframe(file_name):
    try:
        df = pd.read_csv('datasets/collaborative_filtering/' + file_name)
    except:
        url = 'https://redis-ai-resources.s3.us-east-2.amazonaws.com/recommenders/datasets/collaborative-filtering/'
        r = requests.get(url + file_name)
        if not os.path.exists('datasets/collaborative_filtering'):
            os.makedirs('datasets/collaborative_filtering')
        with open('datasets/collaborative_filtering/' + file_name, 'wb') as f:
            f.write(r.content)
        df = pd.read_csv('datasets/collaborative_filtering/' + file_name)
    return df

# for a larger dataset use 'ratings.csv'
ratings_df = fetch_dataframe('ratings_small.csv')

# only keep the columns we need: userId, movieId, rating
ratings_df = ratings_df[['userId', 'movieId', 'rating']]

reader = Reader(rating_scale=(0.0, 5.0))
ratings_data = Dataset.load_from_df(ratings_df, reader)

# split the data into training and testing sets (80% train, 20% test)
train_set, test_set = train_test_split(ratings_data, test_size=0.2)

# use SVD (Singular Value Decomposition) for collaborative filtering
svd = SVD(n_factors=100, biased=False)  # we'll set biased to False so that predictions are of the form "rating_prediction = user_vector * item_vector"

# train the algorithm on the train_set
svd.fit(train_set)

user_vectors = svd.pu  # user latent features (matrix)
movie_vectors = svd.qi  # movie latent features (matrix)

print(f'we have {user_vectors.shape[0]} users with feature vectors of size {user_vectors.shape[1]}')
print(f'we have {movie_vectors.shape[0]} movies with feature vectors of size {movie_vectors.shape[1]}')

we have 671 users with feature vectors of size 100
we have 8397 movies with feature vectors of size 100

# surprise casts userId and movieId to inner ids, so we have to use their mapping to know which rows to use
inner_uid = train_set.to_inner_uid(347) # userId
inner_iid = train_set.to_inner_iid(5515) # movieId

# predict one user's rating of one film
predicted_rating = np.dot(user_vectors[inner_uid], movie_vectors[inner_iid])
print(f'the predicted rating of user {347} on movie {5515} is {predicted_rating}')

the predicted rating of user 347 on movie 5515 is 1.1069607933289707

# fetch and clean the movies data
import datetime
movies_df = fetch_dataframe('movies_metadata.csv')

movies_df.drop(columns=['homepage', 'production_countries', 'production_companies', 'spoken_languages', 'video', 'original_title', 'video', 'poster_path', 'belongs_to_collection'], inplace=True)

# drop rows that have missing values
movies_df.dropna(subset=['imdb_id'], inplace=True)

movies_df['original_language'] = movies_df['original_language'].fillna('unknown')
movies_df['overview'] = movies_df['overview'].fillna('')
movies_df['popularity'] = movies_df['popularity'].fillna(0)
movies_df['release_date'] = movies_df['release_date'].fillna('1900-01-01').apply(lambda x: datetime.datetime.strptime(x, "%Y-%m-%d").timestamp())
movies_df['revenue'] = movies_df['revenue'].fillna(0)
movies_df['runtime'] = movies_df['runtime'].fillna(0)
movies_df['status'] = movies_df['status'].fillna('unknown')
movies_df['tagline'] = movies_df['tagline'].fillna('')
movies_df['title'] = movies_df['title'].fillna('')
movies_df['vote_average'] = movies_df['vote_average'].fillna(0)
movies_df['vote_count'] = movies_df['vote_count'].fillna(0)
movies_df['genres'] = movies_df['genres'].apply(lambda x: [g['name'] for g in eval(x)] if x != '' else []) # convert to a list of genre names
movies_df['imdb_id'] = movies_df['imdb_id'].apply(lambda x: x[2:] if str(x).startswith('tt') else x).astype(int) # remove leading 'tt' from imdb_id

links_df = fetch_dataframe('links_small.csv') # for a larger example use 'links.csv' instead

movies_df = movies_df.merge(links_df, left_on='imdb_id', right_on='imdbId', how='inner')

# place movie vectors and their movieIds in a dataframe
movie_vectors_and_ids = {train_set.to_raw_iid(inner_id): movie_vectors[inner_id].tolist() for inner_id in train_set.all_items()}
movie_vector_df = pd.Series(movie_vectors_and_ids).to_frame('movie_vector')

# merge the movie vector series with the movies dataframe using movieId and id fields
movies_df = movies_df.merge(movie_vector_df, left_on='movieId', right_index=True, how='inner')
movies_df['movieId'] = movies_df['movieId'].apply(lambda x: str(x)) # need to cast to a string as this is a tag field in our search schema
movies_df.head()
his is a tag field in our search schema
movies_df.head()

from redis import Redis
from redisvl.schema import IndexSchema
from redisvl.index import SearchIndex

client = Redis.from_url(<REDIS_URL>) # ex: "redis://localhost:6379"

schema = {
  "index": {
    "name": "movies",
    "prefix": "movie",
    "storage_type": "json"
  },
  "fields": [
    {"name": "title",
     "type": "text",
    },
    {"name": "genres",
     "type": "tag"
    },
    {"name": "revenue",
     "type": "numeric"
    },
    {"name": "release_date",
     "type": "numeric"
    },
    {"name": "popularity",
     "type": "numeric"
    },
    {"name": "vote_average",
     "type": "numeric"
    },
    {"name": "movie_vector",
     "type": "vector",
     "attrs": {
            "dims": 100,
            "distance_metric": "ip",
            "algorithm": "flat",
            "datatype": "float32"
        }
    }
  ]
}

movie_schema = IndexSchema.from_dict(schema)

movie_index = SearchIndex(movie_schema, redis_client=client)
movie_index.create(overwrite=True, drop=True)

movie_keys = movie_index.load(movies_df.to_dict(orient='records'))

from redis.commands.json.path import Path

# collect the user vectors and their userIds
user_vectors_and_ids = {train_set.to_raw_uid(inner_id): user_vectors[inner_id].tolist() for inner_id in train_set.all_users()}

# use a Redis pipeline to store user data and verify it in a single transaction
with client.pipeline() as pipe:
 for user_id, user_vector in user_vectors_and_ids.items():
        user_key = f"user:{user_id}"
        watched_list_ids = ratings_df[ratings_df['userId'] == user_id]['movieId'].tolist()

        user_data = {
            "user_vector": user_vector,
            "watched_list_ids": watched_list_ids
        }
        pipe.json().set(user_key, Path.root_path(), user_data)
        pipe.execute()

from redisvl.query import RangeQuery

user_vector = client.json().get(f"user:{352}")["user_vector"]

query = RangeQuery(vector=user_vector,
                    vector_field_name='movie_vector',
                    num_results=5,
                    return_score=True,
                    return_fields=['title', 'genres']
                    )

results = movie_index.query(query)

for r in results:
    # compute our predicted rating on a scale of 0 to 5 from vector distance
    r['predicted_rating'] = - float(r['vector_distance']) + 1.
    print(f"vector distance: {float(r['vector_distance']):.08f},\t predicted rating: {r['predicted_rating']:.08f},\t title: {r['title']}, ")



vector distance: -3.63527393,	 predicted rating: 4.63527393,	 title: Fight Club, 
vector distance: -3.60445881,	 predicted rating: 4.60445881,	 title: All About Eve, 
vector distance: -3.60197020,	 predicted rating: 4.60197020,	 title: Lock, Stock and Two Smoking Barrels, 
vector distance: -3.59518766,	 predicted rating: 4.59518766,	 title: Midnight in Paris, 
vector distance: -3.58543396,	 predicted rating: 4.58543396,	 title: It Happened One Night, 

from redisvl.query.filter import Tag, Num, Text

def get_recommendations(user_id, filters=None, num_results=10):
    user_vector = client.json().get(f"user:{user_id}")["user_vector"]
query = RangeQuery(vector=user_vector,
                       vector_field_name='movie_vector',
                       num_results=num_results,
                       filter_expression=filters,
                       return_fields=['title', 'overview', 'genres'])

    results = movie_index.query(query)

    return [(r['title'], r['overview'], r['genres'], r['vector_distance']) for r in results]

Top_picks_for_you = get_recommendations(user_id=42) # general SVD results, no filter

block_buster_filter = Num('revenue') > 30_000_000
block_buster_hits = get_recommendations(user_id=42, filters=block_buster_filter)

classics_filter = Num('release_date') < datetime.datetime(1990, 1, 1).timestamp()
classics = get_recommendations(user_id=42, filters=classics_filter)

popular_filter = (Num('popularity') > 50) & (Num('vote_average') > 7)
Whats_popular = get_recommendations(user_id=42, filters=popular_filter)

indie_filter = (Num('revenue') < 1_000_000) & (Num('popularity') > 10)
indie_hits = get_recommendations(user_id=42, filters=indie_filter)
fruity = Text('title') % 'apple|orange|peach|banana|grape|pineapple'
fruity_films = get_recommendations(user_id=42, filters=fruity)
# put all these titles into a single pandas dataframe, where each column is one category
all_recommendations = pd.DataFrame(columns=["top picks", "block busters", "classics", "what's popular", "indie hits", "fruity films"])
all_recommendations["top picks"] = [m[0] for m in Top_picks_for_you]
all_recommendations["block busters"] = [m[0] for m in block_buster_hits]
all_recommendations["classics"] = [m[0] for m in classics]
all_recommendations["what's popular"] = [m[0] for m in Whats_popular]
all_recommendations["indie hits"] = [m[0] for m in indie_hits]
all_recommendations["fruity films"] = [m[0] for m in fruity_films]
all_recommendations.head(10)

# rewrite the get_recommendations() function to use a bloom filter and apply it before we return results
def get_unique_recommendations(user_id, filters=None, num_results=10):
    user_data = client.json().get(f"user:{user_id}")
    user_vector = user_data["user_vector"]
    watched_movies = user_data["watched_list_ids"]

    # filter out movies that the user has already watched
    client.bf().insert('user_watched_list', [f"{user_id}:{movie_id}" for movie_id in watched_movies])

    query = RangeQuery(vector=user_vector,
                       vector_field_name='movie_vector',
                       num_results=num_results * 5,  # fetch more results to filter out watched movies
                       filter_expression=filters,
                       return_fields=['title', 'overview', 'genres', 'movieId'],
    )
    results = movie_index.query(query)

    matches = client.bf().mexists("user_watched_list", *[f"{user_id}:{r['movieId']}" for r in results])

    recommendations = [
        (r['title'], r['overview'], r['genres'], r['vector_distance'], r['movieId'])
        for i, r in enumerate(results) if matches[i] == 0
    ][:num_results]

    # add these recommendations to the bloom filter so they don't appear again
    client.bf().insert('user_watched_list', [f"{user_id}:{r[4]}" for r  in recommendations])
    return recommendations

# put all these titles into a single pandas dataframe , where each column is one category
all_recommendations = pd.DataFrame(columns=["top picks", "block busters", "classics", "what's popular", "indie hits"])
all_recommendations["top picks"] = [m[0] for m in top_picks_for_you]
all_recommendations["block busters"] = [m[0] for m in block_buster_hits]
all_recommendations["classics"] = [m[0] for m in classics]
all_recommendations["what's popular"] = [m[0] for m in whats_popular]
all_recommendations["indie hits"] = [m[0] for m in indie_hits]