Как использовать фреймворк Perseus для решения задач
Привет! Я Аня Никифорова, ML-разработчик по направлению рекомендательных систем в Т-Банке. Этим летом на Turbo ML Conf 2026 мы представили фреймворк Perseus. Он подходит для работы с гетерогенными последовательностями событий пользователей, хорошо масштабируется под различные задачи и из коробки поддерживает кросс-доменные сценарии.
В силу своей гибкости фреймворк может показаться сложным, поэтому мы решили поделиться подробным туториалом, где по шагам рассказываем, как использовать Perseus для своих задач. Посмотрим, как с помощью Perseus реализовать кандидатогенерацию, ранжирование, классификацию и регрессию поверх истории действий пользователей в сервисах: как подготовить данные, какие команды вызвать и как оценить результат. На примере кандидатогенерации подробно рассмотрим, как добавлять в модель дополнительные фичи, обогащать ее событиями и как настраивать модель, меняя лишь пару строк в конфиге.
План туториала
Код туториала опубликован, — можно изучать и пользоваться.
Будем работать с датасетом T-ECD, он был опубликован нами в сентябре 2025 и недавно представлен на конференции KDD-2026 (A*) в Южной Корее. T-ECD основан на данных сервисов, где Perseus уже доказал свою эффективность в продакшене. Речь идет о повышенном кэшбеке, Шопинге и Супермаркетах, которые можно найти в разделе «Город» мобильного приложения Т-Банка.
Датасет хорошо иллюстрирует, что такое экосистема: у нас есть разнообразные сервисы, в которых клиенты совершают действия, и некоторые клиенты пользуются сразу несколькими сервисами. Более того, события, различные по своей природе, могут содержать признаки, указывающие на одну сущность. Например, бренд товара фигурирует и при покупке в магазине, и при заказе на сайте. Всего в датасете представлено пять доменов, и Perseus позволяет легко использовать все многообразие экосистемных данных для улучшения качества на целевой задаче в конкретном домене.
План туториала:
Показать, как Perseus выглядит с точки зрения ML-разработчика и что нужно знать, чтобы работать с фреймворком.
Подготовить данные для дальнейших экспериментов.
Посмотреть, как решать задачу кандидатогенерации.
Построить базовую модель исключительно на последовательности item_id.
Обогатить модель дополнительными фичами, событиями и настроить архитектурные компоненты модели (поменять тип бэкбона и пулинга).
Построить пайплайн ранжирования с помощью Perseus.
Собрать пайплайн классификации.
Обучить регрессию.
В каждой из задач обязательно сравнимся с бейзлайнами.
Что важно знать про работу с Perseus
Зона ответственности ML-инженера при работе с фреймворком ограничивается четырьмя шагами:
Загрузкой списка событий, которые модель сможет использовать для обучения, в хранилище (Event Hub).
Подготовкой представления данных (timestamp, client_id, target) в формате, требуемом для конкретной задачи: кандидатогенерации, ранжирования, классификации или регрессии.
Созданием YAML-конфига, задающего параметры модели.
Запуском команд обучения и инференса.
Все остальные операции автоматически выполняются фреймворком.
Мы собрали глоссарий, чтобы все термины воспринимались в нужном контексте.
Данные:
Событие — факт действия клиента в определенный момент (timestamp, client_id). У события всегда есть тип и опционально атрибуты, например id или бренд товара, с которым пользователь провзаимодействовал. События могут быть любыми — покупка товара, прослушивание музыки, обращение в поддержку — и не обязаны иметь одинаковую схему.
Event Hub — единое хранилище событий, из которого Perseus собирает историю клиента. Если нужного события в нем нет, ML-разработчик добавляет его туда сам. Event Hub достаточно собрать единожды, а затем обращаться к нему при различных задачах. Тем не менее Event Hub не является неизменным и его в любой момент можно обогащать новыми событиями.
Признак — то, что модель учитывает. Признак может лежать в событии, контексте или артефактах, и в конфиге для каждого признака это указывается явно (located_in).
Энкодер превращает признак в эмбеддинг: id для категориальных, ple для числовых, bag-of-words для текстовых. Один энкодер можно переиспользовать для нескольких признаков. Например, бренд из разных доменов попадет в общее пространство.
Устройство базиса:
Базис — постановка ML-задачи, которая задается как датасет объектов, на которых модель обучается и инференсится. Для обучения базис нужно разделить на train- и test-фолды, лучше всего по времени. По test-фолду фреймворк отслеживает метрики от эпохи к эпохе.
Объект базиса (сэмпл) — пара (client_id, timestamp): клиент в фиксированный момент. Именно для него модель формирует эмбеддинг и делает предсказание.
Контекст — признаки уровня объекта базиса, то есть те, что относятся ко всему сэмплу целиком, а не к отдельному событию. Например, соцдем-кластер клиента или флаг, является ли дата праздничным днем. В контексте можно указать только те признаки, которые также будут доступны на инференсе.
Таргет — правильный ответ для объекта базиса. Его вид зависит от задачи: список айтемов для кандидатогенерации, список айтемов с релевантностями для ранжирования, метка класса для классификации, число для регрессии.
Артефакты — дополнительная информация о таргете, которую неудобно хранить в самих сэмплах. Для кандидатогенерации и ранжирования это таблица айтемов (и их признаков), у классификации и регрессии артефактов нет.
Группы — срезы, в которых дополнительно (помимо overall) считаются метрики. Актуальны только для test-фолда.
Архитектура модели:
Бэкбон сводит историю событий и контекст в один эмбеддинг клиента. Сначала event_aggregator векторизует каждое событие, context_aggregator — контекст, а затем history_aggregator обрабатывает полученную последовательность и выдает итоговый вектор. history_aggregator — это ядро бэкбона, именно он отвечает за sequence modeling. Доступные варианты: modern_bert (дефолт), bert, ligr, danet, hstu и mamba. Почти все они основаны на трансформерах, не считая mamba.
Голова — то, что считается поверх эмбеддинга клиента для получения предсказания. Обучается end-to-end вместе с бэкбоном.
Пайплайн:
Конфиг — один YAML-файл, в котором описано все перечисленное: задача и метрики, используемые события и их атрибуты, признаки и энкодеры, бэкбон и гиперпараметры обучения и инференса.
Workdir — рабочая директория, с которой работают команды фреймворка. ML-разработчик кладет в нее базис и конфиг, а Perseus складывает туда все, что считает (ее структура — в конце раздела). Важная оговорка, что Event Hub и Workdir — разные сущности, которые не обязаны физически соседствовать.
Perseus для каждого сэмпла базиса из Event Hub собирает предшествующий ему набор событий клиента. Сиквенс событий и контекст пропускаются через энкодеры и бэкбон. Полученное скрытое состояние проходит через голову, которая формирует предсказание, это предсказание сравнивается с таргетом, и считается лосс. Event Hub связывается с базисом через (timestamp, client_id).
Хранилище событий — Event Hub
Event Hub — единое хранилище событий, из которых Perseus собирает сиквенс для клиента. Путь до Event Hub нужно прописать в переменные окружения.
Каждый тип событий в Event Hub хранится в отдельной папке, а сами события сгруппированы по дням и партиционированы для удобства обращения к ним фреймворка.
Для работы с Event Hub удобно использовать следующие команды:
python -m perseus event-hub add-events events.pq --name transactions-purchase # добавить события из parquet-файла
python -m perseus event-hub list-events # показать доступные события (диапазон дат + атрибуты)
python -m perseus event-hub delete-events --name transactions-purchase --date-from 2025-01-01 --date-to 2025-06-30 # удалить события за периодПостановка задачи
Тип решаемой задачи выбирает ML-разработчик. Задача задает вид таргета и артефактов, набор доступных метрик и голову. Несколько примеров, чтобы понять, какую постановку ML-задачи выбрать:
Бизнес хочет рекомендовать клиентам те товары, которые они с наибольшей вероятностью купят в супермаркете. Это классическая задача рекомендаций. Здесь важно не просто предсказать вероятность покупки, а предложить клиенту ограниченный набор наиболее релевантных товаров. Такую задачу можно сформулировать как задачу кандидатогенерации — отбора кандидатов.
Есть готовый пул товаров, которые нужно упорядочить в ленте так, чтобы на самых верхних позициях оказались те, по которым клиент с наибольшей вероятностью совершит покупку. В отличие от предыдущего примера здесь важен не факт попадания товара в подборку, а именно порядок: чем выше релевантный товар, тем лучше и ошибки на первых позициях критичнее, чем на последних. Это классическая задача ранжирования.
Пусть от бизнеса пришла задача — предсказывать, допустит ли клиент дефолт по своим обязательствам. По своей природе это задача с двумя исходами: дефолт наступит или нет. Поэтому с точки зрения ML-моделирования здесь логично рассматривать бинарную классификацию.
Нужно предсказать, сколько клиент потратит в следующем месяце по всем своим счетам. Здесь целевая переменная принимает непрерывные значения, поэтому постановка задачи — регрессия.
Perseus работает с представлением пользователя, поэтому задачи регрессии и классификации должны быть связаны с пользователями. Так, с помощью Perseus нельзя предсказать цену товара, но можно предсказать суммарные траты пользователя в следующем месяце.
Еще одно решение, которое принимает ML-разработчик, — насколько дробным брать timestamp в базисе. От этого зависит и сама постановка, и количество обучающих сэмплов. Можно предсказывать следующее видео, которое лайкнет пользователь (timestamp с точностью до наносекунд), а можно — видео, которые пользователь лайкнет в течение следующего дня (timestamp, округленный до даты).
Общая рекомендация — группировать базис в соответствии с тем, как часто обновляются рекомендации в сервисе: если раз в сутки, имеет смысл округлять timestamp до даты. То же самое касается задержек при поставке продакшен-данных. Если в сервисе данные приходят с задержкой в час, правильно будет вычесть этот час из timestamp базиса — иначе на обучении модель будет видеть историю, которой в момент предсказания в продакшене еще не окажется.
Запуск пайплайна
Определившись с постановкой задачи, ML-разработчик должен подготовить базис (basis/) и конфиг (config.yaml) в рабочей директории. А дальше все делается командами фреймворка. Обучение:
python -m perseus train prepare-dataset --workdir <workdir> # собрать датасет из базиса и событий
accelerate launch -m perseus train fit-model --workdir <workdir> # обучить модель, результат — в checkpoint/В итоге рабочая директория выглядит так. ML-разработчик готовит только basis/ и config.yaml, все остальное появляется само по мере вызова команд:
workdir/
├── basis/ # ML-разработчик
│ ├── train/ # фолд для обучения
│ │ ├── samples.pq # объекты базиса с таргетом
│ │ └── artifacts/items.pq # айтемы и их признаки
│ ├── test/ # фолд для валидации, структура та же
│ │ ├── samples.pq
│ │ └── artifacts/items.pq
│ ├── samples.pq # базис для инференса
│ └── artifacts/items.pq # айтемы для инференса
├── config.yaml # ML-разработчик
├── dataset/ # Perseus, prepare-dataset: датасет для обучения
└── checkpoint/ # Perseus, prepare-dataset и fit-model: препроцессоры, а после завершения обучения — веса моделиДля инференса можно завести отдельную рабочую директорию, скопировав в нее чекпойнт обученной модели, либо продолжить работать с той же директорией, которая использовалась во время обучения. Еще нужно подготовить базис для инференса — от базиса, использующегося для обучения, он отличается только отсутствием таргета. Команды инференса:
python -m perseus inference distribute-samples --workdir <workdir> # разложить базис по партициям
python -m perseus inference make-items-embeddings --workdir <workdir> # посчитать эмбеддинги айтемов (необязательный шаг)
python -m perseus inference make-backbone-embeddings --workdir <workdir> # посчитать эмбеддинги клиентов
python -m perseus inference make-head-predictions --workdir <workdir> # получить предсказания в predictions/Рабочая директория после инференса выглядит так:
workdir/
├── basis/ # ML-разработчик
│ ├── samples.pq # базис для инференса
│ └── artifacts/items.pq # айтемы для инференса
├── checkpoint/ # Perseus, prepare-dataset и fit-model: препроцессоры и веса модели
├── samples/ # Perseus, distribute-samples: базис для инференса, разложенный по партициям
├── embeddings/ # Perseus, make-backbone-embeddings: эмбеддинги клиентов
├── items/ # Perseus, make-items-embeddings: эмбеддинги айтемов
└── predictions/ # Perseus, make-head-predictions: итоговые предсказанияАртефакты (artifacts/items.pq) нужны только для кандидатогенерации и ранжирования — у классификации и регрессии в базисе лежат только samples.pq. Метрики из конфига Perseus считает сам, но только на этапе обучения и только на test-фолде. Метрики на инференсе, в том числе сравнение с бейзлайном, — уже ответственность ML-разработчика.
Важно для честного сравнения: предсказания получают только те объекты базиса, по которым есть хотя бы одно событие раньше timestamp сэмпла, иначе сэмпл выпадает и строк в predictions/ оказывается меньше, чем в базисе. Поэтому в туториале мы считаем бейзлайн на полном базисе и им же заполняем пропуски в предсказаниях модели.
Эксперименты на T-ECD
Пройдем по всем четырем типам задач. Порядок действий в каждой из них одинаковый: соберем базис, посчитаем бейзлайн, обучим модель, проинференсим ее и сравним метрики на одном и том же inference-базисе. Все эксперименты мы запускали на одной H100, линейный прогон занимает около 8 часов.
Работать будем с малой версией T-ECD, домен Marketplace: события четырех типов (view, click, like, clickout), справочник товаров с брендом и ценой и справочник пользователей с соцдем-кластером. Добавим данные из доменов Retail и Offers.
Listing
snapshot_download(
repo_id="t-tech/T-ECD",
repo_type="dataset",
allow_patterns="dataset/small/marketplace/",
local_dir=download_dir
)
snapshot_download(
repo_id="t-tech/T-ECD",
repo_type="dataset",
allow_patterns="dataset/small/users.pq",
local_dir=download_dir
)
snapshot_download(
repo_id="t-tech/T-ECD",
repo_type="dataset",
allow_patterns="dataset/small/retail/",
local_dir=download_dir
)
snapshot_download(
repo_id="t-tech/T-ECD",
repo_type="dataset",
allow_patterns="dataset/small/offers/",
local_dir=download_dir
)
EVENTS = pl.read_parquet(download_dir / "dataset/small/marketplace/events")
EVENTS = EVENTS.select(
pl.col("action_type").alias("event"),
pl.col("timestamp").dt.total_microseconds().cast(pl.Datetime("us")).dt.truncate("1s").cast(pl.Datetime("ns")).alias("timestamp"),
pl.col("user_id").cast(pl.String).alias("client_id"),
"item_id",
"subdomain",
).unique()
CLIENTS = pl.read_parquet(download_dir / "dataset/small/users.pq", columns=["user_id", "socdem_cluster"])
CLIENTS = CLIENTS.with_columns(pl.col("user_id").cast(pl.String).alias("client_id")).drop("user_id")
ITEMS = pl.read_parquet(download_dir / "dataset/small/marketplace/items.pq", columns=["item_id", "brand_id", "price"])
retail_events = (
pl.scan_parquet(download_dir / "dataset/small/retail/events")
.filter(pl.col("action_type").eq("order"))
.select(
pl.col("action_type").alias("event"),
pl.col("timestamp").dt.total_microseconds().cast(pl.Datetime("us")).dt.truncate("1s").cast(pl.Datetime("ns")).alias("timestamp"),
pl.col("user_id").cast(pl.String).alias("client_id")
).unique()
).collect()
offers_events = (
pl.scan_parquet(download_dir / "dataset/small/offers/events")
.filter(~pl.col("action_type").eq("view"))
.select(
pl.col("action_type").alias("event"),
pl.col("timestamp").dt.total_microseconds().cast(pl.Datetime("us")).dt.truncate("1s").cast(pl.Datetime("ns")).alias("timestamp"),
pl.col("user_id").cast(pl.String).alias("client_id"),
pl.col("item_id")
).unique()
.join(
pl.scan_parquet(download_dir / "dataset/small/offers/items.pq").select("item_id", "brand_id"),
on="item_id",
how="left"
)
.drop("item_id")
).collect()Создание Event Hub
Сначала подготовим события. В нашем случае Event Hub — локальная директория, путь до которой указывается в env-файле. Каждое событие обязано содержать timestamp (тип ns) и client_id (строка), а все остальное — необязательные атрибуты, которые дальше можно будет использовать как признаки: для Marketplace это item_id и subdomain (рекомендательная поверхность, где было совершено событие).
Каждый тип события загружается отдельной командой под своим именем — так в Event Hub появятся marketplace-view, marketplace-click, marketplace-like и marketplace-clickout.
Listing
env = {**os.environ, "INTERNAL_STORAGE_EVENT_HUB": "/event-hub/"}
for (event,), group in EVENTS.group_by("event"):
event_name = f"marketplace-{event}"
group = group.drop("event")
with tempfile.TemporaryDirectory() as tmp:
staging = Path(tmp) / "events.pq"
group.write_parquet(staging)
subprocess.run(
[
"uv", "run", "python", "-m", "perseus",
"event-hub", "add-events",
str(staging),
"--name", event_name,
"--source", "event_hub"
],
env=env,
check=True
)
Аналогично добавим события Offers (offers-click, offers-clickout, offers-like) и Retail (retail-order).
Listing
for (event,), group in retail_events.group_by("event"):
event_name = f"retail-{event}"
group = group.drop("event")
with tempfile.TemporaryDirectory() as tmp:
staging = Path(tmp) / "events.pq"
group.write_parquet(staging)
subprocess.run(
[
"uv", "run", "python", "-m", "perseus",
"event-hub", "add-events",
str(staging),
"--name", event_name,
"--source", "event_hub"
],
env=env,
check=True
)
for (event,), group in offers_events.group_by("event"):
event_name = f"offers-{event}"
group = group.drop("event")
with tempfile.TemporaryDirectory() as tmp:
staging = Path(tmp) / "events.pq"
group.write_parquet(staging)
subprocess.run(
[
"uv", "run", "python", "-m", "perseus",
"event-hub", "add-events",
str(staging),
"--name", event_name,
"--source", "event_hub"
],
env=env,
check=True
)Соцдем-кластер клиента мы положим в контекст базиса, а признаки товаров — в артефакты. Существует возможность также приджойнить эти признаки к событиям Event Hub, чтобы они обрабатывались энкодером на уровне события.
Кандидатогенерация
Будем предсказывать, на какие товары пользователь наиболее вероятно кликнет в Marketplace на следующий день. В качестве метрик возьмем Recall@100, NDCG@100 и Coverage@100.
Базису для кандидатогенерации нужны артефакты — таблица items.pq с обязательной колонкой item. Остальные ее колонки можно использовать как признаки айтема. В train/samples.pq таргет — список айтемов, с которыми клиент провзаимодействовал полезным для бизнеса образом. Негативы фреймворк сгенерирует сам во время обучения. В test/samples.pq у каждого айтема в таргете дополнительно указывается релевантность (везде 1, так как товары не различаются по уровню релевантности) — она используется при расчете метрики NDCG. В артефактах каждого фолда должны быть все айтемы, встречающиеся в его таргете.
Соберем базис из событий кликов за последние 120 дней. timestamp округлим до даты и сдвинем на −12 часов (ограничение T-ECD). Таргет сэмпла — все товары, на которые клиент кликнул в этот день. Разделим базис на train и test в отношении 80/20 по времени. В тестовой части проставим всем айтемам релевантность 1, так как товары не отличаются друг от друга по степени полезности. В контекст положим соцдем-кластер клиента, в артефакты — айтемы из train-части вместе с брендом. Тестовые артефакты продублируют тренировочные: модель завязана на id товара и не сможет рекомендовать то, что не видела при обучении.
Listing
samples = pl.read_parquet("/event-hub/marketplace-click")
start_date = samples["date"].max() - timedelta(days=120)
samples = (
samples
.filter(pl.col("timestamp") >= start_date)
.with_columns((pl.col("date").cast(pl.Datetime) - timedelta(hours=12)).cast(pl.Datetime("ns")).alias("timestamp"))
.drop("date")
.group_by("timestamp", "client_id").agg(pl.col("item_id").unique().alias("target"))
.sort("timestamp")
)
train_ratio = 0.8
split_idx = int(len(samples) * train_ratio)
split_timestamp = samples["timestamp"][split_idx]
train_samples = samples.filter(pl.col("timestamp") < split_timestamp)
test_samples = samples.filter(pl.col("timestamp") >= split_timestamp)
test_samples = (
test_samples
.with_columns(
pl.col("target").list.eval(
pl.struct([
pl.element().alias("item"),
pl.lit(1).alias("relevance")
])
).alias("target")
)
)
train_samples = train_samples.join(CLIENTS, on="client_id", how="left")
train_samples = train_samples.with_columns(
pl.struct(["socdem_cluster"]).alias("context")
).drop(["socdem_cluster"])
test_samples = test_samples.join(CLIENTS, on="client_id", how="left")
test_samples = test_samples.with_columns(
pl.struct(["socdem_cluster"]).alias("context")
).drop(["socdem_cluster"])
train_samples.write_parquet("../retrieval/basis/train/samples.pq")
test_samples.write_parquet("../retrieval/basis/test/samples.pq")
artifacts = pl.DataFrame(train_samples["target"].explode().unique())
artifacts = artifacts.join(ITEMS, left_on="target", right_on="item_id", how="left").rename({"target": "item"}).select("item", "brand_id")
artifacts.write_parquet("../retrieval/basis/train/artifacts/items.pq")
artifacts.write_parquet("../retrieval/basis/test/artifacts/items.pq")Получается такой train/samples.pq:
В test/samples.pq к каждому айтему таргета добавляется релевантность:
А artifacts/items.pq, одинаковый для обоих фолдов, выглядит так:
Базис для инференса — копия тестовой части вместе с артефактами. Подготовим его один раз, до бейзлайна и первого обучения. Базис при этом не меняется, поэтому метрики всех вариантов модели останутся сравнимыми между собой и с бейзлайном.
Listing
inference_basis = pl.read_parquet("../retrieval/basis/test/samples.pq")
inference_basis.write_parquet("../retrieval/basis/samples.pq")
inference_artifacts = pl.read_parquet("../retrieval/basis/test/artifacts/items.pq")
inference_artifacts.write_parquet("../retrieval/basis/artifacts/items.pq")uv run python -m perseus inference distribute-samples --workdir ../retrievalВ качестве бейзлайна возьмем топ-100 самых популярных айтемов из train-части базиса и порекомендуем их всем пользователям. Скор айтема — его позиция в топе, так что порядок внутри рекомендаций тоже определен.
Listing
toppop_items = (
train_samples.explode("target")["target"]
.value_counts().sort("count", descending=True).head(100)["target"].to_list()
)
toppop_prediction = [
{"item": item, "score": float(len(toppop_items) - rank)}
for rank, item in enumerate(toppop_items)
]Посчитаем бейзлайн сразу, до обучения модели: на инференсе сэмплы без истории отбрасываются, поэтому предсказания модели мы потом приджойним к полному базису и заполним пропуски бейзлайном. Так обе оценки окажутся на одном и том же наборе сэмплов. Дальше во всех задачах будем поступать точно так же:
Вариант | Recall@100 | NDCG@100 | Coverage@100 |
Топ популярных (бейзлайн) | 0,1275 | 0,0430 | 0,0005 |
Первая модель будет использовать единственный признак — последовательность item_id. В конфиге опишем задачу и метрики, целевое событие, энкодер айтема, бэкбон (ModernBERT на 4 слоя, сумму как агрегатор событий) и параметры обучения и инференса.
config1.yaml
task:
type: retrieval
metrics:
recall@100:
type: recall_at_k
params:
k: 100
ndcg@100:
type: ndcg_at_k
params:
k: 100
coverage@100:
type: coverage_at_k
params:
k: 100
events:
marketplace-click:
attributes:
item_id:
max_duration_per_sequence: 365d
max_events_per_sequence: 512
encoders:
item:
type: id
features:
item_id:
located_in:
event: true
encoder: item
item:
located_in:
artifacts: true
encoder: item
backbone:
dim: 256
history_aggregator:
type: modern_bert
params:
num_layers: 4
num_heads: 4
dropout: 0.1
event_aggregator:
type: sum
training:
num_epochs: 10
log_every_n_train_steps: 1000
dataloader:
batch_size: 32
num_workers: 2
pin_memory: true
shuffle: true
test_dataloader:
batch_size: 32
num_workers: 2
pin_memory: true
early_stopping:
metric: recall@100
mode: max
patience: 3
optimizer:
params:
lr: 0.0003
inference:
backbone_dataloader:
batch_size: 64
num_workers: 0
pin_memory: true
head_dataloader:
batch_size: 64
num_workers: 0
pin_memory: true
predict_kwargs:
k: 1000Дальше достаточно двух команд: собрать датасет и обучить модель:
!uv run python -m perseus train prepare-dataset --workdir ../retrieval
!uv run accelerate launch -m perseus train fit-model --workdir ../retrievalВесь процесс обучения будет автоматически документироваться в виде текстовых логов.
При настроенном ClearML увидим следующую картину.
После обучения проинференсим модель и посчитаем по ее предсказаниям те же метрики, что и для бейзлайна:
uv run python -m perseus inference make-backbone-embeddings --workdir ../retrieval
uv run python -m perseus inference make-head-predictions --workdir ../retrievalВариант | Recall@100 | NDCG@100 | Coverage@100 |
Топ популярных (бейзлайн) | 0,1275 | 0,0430 | 0,0005 |
Perseus, только item_id | 0,1518 | 0,0545 | 0,0056 |
Отдельной командой можно получить эмбеддинги айтемов. Они могут пригодиться, например, для поиска похожих товаров или как признаки в другой модели.
uv run python -m perseus inference make-items-embeddings --workdir ../retrievalВ Marketplace представлен не только item_id. Добавим в модель подраздел сервиса (subdomain) из событий, бренд товара из артефактов и соцдем-кластер клиента из контекста. Новой подготовки данных не потребуется: все это мы сохранили еще на этапе сбора Event Hub и базиса, поэтому достаточно дописать признаки в конфиг, указав для каждого, где он расположен и каким энкодером кодируется.
config2.yaml
task:
type: retrieval
metrics:
recall@100:
type: recall_at_k
params:
k: 100
ndcg@100:
type: ndcg_at_k
params:
k: 100
coverage@100:
type: coverage_at_k
params:
k: 100
events:
marketplace-click:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
max_events_per_sequence: 512
encoders:
item:
type: id
subdomain:
type: id
brand_id:
type: id
socdem_cluster:
type: id
features:
item_id:
located_in:
event: true
encoder: item
item:
located_in:
artifacts: true
encoder: item
subdomain:
located_in:
event: true
encoder: subdomain
brand_id:
located_in:
artifacts: true
encoder: brand_id
socdem_cluster:
located_in:
context: true
encoder: socdem_cluster
backbone:
dim: 256
history_aggregator:
type: modern_bert
params:
num_layers: 4
num_heads: 4
dropout: 0.1
event_aggregator:
type: sum
context_aggregator:
type: identity
training:
num_epochs: 10
log_every_n_train_steps: 1000
dataloader:
batch_size: 32
num_workers: 2
pin_memory: true
shuffle: true
test_dataloader:
batch_size: 32
num_workers: 2
pin_memory: true
early_stopping:
metric: recall@100
mode: max
patience: 3
optimizer:
params:
lr: 0.0003
inference:
backbone_dataloader:
batch_size: 64
num_workers: 0
pin_memory: true
head_dataloader:
batch_size: 64
num_workers: 0
pin_memory: true
predict_kwargs:
k: 1000Датасет пересоберем, модель обучим и проинференсим заново:
uv run python -m perseus inference make-backbone-embeddings --workdir ../retrieval
uv run python -m perseus inference make-head-predictions --workdir ../retrieval
uv run python -m perseus inference make-items-embeddings --workdir ../retrievalВариант | Recall@100 | NDCG@100 | Coverage@100 |
Perseus, только item_id | 0,1518 | 0,0545 | 0,0056 |
Perseus + признаки | 0,1553 | 0,0572 | 0,0099 |
Добавление признаков позволило немного улучшить качество модели. Теперь обогатим модель событиями. Помимо кликов добавим лайки и кликауты Marketplace, а также события соседних доменов: заказы в Retail и клики, лайки и кликауты в Offers. Дополним конфиг.
Обратим внимание на две вещи. У событий Offers нет item_id, зато есть бренд — тот же признак, что и в артефактах Marketplace, поэтому кодировать его будем общим энкодером и информация из другого домена попадет в то же пространство. А еще событий стало значительно больше, а длина истории ограничена, поэтому зададим целевому событию более высокий приоритет. Иначе клики по товарам вытеснятся из последовательности остальными событиями.
config3.yaml
task:
type: retrieval
metrics:
recall@100:
type: recall_at_k
params:
k: 100
ndcg@100:
type: ndcg_at_k
params:
k: 100
coverage@100:
type: coverage_at_k
params:
k: 100
events:
marketplace-click:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 2
marketplace-like:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 1
marketplace-clickout:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 1
retail-order:
max_duration_per_sequence: 365d
offers-click:
attributes:
brand_id:
max_duration_per_sequence: 365d
offers-like:
attributes:
brand_id:
max_duration_per_sequence: 365d
offers-clickout:
attributes:
brand_id:
max_duration_per_sequence: 365d
max_events_per_sequence: 512
encoders:
item:
type: id
subdomain:
type: id
brand_id:
type: id
socdem_cluster:
type: id
features:
item_id:
located_in:
event: true
encoder: item
item:
located_in:
artifacts: true
encoder: item
subdomain:
located_in:
event: true
encoder: subdomain
brand_id:
located_in:
event: true
artifacts: true
encoder: brand_id
socdem_cluster:
located_in:
context: true
encoder: socdem_cluster
backbone:
dim: 256
history_aggregator:
type: modern_bert
params:
num_layers: 4
num_heads: 4
dropout: 0.1
event_aggregator:
type: sum
context_aggregator:
type: identity
training:
num_epochs: 10
log_every_n_train_steps: 1000
dataloader:
batch_size: 32
num_workers: 2
pin_memory: true
shuffle: true
test_dataloader:
batch_size: 32
num_workers: 2
pin_memory: true
early_stopping:
metric: recall@100
mode: max
patience: 3
optimizer:
params:
lr: 0.0003
inference:
backbone_dataloader:
batch_size: 64
num_workers: 0
pin_memory: true
head_dataloader:
batch_size: 64
num_workers: 0
pin_memory: true
predict_kwargs:
k: 1000Снова соберем датасет, обучим и проинференсим модель с помощью уже знакомых команд:
uv run python -m perseus inference make-backbone-embeddings --workdir ../retrieval
uv run python -m perseus inference make-head-predictions --workdir ../retrieval
uv run python -m perseus inference make-items-embeddings --workdir ../retrievalВариант | Recall@100 | NDCG@100 | Coverage@100 |
Perseus + признаки | 0,1553 | 0,0572 | 0,0099 |
Perseus + события | 0,1638 | 0,0634 | 0,0046 |
Видим приросты в качестве относительно предыдущей версии модели. В этом и есть основная сила Perseus: он позволяет учитывать в пользовательской истории события из разных доменов с разными схемами. Более того, из обширного Event Hub можно подключать только нужный набор событий, тем самым обучая модели на разных срезах данных без необходимости их перезаписи.
Наконец, изменим архитектуру: заменим ModernBERT на HSTU (он учитывает не только порядок событий, но и время между ними), а сумму в агрегаторе событий — на взвешенную сумму. Для этого достаточно поменять пару строк в конфиге.
config4.yaml
task:
type: retrieval
metrics:
recall@100:
type: recall_at_k
params:
k: 100
ndcg@100:
type: ndcg_at_k
params:
k: 100
coverage@100:
type: coverage_at_k
params:
k: 100
events:
marketplace-click:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 2
marketplace-like:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 1
marketplace-clickout:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 1
retail-order:
max_duration_per_sequence: 365d
offers-click:
attributes:
brand_id:
max_duration_per_sequence: 365d
offers-like:
attributes:
brand_id:
max_duration_per_sequence: 365d
offers-clickout:
attributes:
brand_id:
max_duration_per_sequence: 365d
max_events_per_sequence: 512
encoders:
item:
type: id
subdomain:
type: id
brand_id:
type: id
socdem_cluster:
type: id
features:
item_id:
located_in:
event: true
encoder: item
item:
located_in:
artifacts: true
encoder: item
subdomain:
located_in:
event: true
encoder: subdomain
brand_id:
located_in:
event: true
artifacts: true
encoder: brand_id
socdem_cluster:
located_in:
context: true
encoder: socdem_cluster
backbone:
dim: 256
history_aggregator:
type: hstu
params:
num_layers: 4
num_heads: 4
dropout: 0.1
event_aggregator:
type: weighted_sum
context_aggregator:
type: identity
training:
num_epochs: 10
log_every_n_train_steps: 1000
dataloader:
batch_size: 32
num_workers: 2
pin_memory: true
shuffle: true
test_dataloader:
batch_size: 32
num_workers: 2
pin_memory: true
early_stopping:
metric: recall@100
mode: max
patience: 3
optimizer:
params:
lr: 0.0003
inference:
backbone_dataloader:
batch_size: 64
num_workers: 0
pin_memory: true
head_dataloader:
batch_size: 64
num_workers: 0
pin_memory: true
predict_kwargs:
k: 1000Данные при этом не изменились, поэтому пересобирать датасет не нужно — сразу запускаем обучение:
cp ../retrieval/config.yaml ../retrieval/checkpoint/config.yaml
uv run accelerate launch -m perseus train fit-model --workdir ../retrieval
uv run python -m perseus inference make-backbone-embeddings --workdir ../retrieval
uv run python -m perseus inference make-head-predictions --workdir ../retrievalВидим заметный прирост метрик.
Вариант | Recall@100 | NDCG@100 | Coverage@100 |
Perseus + события | 0,1638 | 0,0634 | 0,0046 |
Perseus + архитектура | 0,1812 | 0,0705 | 0,0542 |
Нужно отметить, что под разные задачи могут подходить разные комбинации гиперпараметров, поэтому необходимо экспериментировать.
Соберем все метрики в одну таблицу.
Вариант | Recall@100 | NDCG@100 | Coverage@100 |
Топ популярных (бейзлайн) | 0,1275 | 0,0430 | 0,0005 |
Perseus, только item_id | 0,1518 | 0,0545 | 0,0056 |
Perseus + признаки | 0,1553 | 0,0572 | 0,0099 |
Perseus + события | 0,1638 | 0,0634 | 0,0046 |
Perseus + архитектура | 0,1812 | 0,0705 | 0,0542 |
Видим, что за счет использования различных возможностей фреймворка получается растить метрики.
Ранжирование
Следующая задача — переупорядочить готовый пул кандидатов. Будем считать, что с точки зрения бизнеса события ранжируются как clickout > like > click > view. В качестве метрик возьмем NDCG@20 и MRR@20.
В Perseus ранжирование реализовано как предсказание вероятностей целевых событий (multi-label classification). По взвешенной сумме этих вероятностей затем можно проранжировать объекты. Веса задаются априорно, а не выучиваются моделью, что позволяет ML-разработчику приоритизировать то или иное событие в зависимости от целей бизнеса.
Базис. Ранжированию, как и кандидатогенерации, нужны артефакты с айтемами, а таргет устроен сложнее. Для каждого айтема в нем хранится не только сам айтем, но и набор булевых меток — какими именно целевыми действиями клиент с ним провзаимодействовал. В тестовой части к меткам добавляется релевантность, а в самих сэмплах появляется обязательная колонка items — тот самый пул кандидатов, который модель будет скорить и переупорядочивать.
Соберем базис из всех четырех типов событий Marketplace за те же 120 дней и с тем же округлением timestamp до даты. Метками будут четыре флага по типам событий, а релевантностью — 0 для просмотра, 1 для клика, 2 для лайка и 3 для кликаута. Оставим только те сэмплы, в которых встречается больше одного уровня релевантности: если все айтемы одинаково хороши, упорядочивать нечего и метрика по такому сэмплу неинформативна. Далее так же, как и в кандидатогенерации: сплит 80/20 по времени, соцдем-кластер в контекст, айтемы с брендом в артефакты.
Listing
samples = pl.concat([
pl.read_parquet("/event-hub/marketplace-clickout").with_columns(pl.lit("clickout").alias("event_type")),
pl.read_parquet("/event-hub/marketplace-click").with_columns(pl.lit("click").alias("event_type")),
pl.read_parquet("/event-hub/marketplace-like").with_columns(pl.lit("like").alias("event_type")),
pl.read_parquet("/event-hub/marketplace-view").with_columns(pl.lit("view").alias("event_type")),
])
start_date = samples["date"].max() - timedelta(days=120)
samples = (
samples
.filter(pl.col("timestamp") >= start_date)
.with_columns((pl.col("date").cast(pl.Datetime) - timedelta(hours=12)).cast(pl.Datetime("ns")).alias("timestamp")).drop("date")
)
samples = samples.with_columns(
pl.struct([
pl.col("item_id").alias("item"),
pl.struct([
(pl.col("event_type") == "view").alias("view"),
(pl.col("event_type") == "like").alias("like"),
(pl.col("event_type") == "click").alias("click"),
(pl.col("event_type") == "clickout").alias("clickout"),
]).alias("labels"),
]).alias("target")
).drop("item_id", "event_type")
samples = (
samples
.group_by("timestamp", "client_id")
.agg(pl.col("target").unique())
.filter(
pl.col("target").list.eval(
pl.when(pl.element().struct.field("labels").struct.field("view")).then(0)
.when(pl.element().struct.field("labels").struct.field("click")).then(1)
.when(pl.element().struct.field("labels").struct.field("like")).then(2)
.when(pl.element().struct.field("labels").struct.field("clickout")).then(3)
.otherwise(-1)
).list.unique().list.len() > 1
)
.sort("timestamp")
)
train_ratio = 0.8
split_idx = int(len(samples) * train_ratio)
split_timestamp = samples["timestamp"][split_idx]
train_samples = samples.filter(pl.col("timestamp") < split_timestamp)
test_samples = samples.filter(pl.col("timestamp") >= split_timestamp)
test_samples = (
test_samples
.explode("target")
.with_columns([
pl.col("target").struct.field("item").alias("item"),
pl.col("target").struct.field("labels").struct.field("view").alias("view"),
pl.col("target").struct.field("labels").struct.field("like").alias("like"),
pl.col("target").struct.field("labels").struct.field("click").alias("click"),
pl.col("target").struct.field("labels").struct.field("clickout").alias("clickout"),
])
.with_columns(
pl.when(pl.col("view")).then(0)
.when(pl.col("click")).then(1)
.when(pl.col("like")).then(2)
.when(pl.col("clickout")).then(3)
.otherwise(-1)
.alias("relevance")
)
.group_by("timestamp", "client_id", "item")
.agg([
pl.col("view").any(),
pl.col("like").any(),
pl.col("click").any(),
pl.col("clickout").any(),
pl.col("relevance").max()
])
.with_columns(
pl.struct([
"item",
pl.struct(["view", "like", "click", "clickout"]).alias("labels"),
"relevance"
]).alias("target")
)
.group_by("timestamp", "client_id")
.agg([
pl.col("target"),
pl.col("item").alias("items")
])
)
train_samples = train_samples.join(CLIENTS, on="client_id", how="left")
train_samples = train_samples.with_columns(
pl.struct(["socdem_cluster"]).alias("context")
).drop(["socdem_cluster"])
test_samples = test_samples.join(CLIENTS, on="client_id", how="left")
test_samples = test_samples.with_columns(
pl.struct(["socdem_cluster"]).alias("context")
).drop(["socdem_cluster"])
train_samples.write_parquet("../ranking/basis/train/samples.pq")
test_samples.write_parquet("../ranking/basis/test/samples.pq")
artifacts = pl.DataFrame(
train_samples.select(pl.col("target").list.eval(pl.element().struct.field("item")).explode()).unique()
).rename({"target": "item"})
artifacts = artifacts.with_columns(pl.col("item")).join(ITEMS, left_on="item", right_on="item_id", how="left").select("item", "brand_id")
artifacts.write_parquet("../ranking/basis/train/artifacts/items.pq")
artifacts.write_parquet("../ranking/basis/test/artifacts/items.pq")train/samples.pq выглядит так:
В test/samples.pq к каждому айтему таргета добавляется релевантность, а рядом появляется колонка items — пул кандидатов (labels для краткости свернуты, в файле они такие же, как в train):
Артефакты те же, что и в кандидатогенерации, — item и бренд:
Пулом кандидатов на инференсе будет колонка items из тестовой части — собственный набор айтемов каждого сэмпла, тот же, на котором модель валидировалась в процессе обучения. В продакшене пул приходил бы от кандидатогенератора, например от retrieval-модели из предыдущего раздела. В туториале мы оцениваем качество переупорядочивания в чистом виде, поэтому в пул кладем айтемы, которые пользователь в этот день видел.
В качестве бейзлайна отранжируем пул по популярности айтема в train-части базиса, считая ее по позитивным событиям: клику, лайку и кликауту.
Listing
item_to_popularity = dict(
train_samples.select(pl.col("target").explode()).unnest("target").unnest("labels")
.filter(pl.col("click") | pl.col("like") | pl.col("clickout"))
["item"].value_counts().iter_rows()
)
inference_basis = pl.read_parquet("../ranking/basis/samples.pq")
inference_artifacts = pl.read_parquet("../ranking/basis/artifacts/items.pq")
inference_basis = inference_basis.with_columns(
baseline_prediction=pl.col("items").list.eval(
pl.struct(
item=pl.element(),
probas=pl.struct(**{label: pl.lit(0.0, pl.Float32) for label in ["view", "click", "like", "clickout"]}),
score=pl.element().replace_strict(item_to_popularity, default=0).cast(pl.Float32),
)
)
)Модель предсказывает вероятность каждого типа события для каждого кандидата, а пул упорядочивается по одному скору — взвешенной сумме этих вероятностей. По умолчанию веса равны, то есть просмотр вносит в скор такой же вклад, что и кликаут. Через label_to_weight зададим веса равными релевантностям: тогда скор — это ожидаемая релевантность и просмотры ее не завышают.
config5.yaml
task:
type: ranking
head:
label_to_weight:
view: 0
click: 1
like: 2
clickout: 3
metrics:
ndcg@20:
type: ndcg_at_k
params:
k: 20
mrr@20:
type: mrr_at_k
params:
k: 20
events:
marketplace-click:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 1
marketplace-like:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 2
marketplace-clickout:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 3
marketplace-view:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 0
max_events_per_sequence: 512
encoders:
item:
type: id
subdomain:
type: id
brand_id:
type: id
socdem_cluster:
type: id
features:
item_id:
located_in:
event: true
encoder: item
item:
located_in:
artifacts: true
encoder: item
subdomain:
located_in:
event: true
encoder: subdomain
brand_id:
located_in:
artifacts: true
encoder: brand_id
socdem_cluster:
located_in:
context: true
encoder: socdem_cluster
backbone:
dim: 256
history_aggregator:
type: modern_bert
params:
num_layers: 4
num_heads: 4
dropout: 0.1
event_aggregator:
type: sum
context_aggregator:
type: identity
training:
num_epochs: 10
log_every_n_train_steps: 1000
dataloader:
batch_size: 32
num_workers: 2
pin_memory: true
shuffle: true
test_dataloader:
batch_size: 32
num_workers: 2
pin_memory: true
early_stopping:
metric: ndcg@20
mode: max
patience: 3
optimizer:
params:
lr: 0.0003
inference:
backbone_dataloader:
batch_size: 64
num_workers: 0
pin_memory: true
head_dataloader:
batch_size: 64
num_workers: 0
pin_memory: trueДальше как в кандидатогенерации: собираем датасет, обучаем модель, инференсим теми же командами:
uv run python -m perseus inference distribute-samples --workdir ../ranking
uv run python -m perseus inference make-backbone-embeddings --workdir ../ranking
uv run python -m perseus inference make-head-predictions --workdir ../rankingМодель возвращает вероятности для каждого типа событий и их взвешенную сумму.
Вариант | NDCG@20 | MRR@20 |
По популярности (бейзлайн) | 0,4812 | 0,4357 |
Perseus | 0,4984 | 0,4578 |
Perseus выигрывает у бейзлана, хотя и с меньшим отрывом, чем было в случае с кандидатогенерацией.
Классификация
Будем предсказывать, совершит ли пользователь хотя бы одно активное действие (клик, лайк или кликаут) в Marketplace в течение 7 дней после даты сэмпла. В качестве метрики возьмем ROC-AUC.
Базис для классификации устроен максимально просто: таргет — строка с названием класса, дополнительных колонок и артефактов не требуется. Соберем базис из дней, в которые пользователь был активен в Marketplace, за те же последние 120 дней и с тем же округлением timestamp.
Таргет посчитаем по окну (t, t + 7 дней], то есть строго в будущем относительно сэмпла: visit, если активность в окне была, и no_visit иначе. Последние 7 дней выборки отбросим: для них окно неполное и таргет оказался бы занижен. Затем, как и раньше, разделим базис 80/20 по времени и положим соцдем-кластер в контекст.
Listing
samples = pl.concat([
pl.read_parquet("/event-hub/marketplace-click"),
pl.read_parquet("/event-hub/marketplace-like"),
pl.read_parquet("/event-hub/marketplace-clickout"),
]).select("date", "timestamp", "client_id")
start_date = samples["date"].max() - timedelta(days=120)
samples = (
samples
.filter(pl.col("timestamp") >= start_date)
.with_columns((pl.col("date").cast(pl.Datetime) - timedelta(hours=12)).cast(pl.Datetime("ns")).alias("timestamp")).drop("date")
.unique(["timestamp", "client_id"])
.sort("client_id", "timestamp")
)
horizon = timedelta(days=7)
next_week_visits = samples.rolling(
index_column="timestamp",
period="7d",
offset="0d",
closed="right",
group_by="client_id",
).agg(pl.len().alias("num_visits"))
samples = (
samples
.filter(pl.col("timestamp") < pl.col("timestamp").max() - horizon)
.join(next_week_visits, on=["client_id", "timestamp"], how="left")
.select(
"timestamp",
"client_id",
pl.when(pl.col("num_visits") > 0).then(pl.lit("visit")).otherwise(pl.lit("no_visit")).alias("target"),
)
.sort("timestamp")
)
train_ratio = 0.8
split_idx = int(len(samples) * train_ratio)
split_timestamp = samples["timestamp"][split_idx]
train_samples = samples.filter(pl.col("timestamp") < split_timestamp)
test_samples = samples.filter(pl.col("timestamp") >= split_timestamp)
train_samples = train_samples.join(CLIENTS, on="client_id", how="left")
train_samples = train_samples.with_columns(
pl.struct(["socdem_cluster"]).alias("context")
).drop(["socdem_cluster"])
test_samples = test_samples.join(CLIENTS, on="client_id", how="left")
test_samples = test_samples.with_columns(
pl.struct(["socdem_cluster"]).alias("context")
).drop(["socdem_cluster"])
train_samples.write_parquet("../classification/basis/train/samples.pq")
test_samples.write_parquet("../classification/basis/test/samples.pq")train/samples.pq и test/samples.pq устроены одинаково:
В качестве бейзлайна воспользуемся следующим правилом: будем предсказывать позитивную метку, если у клиента было хотя бы одно положительное взаимодействие за 30-дневный период, предшествующий timestamp-у сэмпла.
Listing
visit_rate = (train_samples["target"] == "visit").mean()
baseline_prediction = pl.struct(
no_visit=pl.lit(1 - visit_rate, pl.Float32),
visit=pl.lit(visit_rate, pl.Float32),
)
inference_basis = pl.read_parquet("../classification/basis/samples.pq")
activity = pl.concat([train_samples, test_samples]).select("client_id", "timestamp").sort("client_id", "timestamp")
visited_before = activity.join(
activity.rolling(
index_column="timestamp",
period="7d",
offset="-30d",
closed="left",
group_by="client_id",
).agg(pl.len().alias("num_prior_visits")),
on=["client_id", "timestamp"],
how="left",
).with_columns(
(pl.col("num_prior_visits").fill_null(0) > 0).cast(pl.Float32).alias("visit")
).select("client_id", "timestamp", "visit")Конфиг получится проще, чем в предыдущих задачах: артефакты не нужны, модель по эмбеддингу пользователя сразу предсказывает распределение по классам. Для ROC-AUC в бинарном случае необходимо указать pos_label.
config6.yaml
task:
type: classification
metrics:
roc_auc:
params:
pos_label: visit
events:
marketplace-click:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 2
marketplace-like:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 1
marketplace-clickout:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 1
retail-order:
max_duration_per_sequence: 365d
offers-click:
attributes:
brand_id:
max_duration_per_sequence: 365d
offers-like:
attributes:
brand_id:
max_duration_per_sequence: 365d
offers-clickout:
attributes:
brand_id:
max_duration_per_sequence: 365d
max_events_per_sequence: 512
encoders:
item:
type: id
subdomain:
type: id
brand_id:
type: id
socdem_cluster:
type: id
features:
item_id:
located_in:
event: true
encoder: item
subdomain:
located_in:
event: true
encoder: subdomain
brand_id:
located_in:
event: true
encoder: brand_id
socdem_cluster:
located_in:
context: true
encoder: socdem_cluster
backbone:
dim: 256
history_aggregator:
type: modern_bert
params:
num_layers: 4
num_heads: 4
dropout: 0.1
event_aggregator:
type: sum
context_aggregator:
type: identity
training:
num_epochs: 10
log_every_n_train_steps: 1000
dataloader:
batch_size: 32
num_workers: 2
pin_memory: true
shuffle: true
test_dataloader:
batch_size: 32
num_workers: 2
pin_memory: true
early_stopping:
metric: roc_auc
mode: max
patience: 3
optimizer:
params:
lr: 0.0003
inference:
backbone_dataloader:
batch_size: 64
num_workers: 0
pin_memory: true
head_dataloader:
batch_size: 64
num_workers: 0
pin_memory: trueКоманды обучения и инференса остаются такими же, как при кандидатогенерации и ранжировании:
uv run python -m perseus train prepare-dataset --workdir ../classification
uv run accelerate launch -m perseus train fit-model --workdir ../classification
uv run python -m perseus inference distribute-samples --workdir ../classification
uv run python -m perseus inference make-backbone-embeddings --workdir ../classification
uv run python -m perseus inference make-head-predictions --workdir ../classificationВариант | ROC-AUC |
Правило (бейзлайн) | 0,52 |
Perseus | 0,59 |
Регрессия
Последняя задача — регрессия. Будем предсказывать суммарную стоимость товаров, с которыми пользователь позитивно провзаимодействует в Marketplace в течение следующего месяца. Просмотры в нее не входят, так как просмотр — это не позитивное взаимодействие. В качестве метрик возьмем MAE и RMSE.
Соберем базис аналогично тому, как делали для классификации. Но теперь нам нужны сами товары и их стоимость, поэтому дедуплицировать события до пар (дата, пользователь) будем только после джойна со справочником товаров. Свернем события в стоимость корзины за день и просуммируем ее в окне (t, t + 30 дней]. Последний месяц выборки отбросим как неполный.
Listing
events = pl.concat([
pl.read_parquet("/event-hub/marketplace-click"),
pl.read_parquet("/event-hub/marketplace-like"),
pl.read_parquet("/event-hub/marketplace-clickout"),
]).select("date", "timestamp", "client_id", "item_id")
start_date = events["date"].max() - timedelta(days=120)
events = (
events
.filter(pl.col("timestamp") >= start_date)
.with_columns((pl.col("date").cast(pl.Datetime) - timedelta(hours=12)).cast(pl.Datetime("ns")).alias("timestamp")).drop("date")
.unique(["timestamp", "client_id", "item_id"])
.join(ITEMS.select("item_id", pl.col("price").cast(pl.Float64)), on="item_id", how="left")
)
horizon = timedelta(days=30)
samples = (
events
.group_by("timestamp", "client_id")
.agg(pl.col("price").sum().alias("daily_spend"))
.sort("client_id", "timestamp")
)
next_month_spend = samples.rolling(
index_column="timestamp",
period="30d",
offset="0d",
closed="right",
group_by="client_id",
).agg(pl.col("daily_spend").sum().alias("target"))
samples = (
samples
.filter(pl.col("timestamp") < pl.col("timestamp").max() - horizon)
.join(next_month_spend, on=["client_id", "timestamp"], how="left")
.select("timestamp", "client_id", pl.col("target").fill_null(0.0))
.sort("timestamp")
)
train_ratio = 0.8
split_idx = int(len(samples) * train_ratio)
split_timestamp = samples["timestamp"][split_idx]
train_samples = samples.filter(pl.col("timestamp") < split_timestamp)
test_samples = samples.filter(pl.col("timestamp") >= split_timestamp)
train_samples = train_samples.join(CLIENTS, on="client_id", how="left")
train_samples = train_samples.with_columns(
pl.struct(["socdem_cluster"]).alias("context")
).drop(["socdem_cluster"])
test_samples = test_samples.join(CLIENTS, on="client_id", how="left")
test_samples = test_samples.with_columns(
pl.struct(["socdem_cluster"]).alias("context")
).drop(["socdem_cluster"])
train_samples.write_parquet("../regression/basis/train/samples.pq")
test_samples.write_parquet("../regression/basis/test/samples.pq")Базис получается такой же формы, что и в классификации, только в target-число:
В качестве бейзлайна возьмем константное предсказание — среднее по train-части базиса.
Listing
mean_target = train_samples["target"].mean()
baseline_prediction = pl.lit(mean_target, pl.Float32)Конфиг почти повторяет конфиг классификации: меняются тип задачи, метрики и голова. Таргет получается скошенным, поэтому отнормируем его для обучения — применим MinMax Scaling, указав соответствующую строчку в конфиге.
config7.yaml
task:
type: regression
metrics:
mae:
rmse:
preprocessor:
scaler: minmax
events:
marketplace-click:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 2
marketplace-like:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 1
marketplace-clickout:
attributes:
item_id:
subdomain:
max_duration_per_sequence: 365d
priority: 1
retail-order:
max_duration_per_sequence: 365d
offers-click:
attributes:
brand_id:
max_duration_per_sequence: 365d
offers-like:
attributes:
brand_id:
max_duration_per_sequence: 365d
offers-clickout:
attributes:
brand_id:
max_duration_per_sequence: 365d
max_events_per_sequence: 512
encoders:
item:
type: id
subdomain:
type: id
brand_id:
type: id
socdem_cluster:
type: id
features:
item_id:
located_in:
event: true
encoder: item
subdomain:
located_in:
event: true
encoder: subdomain
brand_id:
located_in:
event: true
encoder: brand_id
socdem_cluster:
located_in:
context: true
encoder: socdem_cluster
backbone:
dim: 256
history_aggregator:
type: modern_bert
params:
num_layers: 4
num_heads: 4
dropout: 0.1
event_aggregator:
type: sum
context_aggregator:
type: identity
training:
num_epochs: 10
log_every_n_train_steps: 1000
dataloader:
batch_size: 32
num_workers: 2
pin_memory: true
shuffle: true
test_dataloader:
batch_size: 32
num_workers: 2
pin_memory: true
early_stopping:
metric: rmse
mode: min
patience: 3
optimizer:
params:
lr: 0.0003
inference:
backbone_dataloader:
batch_size: 64
num_workers: 0
pin_memory: true
head_dataloader:
batch_size: 64
num_workers: 0
pin_memory: trueКоманды те же, что и в предыдущих задачах:
uv run python -m perseus train prepare-dataset --workdir ../regression
uv run accelerate launch -m perseus train fit-model --workdir ../regression
uv run python -m perseus inference distribute-samples --workdir ../regression
uv run python -m perseus inference make-backbone-embeddings --workdir ../regression
uv run python -m perseus inference make-head-predictions --workdir ../regressionВариант | MAE | RMSE |
Среднее (бейзлайн) | 7,7591 | 13,9357 |
Perseus | 7,1497 | 13,5175 |
Perseus показал себя немного лучше бейзлайна. Возможно, изменение типа бэкбона позволит улучшить метрики, как было в случае с кандидатогенерацией, но проверку этого мы оставим читателям в качестве практического задания.
Заключение
Мы рассмотрели, как работать с фреймворком Perseus. На данных датасета T-ECD, хорошо отражающих сложность и многогранность реальной системы, мы разобрали четыре сценария: кандидатогенерацию, ранжирование, классификацию и регрессию. В каждом случае строили модель, сравнивали с бейзлайном и смотрели, как меняется качество.
Туториал иллюстрирует гибкость фреймворка. Добавить фичи? Подключить события из соседнего домена? Попробовать другой бэкбон или тип пулинга? Достаточно поменять пару строк в YAML-конфиге — никакого переписывания кода с нуля.
Мы постарались показать Perseus с практической стороны, без лишней теории. Конечно, чтобы освоиться, потребуется разобраться в форматах данных и структуре конфигов, но, надеюсь, наш туториал станет хорошей точкой входа. Мы уверены, что Perseus стоит того, чтобы потратить на него время, особенно если вы работаете с мультидоменными данными.
Ждем ваших впечатлений, комментариев и вопросов!
Полезные ссылки:
KioskNews shows a cleaned-up reading view extracted from the publisher’s page — the original always lives on their site, not ours.