Знаток концертов по городам Золотого кольца России на Python с применением LLM

Содержание:

1. Знакомство с сервисом
2. Архитектура
3. Важные нюансы
4. Развёртывание в PROD

1. Знакомство с сервисом

Как это работает?

Представим, что «Знаток концертов» — Ваш умный библиотекарь 📚. Он настоящий специалист в своём деле, и по каждой книге (концерту) у него есть заметка с кратким содержанием 📝

Вы приходите к библиотекарю с мыслью "что-то этакое почитать" 💭. Размахиваете руками и говорите ему ваши желания. Библиотекарь вас внимательно выслушивает и записывает ✍️. Когда вы договорили, он уходит покопаться в своих заметках, чтобы сопоставить ваши желания с теми заметками, что у него есть 🕵️‍♂️. Он читает краткие содержания книг и находит наиболее подходящие для вас.

Когда библиотекарь находит нужное количество книг, он возвращается с ними к вам и с улыбкой на лице даёт почитать 😊

Вот такая добрая и ненавязчивая история )

Как им пользоваться ?

При первом заходе в «Знаток концертов» и нажатии /start Вам будет предложено выбрать город

Фиксация выбранного города

Вводим текстовый запрос и получаем наиболее подходящие мероприятия в выбранном Вами городе

Какие нюансы ?

«Знаток концертов» хранит в сжатой форме информацию о названии, месте проведения и описании мероприятия 📌. Поэтому он НЕ чувствителен к информации о времени проведения концерта ⏰

Так сделано, чтобы информация о времени не перекрывала содержательную информацию о мероприятии.

Все рекомендуемые события актуальны ✅, и если есть желание найти событие по конкретной дате, то удобно воспользоваться Календарём 📅

2. Архитектура

Два основных контура:

Контур наполнения БД:

1. Модуль парсинга получает данные о мероприятиях из внешних источников 🌐 и вызывает текстовый модуль (Selenium + BeautifulSoup4)

2. Текстовый модуль обрабатывает данные о мероприятиях: очищает их и нормализует (pymorphy3) 🧹

3. Подготавливает текстовые запросы для Yandex Cloud AI по шаблону:
   "{names} {places} {descriptions}"

4. Отправляет запросы к "text-search-doc" модели в Yandex Cloud AI с помощью библиотеки yandex_cloud_ml_sdk. От неё получает embeddings — векторные представления информации о мероприятиях.

5. Текстовый модуль возвращает контроль выполнения модулю парсинга 🔄

6. Модуль парсинга заносит мероприятия с embeddings в БД (PostgreSQL) 💾

Контур обработки запросов пользователя:

1. Пользователь отправляет текстовый запрос 💬 в "Знаток концертов"

2. Запрос регистрируется хендлером Telegram-бота (AIOgram)

3. Telegram-бот оращается к текстовому модулю 🔗

4. Текстовый модуль обрабатывает данные из запроса: их очищает и нормализует

5. Отправляет запрос к "text-search-query" модели в Yandex Cloud AI и получает embedding

6. Достаёт из БД НЕ прошедшие (актуальные ✅) мероприятия в городе, который выбран у пользователя

7. Подсчитывает косинусные расстояния 📐 между embedding запроса пользователя и embeddings мероприятий

8. Определяет события с ближайшим косинусным расстоянием - это и есть наиболее подходящие события по запросу пользователя

9. Текстовый модуль возвращает найденнные мероприятия в модуль Telegram-бота

10. Telegram-бот отправляет пользователю найденные мероприятия в удобном формате

11. Пользователь счастлив 🍾

3. Важные нюансы

Парсинг:

📌 НЕ забывать производить очистку временных файлов хромдрайвера

def clean_chrome_tmp(): shutil.rmtree("/tmp/.com.google.*", ignore_errors=True) shutil.rmtree("/tmp/.org.chromium.*", ignore_errors=True)

Лучше делать очистку как минимум перед каждым новым стартом парсера. Если разворачивать сервис на удалённом сервере с ограниченными ресурсами, то временные файлы могут сожрать много места. У меня так и было )

📌 НЕ создавать лишних хромдрайверов для парсинга

Я изначально наткнулся на то, что для перехода на страницу с конкретным мероприятием создавался отдельный драйвер, а потом он закрывался. Это действие лишнее: достаточно одного драйвера, который перейдёт на целевую страницу с событием и получит необходимую информацию, а потом его можно вернуть обратно через метод driver.back()

📌 Используйте WebDriverWait и обработку WebDriverException

Это Вам даст стабильность работы парсера, т. к. сбои и задержки - обычное дело.

📌 Используйте многопоточность для ускорения парсинга данных

parsing_indexes = get_parsing_indexes( start_ind=1, end_ind=number_of_posters+1, count_bot=NUMBER_PARSER_THREADS, ) threads = [Thread(target=parse_events, args=( website, base_site_path, bot_i, parsing_indexes[bot_i], parsing_indexes[bot_i + 1], target_table_in_DB, model) ) for bot_i in range(NUMBER_PARSER_THREADS)] for t in threads: t.start() for t in threads: if t.is_alive(): logger_parser.info(f'Thread №{t} ALIVE') else: logger_parser.info(f'Thread №{t} DEAD') for t in threads: t.join() for t in threads: if t.is_alive(): logger_parser.info(f'Thread №{t} ALIVE') else: logger_parser.info(f'Thread №{t} DEAD')

