Создание систем рекомендаций видеоигр с помощью FastAPI, PostgreSQL и Render: Часть 1

Разработка сервиса рекомендаций видеоигр с использованием API Steam

Делиться

Введение

Рекомендательные системы позволяют приложению генерировать интеллектуальные предложения для пользователя, эффективно отфильтровывая релевантный контент. В этой статье мы создаём и внедряем динамическую рекомендательную систему видеоигр, использующую PostgreSQL, FastAPI и Render, которая рекомендует пользователю новые игры на основе тех, с которыми он взаимодействовал. Цель — предоставить наглядный пример того, как можно создать автономную рекомендательную систему, которую затем можно будет интегрировать во фронтенд-систему или другое приложение.

В этом проекте мы используем данные о видеоиграх, доступные через API Steam, но их можно легко заменить любыми другими интересующими вас данными о продукте. Основные этапы останутся теми же. Мы рассмотрим, как хранить эти данные в базе данных, векторизовать теги игр, генерировать оценки сходства на основе игр, с которыми взаимодействовал пользователь, и возвращать ряд релевантных рекомендаций. В конце статьи мы развернем эту рекомендательную систему в виде веб-приложения с FastAPI, чтобы при каждом взаимодействии пользователя с новой игрой мы могли динамически генерировать и сохранять новый набор рекомендаций для этого пользователя.

Будут использоваться следующие инструменты:

• PostgreSQL
• FastAPI
• Докер
• Оказывать

Те, кому просто интересен репозиторий GitHub, могут найти его здесь.

Оглавление

Из-за объёма проекта он разделён на две статьи. Первая часть посвящена настройке и теории проекта (шаги 1–5 показаны ниже), а вторая — его развёртыванию. Если вам нужна вторая часть, она находится здесь.

Часть 1

1. Обзор набора данных
2. Общая архитектура системы
3. Настройка базы данных
4. Настройка FastAPI
   – Модели
   – Маршруты
5. Построение трубопровода подобия

Часть 2:

1. Развертывание базы данных PostgreSQL на Render
2. Развертывание приложения FastAPI как приложения для рендеринга веб-сайтов
   – Докеризация нашего приложения
   – Отправка образа Docker в DockerHub
   – Извлечение из DockerHub в Render

Обзор набора данных

Набор данных для этого проекта содержит данные примерно о 2000 лучших играх из API Steamworks. Эти данные предоставляются бесплатно и лицензированы для личного и коммерческого использования в соответствии с условиями обслуживания. Существует ограничение на количество запросов в 200 за 5 минут, поэтому мы работаем только с подмножеством данных. С условиями обслуживания можно ознакомиться здесь.

Обзор набора данных по играм представлен ниже. Большинство полей относительно информативны; главное, что следует отметить, — уникальный идентификатор продукта appid. Помимо этого набора данных, у нас также есть несколько дополнительных таблиц, которые мы подробно рассмотрим ниже. Наиболее важной для нашей рекомендательной системы является таблица игровых тегов, которая содержит значения appid, сопоставленные с каждым тегом, связанным с игрой (стратегия, ролевая игра, карточная игра и т. д.). Эти значения были взяты из поля «Категории», показанного в обзоре данных, и затем преобразованы для создания таблицы game_tags, чтобы для каждой комбинации appip:category была уникальная строка.

Более подробный обзор структуры нашего проекта можно увидеть на схеме ниже.

Сейчас мы сделаем краткий обзор архитектуры этого проекта, а затем рассмотрим, как заполнить нашу базу данных.

Архитектура

Для нашей системы рекомендаций мы будем использовать базу данных PostgreSQL с доступом к данным и уровнем обработки FastAPI, что позволит добавлять и удалять игры из списка игр пользователя. Пользователи, вносящие изменения в свою библиотеку игр с помощью POST-запроса FastAPI, также запускают конвейер рекомендаций, используя функцию фоновых задач FastAPI. Эта функция будет запрашивать понравившиеся игры из базы данных, вычислять показатель сходства с непонравившимися играми и обновлять таблицу user_recommendation, добавляя новые N рекомендуемых игр. Наконец, база данных PostgreSQL и сервис FastAPI будут развернуты на Render, чтобы к ним можно было получить доступ за пределами нашей локальной среды. Для этого этапа развертывания можно было бы использовать любой облачный сервис, но в данном случае мы выбрали Render из-за его простоты.

Подводя итог, наш общий рабочий процесс с точки зрения пользователя будет выглядеть следующим образом:

1. Пользователь добавляет игру в свою библиотеку, отправляя POST-запрос из FastAPI в нашу базу данных.
   • Если бы мы хотели прикрепить нашу рекомендательную систему к внешнему приложению, мы могли бы легко привязать этот Post API к пользовательскому интерфейсу.
