README.md

@@ -17,9 +17,7 @@
 | 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

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

assets/recsys/requirements.txt

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

assets/recsys/service/events/Dockerfile

@@ -1,9 +0,0 @@
-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

@@ -1,54 +0,0 @@
-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

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

assets/recsys/service/features/Dockerfile

@@ -1,9 +0,0 @@
-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

@@ -1,57 +0,0 @@
-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

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

assets/recsys/service/recommendations/Dockerfile

@@ -1,9 +0,0 @@
-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

@@ -1,43 +0,0 @@
-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

@@ -1,116 +0,0 @@
-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

@@ -1,12 +0,0 @@
-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