lec12 recsys

README.md

@@ -17,7 +17,9 @@
 | 19.11.2024 | [Docker compose](./lectures/lec10-docker_compose.pptx) |
 | 21.11.2024 | [Мониторинг сервиса. Prometheus. Graphana](./lectures/lec11-monitoring.pptx) |
 | 28.11.2024 | [Работа с БД](./lectures/lec12-database.pptx) |
-
+| 05.12.2024 | [Рекомендательные системы](./lectures/lec13-recsys.pptx) |
+| 12.12.2024 | Создание сервиса рекомендаций - код в [директории](./assets/recsys)|
+| 19.12.2024 | Создание сервиса рекомендаций - код в [директории](./assets/recsys)|
 
 ## <span style="color:red">Перенос занятий</span>
 ### Лекции

assets/recsys/.gitignore

@@ -0,0 +1,2 @@
+*.parquet
+*__pycache__*

assets/recsys/requirements.txt

@@ -0,0 +1,9 @@
+pandas
+matplotlib
+seaborn
+pyarrow==13.0.0
+mlflow==2.7.1 
+implicit==0.7.2
+catboost
+fastapi
+uvicorn

assets/recsys/service/events/Dockerfile

@@ -0,0 +1,9 @@
+FROM python:3.11-slim
+
+COPY . /events_app
+WORKDIR /events_app
+RUN pip install -r requirements.txt
+EXPOSE 8020
+
+CMD ["uvicorn", "events_service:app", "--port", "8020", "--host", "0.0.0.0"] 
+

assets/recsys/service/events/events_service.py

@@ -0,0 +1,54 @@
+from fastapi import FastAPI
+
+
+class EventStore:
+
+    def __init__(self, max_events_per_user=10):
+        self.events = {}
+        self.max_events_per_user = max_events_per_user
+
+
+    def get(self, user_id):
+        """
+        Возвращает события для пользователя
+        """
+        
+        if user_id in self.events:
+            user_events =  self.events[user_id]
+        else:
+            user_events = []
+
+        return user_events
+    
+    
+    def put(self, user_id, item_id):
+        """
+        Сохраняет событие для пользователя
+        """
+        user_events = self.get(user_id)
+        self.events[user_id] = [item_id] + user_events[: self.max_events_per_user]
+
+
+events_store = EventStore()
+
+app = FastAPI(title="events")
+
+@app.post("/put")
+async def put(user_id: int, item_id: int):
+    """
+    Сохраняет событие для user_id, item_id
+    """
+    events_store.put(user_id, item_id)
+
+    return {"result": "ok"}
+
+
+@app.get("/get")
+async def get(user_id: int, k: int = 10):
+    """
+    Возвращает список последних k событий для пользователя user_id
+    """
+    events = events_store.get(user_id)[:k]
+
+    return {"events": events}
+

assets/recsys/service/events/requirements.txt

@@ -0,0 +1,2 @@
+fastapi
+uvicorn

assets/recsys/service/features/Dockerfile

@@ -0,0 +1,9 @@
+FROM python:3.11-slim
+
+COPY . /features_app
+WORKDIR /features_app
+RUN pip install -r requirements.txt
+
+EXPOSE 8010
+
+CMD ["uvicorn", "feature_service:app", "--port", "8010", "--host", "0.0.0.0"] 

assets/recsys/service/features/feature_service.py

@@ -0,0 +1,57 @@
+import logging
+
+import pandas as pd
+from fastapi import FastAPI
+
+PATH_TO_SIMILAR_ITEMS = '../../recommendations/similar_items.parquet'
+
+logger = logging.getLogger("uvicorn.error")
+
+class SimilarItems:
+
+    def __init__(self, path, **kwargs):
+
+        """
+        Загружаем данные из файла
+        """
+        logger.info(f"Loading data")
+        self._similar_items = pd.read_parquet(path)
+        self._similar_items = self._similar_items[kwargs['columns']]
+        self._similar_items = self._similar_items.set_index('item_id')
+        logger.info(f"Loaded")
+
+
+    def get(self, item_id: int, k: int = 10):
+        """
+        Возвращает список k похожих объектов
+        """
+
+        try:
+            i2i = self._similar_items.loc[item_id].head(k)
+            i2i = {"item_id_2": i2i["sim_item_id"].tolist(), "score": i2i['score'].tolist()}
+        except KeyError:
+            logger.error("No recommendations found")
+            i2i = {"item_id_2": [], "score": []}
+        except: 
+            logger.error("problem with similar recomendations")
+
+        return i2i
+
+
+sim_items_store = SimilarItems( 
+    PATH_TO_SIMILAR_ITEMS,
+    columns=["item_id", "sim_item_id", "score"])
+
+logger.info("Ready!")
+
+# создаём приложение FastAPI
+app = FastAPI(title="features")
+
+@app.get("/similar_items")
+async def recommendations(item_id: int, k: int = 10):
+    """
+    Возвращает список похожих объектов длиной k для item_id
+    """
+    i2i = sim_items_store.get(item_id, k)
+
+    return i2i

assets/recsys/service/features/requirements.txt

@@ -0,0 +1,6 @@
+pandas
+matplotlib
+pyarrow
+implicit==0.7.2
+fastapi
+uvicorn

assets/recsys/service/recommendations/Dockerfile

@@ -0,0 +1,9 @@
+FROM python:3.11-slim
+
+COPY . /recs_app
+WORKDIR /recs_app
+RUN pip install -r requirements.txt
+RUN export $(cat ./.env | xargs)
+EXPOSE 8000
+
+CMD ["uvicorn", "recommendation_service:app", "--port", "8000", "--host", "0.0.0.0"]