2. Этот запрос запускает фоновую задачу FastAPI, которая запускает наш рекомендательный конвейер.
3. Рекомендательный конвейер запрашивает в нашей базе данных список игр пользователя и глобальный список игр.
4. Затем рассчитывается показатель схожести между играми пользователя и всеми играми, использующими наши игровые теги.
5. Наконец, наш рекомендательный конвейер отправляет запрос в базу данных, чтобы обновить таблицу рекомендуемых игр для этого пользователя.

Настройка базы данных

Прежде чем создавать нашу рекомендательную систему, первым шагом будет настройка базы данных. Наша базовая схема базы данных показана на рисунке 5. Мы уже обсуждали нашу игровую таблицу; это базовый набор данных, из которого, как правило, берутся все остальные данные. Полный список наших таблиц представлен здесь:

• Таблица игр: содержит базовые данные для каждой уникальной игры в нашей базе данных.
• Таблица пользователей: фиктивная таблица пользователей, содержащая примерную информацию, заполненную для примера.
• Таблица User_Game: содержит сопоставления между всеми играми, которые «понравились» пользователю; эта таблица является одной из базовых таблиц, используемых для формирования рекомендаций путем сбора информации о том, какие игры интересуют пользователя.
• Таблица Game_Tags: содержит сопоставление appid:game_tag, где тег игры может быть чем-то вроде «стратегия», «РПГ», «комедия» — описательным тегом, отражающим суть игры. Каждому appid сопоставлено несколько тегов.
• Таблица рекомендаций пользователя: это наша целевая таблица, которая будет обновляться нашим конвейером. Каждый раз, когда пользователь взаимодействует с новой игрой, наш конвейер рекомендаций запускается и генерирует новую серию рекомендаций для этого пользователя, которые будут храниться здесь.

Чтобы настроить эти таблицы, достаточно просто запустить файл src/load_database.py. Этот файл создаст и заполнит наши таблицы за несколько шагов, описанных ниже. Обратите внимание: сейчас мы сосредоточимся на том, как записать эти данные в общую базу данных, поэтому вам нужно знать только, что External_Database_Url (см. ниже) — это URL-адрес любой базы данных, которую вы хотите использовать. Во второй половине статьи мы рассмотрим, как настроить базу данных в Render и скопировать URL-адрес в ваш файл .env.

from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, Session from sqlalchemy.ext.declarative import declarative_base import os from dotenv import load_dotenv from utils.db_handler import DatabaseHandler import pandas as pd import uuid import sys from sqlalchemy.exc import OperationalError import psycopg2 # Загрузка переменных окружения load_dotenv(override=True) # Создание URL-адреса подключения PostgreSQL для рендеринга URL_database = os.environ.get(«External_Database_Url») # Инициализация DatabaseHandler с помощью нашего URL engine = DatabaseHandler(URL_database) # загрузка начальных данных пользователя users_df = pd.read_csv(«Data/users.csv») games_df = pd.read_csv(«Data/games.csv») user_games_df = pd.read_csv(«Данные/user_games.csv») user_recommendations_df = pd.read_csv(«Данные/user_recommendations.csv») game_tags_df = pd.read_csv(«Данные/game_tags.csv»)

Сначала мы загружаем пять CSV-файлов из папки Data в фреймы данных; у нас есть по одному файлу для каждой таблицы, показанной на схеме нашей базы данных. Мы также устанавливаем соединение с нашими данными, объявляя переменную движка; эта переменная движка использует пользовательский класс DataBaseHandler с методом инициализации, показанным ниже. Этот класс принимает строку подключения к нашей базе данных в Render (или предпочитаемом вами облачном сервисе), переданную из нашего env-файла, и содержит все функции подключения, обновления, удаления и тестирования нашей базы данных.

После загрузки данных и создания экземпляра класса DatabaseHandler нам необходимо определить запрос на создание каждой из пяти таблиц и выполнить эти запросы с помощью функции DatabaseHandler.create_table. Это очень простая функция, которая подключается к нашей базе данных, выполняет запрос и закрывает соединение, оставляя пять таблиц, показанных на схеме базы данных. Однако в данный момент они пусты.

