Главная / Статьи / Потоковая обработка LLM с возможностью возобновления с использованием Redis Streams в качестве точки встречи для каждого шага обработки

redis streaming llm agents gateway realtime

Потоковая обработка LLM с возможностью возобновления с использованием Redis Streams в качестве точки встречи для каждого шага обработки

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

2049 слов

Чего мы хотели

Ответы агента могут занимать десятки секунд: необходимо время на размышление, загрузку навыков, вызов инструментов, ожидание и передачу токенов. В демо-версиях поддерживается одно открытое соединение. В производственной среде передача данных прерывается на полпути, приходится перезапускать шлюзы, и возникают паузы из-за инструментов, возвращающих данные лишь через несколько минут. Цель — продолжаемая реальная временная передача данных, чтобы клиент мог снова подключиться и продолжить ту же беседу.

Путь развития

Итерация 1: клиент общается с агентом напрямую

Просто, но хрупко. Любые сбои в сети прерывают передачу данных. Горизонтальное масштабирование требует использования «липких сессий» или приводит к потере данных.

Итерация 2: стриминг через gRPC между сервисами

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

Почему не Kafka?

Отлично подходит для долговечных логов; слишком тяжело для встречи за один ход с коротким временем сохранения и группами потребителей, которые плохо соответствуют концепции «одной вкладки браузера».

Итерация 3 (то, что сработало): именованная встреча за каждый ход

// Agent output
{
  "type": "tool_result",
  "tool": "product_search",
  "data": {
    "items": [...]
  }
}

// Gateway -> TV
{
  "type": "product_carousel",
  "items": [...]
}

// Gateway -> Mobile
{
  "type": "product_list",
  "items": [...]
}

cursor = last_event_id or "0-0"

while True:
    entries = xread({key: cursor}, block=30_000)

    if not entries:          # the only timeout check point
        check_timeouts()
        continue

    for entry_id, event in entries:
        # writes to the socket; not an ack that the client received it
        sse.send(id=entry_id, data=event.payload)
        cursor = entry_id
        if event.type in TERMINAL:
            return

id: 1755600000123-0
data: {"type":"tool_selected","tool":"search"}

id: 1755600000871-0
data: {"type":"response_block","block":{...}}

GET /sessions/{sid}/turns/{tid}/stream
Last-Event-ID: 1755600000871-0

# turn starts: one atomic step (MULTI/EXEC, or a Lua script)
xadd(key, first_event)
expire(key, GENEROUS_TTL)

# producer finishes: bring it in
expire(key, RECONNECT_TTL)

Каждый ход пользователя сопровождается Redis Stream (или паттерном stream+группа потребителей) с ключом turn_id. Агент публикует события токенов/инструментов; шлюз следит за последним id от клиента. Переподключение возобновляется с этого курсора. Поды могут умереть; стрим сохраняет достаточно истории для данного хода.

Важные детали

Дисциплина курсора — клиенты подтверждают последний увиденный ID стрима; после первого подключения никогда не начинают с 0.

События сердцебиения — предотвращают закрытие промежуточными компонентами неактивных соединений во время ожидания инструментом.

Цикл жизни операции — четкие сигналы turn_started / turn_paused / turn_completed / turn_failed.

TTL — истечение срока действия потоков после завершения операции, чтобы Redis не превращался в бесконечный архив.

Аутентификация — turn_id не является механизмом авторизации; операции привязываются к аутентифицированной сессии.

Ситуация, которая решила проблему: нескомнатинутая пауза

Инструмент передал задачу другому потоку обработки, который ответил лишь спустя несколько минут. Прямой HTTP-стриминг прекратился. С помощью Redis Streams агент опубликовал событие паузы, клиент продолжил отображать статус «работает», а позже токены возобновили работу при повторном подключении без необходимости перезапуска всего плана.

Что мы отправляем клиенту

Типы событий: токены, информация о начале работы инструмента (tool_start), краткое описание результата работы инструмента (tool_result, никогда не секретная информация), ошибки и сигналы завершения. Сохраняйте размер данных небольшим; храните объемные данные в облачном хранилище объектов и отправляйте только ссылки на них.

Затраты, ограничения, моменты, на которые следует обратить внимание

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

Конечное состояние

Шлюз выполняет функции читателя, учитывающего состояние выполнения задач; агент — функции записчика; Redis Streams служит местом встречи данных. Благодаря этой структуре реальное время отклика сохраняется даже при возникновении незначительных сбоев, которые обычно разрушают архитектуры демо-версий.

Совет по эксплуатации: храните последний ID потока в куки сессии или памяти клиента, а также на стороне сервера для возможности воспроизведения. Когда пользователь сообщает, что система замерла, служба поддержки должна восстановить ход диалога из потока, не требуя от пользователя повторного ожидания в течение десяти минут. Добавьте панели управления с данными о частоте возобновления работы, числе прерванных диалогов и средней продолжительности пауз, чтобы команда разработки могла определить, медленно ли работают агенты или нестабильна сеть.

Все статьи Прислать бриф