При парсинге основная время расходуется на ожидание прогрузки страниц и возможные сетевые задержки. Добиться прироста в производительности можно запуском нескольких потоков - в каждом из которых будет создан свой хромдрайвер, несущий ответственность за парсинг отведённых ему событий. Это хороший пример решения I/O bound задачи.

Работа с текстом и LLM

📌 Выбор моделей Yandex Cloud AI для получения эмбеддингов

from yandex_cloud_ml_sdk import YCloudML from conf.settings import FOLDER_ID, AUTH_TOKEN_CLOUD sdk = YCloudML(folder_id=FOLDER_ID, auth=AUTH_TOKEN_CLOUD) model_query = sdk.models.text_embeddings("query") model_doc = sdk.models.text_embeddings("doc") class EmbeddingEngine: def __init__(self, model_query, model_doc): self.model_query = model_query self.model_doc = model_doc def get_query_embedding(self, text): return self.model_query.run(text) def get_doc_embedding(self, text): return self.model_doc.run(text) embedding_engine = EmbeddingEngine(model_query, model_doc)

На начальном этапе я решил использовать Yandex Cloud модели, чтобы не тратить время и ресурсы на обучение собственных. В перспективе планируется использовать собственные модели - это позволит избежать тарификации запросов и более тонко их настроить. Так же это даст независимость от внешнего иснтруммента.

Правила тарификации для Yandex AI Studio: https://yandex.cloud/ru/docs/ai-studio/pricing

📌Нормализовать текст перед векторизацией

Нормализованный и очищенный текст будет преобразован в более точный embedding. Это повысит точность работы рекомендательной системы.

📌 Обращать внимание на лимиты при работе с Yandex AI Studio

Яндекс устанавливает лимиты на кол-во запросов в секунду. Учитывайте лимиты, чтобы избежать блокировок или неожиданного поведения сервиса.

Лимиты для Yandex Cloud моделей: https://yandex.cloud/ru/docs/ai-studio/concepts/limits

📌Кешируйте тяжёлые запросы

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

Telegram-бот

📌 Используйте FSM (Машину состояний)

Чтобы режимы не пересекались и все сценарии с пользователем работали корректно используйте машину состояний. На её основе разграничены сценарии работы обратной связи, календаря и текстового поиска.

📌 Жёстко регистрируйте хендлеры в нужном порядке

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

4. Развёртывание в PROD

Важно чтобы сервис стабильно работал, самостоятельно восстанавливался после сбоев и не требовал постоянного ручного вмешательства. В этом блоке я покажу, как упакован «Знаток концертов» для работы в production.

Инфраструктура

Сервис поднят в timeweb.cloud

Расход ресурсов

На данный момент парсер даёт основную нагрузку по ресурсам ⚡

Он запускается периодически, чтобы поддерживать актуальность базы 🔄

На графиках приведён мониторинг следующих ресурсов 📊:

📌 Нагрузка на процессор 🔥

📌 Трафик 📡

📌 Оперативная память 💾

Контейнеризация (Docker) и оркестрация (Docker Compose)

Каждый компонент системы находится в своём контейнере 📦 и это даёт:

• Изоляцию зависимостей 🛡️

• Упрощение развёртывания 🚀

• Лёгкое масштабирование 📈

• Повышение отказоустойчивости всей системы 💪

Основные (продакшен) контейнеры:

1. app (главное приложение)

   Telegram-бот на AIOgram

   Роль: принимает запросы от пользователей, управляет диалогом с использованием FSM, координирует поиск событий и формирует ответы.

   Зависит от: pg (база данных), redis (кеш).

2. parsers (парсер мероприятий)

   Многопоточный парсер на Selenium + BeautifulSoup4

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

   Ключевая особенность: имеет жёсткие лимиты памяти, так как запускает несколько драйверов Chrome🌐, чтобы избежать ошибки Out of memory💥

3. pg (хранилище данных)

   Роль: хранит информацию о событиях с эмбеддингами, данные о пользователях и сервисную информацию ⚙️

4. redis (оперативный кеш)

   Роль: кеширует результаты семантического поиска. Снижает нагрузку на базу данных и Yandex Cloud AI API при повторяющихся запросах 🔄

Конетейнеры для разработки и отладки:

1. app_debug

2. parsers_debug

Позволяют отлаживать работу через debugpy из IDE 🖥️ непосредственно в контейнерах.

Как они работают вместе: контейнеры связаны в единую сеть Docker Compose🌐. Основной цикл выглядит так: parsers наполняет pg, app по запросу извлекает данные из pg, используя redis для ускорения 📈. Отладочные контейнеры запускаются только по необходимости.

Основные принципы:

1. DRY через шаблоны YAML: Вместо дублирования настроек для похожих компонентов (бот, парсер) я использовал якоря (&app_base_template) и алиасы (<<: *app_base_template). Это в разы сократило файл и сделало его поддержку проще.