# Определение запросов на создание таблиц user_table_creation_query = «»»СОЗДАТЬ ТАБЛИЦУ IF NOT EXISTS users ( id UUID PRIMARY KEY, username VARCHAR(255) UNIQUE NOT NULL, password VARCHAR(255) NOT NULL, email VARCHAR(255) NOT NULL, role VARCHAR(50) NOT NULL ) «»» game_table_creation_query = «»»СОЗДАТЬ ТАБЛИЦУ IF NOT EXISTS games ( id UUID PRIMARY KEY, appid VARCHAR(255) UNIQUE NOT NULL, name VARCHAR(255) NOT NULL, type VARCHAR(255), is_free BOOLEAN DEFAULT FALSE, short_description TEXT, detailed_description TEXT, developers VARCHAR(255), publishers VARCHAR(255), price VARCHAR(255), жанры VARCHAR(255), категории VARCHAR(255), дата_релиза VARCHAR(255), платформы TEXT, metacritic_score FLOAT, рекомендации INTEGER ) «»» user_games_query = «»»СОЗДАТЬ ТАБЛИЦУ, ЕСЛИ НЕ СУЩЕСТВУЕТ user_games ( id UUID ПЕРВИЧНЫЙ КЛЮЧ, имя пользователя VARCHAR(255) NOT NULL, appid VARCHAR(255) NOT NULL, полка VARCHAR(50) ПО УМОЛЧАНИЮ 'Wish_List', рейтинг FLOAT ПО УМОЛЧАНИЮ 0.0, отзыв TEXT ) «»» advice_table_creation_query = «»»СОЗДАТЬ ТАБЛИЦУ, ЕСЛИ НЕ СУЩЕСТВУЕТ user_recommendations ( id UUID ПЕРВИЧНЫЙ КЛЮЧ, имя пользователя VARCHAR(255), appid VARCHAR(255), сходство FLOAT ) «»» game_tags_creation_query = «»»СОЗДАТЬ ТАБЛИЦУ, ЕСЛИ НЕ СУЩЕСТВУЕТ game_tags ( id UUID ПЕРВИЧНЫЙ КЛЮЧ, appid VARCHAR(255) NOT NULL, category VARCHAR(255) NOT NULL ) «»» # Выполнение запросов для создания таблиц engine.delete_table('user_recommendations') engine.delete_table('user_games') engine.delete_table('game_tags') engine.delete_table('games') engine.delete_table('users') # Создание таблиц engine.create_table(user_table_creation_query) engine.create_table(game_table_creation_query) engine.create_table(user_games_query) engine.create_table(recommendation_table_creation_query) engine.create_table(запрос_создания_тегов_игры)

После первоначальной настройки таблицы мы проводим проверку качества, чтобы убедиться, что в каждом наборе данных есть требуемый столбец ID, заполняем данные из фреймов данных в соответствующую таблицу и затем проверяем правильность заполнения таблиц. Функция test_table возвращает словарь в формате {'table_exists': True, 'table_has_data': True}, если настройка выполнена правильно.

# Гарантия того, что каждая строка каждого кадра данных имеет уникальный идентификатор если 'id' не входит в users_df.columns: users_df['id'] = [str(uuid.uuid4()) for _ in range(len(users_df))] если 'id' не входит в games_df.columns: games_df['id'] = [str(uuid.uuid4()) for _ in range(len(games_df))] если 'id' не входит в user_games_df.columns: user_games_df['id'] = [str(uuid.uuid4()) for _ in range(len(user_games_df))] если 'id' не входит в user_recommendations_df.columns: user_recommendations_df['id'] = [str(uuid.uuid4()) for _ in range(len(user_recommendations_df))] if 'id' not in game_tags_df.columns: game_tags_df['id'] = [str(uuid.uuid4()) for _ in range(len(game_tags_df))] # Заполняет 4 таблицы данными из фреймов данных engine.populate_table_dynamic(users_df, 'users') engine.populate_table_dynamic(games_df, 'games') engine.populate_table_dynamic(user_games_df, 'user_games') engine.populate_table_dynamic(user_recommendations_df, 'user_recommendations') engine.populate_table_dynamic(game_tags_df, 'game_tags') # Проверяет, были ли таблицы созданы и заполнены правильно print(engine.test_table('users')) print(engine.test_table('games')) print(engine.test_table('user_games')) print(engine.test_table('user_recommendations')) print(engine.test_table('game_tags'))

Начало работы с FastAPI

Теперь, когда наша база данных настроена и заполнена, нам нужно создать методы для доступа, обновления и удаления данных с помощью FastAPI. FastAPI позволяет легко создавать стандартизированные (и быстрые) API для взаимодействия с нашей базой данных. В документации FastAPI представлено отличное пошаговое руководство, которое можно найти здесь. Вкратце, есть несколько замечательных функций, которые делают FastAPI идеальным средством взаимодействия между базой данных и клиентским приложением.

