Dockerfile and recommendation api with example data

.gitattributes

@@ -33,3 +33,4 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
+*.csv filter=lfs diff=lfs merge=lfs -text

.gitignore

@@ -0,0 +1,4 @@
+data/*
+!data/music_info.csv
+!data/user_listening_history_10k.csv
+recommendation-api/__pycache__

Dockerfile

@@ -0,0 +1,13 @@
+FROM python:3.10.9-slim
+
+WORKDIR /app
+
+COPY recommendation-api .
+COPY data/music_info.csv data/music_info.csv
+COPY data/user_listening_history_10k.csv data/user_listening_history_10k.csv
+
+# Install dependencies from the requirements.txt file located within the recommendation-api directory
+RUN pip install -r requirements.txt
+
+# Command to run the server, assuming server.py is inside the recommendation-api directory
+CMD ["python", "server.py"]

data/music_info.csv

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:d930430f811ba3c77f217b3f456f2b6271c238b828d6d9ad76e889b5d725f187
+size 14985870

data/user_listening_history_10k.csv

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:47a82d52ec512f00bf1a3416ecbf153aaa478266e87f6d3c0c4bff85ce4e1d4a
+size 620427

recommendation-api/learner.py

@@ -0,0 +1,35 @@
+from fastai.collab import load_learner
+from fastai.tabular.all import *
+
+def create_params(size):
+    return nn.Parameter(torch.zeros(*size).normal_(0, 0.01))
+
+class DotProductBias(Module):
+    def __init__(self, n_users, n_items, n_factors, y_range=(0, 1.5)):
+        super().__init__()
+        self.user_factors = create_params([n_users, n_factors])
+        self.user_bias = create_params([n_users])
+        self.item_factors = create_params([n_items, n_factors])
+        self.item_bias = create_params([n_items])
+        self.y_range = y_range
+
+    def forward(self, x):
+        users = self.user_factors[x[:, 0]]
+        items = self.item_factors[x[:, 1]]
+        res = (users * items).sum(dim=1)
+        res += self.user_bias[x[:, 0]] + self.item_bias[x[:, 1]]
+        return sigmoid_range(res, *self.y_range)
+
+def custom_accuracy(prediction, target):    
+    # set all predictions above 0.95 as true positive (correct prediction)
+    prediction = torch.where(prediction > 0.95, torch.tensor(1.0), prediction)
+    # shape [64, 1] to [64]
+    target = target.squeeze(1)
+    correct = (prediction == target).float()
+    accuracy = correct.sum() / len(target)
+    return accuracy
+
+async def setup_learner(model_filename: str):
+    learn = load_learner(model_filename)
+    learn.dls.device = 'cpu'
+    return learn

recommendation-api/model.pkl

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:572c13e535c670418a3391e69edfeaa6249964a247bf9dcf978ca15333b9b494
+size 10347577

recommendation-api/recommender.py

@@ -0,0 +1,25 @@
+from fastai.learner import Learner
+import pandas as pd
+
+from tracks import get_unlistened_tracks_for_user, predictions_to_tracks
+
+def get_recommendations_for_user(learn: Learner, user_id: str, limit: int = 5):
+    not_listened_tracks = get_unlistened_tracks_for_user(user_id)
+
+    # Get predictions for the tracks user hasn't listened yet
+    input_dataframe = pd.DataFrame({'user_id': [user_id] * len(not_listened_tracks), 'entry': not_listened_tracks})
+    test_dl = learn.dls.test_dl(input_dataframe)
+    predictions = learn.get_preds(dl=test_dl)
+
+    # Associate them with prediction score and sort
+    tracks_with_predictions = list(zip(not_listened_tracks, predictions[0].numpy()))
+    tracks_with_predictions.sort(key=lambda x: x[1], reverse=True)
+
+    # Pick n and return as full tracks
+    recommendations = predictions_to_tracks(tracks_with_predictions[:limit])
+
+    return {
+        "user_id": user_id,
+        "limit": limit,
+        "recommendations": recommendations
+    }

recommendation-api/requirements.txt

@@ -0,0 +1,4 @@
+fastai
+fastapi
+uvicorn
+asyncio

recommendation-api/server.py

@@ -0,0 +1,34 @@
+from fastapi import FastAPI, Query
+import asyncio
+import uvicorn
+import os
+
+from tracks import get_top_tracks_for_user, get_users_with_track_interactions
+from recommender import get_recommendations_for_user
+from learner import setup_learner, custom_accuracy # Note that DotProductBias must be imported to global namespace
+
+app = FastAPI()
+model_filename = os.path.join(os.path.dirname(os.path.realpath(__file__)), 'model.pkl')
+learn = None
+
+@app.on_event("startup")
+async def startup_event():
+    global learn
+    tasks = [asyncio.ensure_future(setup_learner(model_filename))]  # assign some task
+    learn = (await asyncio.gather(*tasks))[0]
+
+@app.get("/users")
+async def get_users(limit: int = Query(10)):
+    return get_users_with_track_interactions(limit=limit)
+
+@app.get('/users/{user_id}')
+async def get_user_track_history(user_id: str, limit:int = Query(5)):
+    user_history = get_top_tracks_for_user(user_id, limit)
+    return {"user_id": user_id, "history": user_history}
+
+@app.get("/recommend/{user_id}")
+async def get_recommendations(user_id: str, limit: int = Query(5)):
+    return get_recommendations_for_user(learn, user_id, limit)
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="0.0.0.0", port=int(os.getenv("PORT", 7860)))

recommendation-api/tracks.py

@@ -0,0 +1,44 @@
+import pandas as pd
+
+# Read the CSV files
+tracks_df = pd.read_csv('data/music_info.csv')
+tracks_df.fillna('', inplace=True)
+tracks_df["entry"] = tracks_df["name"] + ", " + tracks_df["artist"] + ", " + tracks_df["year"].astype(str)
+track_interactions_df = pd.read_csv('data/user_listening_history_10k.csv')
+
+# Merge data on those two csvs
+dataframe = pd.merge(tracks_df, track_interactions_df, on='track_id', how='left')
+# Convert all columns to string type
+dataframe = dataframe.astype(str)
+# Create a history lookup dictionary by 'user_id'
+user_to_track_history_dict = {user_id: group.drop('user_id', axis=1).to_dict('records')
+                              for user_id, group in dataframe.groupby('user_id')}
+
+def get_users_with_track_interactions(ascending=False, limit=10):
+    playcount_summary = track_interactions_df.groupby('user_id').size().reset_index(name='track_interactions')
+    playcount_summary.sort_values(by='track_interactions', ascending=ascending, inplace=True)
+    if limit is not None:
+        playcount_summary = playcount_summary.head(limit)
+    return playcount_summary.to_dict(orient='records')
+
+def get_top_tracks_for_user(user_id: str, limit=10):
+    track_list = user_to_track_history_dict.get(user_id, [])
+    sorted_tracks = sorted(track_list, key=lambda x: int(x['playcount']) if 'playcount' in x and x['playcount'].isdigit() else 0, reverse=True)
+    if limit is not None:
+        sorted_tracks = sorted_tracks[:limit]
+    return sorted_tracks
+
+def get_unlistened_tracks_for_user(user_id:str):
+    all_tracks = tracks_df['entry'].tolist()
+    listened_tracks = [track['entry'] for track in user_to_track_history_dict.get(user_id, [])]
+    return list(set(all_tracks) - set(listened_tracks))
+
+def predictions_to_tracks(entries_and_predictions):
+    tracks = []
+    for entry, score in entries_and_predictions:
+        track_info = tracks_df[tracks_df['entry'] == entry]
+        if not track_info.empty:
+            track_dict = track_info.to_dict('records')[0]
+            track_dict['score'] = score.astype(str)
+            tracks.append(track_dict)
+    return tracks