assets/recsys/service/recommendations/rec_handler.py

@@ -0,0 +1,43 @@
+import logging as logger
+import pandas as pd
+
+class Recommendations:
+
+    def __init__(self):
+
+        self._recs = {"personal": None, "default": None}
+
+    def load(self, rec_type, path, **kwargs):
+        """
+        Загружает рекомендации из файла
+        """
+
+        logger.info(f"Loading recommendations, type: {rec_type}")
+        self._recs[rec_type] = pd.read_parquet(path, **kwargs)
+        if rec_type == "personal":
+            self._recs[rec_type] = self._recs[rec_type].set_index("user_id")
+        logger.info(f"Loaded")
+
+    def get(self, user_id: int, k: int=100):
+        """
+        Возвращает список рекомендаций для пользователя
+        """
+        try:
+            recs = self._recs["personal"].loc[user_id]
+            recs = recs["item_id"].to_list()[:k]
+        except KeyError:
+            recs = self._recs["default"]
+            recs = recs["item_id"].to_list()[:k]
+        except:
+            logger.error("No recommendations found")
+            recs = []
+
+        if not recs:
+            logger.warning(f"No default recommendations available for user {user_id}")
+            recs = []
+        else:
+            logger.info(f'recs: {recs}')
+            
+        return recs
+        
+

assets/recsys/service/recommendations/recommendation_service.py

@@ -0,0 +1,116 @@
+import logging
+import requests
+import os
+import pandas as pd
+
+from fastapi import FastAPI
+from rec_handler import Recommendations
+
+PATH_TO_RECOMENDATIONS = '../../recommendations/'
+FILENAME_PERS_RECOMENDATIONS = 'personal_als.parquet'
+FILENAME_TOP_POPULAR = 'top_popular.parquet'
+
+logger = logging.getLogger("uvicorn.error")
+
+rec_store = Recommendations()
+features_store_url = "http://localhost:8010"
+events_store_url = "http://localhost:8020" 
+
+
+logger.info("Starting")
+
+rec_store.load(
+"personal",
+PATH_TO_RECOMENDATIONS+FILENAME_PERS_RECOMENDATIONS,
+columns=["user_id", "item_id", "score"],
+)
+rec_store.load(
+    "default",
+    PATH_TO_RECOMENDATIONS+FILENAME_TOP_POPULAR,
+    columns=["item_id", "rank"],
+)
+
+
+# создаём приложение FastAPI
+app = FastAPI(title="recommendations")
+
+
+@app.post("/recommendations_offline")
+async def recommendations_offline(user_id: int, k: int = 100):
+    """
+    Возвращает список рекомендаций длиной k для пользователя user_id
+    """
+
+    recs = rec_store.get(user_id=user_id, k=k)
+    return {"recs": recs}
+
+
+@app.post("/recommendations_online")
+async def recommendations_online(user_id: int, k: int = 100):
+    """
+    Возвращает список онлайн-рекомендаций длиной k для пользователя user_id
+    """
+
+    headers = {"Content-type": "application/json", "Accept": "text/plain"}
+
+    # получаем последние события пользователя
+    params = {"user_id": user_id, "k": 3}
+    resp = requests.get(events_store_url + "/get", headers=headers, params=params)
+    events = resp.json()
+    events = events["events"]
+
+    # получаем список похожих объектов
+    if len(events) > 0:
+        items = []
+        scores = []
+        for item_id in events:
+            params = {"item_id": item_id, "k": k}
+            headers = {"Content-type": "application/json", "Accept": "text/plain"}
+            resp = requests.get(features_store_url + "/similar_items", headers=headers, params=params)
+            if resp.status_code == 200:
+                similar_items = resp.json()
+            else:
+                similar_items = None
+                print(f"status code: {resp.status_code}")
+            items += similar_items["item_id_2"]
+            scores += similar_items["score"]
+        combined = list(zip(items, scores))
+        combined = sorted(combined, key=lambda x: x[1], reverse=True)
+        combined = [item for item, _ in combined]
+        recs = combined[:k]
+
+    else:
+        recs = []
+
+    return {"recs": recs} 
+
+
+@app.post("/recommendations")
+async def recommendations(user_id: int, k: int = 100):
+    """
+    Возвращает список рекомендаций длиной k для пользователя user_id
+    """
+
+    recs_offline = await recommendations_offline(user_id, k)
+    recs_online = await recommendations_online(user_id, k)
+
+    recs_offline = recs_offline["recs"]
+    recs_online = recs_online["recs"]
+    recs_blended = []
+
+    min_length = min(len(recs_offline), len(recs_online))
+    # чередуем элементы из списков, пока позволяет минимальная длина
+    for i in range(min_length):
+        recs_blended.append(recs_online[i])
+        recs_blended.append(recs_offline[i])
+
+    # добавляем оставшиеся элементы в конец
+    recs_blended += recs_online[min_length:]
+    recs_blended += recs_offline[min_length:]
+
+    # оставляем только первые k рекомендаций
+    recs_blended = recs_blended[:k]
+
+    return {"recs": recs_blended}
+ 
+    

assets/recsys/service/recommendations/requirements.txt

@@ -0,0 +1,12 @@
+pandas
+matplotlib
+pyarrow==13.0.0
+mlflow==2.7.1 
+psycopg==3.1.12
+psycopg[binary,pool]
+boto3==1.34.78
+implicit==0.7.2
+fastapi
+uvicorn
+prometheus-fastapi-instrumentator 
+python-dotenv