1. Стандартизация: FastAPI позволяет нам определять маршруты для взаимодействия с нашими таблицами стандартизированным способом с использованием методов GET, POST, DELETE, UPDATE и т. д. Эта стандартизация позволяет нам создать уровень доступа к данным на чистом Python, который затем может взаимодействовать с широким спектром front-end-приложений. Мы просто вызываем нужные нам методы API во front-end, независимо от того, на каком языке он написан.
2. Валидация данных: Как мы покажем ниже, нам необходимо определить модель данных Pydantic для каждого объекта, с которым мы взаимодействуем (например, наши игры и таблицы пользователей). Главное преимущество этого подхода заключается в том, что он гарантирует наличие определённых типов данных для всех наших переменных. Например, если мы определим объект Game таким образом, что поле рейтинга будет иметь тип float, а пользователь попытается сделать запрос на добавление новой записи с оценкой «отлично», это не сработает. Встроенная валидация данных поможет нам предотвратить всевозможные проблемы с качеством данных, возникающие при масштабировании нашей системы.
3. Асинхронность: функции FastAPI могут выполняться асинхронно, то есть одна из них не зависит от завершения другой. Это может значительно повысить производительность, поскольку медленная задача Fast не будет ожидать завершения медленной.
4. Встроенная документация Swagger: FastAPI имеет встроенный пользовательский интерфейс, к которому мы можем перейти на локальном хосте, что позволяет нам легко тестировать и взаимодействовать с нашими маршрутами.

Модели FastAPI

Часть FastAPI нашего проекта опирается на два основных файла: models.py, который определяет модели данных, с которыми мы будем взаимодействовать (игры, пользователи и т. д.), и main.py, который определяет наше приложение FastAPI и содержит маршруты. В контексте FastAPI маршруты определяют различные пути обработки запросов. Например, у нас может быть маршрут /games для запроса игр из нашей базы данных.

Сначала давайте обсудим наш файл models.py. В этом файле мы определяем все наши модели. Хотя у нас есть разные модели для разных объектов, общий подход будет одинаковым, поэтому мы подробно рассмотрим только модель games, показанную ниже. Первое, что вы заметите ниже, — это то, что для нашего объекта Game определены два реальных класса: класс GameModel, наследующий от базовой модели Pydantic, и класс Game, наследующий от declarative_base sqlalchemy. Возникает естественный вопрос: почему у нас два класса для одной структуры данных (структуры данных нашей игры)?

Если бы мы не использовали базу данных SQL в этом проекте, а вместо этого считывали каждый CSV-файл в фрейм данных при каждом запуске main.py, то нам не понадобился бы класс Game, а только класс GameModel. В этом случае мы бы считывали данные из фрейма данных games.csv, а FastAPI использовал бы класс GameModel для обеспечения корректного соблюдения типов данных.

Однако, поскольку мы используем базу данных SQL, имеет смысл создать отдельные классы для нашего API и нашей базы данных, поскольку эти два класса выполняют несколько разные задачи. Наш класс API отвечает за проверку данных, сериализацию и необязательные поля, а класс базы данных отвечает за специфические для базы данных задачи, такие как определение первичных и внешних ключей, определение таблицы, с которой сопоставлен объект, и защиту защищённых данных. Повторим последний пункт: в нашей базе данных могут быть конфиденциальные поля, предназначенные только для внутреннего использования, и мы не хотим предоставлять их пользователю через API (например, пароль). Эту проблему можно решить, создав отдельный класс Pydantic для пользователя и внутренний класс SQL Alchemy.

Ниже приведен пример того, как это можно реализовать для нашего объекта Games; у нас есть отдельные классы, определенные для других наших таблиц, которые можно найти здесь; однако общая структура та же.

from pydantic import BaseModel from uuid import UUID,uuid4 from typing import Optional from enum import Enum from sqlalchemy import Column, String, Float, Integer import sqlalchemy.dialects.postgresql as pg from sqlalchemy.dialects.postgresql import UUID as SA_UUID from sqlalchemy.ext.declarative import declarative_base import uuid from uuid import UUID # загрузка модели SQL from sqlmodel import Field, Session, SQLModel, create_engine, select # Инициализация базового класса для моделей SQLAlchemy Base = declarative_base() # Это модель Game для базы данных class Game(Base): __tablename__ = «optigame_products» # Имя таблицы в базе данных PostgreSQL id = Column(pg.UUID(as_uuid=True), primary_key=True, default=uuid.uuid4, unique=True, nullable=False) appid = Column(String, unique=True, nullable=False) name = Column(String, nullable=False) type = Column(String, nullable=True) is_free = Column(pg.BOOLEAN, nullable=True, default=False) # short_description = Column(String, nullable=True) detailed_description = Column(String, nullable=True) developer= Column(String, nullable=True) publishers= Column(String, nullable=True) price= Column(String, nullable=True) genres= Column(String, nullable=True) categories= Column(String, nullable=True) release_date= Column(String, nullable=True) platform= Column(String, nullable=True) metacritic_score= Column(Float, nullable=True) suggestions = Column(Integer, nullable=True) class GameModel(BaseModel): id: Optional[UUID] = None appid: str name: str type: Optional[str] = None is_free: Optional[bool] = False short_description: Optional[str] = None detailed_description: Optional[str] = None developer: Optional[str] = None publishers: Optional[str] = None price: Optional[str] = None genres: Optional[str] = None categories: Optional[str] = None release_date: Optional[str] = None platform: Optional[str] = None metacritic_score: Optional[float] = None suggestions: Optional[int] = None class Config: orm_mode = True # Включить режим ORM для работы с объектами SQLAlchemy from_attributes = True # Включить доступ к атрибутам для объектов SQLAlchemy

