Блайндбоксы в мире ML: насколько Labubu поднял конверсию, и как разработчики чуть всё не уронили

В прошлом посте мы разобрали, как работает микросервис Labubu. Теперь разберём, к чему привело внедрение Labubu в компанию.

📌 Как обработка коммуникаций едва не заняла все ресурсы

Как мы помним, инференс классификатора — дорогое удовольствие. Если NLP-инженер неправильно задал промпты или тимлид выбрал не те критерии, микросервис потратит ресурсы воркеров впустую. Поэтому перед полной выгрузкой проводится проверка: вместо желаемого количества подходящих коммуникаций выдаётся 20.

Пользователь Labubu анализирует выданные данные и оценивает качество отбора. Если всё в порядке, запускает процесс для остальных данных.

В итоге у Labubu есть две задачи:

1) Быстро выдать 20 коммуникаций для проверки.
2) Быстро обработать весь запрашиваемый объем.

Проблема: пользователь может запросить выборку из 5000 коммуникаций, что требует обработки около 500 000 диалогов.

Почему это критично? В сервисе запущено 8 воркеров. Каждый отправляет запросы с параллелизмом в 10 единиц. В сумме они могут создавать нагрузку до 4800 RPM, что анимало больше половины лимитов на весь инференс компании вместе взятой. Из-за этого другие команды получали ошибки.

Решение: пришлось ограничить ресурсы для Labubu до 600 RPM, чтобы микросервис больше не перетягивал на себя все мощности.

Новая задача — не выйти за пределы лимита в 600 RPM

Чтобы выполнить задачу, нужно синхронизировать запросы в 8 воркерах. Для этого команда Labubu изменила параметры параллелизма так, чтобы один воркер не кидал больше 10 потоков одновременно. Вот так это реализовали с помощью кода:

Вариант 1. Спавним кучу тасок и лочим их через семафор
Проблема. Способ работает, если тасок меньше тысячи. Потому что у каждой таски есть overhead по памяти — сервис может упереться в ограничение по памяти.

Код:

import asyncio

async def worker(i, sem):
    async with sem:
        return await job(i)

async def main():
    sem = asyncio.Semaphore(10)
    await asyncio.gather(*(worker(i, sem) for i in range(10_000)))

asyncio.run(main())

Вариант 2. Закидываем все таски в очередь и создаём воркеры

Проблема. Способ оказался сложнее по реализации, чем предыдущий.

Код:

import asyncio

async def worker(q):
    while (i := await q.get()) is not None:
        try: await job(i)
        finally: q.task_done()

async def main():
    q = asyncio.Queue()
    for i in range(10_000): q.put_nowait(i)

    workers = [asyncio.create_task(worker(q)) for _ in range(100)]
    for _ in workers: q.put_nowait(None)

    await q.join()
    await asyncio.gather(*workers)

asyncio.run(main())

В итоге выбрали вариант 2. Способ сработал: воркеры не кидали больше 10 потоков, а сервис не выходил за 600 RPM. Задача выполнена? Оказалось, что нет.

Что пошло не так:
• Labubu постоянно превышал 600 RPM. Примерно 15 секунд из каждой минуты он стабильно натыкался на 429-ю ошибку.
• Back off плохо реализовали. Он сразу ретраил итерацию — сервис после 429 ошибки сразу шёл искать новую 429 ошибку.
• У соседней команды легла ручка проверки квот — она начала отвечать по 300–500 мс из-за бесконечного цикла итераций.

Проблему решили на лету, а соседняя команда переехала на выделенный кластер СУБД.

📌 Какую ценность принёс Labubu?
Несмотря на инфраструктурные сложности, сервис оправдал вложения в разработку и поддержку.

Благодаря таргетированным коммуникациям, Labubu помог найти и привлечь 174 новые компании в комьюнити и поднять конверсию с 1% до 3,91%.

А ещё он подарил бесценный опыт своей команде разработчиков.

💜 Этот пост написал Макс Умутаев, бэкенд-разработчик в Точка Банк