Как в RabbitMQ обрабатывать долгую генерацию LLMкой сообщений?

Question

[PerseforeComplete]

Упрощённо задача выглядит так
В веб интерфейсе LLMке передаётся запрос, на который она генерирует ответ. Publisher посылает запрос на генерацию. Consumer забирает запрос, запускает LLM для генерации ответа и возвращает ответ.

Генерация ответа очень долгая - LLM запускается на CPU. В очереди может скопиться некоторое количество сообщений для генерации и висеть там часами.

Проблема - RabbitMQ обрывает соединение при очень долгой генерации. Условно - если я передам 3 запроса на генерацию, и LLM на каждый генерирует ответ 20 минут, то для первого ответ сгенерируется и consumer без проблем вернёт ответ, а два других отвалятся с ошибкой, потому что соединение RabbitMQ уже закрыто

[warning] Consumer 1 on channel 1 has timed out waiting for delivery acknowledgement. 
[error] Channel error on connection, channel 1:
[error] operation none caused a channel exception precondition_failed: delivery acknowledgement on channel 1 timed out.

Есть симптоматическое решение - в rabbitmq.conf прописать

consumer_timeout = 1200000 # 2 hours in milliseconds

Это поможет в данном конкретном случае, но легко представить себе случай, в котором это не сработает - если в consumer придёт 4 запроса и LLM будет генерировать ответ по 50 минут, то для первых двух запросов ответ будет сгенерирован и опубликован, а два оставшихся не смогут положиться в очередь, потому что соединение уже будет закрыто с указанной выше ошибкой.

Я не могу увеличивать consumer_timeout вечно. Вопрос в том, как правильно организовать систему что бы был 1 consumer, который из очереди достаёт сообщения и часами их обрабатывает, возвращает в очередь ответ и соединение в очереди при этом не закрывается?

Answer 1

[Wexter]

так отправляйте ACK сразу после чтения из очереди и обрабатывайте сколько хотите

Comments

[PerseforeComplete]

Вроде такая практика может привести к проблемам, если сделать ACK, но потом возникнет ошибка при обработке

[Wexter]

PerseforeComplete, тогда определяйте для себя максимальное время ответа для ллм (1 день? 1 неделю?) и задавайте его в качестве тайм-аута

Answer 2

[Алексей Уколов]

А как у вас идёт работа в этом веб-интерфейсе? Пользователь его часами не закрывает в ожидании ответа?
По-хорошему, тут надо полностью переделать модель взаимодействия - есть персистентная сущность "чат", туда сначала записывается вопрос пользователя, а потом ответ LLM. И тогда Rabbit не блокируется совсем - consumer возвращает ответ не в то же соединение, в котором пришёл запрос, а сохраняет его в БД и, опционально, кидает новое событие в очередь, которое обрабатывает какой-то уже другой consumer, который отправляет пользователю уведомление (ну или можно по websocket и сам ответ LLM прислать, если пользователь окно чата не закрыл).

Comments

[PerseforeComplete]

А как у вас идёт работа в этом веб-интерфейсе? Пользователь его часами не закрывает в ожидании ответа?

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

[Алексей Уколов]

А, я подумал, что у вас RPC и именно в нём таймаут возникает.
Тогда я что-то не понимаю, как у вас сейчас всё устроено и где возникает проблема.
В consumer_timeout должно быть значение, которое превышает максимальное время обработки одного сообщения. Т.е. это время, за которое вы должны отдать ACK. Но при чём тут тогда следующие сообщения и какое соединение там закрывается? Если у вас consumer работает в один поток, то Rabbit спокойно принимает сообщения в очередь, они там лежат сколько угодно долго, пока их кто-то не возьмёт в обработку, и только в этот момент начинает тикать consumer_timeout.

[PerseforeComplete]

У меня как раз таки RPC.

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

В том то и проблема. У меня стоит автоматический ACK, я руками не делаю. Если я, условно, открою 3 вкладки сваггера, сделаю в каждой по 3 запроса сразу же, consumer_timeout стоит 30 минут, ллмка генерит ответ 20 минут, то на 2ом запросе будет ошибка с delivery ack timeout, хотя казалось бы это же для одного запроса timeout? Это только если отправлять сразу несколько запросов. Если послать 1, подождать пока он завершится и тогда уже слать 2ой, то всё работает нормально

[Алексей Уколов]

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

И во сколько потоков у вас consumer запущен?

[PerseforeComplete]

Если я правильно понимаю, в асинхронно в 1 поток

[Алексей Уколов]

А входящих потоков у вас N. Каждый из них кидает в очередь сообщение и открывает туннель (RPC), по которому ждёт ответа (и последующие не дожидаются). Это неправильно - зачем им что-то ждать, если сам пользователь ушёл. Тут вообще не нужен RPC и тогда, скорее всего, никаких таймаутов вы ловить не будете.