Настройка маршрутов FastAPI

После определения наших моделей мы можем создать методы для взаимодействия с этими моделями и запроса данных из базы данных (GET), добавления данных в базу данных (POST) или удаления данных из базы данных (DELETE). Ниже приведен пример определения запроса GET для нашей игровой модели. В начале функции main.py у нас есть начальная настройка для получения URL-адреса базы данных и подключения к ней. Затем мы инициализируем наше приложение и добавляем промежуточное ПО для определения URL-адресов, с которых мы будем принимать запросы. Поскольку мы будем развертывать проект FastAPI на Render и отправлять ему запросы с нашей локальной машины, единственным разрешенным источником является порт localhost 8000. Затем мы определяем наш метод app.get с именем fetch_products, который принимает на входе appid, запрашивает в нашей базе данных объекты Game, где appid равен нашему отфильтрованному appid, и возвращает эти продукты.

Обратите внимание, что представленный ниже фрагмент содержит только настройку и метод first get, остальные довольно похожи и доступны в репозитории, поэтому мы не будем давать здесь подробное объяснение каждого из них.

from fastapi import FastAPI, Depends, HTTPException, BackgroundTasks from uuid import uuid4, UUID from sqlalchemy import create_engine from sqlalchemy.ext.declarative import declarative_base from sqlalchemy.orm import sessionmaker, Session from dotenv import load_dotenv import os # Загрузка переменных среды load_dotenv() # импорт безопасности from fastapi.middleware.cors import CORSMiddleware from fastapi.security import OAuth2PasswordBearer # пользовательский импорт from src.models import User, Game, GameModel, UserModel, UserGameModel, UserGame, Game Similarity,Game SimilarityModel, UserRecommendation, UserRecommendationModel from src.similarity_pipeline import UserRecommendationService # Загрузка строки подключения к базе данных из переменной среды или файла .env DATABASE_URL = os.environ.get(«Internal_Database_Url») # создание подключения к базе данных engine = create_engine(DATABASE_URL) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base() # Создание таблиц базы данных (если они еще не существуют) Base.metadata.create_all(bind=engine) # Зависимость для получения сеанса базы данных def get_db(): db = SessionLocal() try: yield db Finally: db.close() # Инициализация приложения FastAPI app = FastAPI(title=»Game Store API», version=»1.0.0″) # Добавление промежуточного ПО CORS для разрешения запросов origins = [«http://localhost:8000»] app.add_middleware( CORSMiddleware, allow_origins=origins, allow_credentials=True, allow_methods=[«*»], allow_headers=[«*»], ) #————————————————-# # ————ЧАСТЬ 1: МЕТОДЫ GET——————-# #————————————————-# @app.get(«/») async def root(): return {«message»: «Hello World»} @app.get(«/api/v1/games/») async def fetch_products(appid: str = None, db: Session = Depends(get_db)): # Запрос к базе данных с использованием игровой модели SQLAlchemy if appid: products = db.query(Game).filter(Game.appid == appid).all() else: products = db.query(Game).all() return [GameModel.from_orm(product) for product in products]

После определения нашего main.py мы наконец можем запустить его из базового каталога проекта с помощью команды ниже.

uvicorn src.main:app —reload

После этого мы можем перейти по адресу http://127.0.0.1:8000/docs и увидеть представленную ниже интерактивную среду FastAPI. На этой странице мы можем протестировать любой из наших методов, определённых в файле main.py. В случае функции fetch_products мы можем передать ей appid и вернуть все соответствующие игры из нашей базы данных.

Создание нашего конвейера подобия

Мы настроили базу данных и можем получать доступ к ней и обновлять её через FastAPI. Теперь пора перейти к центральной функции этого проекта: рекомендательному конвейеру. Рекомендательные системы — хорошо изученная область, и мы не предлагаем здесь никаких инноваций. Однако это наглядный пример реализации базовой рекомендательной системы с использованием FastAPI.