2. Защита от падений: Для всех контейнеров задана политика перезапуска on-failure с 30-секундной задержкой и неограниченным числом попыток. Если парсер упадёт из-за временной сетевой проблемы, Docker Compose сам его поднимет.

3. Жёсткие лимиты памяти: Каждому контейнеру прописаны deploy.resources.limits. Это предотвращает ситуацию, когда один «прожорливый» модуль (тот же Chrome в парсере) лишит ресурсов другие контейнеры и «уронит» удалённый сервер.

4. Отладка без боли: Рядом с основными контейнерами (app, parsers) есть их debug-версии (app_debug, parsers_debug). Они запускают контейнер с пробросом порта для debugpy. Если на сервере что-то пошло не так, я могу подключиться из IDE к удалённому контейнеру и отладить код так же, как на локальной машине. Это экономит время.

# Пример debug-контейнера app_debug: <<: *app_base_template ports: - "6789:6789" command: python -m debugpy --listen 0.0.0.0:6789 --wait-for-client music_events_bot.py

Автоматизация: Makefile как единая точка входа

Makefile — это не просто туториал, а рабочий инструмент инженера. Я максимально упаковал в него все рутинные операции. Ниже приведу основные:

• make init_env — магия для нового разработчика. Команда сама ставит нужную версию Python, создаёт виртуальное окружение и через poetry ставит зависимости. Вся настройка локальной среды — одна команда

• make run_service / make run_parsers — запуск отдельных частей системы в продакшене. Под капотом: сборка образов и docker-compose up -d

• make run_app_debug / make run_parsers_debug — запуск контейнеров для отладки

• make stop_all_and_remove — полный сброс для чистого перезапуска

Тестирование обеспечит спокойный сон

Код должен быть защищен тестами 🛡️. Это НЕ бюрократическая процедура, а страховой полис для разработчика. Тесты гарантируют, что сегодняшнее изменение НЕ сломает вчерашнюю функциональность, и позволяют писать код со спокойной душой 🙌

В качестве библиотеки для тестирования использовалась Pytest.

Структура и принципы тестовой среды

1. Разделение ответственности через типы тестов:

   Юнит-тесты 🧩 проверяют работу отдельных функций в изоляции (нормализация текста, расчёт косинусного расстояния).

   Интеграционные тесты 🔗 проверяют взаимодействие модулей (например, как парсер сохраняет данные в БД, как бот обрабатывает цепочку команд).

2. Полная изоляция тестов 🔒: для тестирования поднимается отдельный контейнер с тестовой базой. Данные инициализируются через фикстуры перед каждым тестом и после полностью очищаются. Это гарантирует, что тесты НЕ зависят от состояния продакшен-базы и НЕ влияют друг на друга.

3. Мокирование внешних зависимостей 🎭: любые обращения к внешним сервисам (Yandex Cloud AI API, Telegram Bot API) заменяются на заглушки unittest.mock. Это делает тесты:
   📌 Быстрыми ⚡ (НЕ ждём сетевых ответов)
   📌 Стабильными 🛡️ (НЕ зависим от доступности сторонних сервисов)
   📌 Дешёвыми 🐸 (НЕ тратим деньги на облачные вызовы)

4. Тестирование критических путей:

   Парсер: проверка, что извлечённые с сайта данные корректно очищаются и укладываются в заданный шаблон:{names} {places} {descriptions}.

   Работа с эмбеддингами: проверка логики поиска ближайших событий (корректность работы top_closest_meaning_event с заранее известными векторами).

   Бот: проверка сценариев FSM — корректность переходов между состояниями (выбор города, обработка текстового запроса, обратная связь и т.д.).

   Кеширование: проверка, что Redis действительно сохраняет и возвращает результаты для одинаковых запросов.

Инвестирование времени на написание тестов — это экономия времени ⏳ в будущем на отладке. Тесты делают код предсказуемым и управляемым. Тестирование - единственный способ сохранить здоровье проекта 💚 в долгосрочной перспективе.

Заключение 🏁

Создание современного полезного продукта с использованием ИИ — это в первую очередь инженерная задача 🛠️. Языковые модели LLM (в нашем случае — Yandex Cloud AI) — это мощный инструмент в руках умелого разработчика. Его сила раскрывается, когда система продумана: качественные данные 📊, грамотное их хранение 📚 и эффективное кеширование ⚡. Это позволяет быстро доставлять качественный результат пользователю 🚀.

«Знаток концертов» 🏛️ — это хороший пример синтеза современных технологий 🔗. Парсинг, облачные ML-модели и микросервисная архитектура объединены в один живой и полезный продукт.

Что дальше? У сервиса есть куда расти:

• Рекомендательная система 🎯

• Продвинутый мониторинг 📈

• Полноценный CI/CD-пайплайн 🔄

Но это тема для следующих статей 📚...

https://vk.com/adepteam

💪 Спасибо, что прошли этот путь !

Если есть вопросы — задавайте в комментариях 💬

Можно писать мне или в группу в ВК 😉