Начало работы — как рекомендовать продукты?

Если задуматься над вопросом «Как бы я рекомендовал новые продукты, которые понравятся пользователю?», то есть два интуитивно понятных подхода.

1. Системы совместных рекомендаций: если у меня есть группа пользователей и группа продуктов, я могу определить пользователей со схожими интересами, посмотрев на их общую корзину продуктов, а затем определить продукты, «отсутствующие» в корзине данного пользователя. Например, если у меня есть пользователи 1–3 и продукты AC, пользователям 1–2 нравятся все три продукта, но пользователю 3 пока понравились только продукты A + B, то я могу порекомендовать ему продукт C. Это логично: у всех трёх пользователей есть высокая степень совпадения в понравившихся им продуктах, но продукт C отсутствует в корзине пользователя 3, есть высокая вероятность, что он ему тоже понравится. Этот процесс генерации рекомендаций путём сравнения пользователей, которые похожи, называется совместной фильтрацией.
2. Система рекомендаций на основе контента: если у меня есть серия товаров, я могу найти товары, похожие на те, которые понравились пользователю, и рекомендовать их. Например, если у меня есть серия тегов для каждой игры, я могу преобразовать серию тегов каждой игры в вектор из единиц и нулей, а затем использовать меру сходства (в данном случае косинусную меру сходства) для измерения сходства между играми на основе этих векторов. После этого я могу вернуть N игр, наиболее похожих на те, которые понравились пользователю, исходя из их степени сходства.

Более подробную информацию о рекомендательных системах можно найти здесь.

Поскольку наш исходный набор данных не содержит большого количества пользователей, у нас нет необходимых данных для предложения товаров на основе сходства пользователей, что известно как проблема холодного старта. Поэтому вместо этого мы разработаем рекомендательную систему на основе контента, поскольку у нас есть значительный объём игровых данных для работы.

Чтобы построить наш конвейер, нам необходимо решить две проблемы: (1) как рассчитать показатели схожести для пользователя и (2) как автоматизировать этот процесс для запуска при каждом обновлении пользователем своих игр?

Мы рассмотрим, как можно запускать конвейер подобия каждый раз, когда пользователь отправляет запрос POST, отмечая игру как понравившуюся, а затем рассмотрим, как построить сам конвейер.

Привязка рекомендательного конвейера к FastAPI

А пока представьте, что у нас есть сервис рекомендаций, который обновляет нашу таблицу user_recommendation. Мы хотим, чтобы этот сервис вызывался каждый раз, когда пользователь обновляет свои настройки. Реализовать это можно в несколько шагов, как показано ниже. Сначала мы определяем функцию generate_recommendations_background, которая отвечает за подключение к нашей базе данных, запуск конвейера сравнения и закрытие соединения. Далее нам нужно обеспечить её вызов, когда пользователь делает запрос на публикацию (например, ставит отметку «Нравится» новой игре). Для этого мы просто добавляем вызов функции в конец функции запроса на публикацию create_user_game.

Результатом этого рабочего процесса является то, что всякий раз, когда пользователь делает запрос на публикацию в нашей таблице user_game, он вызывает функцию create_user_game, добавляет новый объект user_game в базу данных, а затем запускает конвейер подобия как фоновую функцию.

Примечание: Метод записи Below и вспомогательная функция хранятся в main.py вместе с остальными методами FastAPI.

# импорт конвейера подобия из src.similarity_pipeline import UserRecommendationService # Функция фоновой задачи def generate_recommendations_background(username: str, database_url: str): «»»Фоновая задача по генерации рекомендаций для пользователя»»» # Создание нового сеанса базы данных для фоновой задачи background_engine = create_engine(database_url) BackgroundSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=background_engine) db = BackgroundSessionLocal() try: advice_service = UserRecommendationService(db, database_url) advice_service.generate_recommendations_for_user(username) Finally: db.close() # Метод Post, который вызывает функцию фоновой задачи @app.post(«/api/v1/user_game/») async def create_user_game(user_game: UserGameModel, background_tasks: BackgroundTasks, db: Session = Depends(get_db)): # Проверяем, существует ли уже запись exist = db.query(UserGame).filter_by(username=user_game.username, appid=user_game.appid).first() if exist: raise HTTPException(status_code=400, detail=»У пользователя уже есть эта игра.») # Подготавливаем данные со значениями по умолчанию user_game_data = { «username»: user_game.username, «appid»: user_game.appid, «shelf»: user_game.shelf if user_game.shelf is not None else «Wish_List», «rating»: user_game.rating if user_game.rating is not None else 0.0, «review»: user_game.review if user_game.review is not None else «» } if user_game.id is not None: user_game_data[«id»] = UUID(str(user_game.id)) # Сохранить игру пользователя в базе данных db_user_game = UserGame(**user_game_data) db.add(db_user_game) db.commit() db.refresh(db_user_game) # Запустить фоновую задачу для генерации рекомендаций для этого пользователя background_tasks.add_task(generate_recommendations_background, user_game.username, DATABASE_URL) return db_user_game

Создание рекомендательного конвейера

Теперь, когда мы понимаем, как наш конвейер схожести может запускаться, когда пользователь обновляет понравившиеся игры, пора разобраться в механизме работы конвейера рекомендаций. Наш конвейер рекомендаций хранится в файле similarity_pipeline.py и содержит класс UserRecommendationService, импорт и создание экземпляра которого мы показали выше. Этот класс содержит ряд вспомогательных функций, которые в конечном итоге вызываются в методе generate_recommendations_for_user. Существует 7 основных шагов, которые мы рассмотрим последовательно.

1. Извлечение игр пользователя: для генерации рекомендаций похожих игр нам необходимо извлечь игры, которые пользователь уже добавил в свою корзину. Это делается с помощью вызова вспомогательной функции fetch_user_games. Эта функция запрашивает таблицу user_games, используя идентификатор пользователя, который отправляет запрос POST, в качестве входных данных и возвращает все игры в его корзине.
2. Извлечение тегов игр: для сравнения игр нам нужно измерение, по которому их можно сравнивать, и это измерение — теги, связанные с каждой игрой (стратегия, настольная игра и т. д.). Чтобы получить сопоставление «игра:тег», мы вызываем функцию fetch_all_game_tags, которая возвращает теги для всех игр в нашей базе данных.
3. Векторизация тегов игр: Чтобы сравнить сходство между играми A и B, сначала нужно векторизовать теги игр с помощью функции create_game_vectors. Эта функция берёт последовательность всех тегов в алфавитном порядке и проверяет, связан ли каждый из них с заданной игрой. Например, если наш общий набор тегов — [boardgame, deckbuilding, resource-management], а с игрой 1 связан только тег boardgame, то её вектор будет [1, 0, 0].
4. Создаём вектор пользователей: как только у нас есть вектор, представляющий каждую игру, нам нужен агрегированный вектор пользователей для сравнения. Для этого мы используем функцию create_user_vector, которая генерирует агрегированный вектор той же длины, что и векторы игр, который затем можно использовать для вычисления степени сходства между нашим пользователем и всеми остальными играми.
5. Расчет сходства: мы используем векторы, созданные на шагах 3 и 4, в нашей функции calculate_user_recommendations, которая вычисляет показатель косинусного сходства в диапазоне от 0 до 1 и измеряет сходство между каждой игрой и нашими совокупными играми пользователей.
6. Удаление старых рекомендаций: прежде чем заполнять таблицу user_recommendations новыми рекомендациями для пользователя, необходимо удалить старые рекомендации с помощью функции delete_existing_recommendations. Это удаляет только рекомендации пользователя, сделавшего запрос на публикацию; остальные рекомендации остаются без изменений.
7. Заполнение новых рекомендаций: после удаления старых рекомендаций мы заполняем новые с помощью save_recommendations.

из sqlalchemy.orm импорт Session из sqlalchemy импорт create_engine, text из src.models импорт UserGame, UserRecommendation из sklearn.metrics.pairwise импорт cosine_similarity импорт pandas as pd импорт uuid из typing импорт List импорт logging # Настройка ведения журнала logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) class UserRecommendationService: def __init__(self, db_session: Session, database_url: str): self.db = db_session self.database_url = database_url self.engine = create_engine(database_url) def fetch_user_games(self, username: str) -> pd.DataFrame: «»»Извлечь все игры для указанного пользователя»»» query = text(«SELECT username, appid FROM user_games WHERE username = :username») с self.engine.connect() в качестве conn: result = conn.execute(query, {«username»: username}) data = result.fetchall() return pd.DataFrame(data, columns=['username', 'appid']) def fetch_all_category(self) -> pd.DataFrame: «»»Извлечь все игровые теги»»» query = text(«SELECT appid, category FROM category») с self.engine.connect() в качестве conn: result = conn.execute(query) data = result.fetchall() return pd.DataFrame(data, columns=['appid', 'category']) def create_game_vectors(self, tag_df: pd.DataFrame) -> tuple[pd.DataFrame, List[str], List[str]]: «»»Создать игру векторы из тегов «»» unique_tags = tag_df['category'].drop_duplicates().sort_values().tolist() unique_games = tag_df['appid'].drop_duplicates().sort_values().tolist() game_vectors = [] для игры в unique_games: tags = tag_df[tag_df['appid'] == game]['category'].tolist() vector = [1 если тег в тегах else 0 для тега в unique_tags] game_vectors.append(vector) return pd.DataFrame(game_vectors, columns=unique_tags, index=unique_games), unique_tags, unique_games def create_user_vector(self, user_games_df: pd.DataFrame, game_vectors: pd.DataFrame, unique_tags: List[str]) -> pd.DataFrame: «»»Создать вектор пользователя из сыгранных им игр»»» if user_games_df.empty: return pd.DataFrame([[0] * len(unique_tags)], columns=unique_tags, index=['unknown_user']) username = user_games_df.iloc[0]['username'] user_games = user_games_df['appid'].tolist() # Сохранять только те игры, которые есть в game_vectors user_games = [g for g in user_games if g in game_vectors.index] if not user_games: user_vector = [0] * len(unique_tags) else: played_game_vectors = game_vectors.loc[user_games] user_vector = played_game_vectors.mean(axis=0).tolist() return pd.DataFrame([user_vector], columns=unique_tags, index=[имя пользователя]) def calculate_user_recommendations(self, user_vector: pd.DataFrame, game_vectors: pd.DataFrame, top_n: int = 20) -> pd.DataFrame: «»»Рассчитать сходство между вектором пользователя и всеми игровыми векторами»»» username = user_vector.index[0] user_vector_data = user_vector.iloc[0].values.reshape(1, -1) # Рассчитать сходство similarities = cosine_similarity(user_vector_data, game_vectors) similarity_df = pd.DataFrame(similarities.T, index=game_vectors.index, columns=[имя пользователя]) # Получить N лучших рекомендаций top_games = similarity_df[имя пользователя].nlargest(top_n) suggestions = [] for appid, similarity in top_games.items(): suggestions.append({ «username»: имя пользователя, «appid»: appid, «similarity»: float(similarity) }) return pd.DataFrame(recommendations) def delete_existing_recommendations(self, username: str): «»»Удалить существующие рекомендации для пользователя»»» self.db.query(UserRecommendation).filter(UserRecommendation.username == username).delete() self.db.commit() def save_recommendations(self, suggestions_df: pd.DataFrame): «»»Сохранить новые рекомендации в базе данных»»» for _, row in suggestions_df.iterrows(): advice = UserRecommendation( id=uuid.uuid4(), username=row['username'], appid=row['appid'], similarity=row['similarity'] ) self.db.add(recommendation) self.db.commit() def generate_recommendations_for_user(self, username: str, top_n: int = 20): «»»Основной метод для генерации рекомендаций для определенного пользователя»»» try: logger.info(f»Начало генерации рекомендаций для пользователя: {username}») # 1. Извлечь игры пользователя user_games_df = self.fetch_user_games(username) if user_games_df.empty: logger.warning(f»Игры не найдены для пользователя: {username}») return # 2. Извлечь все теги игр tag_df = self.fetch_all_category() if tag_df.empty: logger.error(«Теги игр не найдены в базе данных») return # 3. Создать векторы игр game_vectors, unique_tags, unique_games = self.create_game_vectors(tag_df) # 4. Создать вектор пользователя user_vector = self.create_user_vector(user_games_df, game_vectors, unique_tags) # 5. Рассчитать рекомендации suggestions_df = self.calculate_user_recommendations(user_vector, game_vectors, top_n) # 6. Удалить существующие рекомендации self.delete_existing_recommendations(username) # 7. Сохраните новые рекомендации self.save_recommendations(recommendations_df) logger.info(f»Успешно сгенерировано {len(recommendations_df)} рекомендаций для пользователя: {username}») except Exception as e: logger.error(f»Ошибка генерации рекомендаций для пользователя {username}: {str(e)}») self.db.rollback() raise

Подведение итогов

В этой статье мы рассмотрели, как настроить базу данных PostgreSQL и приложение FastAPI для запуска системы рекомендаций игр. Однако мы пока не рассматривали, как развернуть эту систему в облачном сервисе, чтобы другие могли с ней взаимодействовать. Подробнее об этом читайте во второй части.

Рисунки : Все изображения, если не указано иное, принадлежат автору.

Ссылки

1. Репозиторий Github для проекта: https://github.com/pinstripezebra/recommender_system
2. Документация FastAPI: https://fastapi.tiangolo.com/tutorial/
3. Рекомендательные системы: https://en.wikipedia.org/wiki/Recommender_system
4. Косинусное подобие: https://en.wikipedia.org/wiki/Cosine_similarity)

Источник: towardsdatascience.com